diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 1974a4ec33..e6ac4c32ce 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1009,3 +1009,67 @@ Admission atomically settles an unclaimed coordinator and admits one fresh turn, preserving history, unknown action outcomes, and attempt counts. Pauses, approvals, budgets, task ownership, and terminal task status still gate admission. No automatic provider replay is authorized by a cancelled startup. + +### Delivering queued messages after a legacy run stops + +The legacy queued-message Interrupt action accepts a null `targetRunId` when +there is no active turn. It validates the queue identity and revision under +the task lock and records durable board intent to send the saved queue. A +run that stops between the queue read and the click is also accepted. The +server never redirects interruption to an unrelated active run. +Intentional interruption does not show the global cancelled/failed run toast; +the queue control supplies its own delivery feedback. + +This click can authorize a fresh conversation for messages written before +the prior run stopped. It preserves the original message content and authors, +and retains process/lease stop proofs, task ownership, pauses, approvals, and +budget checks. Queue edits and discards remain authoritative until dispatch. +Dispatch revalidates the consumed queue receipt against the operator, task, +agent, message, and successor run; the operator need not be the message author. +Repeated delivery attempts cannot create another successor after the queue +is consumed. Native same-turn steering retains its active-target contract. + +Legacy finalization retries deferred input after adapter and lease cleanup. +The scheduler also revisits bounded batches of stranded queues after restart +or a late enqueue. Both use normal admission; an existing queued successor +owns the next turn even before it acquires the task execution lock. A recovery +hold or a plain operator Stop does not by itself authorize old input. The +successor guard is scoped to the same agent so another agent's review +participation keeps its independent recovery path. + +An explicit queued-message Interrupt also grants one scoped cleanup retry for +the stopped run. Old ephemeral leases whose cleanup predates provider stop +receipts are rechecked through the recorded provider teardown path. Retained +resources and sandboxes owned by another lease are not rechecked this way. +Delivery still requires the provider's verified stop receipt. Periodic queue +retries do not gain extra cleanup attempts, and the queue displays the server's +waiting reason while cleanup remains unresolved. + +### Operator identity and permission for manual dispatch + +A legacy queued-message Interrupt is a new instruction from the user who clicks +it. The new run uses that user's execution identity, including when someone else +wrote the queued messages. Message bodies and historical authors stay unchanged. +The task page and pipeline conversations both permit Interrupt after the target +run stops and submit the queue's current revision. +Startup validates the consumed queue receipt against the new run, company, +agent, task, clicking user, and delivered message IDs. Automatic retries inherit +the resulting execution identity through the ordinary run identity history. + +Starting an existing agent requires `agent:wake`, which active non-viewer board +members have within their company. Both wake endpoints use this action instead +of `agents:create`. An exact task retry also checks `issue:comment` on the task +from the stored failed run and verifies that its assigned agent has not changed. +External chat retries retain their additional conversation authorization. +Ordinary board wake requests also persist the clicking user's identity, so +adopting another author's queued message cannot change their execution authority. +If that wake merges into an older deferred request, the same transaction updates +the request's execution requester to the clicking user. +Manual wake requests wait for their own run and execution identity. They do not +merge into an agent's active run, with or without a task. +Private agent conversations retain their owner-only wake and retry checks. + +These actions do not grant permission to hire agents or change their settings. +Each action during execution still checks the agent's authority and the +responsible user's authority. A denied retry returns before dispatch; it does +not create a new failed run or change the task's state. diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-credentials.test.ts b/packages/paperclip-runner/src/drivers/acpx/codex-credentials.test.ts index 214f1a4b00..8a18b20012 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-credentials.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-credentials.test.ts @@ -168,12 +168,10 @@ describe("managed Codex credentials", () => { ] as const)( "tolerates one unrelated silent %s quorum listener", async (_label, occupiedIndex) => { - const prepared = - occupiedIndex === 0 ? await silentPrimaryQuorumFixture() : null; - const fixture = prepared?.fixture ?? (await credentialFixture()); - const ports = credentialLeasePorts(await realpath(fixture.home)); - const occupied = - prepared?.occupied ?? (await listenSilently(ports[occupiedIndex])); + const { fixture, occupied } = await silentPrimaryQuorumFixture( + credentialFixture, + occupiedIndex, + ); try { const lease = await stageManagedCodexCredential({ agentHomeDirectory: fixture.home, diff --git a/server/src/__tests__/agent-live-run-routes.test.ts b/server/src/__tests__/agent-live-run-routes.test.ts index 55a74aa750..7520942040 100644 --- a/server/src/__tests__/agent-live-run-routes.test.ts +++ b/server/src/__tests__/agent-live-run-routes.test.ts @@ -425,7 +425,7 @@ describe("agent live run routes", () => { expect(res.body).not.toHaveProperty("resultJson"); expect(res.body).not.toHaveProperty("contextSnapshot"); expect(res.body).not.toHaveProperty("logRef"); - }, 10_000); + }); it("ignores a stale execution run from another issue and falls back to the assignee's matching run", async () => { mockHeartbeatService.getRunIssueSummary.mockResolvedValue({ @@ -832,6 +832,7 @@ describe("agent live run routes", () => { // Optional wake fields retain their existing shape; execution identity // always comes from the authenticated caller. expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith(routeAgentId, { + manualUserWake: true, source: "on_demand", triggerDetail: "manual", reason: "issue_assigned", @@ -863,6 +864,7 @@ describe("agent live run routes", () => { expect(res.status, JSON.stringify(res.body)).toBe(202); expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith(routeAgentId, { + manualUserWake: true, source: "on_demand", triggerDetail: "manual", requestedByActorType: "user", @@ -876,6 +878,18 @@ describe("agent live run routes", () => { }); }); + it.each(["wakeup", "heartbeat/invoke"])("lets an operator start an existing agent via %s without creating agents", async (endpoint) => { + mockAccessService.decide.mockImplementation(async ({ action }) => ({ + allowed: action === "agent:wake", explanation: "Missing permission: agents:create", + })); + const res = await requestApp(await createApp(undefined, { + type: "board", userId: "operator", source: "session", companyIds: ["company-1"], + }), url => request(url).post(`/api/agents/${routeAgentId}/${endpoint}`).send({})); + expect(res.status, JSON.stringify(res.body)).toBe(202); + expect(mockAccessService.decide).toHaveBeenCalledWith(expect.objectContaining({ action: "agent:wake" })); + expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith(routeAgentId, expect.objectContaining({ manualUserWake: true })); + }); + describe("exact failed chat run retry", () => { const retryBody = { failedRunId: failedChatRunId, @@ -901,6 +915,10 @@ describe("agent live run routes", () => { companyId: "company-1", }); mockHeartbeatService.getRun.mockResolvedValue(selectedRun); + mockIssueService.getById.mockResolvedValue({ + id: failedChatIssueId, companyId: "company-1", assigneeAgentId: routeAgentId, + assigneeUserId: null, projectId: null, parentId: null, status: "blocked", + }); mockChatRunRetries.prepareFailedChatRunRetry.mockResolvedValue({ actionId: retryActionId, issueId: failedChatIssueId, @@ -913,6 +931,50 @@ describe("agent live run routes", () => { }); }); + it("retries a task for an operator without agent-creation permission", async () => { + const fixture = createFailedChatRetryDb(false); + mockHeartbeatService.getRun.mockResolvedValue({ ...selectedRun, contextSnapshot: { + issueId: failedChatIssueId, + } }); + mockAccessService.decide.mockImplementation(async ({ action }) => ({ + allowed: action === "issue:comment" || action === "agent:wake", explanation: "Missing permission: agents:create", + })); + const res = await requestApp(await createApp(fixture.db, { + type: "board", userId: "operator", source: "session", companyIds: ["company-1"], + }), url => request(url).post(`/api/agents/${routeAgentId}/wakeup`).send(retryBody)); + expect(res.status, JSON.stringify(res.body)).toBe(202); + expect(mockAccessService.decide).toHaveBeenCalledWith(expect.objectContaining({ + action: "issue:comment", resource: expect.objectContaining({ + type: "issue", companyId: "company-1", issueId: failedChatIssueId, + }), + })); + expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith(routeAgentId, expect.objectContaining({ + requestedByActorType: "user", requestedByActorId: "operator", failedRunId: failedChatRunId, + payload: { issueId: failedChatIssueId }, + })); + }); + + it.each(["viewer", "missing", "other-company", "reassigned", "other-chat-owner"])( + "rejects a %s task retry without dispatching or requiring agent creation", async (fault) => { + const fixture = createFailedChatRetryDb(false); + mockHeartbeatService.getRun.mockResolvedValue({ ...selectedRun, contextSnapshot: { issueId: failedChatIssueId } }); + if (fault === "viewer") mockAccessService.decide.mockResolvedValue({ + allowed: false, explanation: "Viewer membership does not grant issue:comment.", + }); + else mockIssueService.getById.mockResolvedValue(fault === "missing" ? null : { + id: failedChatIssueId, companyId: fault === "other-company" ? "elsewhere" : "company-1", + assigneeAgentId: "other-agent", assigneeUserId: null, projectId: null, parentId: null, status: "blocked", + ...(fault === "other-chat-owner" ? { conversationAgentId: routeAgentId, conversationUserId: "someone-else" } : {}), + }); + const res = await requestApp(await createApp(fixture.db), url => + request(url).post(`/api/agents/${routeAgentId}/wakeup`).send(retryBody)); + expect(res.status).toBe(fault === "viewer" || fault === "other-chat-owner" ? 403 : fault === "reassigned" ? 409 : 404); + expect(mockAccessService.decide.mock.calls.every(([input]) => input.action !== "agents:create")).toBe(true); + expect(mockHeartbeatService.wakeup).not.toHaveBeenCalled(); + expect(mockChatRunRetries.prepareFailedChatRunRetry).not.toHaveBeenCalled(); + }, + ); + it.each([ ["failed", "deferred", null], ["timed_out", "running", "55555555-5555-4555-8555-555555555555"], @@ -1073,7 +1135,7 @@ describe("agent live run routes", () => { ); it.each(["agent", "company", "permission"])( - "denies %s authority before retry selection", + "denies %s authority before retry admission", async (denial) => { const fixture = createFailedChatRetryDb(); const actor = @@ -1615,7 +1677,7 @@ describe("agent live run routes", () => { id: "trace-1", status: "incomplete", deletedAt: null, - expiresAt: new Date(Date.now() + 60_000), + expiresAt: new Date("2099-01-01T00:00:00.000Z"), }, "trace_incomplete", ], diff --git a/server/src/__tests__/authorization-service.test.ts b/server/src/__tests__/authorization-service.test.ts index 7a303268f4..b76683562e 100644 --- a/server/src/__tests__/authorization-service.test.ts +++ b/server/src/__tests__/authorization-service.test.ts @@ -1462,8 +1462,8 @@ describeEmbeddedPostgres("authorization service", () => { })).resolves.toMatchObject({ allowed: false, reason: "deny_missing_membership" }); }); - it("keeps denying self-gated null-mapped actions for board members", async () => { - const company = await createCompany(db, "BoardWakeDenied"); + it("allows legacy member roles to wake agents while rejecting incomplete task mutation scope", async () => { + const company = await createCompany(db, "BoardWake"); const userId = `user-${randomUUID()}`; const targetAgent = await createAgent(db, company.id, { role: "engineer" }); await db.insert(companyMemberships).values({ @@ -1481,8 +1481,8 @@ describeEmbeddedPostgres("authorization service", () => { action: "agent:wake", resource: { type: "agent", companyId: company.id, agentId: targetAgent.id }, })).resolves.toMatchObject({ - allowed: false, - reason: "deny_unsupported_action", + allowed: true, + reason: "allow_simple_company_member", }); const issue = await createIssue(db, company.id, { title: "Wake denied issue" }); await expect(authorization.decide({ @@ -1749,6 +1749,25 @@ describeEmbeddedPostgres("authorization service", () => { }); }); + it.each(["session", "cloud_tenant"] as const)("allows %s operators to start agents, without granting hiring rights", async (source) => { + const company = await createCompany(db, "wake"); + const agent = await createAgent(db, company.id); + const userId = await createUser(db); + await db.insert(companyMemberships).values({ companyId: company.id, + principalType: "user", principalId: userId, status: "active", membershipRole: "operator" }); + const auth = authorizationService(db); + const actor = { type: "board" as const, source, userId, companyIds: [company.id] }; + const resource = { type: "agent" as const, companyId: company.id, agentId: agent.id }; + expect(await auth.decide({ actor, action: "agent:wake", resource })).toMatchObject({ allowed: true }); + expect(await auth.decide({ actor, action: "agents:create", resource: { type: "company", companyId: company.id } })).toMatchObject({ allowed: false }); + await db.update(companyMemberships).set({ membershipRole: "viewer" }).where(eq(companyMemberships.principalId, userId)); + expect(await auth.decide({ actor, action: "agent:wake", resource })).toMatchObject({ allowed: false }); + await db.update(companyMemberships).set({ membershipRole: "operator", status: "suspended" }).where(eq(companyMemberships.principalId, userId)); + expect(await auth.decide({ actor, action: "agent:wake", resource })).toMatchObject({ allowed: false }); + const otherCompany = await createCompany(db, "other-wake"); + expect(await auth.decide({ actor, action: "agent:wake", resource: { ...resource, companyId: otherCompany.id } })).toMatchObject({ allowed: false }); + }); + it("limits viewer members to read-only visibility actions", async () => { const company = await createCompany(db, "BoardViewerVisibility"); const userId = `user-${randomUUID()}`; diff --git a/server/src/__tests__/heartbeat-responsible-user-invariant.test.ts b/server/src/__tests__/heartbeat-responsible-user-invariant.test.ts index 8470b456b5..0fd788d258 100644 --- a/server/src/__tests__/heartbeat-responsible-user-invariant.test.ts +++ b/server/src/__tests__/heartbeat-responsible-user-invariant.test.ts @@ -149,6 +149,142 @@ describeEmbeddedPostgres("heartbeat responsible-user invariant", () => { return { companyId, ownerUserId, agentId }; } + it("dispatches an interrupted queue under the clicking operator through the real startup path", async () => { + const { companyId, agentId, ownerUserId } = await seedCompany(); + const operatorId = `operator-${randomUUID()}`, issueId = randomUUID(), commentId = randomUUID(), queueId = randomUUID(); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: operatorId, + membershipRole: "operator", status: "active" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Interrupted queue", status: "todo", + assigneeAgentId: agentId, responsibleUserId: ownerUserId }); + await db.insert(issueComments).values({ id: commentId, companyId, issueId, authorUserId: ownerUserId, body: "Continue the task" }); + await db.insert(agentWakeupRequests).values({ id: queueId, companyId, agentId, + source: "automation", status: "deferred_issue_execution", requestedByActorType: "system", + payload: { issueId, commentId, queuedCommentInterrupt: { actorId: operatorId, requestedAt: new Date().toISOString() }, + _paperclipWakeContext: { wakeCommentIds: [commentId], responsibleUserId: ownerUserId, + retryOfRunId: randomUUID(), originIdentityContextId: randomUUID() } }, + }); + await heartbeat.resumeQueuedCommentInterrupt(companyId, queueId); + const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, queueId)); + expect(receipt.status).toBe("coalesced"); + const completed = await waitForRun(db, receipt.runId!); + expect(completed).toMatchObject({ status: "succeeded", responsibleUserId: operatorId }); + expect(completed?.activeIdentityContextId).toBeTruthy(); + expect(completed?.contextSnapshot?.originIdentityContextId).toBeUndefined(); + expect(completed?.contextSnapshot?.retryOfRunId).toBeUndefined(); + expect(mockAdapterExecute).toHaveBeenCalled(); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId)); + expect(runs.every(run => run.responsibleUserId === operatorId && run.status === "succeeded")).toBe(true); + expect((await db.select().from(issueComments).where(eq(issueComments.id, commentId)))[0].authorUserId).toBe(ownerUserId); + }); + + it("keeps a board manual wake under its caller even when it adopts someone else's queue", async () => { + const { companyId, agentId, ownerUserId } = await seedCompany(); + const operatorId = `operator-${randomUUID()}`, issueId = randomUUID(), commentId = randomUUID(), queueId = randomUUID(); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: operatorId, + membershipRole: "operator", status: "active" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Manual wake", status: "todo", + assigneeAgentId: agentId, responsibleUserId: ownerUserId }); + await db.insert(issueComments).values({ id: commentId, companyId, issueId, authorUserId: ownerUserId, body: "Pending work" }); + await db.insert(agentWakeupRequests).values({ id: queueId, companyId, agentId, + source: "automation", reason: "issue_commented", status: "deferred_issue_execution", requestedByActorType: "user", requestedByActorId: ownerUserId, + payload: { issueId, commentId, _paperclipWakeContext: { wakeCommentIds: [commentId] } }, + }); + const run = await heartbeat.wakeup(agentId, { manualUserWake: true, source: "on_demand", triggerDetail: "manual", + payload: { issueId }, requestedByActorType: "user", requestedByActorId: operatorId, + contextSnapshot: { responsibleUserId: operatorId } }); + expect(run?.responsibleUserId).toBe(operatorId); + const completed = await waitForRun(db, run!.id); + expect(completed).toMatchObject({ status: "succeeded", responsibleUserId: operatorId }); + expect(completed?.contextSnapshot?.wakeCommentIds).toEqual([commentId]); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + expect((await db.select().from(heartbeatRuns)).every(row => row.responsibleUserId === operatorId)).toBe(true); + }); + + it("keeps the clicking user when a manual wake merges into an older deferred receipt", async () => { + const { companyId, agentId, ownerUserId } = await seedCompany(); + const operatorId = `operator-${randomUUID()}`, issueId = randomUUID(), commentId = randomUUID(), queueId = randomUUID(); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: operatorId, + membershipRole: "operator", status: "active" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Deferred manual wake", status: "todo", + assigneeAgentId: agentId, responsibleUserId: ownerUserId }); + let finish!: () => void; + const blocked = new Promise(resolve => { finish = resolve; }); + const execute = mockAdapterExecute.getMockImplementation()!; + mockAdapterExecute.mockImplementationOnce(async () => { await blocked; return execute(); }); + const first = await heartbeat.wakeup(agentId, { payload: { issueId }, + requestedByActorType: "user", requestedByActorId: ownerUserId }); + try { + await vi.waitFor(() => expect(mockAdapterExecute).toHaveBeenCalled(), { timeout: 5_000 }); + await db.insert(issueComments).values({ id: commentId, companyId, issueId, authorUserId: ownerUserId, body: "Pending work" }); + await db.insert(agentWakeupRequests).values({ id: queueId, companyId, agentId, + source: "automation", reason: "issue_commented", status: "deferred_issue_execution", + requestedByActorType: "user", requestedByActorId: ownerUserId, + payload: { issueId, commentId, _paperclipWakeContext: { wakeCommentIds: [commentId] } }, + }); + expect(await heartbeat.wakeup(agentId, { manualUserWake: true, source: "on_demand", triggerDetail: "manual", + payload: { issueId }, requestedByActorType: "user", requestedByActorId: operatorId, + contextSnapshot: { responsibleUserId: operatorId } })).toBeNull(); + const [pending] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, queueId)); + expect(pending).toMatchObject({ requestedByActorType: "user", requestedByActorId: operatorId, + payload: { manualUserWake: true } }); + } finally { + finish(); + } + await drainHeartbeatRunsToQuiescence(db, heartbeat); + const successors = (await db.select().from(heartbeatRuns)).filter(run => run.id !== first!.id); + expect(successors.length).toBeGreaterThan(0); + expect(successors.every(run => run.responsibleUserId === operatorId && run.status === "succeeded")).toBe(true); + expect((await db.select().from(issueComments).where(eq(issueComments.id, commentId)))[0].authorUserId).toBe(ownerUserId); + }); + + it("starts an unscoped manual wake with its own user instead of joining another user's run", async () => { + const { companyId, agentId, ownerUserId } = await seedCompany(); + const operatorId = `operator-${randomUUID()}`; + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: operatorId, + membershipRole: "operator", status: "active" }); + let finish!: () => void; + const blocked = new Promise(resolve => { finish = resolve; }); + const execute = mockAdapterExecute.getMockImplementation()!; + mockAdapterExecute.mockImplementationOnce(async () => { await blocked; return execute(); }); + const first = await heartbeat.wakeup(agentId, { manualUserWake: true, source: "on_demand", triggerDetail: "manual", + requestedByActorType: "user", requestedByActorId: ownerUserId }); + let second: Awaited>; + try { + await vi.waitFor(() => expect(mockAdapterExecute).toHaveBeenCalled(), { timeout: 5_000 }); + second = await heartbeat.wakeup(agentId, { manualUserWake: true, source: "on_demand", triggerDetail: "manual", + requestedByActorType: "user", requestedByActorId: operatorId }); + expect(second?.id).not.toBe(first!.id); + expect(second?.responsibleUserId).toBe(operatorId); + } finally { + finish(); + } + await drainHeartbeatRunsToQuiescence(db, heartbeat); + expect(await waitForRun(db, second!.id)).toMatchObject({ status: "succeeded", responsibleUserId: operatorId }); + expect(await waitForRun(db, first!.id)).toMatchObject({ status: "succeeded", responsibleUserId: ownerUserId }); + }); + + it("denies a manual wake of another user's private conversation", async () => { + const { companyId, agentId, ownerUserId } = await seedCompany(); + const issueId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Private conversation", status: "todo", + assigneeAgentId: agentId, responsibleUserId: ownerUserId, conversationAgentId: agentId, conversationUserId: ownerUserId, conversationState: "active" }); + await expect(heartbeat.wakeup(agentId, { manualUserWake: true, source: "on_demand", triggerDetail: "manual", + payload: { issueId }, requestedByActorType: "user", requestedByActorId: "another-user" })).rejects.toThrow("conversation owner"); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + expect(await db.select().from(heartbeatRuns)).toHaveLength(0); + }); + + it("does not accept a caller-supplied manual-wake authority marker", async () => { + const { companyId, agentId, ownerUserId } = await seedCompany(); + const run = await heartbeat.wakeup(agentId, { source: "on_demand", triggerDetail: "manual", + requestedByActorType: "agent", requestedByActorId: agentId, payload: { manualUserWake: true }, + contextSnapshot: { responsibleUserId: ownerUserId } }); + expect((await waitForRun(db, run!.id))?.status).toBe("succeeded"); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, run!.wakeupRequestId!)); + expect(wake.payload?.manualUserWake).toBeUndefined(); + }); + it("uses the issue responsible user for automated dependency wakes without a message context", async () => { const { companyId, agentId } = await seedCompany(); const issueResponsibleUserId = `issue-owner-${randomUUID()}`; diff --git a/server/src/__tests__/issue-queued-comments-routes.test.ts b/server/src/__tests__/issue-queued-comments-routes.test.ts index 6a4f23a638..4b3bae7816 100644 --- a/server/src/__tests__/issue-queued-comments-routes.test.ts +++ b/server/src/__tests__/issue-queued-comments-routes.test.ts @@ -14,13 +14,14 @@ import { createDb, heartbeatRuns, issueComments, + issueRecoveryActions, issues, runIdentityContexts, } from "@paperclipai/db"; import { errorHandler } from "../middleware/index.js"; import { issueRoutes } from "../routes/issues.js"; import { heartbeatService } from "../services/heartbeat.js"; -import { reconcileSteeredIdentity } from "../services/run-identity.js"; +import { initializeRunIdentity, reconcileSteeredIdentity } from "../services/run-identity.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, @@ -195,6 +196,145 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { }, ); + it("does not accept interruption authority from an agent wake payload", async () => { + const seeded = await seedQueue(); + 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)); + await heartbeatService(db).wakeup(seeded.agentId, { + source: "on_demand", reason: "issue_commented", + requestedByActorType: "agent", requestedByActorId: seeded.agentId, + payload: { issueId: seeded.issueId, commentId: seeded.commentIds[1], + queuedCommentInterrupt: { actorId: "other-operator", requestedAt: new Date().toISOString() } }, + contextSnapshot: { issueId: seeded.issueId, wakeCommentId: seeded.commentIds[1] }, + }); + const wakes = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, seeded.companyId)); + expect(wakes.length).toBeGreaterThan(0); + expect(wakes.every(wake => !wake.payload?.queuedCommentInterrupt)).toBe(true); + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, seeded.companyId)); + expect(runs.find(run => run.id === seeded.runId)?.status).toBe("running"); + expect(runs.every(run => !run.contextSnapshot?.explicitUserContinuation)).toBe(true); + }); + + it("denies a viewer's interrupt before persisting intent or cancelling a run", async () => { + const seeded = await seedQueue(); + await db.update(companyMemberships).set({ membershipRole: "viewer" }) + .where(eq(companyMemberships.principalId, "other-operator")); + const client = app(seeded.companyId, "other-operator"); + 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, revision: queue.body.revision, targetRunId: seeded.runId, + }).expect(403); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)); + expect(wake.payload?.queuedCommentInterrupt).toBeUndefined(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)))[0].status).toBe("running"); + }); + + it.each([null, "stopped-target", "system-receipt"])("sends a stopped legacy queue once with target %s", async (target) => { + const seeded = await seedQueue(); + if (target === "system-receipt") await db.update(agentWakeupRequests).set({ + requestedByActorType: "system", requestedByActorId: "heartbeat", + }).where(eq(agentWakeupRequests.id, seeded.wakeId)); + 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", status: "succeeded", + finishedAt: new Date("2026-08-22T15:03:00.000Z"), + }).where(eq(heartbeatRuns.id, seeded.runId)); + await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, seeded.issueId)); + // Occupy this agent on a different task so the actual successor remains + // queued and the test never launches a provider. + await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId, + status: "running", contextSnapshot: { issueId: randomUUID() }, + }); + const client = app(seeded.companyId, "other-operator"); + const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + expect(queue.body.targetRunId).toBeNull(); + const body = { queueId: seeded.wakeId, revision: queue.body.revision, + targetRunId: target === "stopped-target" ? seeded.runId : null }; + await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`).send(body).expect(200); + 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.status).toBe("queued"); + expect(successor.responsibleUserId).toBe("other-operator"); + const identity = await initializeRunIdentity(db, { + companyId: seeded.companyId, issueId: seeded.issueId, + runId: successor.id, messageIds: seeded.commentIds, responsibleUserId: "queue-owner", cause: "dispatch", + }); + expect(identity.responsibleUserId).toBe("other-operator"); + expect(identity.cause).toBe("queued_comment_interrupt"); + expect(successor.contextSnapshot?.wakeCommentIds).toEqual(seeded.commentIds); + await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, seeded.companyId))).toHaveLength(3); + await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`).send(body).expect(409); + }); + + it.each(["user", "system"])("keeps stopped-run interruption intent on a %s receipt across restart until the process stops, then delivers once", async (actorType) => { + const seeded = await seedQueue(); + await db.update(agentWakeupRequests).set({ requestedByActorType: actorType }) + .where(eq(agentWakeupRequests.id, seeded.wakeId)); + 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", status: "failed", + processPid: process.pid, errorCode: "process_lost", + finishedAt: new Date("2026-08-22T15:03:00.000Z"), + }).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: "legacy_execution_requires_reconciliation", fingerprint: seeded.runId, + status: "resolved", outcome: "blocked", nextAction: "Automatic recovery stopped.", + 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, "other-operator"); + 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, revision: queue.body.revision, targetRunId: null, + }).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?.queuedCommentInterrupt).toMatchObject({ actorId: "other-operator" }); + expect(waiting.payload?.executionWait).toMatchObject({ reason: "process_running" }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, seeded.companyId))).toHaveLength(2); + await db.update(heartbeatRuns).set({ processPid: 999999999 }).where(eq(heartbeatRuns.id, seeded.runId)); + await db.update(agentWakeupRequests).set({ updatedAt: new Date(0) }).where(eq(agentWakeupRequests.id, seeded.wakeId)); + // New service instances have no memory of the HTTP request. Concurrent + // periodic workers must consume its durable receipt exactly once. + await Promise.all([heartbeatService(db).resumeQueuedRuns(), heartbeatService(db).resumeQueuedRuns()]); + const [delivered] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)); + expect(delivered.status).toBe("coalesced"); + const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, delivered.runId!)); + expect(successor.contextSnapshot).toMatchObject({ wakeCommentIds: seeded.commentIds, + previousRunId: seeded.runId, forceFreshSession: true }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, seeded.companyId))).toHaveLength(3); + }); + + it("recovers a message deferred after legacy finalization released the task lock", async () => { + const seeded = await seedQueue(); + 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", status: "succeeded", + finishedAt: new Date("2026-08-22T15:03:00.000Z"), + }).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?.wakeCommentIds).toEqual(seeded.commentIds); + await heartbeatService(db).resumeQueuedRuns(); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, seeded.companyId))).toHaveLength(3); + }); + async function promoteQueue(seeded: Awaited>) { const queueRunId = randomUUID(); const wake = await db diff --git a/server/src/__tests__/run-identity.test.ts b/server/src/__tests__/run-identity.test.ts index 634d6a70cc..9155a9b9fa 100644 --- a/server/src/__tests__/run-identity.test.ts +++ b/server/src/__tests__/run-identity.test.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import { eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { agents, companies, createDb, heartbeatRuns, heartbeatRunEvents, issueComments, issueThreadInteractions, issues } from "@paperclipai/db"; +import { agentWakeupRequests, agents, companies, createDb, heartbeatRuns, heartbeatRunEvents, issueComments, issueThreadInteractions, issues } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { acceptSteeredIdentity, captureRunIdentity, initializeRunIdentity, listRunIdentityContexts, rejectSteeredIdentity, reserveSteeredIdentity } from "../services/run-identity.js"; @@ -41,6 +41,71 @@ const support = await getEmbeddedPostgresTestSupport(); await initializeRunIdentity(db, { ...input, responsibleUserId: "B", cause: "restart" }); expect(await listRunIdentityContexts(db, input.companyId, input.runId)).toHaveLength(4); }); + async function seedInterrupt() { + const input = await seed(); + const queueId = randomUUID(), wakeupRequestId = randomUUID(); + const contextSnapshot = { issueId: input.issueId, wakeCommentIds: input.messageIds }; + await db.insert(agentWakeupRequests).values([ + { id: queueId, companyId: input.companyId, agentId: input.agentId, source: "automation", + status: "coalesced", runId: input.runId, requestedByActorType: "system", + payload: { issueId: input.issueId, _paperclipWakeContext: { wakeCommentIds: input.messageIds }, + queuedCommentInterrupt: { actorId: "operator", requestedAt: new Date().toISOString() } } }, + { id: wakeupRequestId, companyId: input.companyId, agentId: input.agentId, source: "on_demand", + status: "queued", runId: input.runId, requestedByActorType: "user", requestedByActorId: "operator", + idempotencyKey: `queued-comment-interrupt:${queueId}` }, + ]); + await db.update(heartbeatRuns).set({ wakeupRequestId, contextSnapshot }).where(eq(heartbeatRuns.id, input.runId)); + return { ...input, queueId, wakeupRequestId, contextSnapshot }; + } + + it("uses the clicking operator through startup and restart without changing message authors", async () => { + const input = await seedInterrupt(); + // A stale originating context cannot replace the explicit click's identity. + const identity = await initializeRunIdentity(db, { + ...input, responsibleUserId: "A", parentContextId: randomUUID(), cause: "dispatch", + }); + expect(identity).toMatchObject({ responsibleUserId: "operator", cause: "queued_comment_interrupt" }); + const history = await listRunIdentityContexts(db, input.companyId, input.runId); + expect(history.map(row => row.responsibleUserId)).toEqual(["operator", "operator", "operator", "operator"]); + expect(await initializeRunIdentity(db, { ...input, responsibleUserId: "B", cause: "restart" })).toEqual(identity); + const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, input.issueId)); + expect(input.messageIds.map(id => comments.find(c => c.id === id)?.authorUserId)).toEqual(["A", "B", "A"]); + expect((await captureRunIdentity(db, input)).context?.responsibleUserId).toBe("operator"); + const retryRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: retryRunId, companyId: input.companyId, + agentId: input.agentId, contextSnapshot: input.contextSnapshot, status: "queued", retryOfRunId: input.runId }); + const retried = await initializeRunIdentity(db, { companyId: input.companyId, runId: retryRunId, + issueId: input.issueId, parentRunId: input.runId, responsibleUserId: "A", cause: "retry" }); + expect(retried.responsibleUserId).toBe("operator"); + }); + + it.each(["malformed", "missing", "unconsumed", "other-run", "other-task", "other-agent", "other-actor", "other-message"])( + "rejects %s interrupt authority before creating any execution identity", async (fault) => { + const input = await seedInterrupt(); + if (fault === "malformed" || fault === "missing") { + await db.update(agentWakeupRequests).set({ + idempotencyKey: `queued-comment-interrupt:${fault === "malformed" ? "not-an-id" : randomUUID()}`, + }).where(eq(agentWakeupRequests.id, input.wakeupRequestId)); + } else if (fault === "other-actor") { + await db.update(agentWakeupRequests).set({ requestedByActorId: "someone-else" }).where(eq(agentWakeupRequests.id, input.wakeupRequestId)); + } else { + const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, input.queueId)); + if (fault === "other-agent") { + const agentId = randomUUID(); + await db.insert(agents).values({ id: agentId, companyId: input.companyId, name: "Other", role: "engineer" }); + await db.update(agentWakeupRequests).set({ agentId }).where(eq(agentWakeupRequests.id, input.queueId)); + } else await db.update(agentWakeupRequests).set( + fault === "unconsumed" ? { status: "deferred_issue_execution" } : + fault === "other-run" ? { runId: null } : + { payload: { ...receipt.payload, ...(fault === "other-task" ? { issueId: randomUUID() } : + { _paperclipWakeContext: { wakeCommentIds: [randomUUID()] } }) } }, + ).where(eq(agentWakeupRequests.id, input.queueId)); + } + await expect(initializeRunIdentity(db, { ...input, responsibleUserId: "A", cause: "dispatch" })).rejects.toThrow("interrupt authority"); + expect(await listRunIdentityContexts(db, input.companyId, input.runId)).toHaveLength(0); + }, + ); + it("holds acquisition during uncertain steering, preserves snapshots, and never rewinds on replay", async () => { const input = await seed(); await initializeRunIdentity(db, { ...input, messageIds: [input.messageIds[0]], responsibleUserId: "A", cause: "instruction" }); diff --git a/server/src/modules/wake-queue/adapters/postgres.test.ts b/server/src/modules/wake-queue/adapters/postgres.test.ts index 9e2567f935..8a5a2b1951 100644 --- a/server/src/modules/wake-queue/adapters/postgres.test.ts +++ b/server/src/modules/wake-queue/adapters/postgres.test.ts @@ -242,6 +242,24 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { }); } + it.each(["queued", "running", "scheduled_retry"])("does not promote another turn behind a %s successor without an execution lock", async (status) => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId }); + const runId = await seedRun({ companyId, agentId, status: "succeeded", contextSnapshot: { issueId } }); + await seedRun({ companyId, agentId, status, contextSnapshot: { issueId } }); + const wakeId = await seedDeferredWake({ companyId, agentId, issueId }); + const adapter = createPostgresWakeQueueAdapter(db, stubDeps); + let drained = false; + await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async () => { + drained = true; + return { outcome: { kind: "released" }, postCommitEffects: [] }; + }); + expect(drained).toBe(false); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId)); + expect(wake.status).toBe("deferred_issue_execution"); + }); + it("leaves deferred work untouched until the effective execution hold clears", async () => { const companyId = await seedCompany(); const agentId = await seedAgent({ companyId }); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 514ddcdcf2..efbcec9e1e 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -894,6 +894,10 @@ export function createWakeAdmissionWriter(): WakeAdmissionWriter { .set({ payload: input.mergedPayload, coalescedCount: input.nextCoalescedCount, + ...(input.manualUserWakeActorId ? { + requestedByActorType: "user", + requestedByActorId: input.manualUserWakeActorId, + } : {}), updatedAt: new Date(), }) .where( @@ -1052,6 +1056,20 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd throw new Error(`wake-queue: pre-drain decision ${preDrain.kind} reached without an issue row`); } + // Enqueue does not stamp executionRunId until dispatch. A concurrent + // queued successor still owns the next turn, including during a late + // finalization/stranded-queue retry under this issue lock. Another + // agent's review participation retains its separate recovery path. + const [successor] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, input.companyId), + eq(heartbeatRuns.agentId, run.agentId), + sql`${heartbeatRuns.id} <> ${run.id}`, + or(eq(heartbeatRuns.nativeIssueId, issueRow.id), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueRow.id}`), + inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]), + )).limit(1); + if (successor) return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot }; + if (preDrain.kind === "blocked") { return { outcome: { diff --git a/server/src/modules/wake-queue/application/ports.ts b/server/src/modules/wake-queue/application/ports.ts index 0dfc4e8f25..c4e4cc0f9e 100644 --- a/server/src/modules/wake-queue/application/ports.ts +++ b/server/src/modules/wake-queue/application/ports.ts @@ -428,6 +428,8 @@ export interface WakeAdmissionWriter { existingDeferredWakeId: string; mergedPayload: Record; nextCoalescedCount: number; + /** A fresh manual click replaces the merged queue's execution requester. */ + manualUserWakeActorId?: string; /** Persist each durable input's own receipt atomically with the merge. */ coalescedReceipt?: CoalescedDeferredAdmissionReceipt; }, 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 71de306851..528acc5391 100644 --- a/server/src/modules/wake-queue/application/use-cases.test.ts +++ b/server/src/modules/wake-queue/application/use-cases.test.ts @@ -848,6 +848,17 @@ describe("admitWakeBehindIssueExecution", () => { expect(writer.coalesceIntoActiveExecutionRun).not.toHaveBeenCalled(); }); + it("gives manual input its own run boundary even when the active receipt has the same requester", async () => { + const writer = createFakeAdmissionWriter(); + const admit = createAdmitWakeBehindIssueExecution({ + reader: createFakeAdmissionReader(), writer, helpers: createFakeAdmissionHelpers(), + }); + expect(await admit(SCOPE, admissionInput({ payload: { issueId: "issue-1", manualUserWake: true } }))) + .toEqual({ kind: "deferred" }); + expect(writer.coalesceIntoActiveExecutionRun).not.toHaveBeenCalled(); + expect(writer.insertNewDeferredWake).toHaveBeenCalledTimes(1); + }); + it("keeps ordinary non-durable coalescing independent of durable actor lookup", async () => { const writer = createFakeAdmissionWriter(); const reader = createFakeAdmissionReader({ diff --git a/server/src/modules/wake-queue/application/use-cases.ts b/server/src/modules/wake-queue/application/use-cases.ts index b9817e53df..c9c35fee7d 100644 --- a/server/src/modules/wake-queue/application/use-cases.ts +++ b/server/src/modules/wake-queue/application/use-cases.ts @@ -342,6 +342,7 @@ async function promoteDeferredWake( const promotedTriggerDetail = workingCandidate.triggerDetail ?? null; const promotedPayload = { ...workingCandidate.payload }; delete promotedPayload["_paperclipWakeContext"]; + delete promotedPayload["queuedCommentInterrupt"]; const promotedContextSeed: Record = { ...workingCandidate.deferredContextSeed }; if (pauseHold.activePauseHold) { @@ -716,6 +717,12 @@ export function createAdmitWakeBehindIssueExecution(deps: { scope: TransactionScope, input: AdmitWakeBehindIssueExecutionInput, ): Promise { + const manualUserWakeActorId = input.payload?.manualUserWake === true + ? readNonEmptyString(input.requestedByActorId) : null; + if (input.payload?.manualUserWake === true && + (input.requestedByActorType !== "user" || !manualUserWakeActorId)) { + throw new Error("wake-queue: manual wake requires an authenticated user"); + } const isSameExecutionAgent = await deps.reader.isSameExecutionAgent(scope, { companyId: input.companyId, activeExecutionRunAgentId: input.activeExecutionRun.agentId, @@ -723,8 +730,10 @@ export function createAdmitWakeBehindIssueExecution(deps: { agentNameKey: input.agentNameKey, }); + // A manual click establishes a fresh execution identity. Even a matching + // requester can have a different originating identity on an exact retry. const shouldDeferFollowupWake = - deps.helpers.shouldDeferFollowupWakeForSameIssue({ + Boolean(manualUserWakeActorId) || deps.helpers.shouldDeferFollowupWakeForSameIssue({ activeRunStatus: input.activeExecutionRun.status, isSameExecutionAgent, wakeCommentId: input.wakeCommentId, @@ -830,6 +839,7 @@ export function createAdmitWakeBehindIssueExecution(deps: { existingDeferredWakeId: existingDeferred.id, mergedPayload, nextCoalescedCount: (existingDeferred.coalescedCount ?? 0) + 1, + ...(manualUserWakeActorId ? { manualUserWakeActorId } : {}), ...(input.durableReceipt ? { coalescedReceipt: { diff --git a/server/src/routes/agents.ts b/server/src/routes/agents.ts index f4f12b7703..1428fae7b8 100644 --- a/server/src/routes/agents.ts +++ b/server/src/routes/agents.ts @@ -1735,6 +1735,17 @@ export function agentRoutes( throw forbidden(decision.explanation, authorizationDeniedDetails(decision)); } + async function assertBoardCanWakeAgent(req: Request, agent: { id: string; companyId: string }) { + assertBoard(req); + if (!hasCompanyAccess(req, agent.companyId)) throw notFound("Agent not found"); + assertCompanyAccess(req, agent.companyId); + const decision = await access.decide({ + actor: req.actor, action: "agent:wake", + resource: { type: "agent", companyId: agent.companyId, agentId: agent.id }, + }); + if (!decision.allowed) throw forbidden(decision.explanation, authorizationDeniedDetails(decision)); + } + // The single owner-authorization helper for the three adapter login routes. It // requires a board actor, company access, and the same configuration // permission as the adapter Test route (`agents:create`). It returns the @@ -5534,7 +5545,7 @@ export function agentRoutes( return; } } else { - await assertBoardCanManageAgentsForCompany(req, agent.companyId); + await assertBoardCanWakeAgent(req, agent); } if (req.body.debug?.providerTrace === "raw") { assertInstanceAdmin(req); @@ -5576,6 +5587,23 @@ export function agentRoutes( typeof failedContext.issueId === "string" ? failedContext.issueId : null; + if (issueId) { + const issue = await issueService(db).getById(issueId); + if (!issue || issue.companyId !== agent.companyId) throw notFound("Task not found"); + if (issue.conversationAgentId && issue.conversationUserId !== req.actor.userId) { + throw forbidden("Only the conversation owner can retry a chat run"); + } + const decision = await access.decide({ + actor: req.actor, action: "issue:comment", + resource: { + type: "issue", companyId: issue.companyId, issueId: issue.id, + projectId: issue.projectId, parentIssueId: issue.parentId, + assigneeAgentId: issue.assigneeAgentId, assigneeUserId: issue.assigneeUserId, status: issue.status, + }, + }); + if (!decision.allowed) throw forbidden(decision.explanation, authorizationDeniedDetails(decision)); + if (issue.assigneeAgentId !== agent.id) throw conflict("The task is no longer assigned to this agent."); + } const chatBinding = issueId ? await db .select({ id: chatConversations.id }) @@ -5642,6 +5670,7 @@ export function agentRoutes( } const run = await heartbeat.wakeup(id, { failedRunId: req.body.failedRunId ?? null, + ...(req.actor.type === "board" && !req.body.failedRunId ? { manualUserWake: true } : {}), source: opts.source, triggerDetail: req.body.triggerDetail ?? "manual", reason: req.body.reason ?? null, @@ -5736,7 +5765,7 @@ export function agentRoutes( return; } } else { - await assertBoardCanManageAgentsForCompany(req, agent.companyId); + await assertBoardCanWakeAgent(req, agent); } const providerTraceRequested = req.body?.debug?.providerTrace === "raw"; if (providerTraceRequested) { @@ -5775,6 +5804,7 @@ export function agentRoutes( } } const wakeOpts: Parameters[1] = { + ...(req.actor.type === "board" ? { manualUserWake: true } : {}), source: "on_demand", triggerDetail: typeof body.triggerDetail === "string" ? body.triggerDetail as "manual" | "system" | "ping" | "callback" : "manual", requestedByActorType: req.actor.type === "agent" ? "agent" : "user", diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 55deb1f8a1..81ed975078 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -357,6 +357,9 @@ const queuedCommentSteeringTargetSchema = queuedCommentMutationTargetSchema.extend({ targetRunId: z.string().min(1), }); +const queuedCommentInterruptTargetSchema = queuedCommentMutationTargetSchema.extend({ + targetRunId: z.string().min(1).nullable(), +}); const editQueuedCommentSchema = queuedCommentMutationTargetSchema.extend({ body: z .string() @@ -6943,9 +6946,10 @@ export function issueRoutes( actor: ReturnType; queueId: string; targetRunId?: string; + allowStoppedTarget?: boolean; }) { - await input.tx - .select({ id: issueRows.id }) + const [currentIssue] = await input.tx + .select() .from(issueRows) .where( and( @@ -6954,6 +6958,8 @@ export function issueRoutes( ), ) .for("update"); + if (!currentIssue) throw notFound("Issue not found"); + input.issue = currentIssue; const wake = await input.tx .select() .from(agentWakeupRequests) @@ -7030,7 +7036,7 @@ export function issueRoutes( and( eq(heartbeatRuns.id, activeRunId), eq(heartbeatRuns.companyId, input.issue.companyId), - eq(heartbeatRuns.status, "running"), + input.allowStoppedTarget ? undefined : eq(heartbeatRuns.status, "running"), ), ) .for("update") @@ -15216,12 +15222,14 @@ export function issueRoutes( router.post( "/issues/:id/queued-comments/interrupt", - validate(queuedCommentSteeringTargetSchema), + validate(queuedCommentInterruptTargetSchema), async (req, res) => { assertBoard(req); if (!req.actor.userId) throw forbidden("Board user context required"); const issue = await getAccessibleResource(req, res, svc.getById(req.params.id as string), "Issue not found"); if (!issue) return; + const decision = await decideIssueAccess(req, issue, "issue:comment"); + if (!decision.allowed) throw forbidden(decision.explanation, authorizationDeniedDetails(decision)); if (issue.conversationAgentId) { if (!(await instanceSettings.getExperimental()).enableAgentChat) throw notFound("Agent Chat is disabled"); if (req.actor.userId !== issue.conversationUserId) { @@ -15229,23 +15237,43 @@ export function issueRoutes( } } const actor = getActorInfo(req); - await db.transaction(async (tx) => { + const runToInterrupt = await db.transaction(async (tx) => { const locked = await lockQueuedCommentState({ - tx, issue, actor, queueId: req.body.queueId, targetRunId: req.body.targetRunId, + tx, issue, actor, queueId: req.body.queueId, targetRunId: req.body.targetRunId ?? undefined, + allowStoppedTarget: true, }); assertQueueMutationTarget({ queue: locked.queue, queueId: req.body.queueId, revision: req.body.revision }); - if (locked.queue.protocol !== "legacy" || locked.activeRun?.agentId !== issue.assigneeAgentId) { + if (locked.queue.protocol !== "legacy" || locked.state !== "deferred" || + !locked.queue.entries.length || + (locked.activeRun && locked.activeRun.agentId !== locked.wake.agentId)) { throw conflict("This queue does not support legacy interruption"); } + if (locked.activeRun && locked.activeRun.status !== "running" && + !["succeeded", "failed", "timed_out", "interrupted", "cancelled"].includes(locked.activeRun.status)) { + throw conflict("The previous run has not stopped"); + } + if (locked.activeRun?.status === "running" && locked.activeRun.id !== req.body.targetRunId) { + throw conflict("The queued message targets a stale run", { code: "queued_comment_stale_target" }); + } + // The click is durable fresh user intent, including when the message + // predates a failed run's stop. Keep its content and original attribution. + await tx.update(agentWakeupRequests).set({ + payload: { ...readObject(locked.wake.payload), queuedCommentInterrupt: { + actorId: actor.actorId, requestedAt: new Date().toISOString(), + } }, + updatedAt: new Date(), + }).where(eq(agentWakeupRequests.id, locked.wake.id)); + return locked.activeRun?.status === "running" ? locked.activeRun.id : null; }); // Never hold the issue lock while joining the adapter. Queue edits and // discards stay authoritative until the dispatcher claims the successor. const options = operatorInterruptCancelOptions({ issueId: issue.id, actor }); - await heartbeat.cancelRun(req.body.targetRunId, "Interrupted to send queued messages", { + if (runToInterrupt) await heartbeat.cancelRun(runToInterrupt, "Interrupted to send queued messages", { ...options, suppressImmediateRecovery: true, resultJson: { ...options.resultJson, queuedCommentInterruptQueueId: req.body.queueId }, }); + await heartbeat.resumeQueuedCommentInterrupt(issue.companyId, req.body.queueId, { retryCleanup: true }); await logActivity(db, { companyId: issue.companyId, actorType: actor.actorType, actorId: actor.actorId, agentId: actor.agentId, runId: actor.runId, agentApiKeyId: actor.agentApiKeyId, diff --git a/server/src/services/authorization.ts b/server/src/services/authorization.ts index a63b1a407c..70b1d2032c 100644 --- a/server/src/services/authorization.ts +++ b/server/src/services/authorization.ts @@ -1784,6 +1784,7 @@ export function authorizationService(db: Db | DbTransaction) { } if ( input.action === "agent:read" || + input.action === "agent:wake" || input.action === "company_scope:read" || input.action === "decision_queue:manage" || input.action === "decision_queue:read" || @@ -1798,6 +1799,7 @@ export function authorizationService(db: Db | DbTransaction) { // Mirroring the tasks:assign carve-out above, viewers keep the // read-only visibility actions but not the privileged ones. const requiresNonViewer = + input.action === "agent:wake" || input.action === "runtime:manage" || input.action === "secrets:read" || input.action === "decision_queue:manage" || diff --git a/server/src/services/execution-continuation.ts b/server/src/services/execution-continuation.ts index 2b4b694c2e..b6dde3f9b1 100644 --- a/server/src/services/execution-continuation.ts +++ b/server/src/services/execution-continuation.ts @@ -1,4 +1,5 @@ -import { and, asc, desc, eq, isNotNull, isNull, sql } from "drizzle-orm"; +import { and, asc, desc, eq, inArray, isNotNull, isNull, sql } from "drizzle-orm"; +import { z } from "zod"; import { agentWakeupRequests, heartbeatRuns, @@ -11,6 +12,7 @@ import { import type { ExecutionContinuationEnvelope } from "@paperclipai/shared"; import { sanitizeQuarantinedCommentForHigherTrust } from "./source-trust.js"; import { hasConversationContinuationPolicy } from "./conversation-continuation.js"; +import { queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js"; const object = (v: unknown): Record => v && typeof v === "object" && !Array.isArray(v) @@ -268,18 +270,39 @@ export async function buildExecutionContinuation(input: { eq(agentWakeupRequests.reason, "retry_failed_run"), eq(agentWakeupRequests.requestedByActorType, "user"), sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, )) : []; - const authorization = reconciliations.map(row => object(row.evidence.explicitUserContinuation)) - .find(value => value.previousRunId === explicitUserSource && + // Admission records the board operator's authority separately from the + // message author. At dispatch, prove that exact queue was adopted by this + // run; caller-supplied continuation context cannot grant this authority. + const continuationAuthorizations = reconciliations.map(row => object(row.evidence.explicitUserContinuation)) + .filter(value => value.previousRunId === explicitUserSource && (!input.runId || value.runId === input.runId) && value.commentId === explicitContinuation.commentId && - priorRuns.some(run => run.id === value.runId) && - (failedRunId + priorRuns.some(run => run.id === value.runId)); + const interruptQueueIds = [...new Set(continuationAuthorizations.flatMap(value => { + const parsed = z.string().guid().safeParse(value.queuedCommentInterruptId); + return parsed.success ? [parsed.data] : []; + }))]; + const interruptQueues = interruptQueueIds.length + ? await db.select().from(agentWakeupRequests).where(and( + inArray(agentWakeupRequests.id, interruptQueueIds), + eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, input.agentId), + eq(agentWakeupRequests.status, "coalesced"), + sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, + sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is not null`, + )) : []; + const authorization = continuationAuthorizations.find(value => failedRunId ? value.failedRunId === failedRunId && retryWakes.some(wake => wake.runId === value.runId && wake.requestedByActorId === value.actorId && priorRuns.some(run => run.id === wake.runId && run.retryOfRunId === failedRunId)) : rows.some(comment => comment.id === value.commentId && - comment.authorType === "user" && comment.authorUserId === value.actorId && - !comment.createdByRunId && !comment.deletedAt))); + comment.authorType === "user" && + (value.queuedCommentInterruptId + ? interruptQueues.some(queue => queue.id === value.queuedCommentInterruptId && + queue.runId === value.runId && + object(object(queue.payload).queuedCommentInterrupt).actorId === value.actorId && + queuedCommentIdsFromWakePayload(queue.payload).includes(comment.id)) + : comment.authorUserId === value.actorId) && + !comment.createdByRunId && !comment.deletedAt)); if (!predecessor || !authorization || explicitUserSource !== sourceRunId) throw new Error("continuation_user_authorization_missing"); } diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 2f43896571..7d320ca9f6 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -42,6 +42,143 @@ const support = await getEmbeddedPostgresTestSupport(); actorType: "user", actorId: "board", reason: "issue_commented" }; } type Fixture = Awaited>; + it.each(["pending", "failed", "historical", "shared", "retained"])("an explicit queued interrupt retries only its stopped sandbox, without granting automatic retries (%s)", async scenario => { + const fails = scenario === "failed"; + const protectedLease = scenario === "shared" || scenario === "retained"; + const f = await seed(), other = await seed(); + await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId)); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + await db.update(heartbeatRuns).set({ runtimeMode: "legacy", nativeIssueId: null, processPid: null }) + .where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" }) + .where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + const queueId = randomUUID(); + await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId, + source: "automation", reason: "issue_commented", status: "deferred_issue_execution", + requestedByActorType: "system", payload: { issueId: f.issueId, commentId: f.commentId, + _paperclipWakeContext: { wakeCommentIds: [f.commentId] }, + queuedCommentInterrupt: { actorId: "board", requestedAt: new Date().toISOString() } }, + }); + const identities = [f, other].map(fixture => ({ id: randomUUID(), companyId: fixture.companyId, + heartbeatRunId: fixture.sourceRunId, provider: "daytona", providerLeaseId: fixture.sourceRunId })); + for (const identity of identities) await db.insert(environmentLeases).values({ ...identity, + status: "pending_cleanup", leasePolicy: "ephemeral", releasedAt: new Date(), cleanupStatus: "failed", + metadata: { pendingCleanupRetryAttempts: 5, pendingCleanupRetryCapWarned: true } }); + if (scenario === "historical" || protectedLease) await db.update(environmentLeases).set({ + status: "failed", cleanupStatus: "success", + }).where(eq(environmentLeases.id, identities[0].id)); + if (scenario === "shared") await db.update(environmentLeases).set({ + providerLeaseId: identities[0].providerLeaseId, status: "active", releasedAt: null, + }).where(eq(environmentLeases.id, identities[1].id)); + if (scenario === "retained") await db.update(environmentLeases).set({ + status: "retained", leasePolicy: "retain_on_failure", + }).where(eq(environmentLeases.id, identities[0].id)); + const attempted: string[] = []; + const heartbeat = heartbeatService(db, { environmentRuntime: { + isPendingCleanupWorkerReady: async () => true, + retryPendingSandboxTeardown: async ({ lease }: { lease: { id: string; providerLeaseId: string } }) => { + attempted.push(lease.id); + if (fails) throw new Error("Provider unavailable"); + return { providerLeaseId: lease.providerLeaseId, state: "destroyed" }; + }, + } as unknown as HeartbeatEnvironmentRuntime }); + try { + await heartbeat.resumeQueuedCommentInterrupt(f.companyId, queueId); + expect(attempted).toEqual([]); + await heartbeat.resumeQueuedCommentInterrupt(f.companyId, queueId, { retryCleanup: true }); + expect(attempted).toEqual(protectedLease ? [] : [identities[0].id]); + await heartbeat.resumeQueuedCommentInterrupt(f.companyId, queueId); + expect(attempted).toHaveLength(protectedLease ? 0 : 1); + const [queue] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, queueId)); + expect(queue.status).toBe(fails || protectedLease ? "deferred_issue_execution" : "coalesced"); + expect(Boolean(queue.runId)).toBe(!fails && !protectedLease); + const [untouched] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, identities[1].id)); + expect(untouched).toMatchObject({ status: scenario === "shared" ? "active" : "pending_cleanup", metadata: { pendingCleanupRetryAttempts: 5 } }); + } finally { + for (const identity of identities) await db.delete(environmentLeases).where(eq(environmentLeases.id, identity.id)); + } + }); + it("a durable queue interrupt authorizes older legacy messages but still requires the provider to stop", async () => { + const f = await seed(); + await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId)); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + await db.update(heartbeatRuns).set({ runtimeMode: "legacy", nativeIssueId: null, + processPid: process.pid, + }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" }) + .where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + await db.update(issueComments).set({ authorUserId: "original-author", createdAt: new Date("2026-09-11T09:00:00Z") }) + .where(eq(issueComments.id, f.commentId)); + const queueId = randomUUID(); + await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId, + source: "on_demand", reason: "issue_commented", status: "deferred_issue_execution", + requestedByActorType: "user", requestedByActorId: "original-author", + payload: { issueId: f.issueId, _paperclipWakeContext: { wakeCommentIds: [f.commentId] }, + queuedCommentInterrupt: { actorId: "board", requestedAt: new Date().toISOString() } }, + }); + const attempt = (queue = queueId) => db.transaction(async tx => { + await tx.select().from(issues).where(eq(issues.id, f.issueId)).for("update"); + return admitExplicitNativeContinuation({ ...f, db: tx as unknown as typeof db, + queuedCommentInterruptId: queue, dryRun: true }); + }); + expect(await attempt()).toBeNull(); + await db.update(heartbeatRuns).set({ processPid: 999999999 }).where(eq(heartbeatRuns.id, f.sourceRunId)); + expect(await attempt(randomUUID())).toBeNull(); + expect(await attempt()).toMatchObject({ previousRunId: f.sourceRunId, commentId: f.commentId }); + await db.update(agentWakeupRequests).set({ status: "cancelled" }).where(eq(agentWakeupRequests.id, queueId)); + expect(await attempt()).toBeNull(); + }); + it("dispatches another user's queued legacy message using the consumed board interrupt receipt", async () => { + const f = await seed(); + await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId)); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + await db.update(heartbeatRuns).set({ runtimeMode: "legacy", nativeIssueId: null }) + .where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" }) + .where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + await db.update(issueComments).set({ authorUserId: "original-author", createdAt: new Date("2026-09-11T09:00:00Z") }) + .where(eq(issueComments.id, f.commentId)); + // Hold adapter startup so the test can exercise the real dispatch envelope + // deterministically, without invoking a provider. + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + const queueId = randomUUID(); + await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId, + source: "automation", reason: "issue_commented", status: "deferred_issue_execution", + requestedByActorType: "system", payload: { issueId: f.issueId, commentId: f.commentId, + _paperclipWakeContext: { wakeCommentIds: [f.commentId] }, + queuedCommentInterrupt: { actorId: "board", requestedAt: new Date().toISOString() } }, + }); + await heartbeatService(db).resumeQueuedCommentInterrupt(f.companyId, queueId); + const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, queueId)); + expect(receipt.status).toBe("coalesced"); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, receipt.runId!)); + const dispatch = (runId = run.id) => buildExecutionContinuation({ db, companyId: f.companyId, + issueId: f.issueId, agentId: f.agentId, runId, context: run.contextSnapshot!, + summary: null, exposeLowTrustRaw: false }); + const envelope = await dispatch(); + expect(envelope.interruptedRunId).toBe(f.sourceRunId); + expect(envelope.originCommentIds).toContain(f.commentId); + expect(envelope.messages).toEqual(expect.arrayContaining([expect.objectContaining({ id: f.commentId, body: "What happened?" })])); + await expect(dispatch(randomUUID())).rejects.toThrow("continuation_user_authorization_missing"); + for (const patch of [ + { status: "cancelled" }, { runId: f.sourceRunId }, + { payload: { ...receipt.payload, issueId: randomUUID() } }, + { payload: { ...receipt.payload, queuedCommentInterrupt: { actorId: "someone-else" } } }, + { payload: { ...receipt.payload, _paperclipWakeContext: { wakeCommentIds: [] }, commentId: undefined } }, + ]) { + await db.update(agentWakeupRequests).set(patch).where(eq(agentWakeupRequests.id, queueId)); + await expect(dispatch()).rejects.toThrow("continuation_user_authorization_missing"); + await db.update(agentWakeupRequests).set({ status: receipt.status, runId: receipt.runId, payload: receipt.payload }) + .where(eq(agentWakeupRequests.id, queueId)); + } + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + await db.update(issueRecoveryActions).set({ evidence: { ...action.evidence, + explicitUserContinuation: { ...(action.evidence.explicitUserContinuation as Record), + queuedCommentInterruptId: "malformed-historical-receipt" }, + } }).where(eq(issueRecoveryActions.id, action.id)); + await expect(dispatch()).rejects.toThrow("continuation_user_authorization_missing"); + }); const admit = (f: Fixture, dryRun = false) => db.transaction(async tx => { await tx.select().from(issues).where(eq(issues.id, f.issueId)).for("update"); const result = await admitExplicitNativeContinuation({ ...f, dryRun, db: tx as unknown as typeof db }); diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 3aeaef4b88..3aedd8e17b 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -5,7 +5,7 @@ import { hasRemoteTerminationReceipt, remoteLeaseCleanupScope } from "./remote-e import { z } from "zod"; import { and, eq, inArray, isNull, ne, or, sql } from "drizzle-orm"; import { - agents, approvals, issueApprovals, issueThreadInteractions, + agents, agentWakeupRequests, approvals, issueApprovals, issueThreadInteractions, environmentLeases, heartbeatRuns, issueComments, issueRecoveryActions, issues, nativeRunFinalizations, type Db, } from "@paperclipai/db"; @@ -15,6 +15,7 @@ import { adapterExecutionControls } from "./adapter-execution-control.js"; import { persistActivity } from "./activity-log.js"; import { historicalAdapterType, isConversationAdapter } from "./conversation-continuation.js"; +import { queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js"; type Run = typeof heartbeatRuns.$inferSelect; const terminal = ["failed", "interrupted", "timed_out", "cancelled"]; @@ -33,6 +34,8 @@ export async function admitExplicitNativeContinuation(input: { actorType: string | null | undefined; actorId: string | null | undefined; reason: string | null; commentId: string | null; successorRunId: string; failedRunId?: string | null; + /** Server-recorded board intent to send an existing legacy message queue. */ + queuedCommentInterruptId?: string; dryRun?: boolean; onBlocked?: (reason: string, message: string) => void; }): Promise<{ previousRunId: string; commentId: string | null; failedRunId?: string } | null> { @@ -47,16 +50,27 @@ export async function admitExplicitNativeContinuation(input: { eq(issues.companyId, companyId), eq(issues.id, issueId), )); if (!task || task.assigneeAgentId !== agentId || ["done", "cancelled"].includes(task.status)) return null; + const [interruptQueue] = input.queuedCommentInterruptId ? await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.id, input.queuedCommentInterruptId), + eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, agentId), + eq(agentWakeupRequests.status, "deferred_issue_execution"), + sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, + sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt'->>'actorId' = ${actorId}`, + )) : []; + const queuedInterrupt = Boolean(interruptQueue && commentId && + queuedCommentIdsFromWakePayload(interruptQueue.payload).includes(commentId)); + if (input.queuedCommentInterruptId && !queuedInterrupt) return null; const [comment] = retry ? [] : await db.select().from(issueComments).where(and( eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId), eq(issueComments.id, commentId!), eq(issueComments.authorType, "user"), - eq(issueComments.authorUserId, actorId), isNull(issueComments.createdByRunId), + 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(); 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; const actions = await db.select().from(issueRecoveryActions).where(and( eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, issueId), executionBlockerPredicate(), @@ -86,7 +100,7 @@ export async function admitExplicitNativeContinuation(input: { if (!run || run.agentId !== agentId || !terminal.includes(run.status) || (run.nativeIssueId ?? run.contextSnapshot?.issueId) !== issueId || !run.finishedAt) return blocked("source_unavailable", "The previous execution has not finished or its owner changed. Your message is saved."); - if (authorizedAt <= run.finishedAt) return blocked("message_predates_stop", "This message arrived before the previous run stopped. Send a new message to continue."); + if (!queuedInterrupt && authorizedAt <= run.finishedAt) return blocked("message_predates_stop", "This message arrived before the previous run stopped. Send a new message to continue."); if (adapterExecutionControls.has(run.id)) return blocked("execution_settling", "Waiting for the previous run to stop. Your message will start automatically."); const unusedAdmission = run.status === "cancelled" && !run.startedAt && run.errorCode === "execution_reconciliation_required" && @@ -94,6 +108,7 @@ export async function admitExplicitNativeContinuation(input: { const legacyUserTurn = run.runtimeMode === "legacy" && action.cause === "legacy_execution_requires_reconciliation" && isConversationAdapter(agent.adapterType); + if (queuedInterrupt && !legacyUserTurn) return null; if (legacyUserTurn) { const historicalAdapter = await historicalAdapterType(db, run); // A settings change never converts a known process/webhook execution into @@ -161,7 +176,8 @@ 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 } : {}), runId: input.successorRunId, + const authorization = { actorId, commentId, ...(retry ? { failedRunId: input.failedRunId } : {}), + ...(queuedInterrupt ? { queuedCommentInterruptId: input.queuedCommentInterruptId } : {}), runId: input.successorRunId, previousRunId: previous.id, recordedAt: new Date().toISOString() }; for (const runId of cancelledStartupIds) { await db.update(nativeRunFinalizations).set({ diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index db550768db..3846948e79 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -2,7 +2,7 @@ import { AGENT_CHAT_DIRECTIVE, conversationReplay, isConversation, isConversatio import { PROCESS_IDENTITY_RECORDED, recordNativeLocalProcessStop } from "./native-local-process-stop.js"; import { legacyControllerBootId, legacyControllerClaim, renewLegacyControllerLease, hasLiveLegacyController, revokeExpiredLegacyController, watchLegacyControllerLease } from "./legacy-controller-lease.js"; import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js"; -import { remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteCleanupScopes } from "./remote-execution-termination.js"; +import { hasRemoteTerminationReceipt, remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteCleanupScopes } from "./remote-execution-termination.js"; import { applyConnectorSkills, prepareConnectorSkillDelivery, resolveConnectorAssignments } from "./connector-runtime.js"; import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js"; import { connectionIntentService } from "./connection-intents.js"; @@ -27,7 +27,7 @@ import { buildHeartbeatRunStatusLiveEventPayload } from "./heartbeat-run-status- export { buildHeartbeatRunStatusLiveEventPayload } from "./heartbeat-run-status-payload.js"; import { buildExecutionContinuation } from "./execution-continuation.js"; import { renderPaperclipWakePrompt } from "@paperclipai/adapter-utils/server-utils"; -import { initializeRunIdentity } from "./run-identity.js"; +import { initializeRunIdentity, explicitOperatorRunIdentity } from "./run-identity.js"; import { assertDurableChatWakeupReceipt, assertDurableChatWakeupRequest, @@ -61,6 +61,7 @@ import { gte, inArray, isNull, + isNotNull, lt, lte, ne, @@ -3501,6 +3502,10 @@ function normalizeMaxConcurrentRuns(value: unknown) { } interface WakeupOptions { + /** Set only by authenticated board wake routes; never copied from caller payloads. */ + manualUserWake?: boolean; + /** Internal resume of a queue with persisted board interruption intent. */ + queuedCommentInterruptId?: string; /** Exact failed run selected by an authenticated board Retry request. */ failedRunId?: string | null; durableChatRequest?: DurableChatWakeupRequest; @@ -10055,6 +10060,10 @@ export function heartbeatService( for (const wake of pending) { if (wake.idempotencyKey?.startsWith("chat-inbound:")) continue; const payload = parseObject(wake.payload); + if (payload.queuedCommentInterrupt) { + await resumeQueuedCommentInterrupt(wake.companyId, wake.id); + continue; + } const context = parseObject(payload[DEFERRED_WAKE_CONTEXT_KEY]); const commentId = deriveCommentId(context, payload); if (legacyContinuation) { @@ -10094,6 +10103,93 @@ export function heartbeatService( } } + async function resumeQueuedCommentInterrupt(companyId: string, queueId: string, opts?: { retryCleanup?: boolean }) { + const [wake] = await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.id, queueId), eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.status, "deferred_issue_execution"), + )); + if (!wake) return; + const payload = parseObject(wake.payload); + const actorId = readNonEmptyString(parseObject(payload.queuedCommentInterrupt).actorId); + const commentIds = queuedCommentIdsFromWakePayload(payload); + const issueId = readNonEmptyString(payload.issueId); + if (!actorId || !issueId || !commentIds.length) return; + const agent = await getAgent(wake.agentId); + if (!agent || agent.companyId !== companyId || agent.adapterType === "paperclip_runner") return; + const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, companyId), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`, + inArray(heartbeatRuns.status, ["running", "queued", "scheduled_retry"]), + )).limit(1); + if (active) return; + if (opts?.retryCleanup) { + // Only the HTTP click grants an extra cleanup attempt. Periodic retries + // reuse the intent to deliver, never a fresh provider teardown budget. + const sourceRun = await db.transaction(async tx => { + const [task] = await tx.select().from(issues).where(and( + eq(issues.companyId, companyId), eq(issues.id, issueId), + )).for("update"); + const [current] = await tx.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.id, queueId), eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.agentId, wake.agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"), + sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, + sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt'->>'actorId' = ${actorId}`, + )); + if (!task || task.assigneeAgentId !== wake.agentId || ["done", "cancelled"].includes(task.status) || + !current || !queuedCommentIdsFromWakePayload(current.payload).length) return null; + const [successor] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, companyId), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`, + inArray(heartbeatRuns.status, ["running", "queued", "scheduled_retry"]), + )).limit(1); + if (successor) return null; + const blocker = await getExecutionBlocker(tx as unknown as Db, companyId, issueId); + const run = blocker?.runId ? await tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, blocker.runId), + eq(heartbeatRuns.agentId, wake.agentId), eq(heartbeatRuns.runtimeMode, "legacy"), + inArray(heartbeatRuns.status, ["failed", "timed_out", "interrupted", "cancelled"]), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`, + )).then(rows => rows[0]) : null; + if (!run || activeRunExecutions.has(run.id) || adapterExecutionControls.has(run.id)) return null; + // Older ephemeral leases recorded successful cleanup without a provider + // receipt. Re-verify them through the recorded teardown path; a timestamp + // alone never certifies termination. Retained/reusable resources stay put. + const historical = await tx.select().from(environmentLeases).where(and( + eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, run.id), + eq(environmentLeases.leasePolicy, "ephemeral"), isNotNull(environmentLeases.releasedAt), + inArray(environmentLeases.status, ["released", "expired", "failed"]), + )).for("update"); + for (const lease of historical) { + if (!lease.provider || lease.provider === "local" || !lease.providerLeaseId || hasRemoteTerminationReceipt(lease)) continue; + const [otherOwner] = await tx.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + ne(environmentLeases.id, lease.id), eq(environmentLeases.provider, lease.provider), + eq(environmentLeases.providerLeaseId, lease.providerLeaseId), + or(isNull(environmentLeases.releasedAt), inArray(environmentLeases.status, ["active", "retained", "pending_cleanup"])), + )).limit(1); + if (otherOwner) continue; + await tx.update(environmentLeases).set({ status: "pending_cleanup", updatedAt: new Date() }) + .where(eq(environmentLeases.id, lease.id)); + } + return run; + }); + if (sourceRun) await sweepPendingCleanupLeases({ explicitRetry: { + companyId, runId: sourceRun.id, actorId, reason: "queued_comment_interrupt", + } }); + } + const deliveryPayload = { ...payload }; + 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), + requestedByActorType: "user", requestedByActorId: actorId, + queuedCommentInterruptId: queueId, + issueStateGuard: { assigneeAgentId: wake.agentId, statuses: ["todo", "in_progress", "in_review", "blocked"] }, + idempotencyKey: `queued-comment-interrupt:${queueId}`, + }, queueId); + } + async function resumeExecutionWaitComments() { if ((await getSchedulingSuppression()).suppressed) return; const waits = await db.select({ wake: agentWakeupRequests }) @@ -10108,6 +10204,7 @@ export function heartbeatService( ))), eq(agentWakeupRequests.status, "deferred_issue_execution"), eq(agentWakeupRequests.requestedByActorType, "user"), sql`${agentWakeupRequests.payload}->'executionWait' is not null`, + sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is null`, lte(agentWakeupRequests.updatedAt, new Date(Date.now() - 30_000)), notInArray(issues.status, ["done", "cancelled"]))) .orderBy(asc(agentWakeupRequests.updatedAt)).limit(50); @@ -10613,7 +10710,8 @@ export function heartbeatService( ReturnType >; }) { - const responsibleUserId = await resolveResponsibleUserIdForRunSeed({ + const operatorIdentity = await explicitOperatorRunIdentity(db, input.run); + const responsibleUserId = operatorIdentity?.actorId ?? await resolveResponsibleUserIdForRunSeed({ companyId: input.run.companyId, contextSnapshot: input.contextSnapshot, issueContext: input.issueContext, @@ -17834,7 +17932,7 @@ export function heartbeatService( * A later user Retry may try again after a provider failure; automatic * sweeps retain their exhausted budget and never gain extra attempts. */ - explicitRetry?: { companyId: string; runId: string; actorId: string }; + explicitRetry?: { companyId: string; runId: string; actorId: string; reason?: "retry_failed_run" | "queued_comment_interrupt" }; }): Promise<{ swept: number; destroyed: number; @@ -17964,7 +18062,7 @@ export function heartbeatService( if (opts?.explicitRetry) await logActivity(db, { companyId: row.companyId, actorType: "user", actorId: opts.explicitRetry.actorId, action: "environment_lease.cleanup_retried", entityType: "environment_lease", entityId: row.id, - runId: opts.explicitRetry.runId, details: { attempt: attempts + 1, reason: "retry_failed_run" }, + runId: opts.explicitRetry.runId, details: { attempt: attempts + 1, reason: opts.explicitRetry.reason ?? "retry_failed_run" }, }); try { @@ -18676,6 +18774,52 @@ export function heartbeatService( if ((await getSchedulingSuppression()).suppressed) return; await resumeExecutionWaitComments(); const cutoff = await getWorktreeExecutionCutoff(); + const pendingInterrupts = await db.select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId }) + .from(agentWakeupRequests).innerJoin(companies, eq(companies.id, agentWakeupRequests.companyId)) + .where(and(eq(agentWakeupRequests.status, "deferred_issue_execution"), + eq(companies.status, "active"), + sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is not null`, + lte(agentWakeupRequests.updatedAt, new Date(Date.now() - 30_000)), + cutoff ? gte(agentWakeupRequests.requestedAt, cutoff) : undefined)) + .orderBy(asc(agentWakeupRequests.updatedAt)).limit(50); + for (const wake of pendingInterrupts) { + await db.update(agentWakeupRequests).set({ updatedAt: new Date() }).where(and( + eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.status, "deferred_issue_execution"), + )); + await resumeQueuedCommentInterrupt(wake.companyId, wake.id).catch(err => { + logger.warn({ err, queueId: wake.id }, "failed to resume interrupted comment queue"); + }); + } + // A server restart or a message/cleanup race can leave a deferred wake + // after its owner has released the issue lock. Revisit it through the same + // release admission, so recovery holds and operator Stops still apply. + const strandedQueues = await db.select({ wake: agentWakeupRequests }) + .from(agentWakeupRequests) + .innerJoin(issues, and(eq(issues.companyId, agentWakeupRequests.companyId), + sql`${issues.id}::text = ${agentWakeupRequests.payload}->>'issueId'`, + eq(issues.assigneeAgentId, agentWakeupRequests.agentId))) + .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`, + 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; + 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)}`, + )).orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)).limit(1); + await db.update(agentWakeupRequests).set({ updatedAt: new Date() }).where(and( + eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.status, "deferred_issue_execution"), + )); + if (!latest || latest.runtimeMode !== "legacy" || !isHeartbeatRunTerminalStatus(latest.status)) continue; + await releaseIssueExecutionAndPromote(latest, { suppressImmediateRecovery: true }).catch(err => { + logger.warn({ err, queueId: wake.id }, "failed to promote stranded legacy comments"); + }); + } // The cancellation marker is durable intent. Retry while its exact queue // is still deferred, including after a failed cleanup promotion or restart. @@ -25142,18 +25286,6 @@ export function heartbeatService( ${JSON.stringify({ startupPreparationSettledAt: new Date().toISOString() })}::jsonb`, }).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, "cancelled"))); } - // Interrupting a queued message explicitly authorizes the pending queue. - // Retry its normal promotion after leases and adapter cleanup have settled; - // the earlier terminal write can still have an execution blocker here. - if ( - latestRun?.status === "cancelled" && - latestRun.runtimeMode !== "native" && - readNonEmptyString(latestRun.resultJson?.queuedCommentInterruptQueueId) - ) { - await releaseIssueExecutionAndPromote(latestRun, { suppressImmediateRecovery: true }).catch((err) => { - logger.error({ err, runId: run.id }, "failed to promote interrupted comment queue after cleanup"); - }); - } } finally { controllerLease.stop(); activeRunExecutions.delete(run.id); @@ -25166,6 +25298,20 @@ export function heartbeatService( adapterExecutionControls.delete(run.id); } } + // Terminalization precedes lease and adapter cleanup. Only now is the + // owner gone; retry pending input for ordinary completions as well as Stop. + if (latestRun?.runtimeMode === "legacy" && isHeartbeatRunTerminalStatus(latestRun.status)) { + const [pending] = await db.select({ id: agentWakeupRequests.id, payload: agentWakeupRequests.payload }).from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId), + eq(agentWakeupRequests.status, "deferred_issue_execution"), + sql`${agentWakeupRequests.payload}->>'issueId' = ${String(latestRun.contextSnapshot?.issueId)}`, + )).limit(1); + if (pending) await (pending.payload?.queuedCommentInterrupt + ? resumeQueuedCommentInterrupt(run.companyId, pending.id) + : releaseIssueExecutionAndPromote(latestRun, { suppressImmediateRecovery: true })).catch(err => { + logger.error({ err, runId: run.id }, "failed to promote legacy comment queue after cleanup"); + }); + } if ( !nativeSessionResumeScheduled && !nativeWorkspaceFinalizeScheduled && @@ -25211,7 +25357,19 @@ export function heartbeatService( ...(opts.contextSnapshot ?? {}), }; const reason = opts.reason ?? null; - const payload = opts.payload ?? null; + let payload = opts.payload ? { ...opts.payload } : null; + // Only the board queue route can record interruption authority on an + // existing receipt. Never accept this internal marker from a wake caller. + if (payload) { + delete payload.queuedCommentInterrupt; + delete payload.manualUserWake; + } + if (opts.manualUserWake) { + if (opts.requestedByActorType !== "user" || !opts.requestedByActorId || opts.failedRunId) { + throw new HttpError(403, "Manual wake requires an authenticated user"); + } + payload = { ...payload, manualUserWake: true }; + } const executionReconciliationWake = contextSnapshot.source === "execution.reconciled" || opts.idempotencyKey?.startsWith("execution-reconciliation:") === true; @@ -25236,6 +25394,9 @@ export function heartbeatService( if (issueId) { const conversation = await getIssueExecutionContext(agent.companyId, issueId); if (isConversation(conversation)) { + if (opts.manualUserWake && conversation!.conversationUserId !== opts.requestedByActorId) { + throw new HttpError(403, "Only the conversation owner can start a chat run"); + } if (isConversationExecutionWake(conversation, reason ?? readNonEmptyString(enrichedContextSnapshot.wakeReason))) return null; if (agent.id !== conversation!.conversationAgentId) return null; if (!(await instanceSettings.getExperimental()).enableAgentChat) return null; @@ -25557,8 +25718,10 @@ export function heartbeatService( const isolatedWorkspacesEnabled = issueId ? (await instanceSettings.getExperimental()).enableIsolatedWorkspaces : false; + let operatorResponsibleUserId: string | null = opts.manualUserWake ? opts.requestedByActorId! : null; let queuedResponsibleUserIdPromise: Promise | null = null; const resolveQueuedResponsibleUserId = () => { + if (operatorResponsibleUserId) return Promise.resolve(operatorResponsibleUserId); queuedResponsibleUserIdPromise ??= (async () => { const queuedIssueContext = issueId ? await getIssueExecutionContext(agent.companyId, issueId) @@ -25738,8 +25901,13 @@ export function heartbeatService( const [pending] = await tx.select().from(agentWakeupRequests).where(and( eq(agentWakeupRequests.id, executionWaitRequestId), eq(agentWakeupRequests.companyId, agent.companyId), eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"), - eq(agentWakeupRequests.requestedByActorType, "user"), - eq(agentWakeupRequests.requestedByActorId, opts.requestedByActorId ?? ""), + // A user message can join a queue originally created by a + // system wake. The recorded board click supplies fresh authority. + opts.queuedCommentInterruptId === executionWaitRequestId + ? undefined : eq(agentWakeupRequests.requestedByActorType, "user"), + opts.queuedCommentInterruptId === executionWaitRequestId + ? sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt'->>'actorId' = ${opts.requestedByActorId ?? ""}` + : eq(agentWakeupRequests.requestedByActorId, opts.requestedByActorId ?? ""), sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, )); // The issue lock serializes cleanup callbacks and periodic workers. @@ -25747,6 +25915,21 @@ export function heartbeatService( if (!pending || !wakeCommentId || !queuedCommentIdsFromWakePayload(pending.payload).includes(wakeCommentId)) { return { kind: "deferred" as const }; } + if (!opts.queuedCommentInterruptId && pending.payload?.manualUserWake === true) { + // A persisted manual wake keeps its actor when an execution wait + // resumes. The locked receipt above has revalidated that actor. + payload = { ...payload, manualUserWake: true }; + operatorResponsibleUserId = opts.requestedByActorId!; + } + if (opts.queuedCommentInterruptId) { + // The locked board receipt supplies execution authority even when + // another user authored the messages. Dispatch revalidates the receipt. + operatorResponsibleUserId = opts.requestedByActorId!; + // Edits/discards between the click and dispatch remain authoritative. + Object.assign(enrichedContextSnapshot, withQueuedCommentIdsInRunContext( + enrichedContextSnapshot, queuedCommentIdsFromWakePayload(pending.payload), + )); + } } let automaticParentRunId: string | null = null; if ( @@ -26124,6 +26307,7 @@ export function heartbeatService( db: tx as unknown as Db, companyId: issue.companyId, issueId: issue.id, agentId, actorType: opts.requestedByActorType, actorId: opts.requestedByActorId, reason, commentId: wakeCommentId ?? null, failedRunId: opts.failedRunId, successorRunId: explicitContinuationRunId, + queuedCommentInterruptId: opts.queuedCommentInterruptId, dryRun: true, onBlocked: (reason, message) => { continuationWait = { reason, message }; }, }))) return deferBlockedExecution(executionBlocker); @@ -26886,6 +27070,7 @@ export function heartbeatService( db: tx as unknown as Db, companyId: issue.companyId, issueId: issue.id, agentId, actorType: opts.requestedByActorType, actorId: opts.requestedByActorId, reason, commentId: wakeCommentId ?? null, failedRunId: opts.failedRunId, successorRunId: explicitContinuationRunId, + queuedCommentInterruptId: opts.queuedCommentInterruptId, }); if (!explicitContinuation && executionBlocker) return deferBlockedExecution(executionBlocker); if (explicitContinuation) { @@ -26929,6 +27114,7 @@ export function heartbeatService( .orderBy(asc(agentWakeupRequests.requestedAt)) : []; const adoptedComments = pendingComments.filter((wake) => { + if (wake.id === opts.queuedCommentInterruptId) return true; const deferredPayload = parseObject(wake.payload); const deferredContext = parseObject( deferredPayload[DEFERRED_WAKE_CONTEXT_KEY], @@ -27103,8 +27289,9 @@ export function heartbeatService( contextSnapshot: enrichedContextSnapshot, wakeCommentId, }); + // Unscoped manual wakes need their own receipt and execution identity too. const rawCoalescedTarget = - opts.allowRunCoalescing === false + opts.allowRunCoalescing === false || opts.manualUserWake ? null : (sameScopeQueuedRun ?? sameScopeScheduledRetryRun ?? @@ -28508,6 +28695,7 @@ export function heartbeatService( releaseEnvironmentLeasesForRun, resumeRemoteStopComments, + resumeQueuedCommentInterrupt, resumeExecutionWaitComments, sweepStaleIssueLocks, diff --git a/server/src/services/run-identity.ts b/server/src/services/run-identity.ts index 117e4744b1..2bfe88ad82 100644 --- a/server/src/services/run-identity.ts +++ b/server/src/services/run-identity.ts @@ -1,5 +1,6 @@ import { and, asc, desc, eq, inArray, sql } from "drizzle-orm"; import { + agentWakeupRequests, heartbeatRuns, heartbeatRunEvents, issueComments, @@ -10,6 +11,48 @@ import { } from "@paperclipai/db"; import { conflict, forbidden } from "../errors.js"; import { isUuidLike } from "@paperclipai/shared"; +import { queuedCommentIdsFromRunContext, queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js"; + +/** Resolve an explicit click from persisted receipts, never caller context or message authors. */ +export async function explicitOperatorRunIdentity( + executor: Pick, + run: Pick, +) { + const [request] = run.wakeupRequestId ? await executor.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.id, run.wakeupRequestId), eq(agentWakeupRequests.companyId, run.companyId), + eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.runId, run.id), + )) : []; + if (request?.payload?.manualUserWake === true) { + if (request.requestedByActorType !== "user" || !request.requestedByActorId) { + throw forbidden("Manual wake requires an authenticated user"); + } + return { actorId: request.requestedByActorId, cause: "manual_user_wake" }; + } + const prefix = "queued-comment-interrupt:"; + if (!request?.idempotencyKey?.startsWith(prefix)) return null; + const queueId = request.idempotencyKey.slice(prefix.length); + // The key only locates a candidate. The consumed queue, actor, run, company, + // agent, task, and delivered messages must all independently agree. + if (!isUuidLike(queueId)) { + throw forbidden("Queued-message interrupt authority is unavailable"); + } + const [receipt] = await executor.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.id, queueId), eq(agentWakeupRequests.companyId, run.companyId), + eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.runId, run.id), + eq(agentWakeupRequests.status, "coalesced"), + )); + const marker = receipt?.payload?.queuedCommentInterrupt; + const actorId = marker && typeof marker === "object" && "actorId" in marker ? marker.actorId : null; + const ids = queuedCommentIdsFromWakePayload(receipt?.payload); + const deliveredIds = queuedCommentIdsFromRunContext(run.contextSnapshot); + if (typeof actorId !== "string" || !actorId || !ids.length || + receipt?.payload?.issueId !== run.contextSnapshot?.issueId || + !ids.every(id => deliveredIds.includes(id)) || + request.requestedByActorType !== "user" || request.requestedByActorId !== actorId) { + throw forbidden("Queued-message interrupt authority is unavailable"); + } + return { actorId, cause: "queued_comment_interrupt" }; +} export type RunIdentityContext = typeof runIdentityContexts.$inferSelect; type Executor = Pick; @@ -147,6 +190,7 @@ export async function initializeRunIdentity( .where(eq(runIdentityContexts.id, run.activeIdentityContextId)); return current!; } + const operatorIdentity = await explicitOperatorRunIdentity(tx, run); const [parent] = input.parentRunId ? await tx .select() @@ -171,10 +215,9 @@ export async function initializeRunIdentity( ), ) : []; - const parentId = - interaction?.sourceIdentityContextId ?? - input.parentContextId ?? - parent?.activeIdentityContextId; + const parentId = operatorIdentity ? null : ( + interaction?.sourceIdentityContextId ?? input.parentContextId ?? parent?.activeIdentityContextId + ); const [origin] = parentId ? await tx .select() @@ -192,12 +235,10 @@ export async function initializeRunIdentity( let current = await append(tx, { companyId: input.companyId, runId: input.runId, - responsibleUserId: origin - ? origin.responsibleUserId - : input.responsibleUserId, + responsibleUserId: operatorIdentity?.actorId ?? (origin ? origin.responsibleUserId : input.responsibleUserId), parentContextId: origin?.id ?? null, cause: - origin?.cause === "company_default" ? "company_default" : input.cause, + operatorIdentity ? operatorIdentity.cause : origin?.cause === "company_default" ? "company_default" : input.cause, correlationId: "dispatch", }); const ids = [...new Set(input.messageIds ?? [])]; @@ -220,10 +261,10 @@ export async function initializeRunIdentity( current = await append(tx, { companyId: input.companyId, runId: input.runId, - responsibleUserId: comment.authorUserId, + responsibleUserId: operatorIdentity?.actorId ?? comment.authorUserId, messageId: id, parentContextId: current.id, - cause: "instruction", + cause: operatorIdentity?.cause ?? "instruction", correlationId: `message:${id}`, }); } diff --git a/ui/src/api/issues.test.ts b/ui/src/api/issues.test.ts index 3d78406b57..31a4548767 100644 --- a/ui/src/api/issues.test.ts +++ b/ui/src/api/issues.test.ts @@ -29,6 +29,23 @@ describe("issuesApi.list", () => { mockApi.patch.mockResolvedValue({}); }); + it.each([null, "stopped-run"])("dispatches a stopped queue using its current revision (%s)", async (target) => { + mockApi.get.mockResolvedValueOnce({ queueId: "queue-1", targetRunId: null, revision: "revision-2" }); + await issuesApi.interruptLatestQueuedComments("issue-1", target); + expect(mockApi.post).toHaveBeenCalledWith("/issues/issue-1/queued-comments/interrupt", { + queueId: "queue-1", targetRunId: null, revision: "revision-2", + }); + }); + + it.each([ + { queueId: null, targetRunId: null, revision: "empty" }, + { queueId: "queue-1", targetRunId: "new-run", revision: "changed" }, + ])("rejects a changed or empty queue before interruption", async queue => { + mockApi.get.mockResolvedValueOnce(queue); + await expect(issuesApi.interruptLatestQueuedComments("issue-1", "old-run")).rejects.toThrow("queued messages changed"); + expect(mockApi.post).not.toHaveBeenCalled(); + }); + it("fetches all pages of tasks created from the source without filtering parentage", async () => { const firstPage = Array.from({ length: 500 }, (_, index) => ({ id: `task-${index}` })); mockApi.get.mockResolvedValueOnce(firstPage).mockResolvedValueOnce([{ id: "last-task" }]); diff --git a/ui/src/api/issues.ts b/ui/src/api/issues.ts index 9abac76847..7d8bf562a3 100644 --- a/ui/src/api/issues.ts +++ b/ui/src/api/issues.ts @@ -386,8 +386,17 @@ export const issuesApi = { ), interruptQueuedComments: ( id: string, - data: { queueId: string; targetRunId: string; revision: string }, + data: { queueId: string; targetRunId: string | null; revision: string }, ) => api.post(`/issues/${id}/queued-comments/interrupt`, data), + interruptLatestQueuedComments: async (id: string, expectedTargetRunId: string | null): Promise => { + const queue = await issuesApi.getQueuedComments(id); + if (!queue.queueId || (queue.targetRunId && queue.targetRunId !== expectedTargetRunId)) { + throw new Error("The queued messages changed. Refresh and try again."); + } + return issuesApi.interruptQueuedComments(id, { + queueId: queue.queueId, revision: queue.revision, targetRunId: queue.targetRunId, + }); + }, steerQueuedComment: ( id: string, commentId: string, diff --git a/ui/src/components/CommentThread.tsx b/ui/src/components/CommentThread.tsx index 0557c37ee9..6f26b23ae9 100644 --- a/ui/src/components/CommentThread.tsx +++ b/ui/src/components/CommentThread.tsx @@ -103,7 +103,7 @@ interface CommentThreadProps { currentAssigneeValue?: string; suggestedAssigneeValue?: string; mentions?: MentionOption[]; - onInterruptQueued?: (runId: string) => Promise; + onInterruptQueued?: (runId: string | null) => Promise; interruptingQueuedRunId?: string | null; composerDisabledReason?: string | null; externalReferences?: MarkdownExternalReferenceMap; diff --git a/ui/src/components/IssueChatThread.test.tsx b/ui/src/components/IssueChatThread.test.tsx index d8f11d9798..23ec7af208 100644 --- a/ui/src/components/IssueChatThread.test.tsx +++ b/ui/src/components/IssueChatThread.test.tsx @@ -3150,6 +3150,24 @@ describe("IssueChatThread", () => { act(() => root.unmount()); }); + it("dispatches queued messages with Interrupt after the target run has stopped", () => { + const root = createRoot(container); + const onInterruptQueued = vi.fn(async () => {}); + act(() => root.render( {}} onInterruptQueued={onInterruptQueued} showComposer={false} + enableLiveTranscriptPolling={false} + />)); + const interrupt = [...container.querySelectorAll("button")].find(button => button.textContent === "Interrupt"); + expect(interrupt).toBeDefined(); + act(() => interrupt!.click()); + expect(onInterruptQueued).toHaveBeenCalledWith(null); + act(() => root.unmount()); + }); + it("shows deferred wake badge only for hold-deferred queued comments", () => { const root = createRoot(container); diff --git a/ui/src/components/IssueChatThread.tsx b/ui/src/components/IssueChatThread.tsx index 545bbba506..cb398cbbb3 100644 --- a/ui/src/components/IssueChatThread.tsx +++ b/ui/src/components/IssueChatThread.tsx @@ -271,7 +271,7 @@ interface IssueChatMessageContext { stoppingRunLabel?: string; stopRunVariant?: "stop" | "pause"; runFinalizationActions?: readonly IssueChatRunFinalizationAction[]; - onInterruptQueued?: (runId: string) => Promise; + onInterruptQueued?: (runId: string | null) => Promise; onCancelQueued?: (commentId: string) => void; onDeleteComment?: (commentId: string) => Promise | void; onImageClick?: (src: string) => void; @@ -649,7 +649,7 @@ interface IssueChatThreadProps { transcriptsByRunId?: ReadonlyMap; hasOutputForRun?: (runId: string) => boolean; includeSucceededRunsWithoutOutput?: boolean; - onInterruptQueued?: (runId: string) => Promise; + onInterruptQueued?: (runId: string | null) => Promise; onCancelQueued?: (commentId: string) => void; /** Authoritative PRP queue. The classic thread intentionally ignores it. */ queuedCommentQueue?: IssueQueuedCommentQueue | null; @@ -2116,7 +2116,7 @@ function IssueChatUserMessage({ > {queueBadgeLabel} - {queueTargetRunId && onInterruptQueued ? ( + {onInterruptQueued ? (