diff --git a/packages/paperclip-runner/protocol/manifest.json b/packages/paperclip-runner/protocol/manifest.json index 3ebb45a9cb..f3de19ef14 100644 --- a/packages/paperclip-runner/protocol/manifest.json +++ b/packages/paperclip-runner/protocol/manifest.json @@ -95,7 +95,7 @@ { "path": "schemas/usage.schema.json", "id": "https://paperclip.dev/schemas/prp/v1/usage.schema.json", - "sha256": "28fdbb3202095144649dddc438e6270c66d599c6f7e86838687f58d644726456" + "sha256": "34b1478d2054f898d64b54b62a9bb738670087893bb1e5967da541539d2daf36" }, { "path": "schemas/workspace-diff.schema.json", diff --git a/packages/paperclip-runner/protocol/schemas/usage.schema.json b/packages/paperclip-runner/protocol/schemas/usage.schema.json index 68c912f2dc..c1a57e17bf 100644 --- a/packages/paperclip-runner/protocol/schemas/usage.schema.json +++ b/packages/paperclip-runner/protocol/schemas/usage.schema.json @@ -10,6 +10,7 @@ "providerSessionId": { "type": ["string", "null"], "maxLength": 240 }, "providerRequestId": { "type": ["string", "null"], "maxLength": 240 }, "cumulative": { "$ref": "#/$defs/measurement" }, + "runDeltaAvailable": { "type": "boolean" }, "runDelta": { "$ref": "#/$defs/measurement" } }, "additionalProperties": false, diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs index f9e7b0d8fe..634f8ddda6 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs @@ -199,8 +199,7 @@ pub fn normalize_codex_notification(method: &str, params: &Value) -> Vec Vec { }); }); + it("retains canonical runner question sets without narrowing their public bounds", () => { + const questionSet = { + schema: "paperclip.question_set.v1" as const, + title: "Runner input", + questions: [{ + id: "deployment-color", + prompt: "Which deployment color should the runner use?", + required: true, + answerMode: "single_select" as const, + options: [{ id: "blue", label: "Blue" }], + }], + }; + const parsed = createIssueThreadInteractionSchema.parse({ + kind: "ask_user_questions", + continuationPolicy: "none", + resolverPolicy: "human_only", + payload: { + version: 1, + questions: [{ + id: "deployment-color", + prompt: "Which deployment color should the runner use?", + selectionMode: "single", + allowOther: false, + options: [{ id: "blue", label: "Blue" }], + }], + questionSet, + }, + }); + expect(parsed.kind).toBe("ask_user_questions"); + if (parsed.kind !== "ask_user_questions") return; + expect(parsed.payload.questionSet).toEqual(questionSet); + + expect(() => paperclipQuestionSetPayloadSchema.parse({ + ...questionSet, + questions: [{ ...questionSet.questions[0], answerMode: "text", options: questionSet.questions[0].options }], + })).toThrow("text questions cannot define options"); + }); + it("rejects unsafe request_confirmation target hrefs", () => { const base = { kind: "request_confirmation", diff --git a/packages/shared/src/types/issue.ts b/packages/shared/src/types/issue.ts index 5456b89fc6..d6a9eff602 100644 --- a/packages/shared/src/types/issue.ts +++ b/packages/shared/src/types/issue.ts @@ -1109,6 +1109,8 @@ export interface AskUserQuestionsQuestion { helpText?: string | null; selectionMode: "single" | "multi"; required?: boolean; + /** False suppresses the legacy free-form fallback for closed select sets. */ + allowOther?: boolean; options: AskUserQuestionsQuestionOption[]; } diff --git a/packages/shared/src/validators/index.ts b/packages/shared/src/validators/index.ts index e76bb8cd2d..cc72371b30 100644 --- a/packages/shared/src/validators/index.ts +++ b/packages/shared/src/validators/index.ts @@ -460,6 +460,7 @@ export { suggestTasksResultSchema, askUserQuestionsQuestionOptionSchema, askUserQuestionsQuestionSchema, + paperclipQuestionSetPayloadSchema, askUserQuestionsPayloadSchema, askUserQuestionsAnswerSchema, askUserQuestionsResultSchema, diff --git a/packages/shared/src/validators/issue.ts b/packages/shared/src/validators/issue.ts index 824d6307db..1a5604b4d2 100644 --- a/packages/shared/src/validators/issue.ts +++ b/packages/shared/src/validators/issue.ts @@ -870,9 +870,9 @@ export const suggestTasksResultSchema = z.object({ }); export const askUserQuestionsQuestionOptionSchema = z.object({ - id: z.string().trim().min(1).max(120), - label: z.string().trim().min(1).max(120), - description: z.string().trim().max(500).nullable().optional(), + id: z.string().trim().min(1).max(160), + label: z.string().trim().min(1).max(1000), + description: z.string().trim().max(4000).nullable().optional(), freeText: z .boolean() .optional() @@ -882,12 +882,13 @@ export const askUserQuestionsQuestionOptionSchema = z.object({ }); export const askUserQuestionsQuestionSchema = z.object({ - id: z.string().trim().min(1).max(120), - prompt: z.string().trim().min(1).max(500), - helpText: z.string().trim().max(1000).nullable().optional(), + id: z.string().trim().min(1).max(160), + prompt: z.string().trim().min(1).max(4000), + helpText: z.string().trim().max(4000).nullable().optional(), selectionMode: z.enum(["single", "multi"]), required: z.boolean().optional(), - options: z.array(askUserQuestionsQuestionOptionSchema).min(1).max(10), + allowOther: z.boolean().optional(), + options: z.array(askUserQuestionsQuestionOptionSchema).min(1).max(129), }); const paperclipQuestionOptionSchema = z.object({ @@ -918,14 +919,54 @@ const paperclipQuestionSchema = z.object({ minimum: z.number().finite().optional(), maximum: z.number().finite().optional(), }).optional(), +}).superRefine((value, ctx) => { + if (value.answerMode === "text" && value.options && value.options.length > 0) { + ctx.addIssue({ code: z.ZodIssueCode.custom, message: "text questions cannot define options", path: ["options"] }); + } + if (value.answerMode !== "text" && (!value.options || value.options.length === 0)) { + ctx.addIssue({ code: z.ZodIssueCode.custom, message: "select questions require options", path: ["options"] }); + } + if (value.answerMode === "text" && value.customAnswer) { + ctx.addIssue({ code: z.ZodIssueCode.custom, message: "text questions cannot define customAnswer", path: ["customAnswer"] }); + } + if ( + value.textValidation?.minLength !== undefined + && value.textValidation.maxLength !== undefined + && value.textValidation.minLength > value.textValidation.maxLength + ) { + ctx.addIssue({ code: z.ZodIssueCode.custom, message: "minLength cannot exceed maxLength", path: ["textValidation"] }); + } + if ( + value.textValidation?.minimum !== undefined + && value.textValidation.maximum !== undefined + && value.textValidation.minimum > value.textValidation.maximum + ) { + ctx.addIssue({ code: z.ZodIssueCode.custom, message: "minimum cannot exceed maximum", path: ["textValidation"] }); + } + if (value.textValidation?.pattern !== undefined) { + try { + new RegExp(value.textValidation.pattern); + } catch { + ctx.addIssue({ code: z.ZodIssueCode.custom, message: "pattern must be a valid regular expression", path: ["textValidation", "pattern"] }); + } + } + const optionIds = value.options?.map((option) => option.id) ?? []; + if (new Set(optionIds).size !== optionIds.length) { + ctx.addIssue({ code: z.ZodIssueCode.custom, message: "option ids must be unique", path: ["options"] }); + } }); -const paperclipQuestionSetSchema = z.object({ +export const paperclipQuestionSetPayloadSchema = z.object({ schema: z.literal("paperclip.question_set.v1"), title: z.string().max(1000).optional(), description: z.string().max(4000).optional(), submitLabel: z.string().max(200).optional(), questions: z.array(paperclipQuestionSchema).min(1).max(64), +}).superRefine((value, ctx) => { + const questionIds = value.questions.map((question) => question.id); + if (new Set(questionIds).size !== questionIds.length) { + ctx.addIssue({ code: z.ZodIssueCode.custom, message: "question ids must be unique", path: ["questions"] }); + } }); export const askUserQuestionsPayloadSchema = z.object({ @@ -933,9 +974,9 @@ export const askUserQuestionsPayloadSchema = z.object({ title: z.string().trim().max(240).nullable().optional(), submitLabel: z.string().trim().max(120).nullable().optional(), supersedeOnUserComment: z.boolean().optional(), - questions: z.array(askUserQuestionsQuestionSchema).min(1).max(10), + questions: z.array(askUserQuestionsQuestionSchema).min(1).max(64), /** Exact canonical presentation retained for a recovered harness request. */ - questionSet: paperclipQuestionSetSchema.optional(), + questionSet: paperclipQuestionSetPayloadSchema.optional(), /** Stable correlation for draft handoff from a live runtime request. */ runtimeRequestId: z.string().trim().min(1).max(255).nullable().optional(), }).superRefine((value, ctx) => { @@ -976,16 +1017,16 @@ export const askUserQuestionsPayloadSchema = z.object({ }); export const askUserQuestionsAnswerSchema = z.object({ - questionId: z.string().trim().min(1).max(120), - optionIds: z.array(z.string().trim().min(1).max(120)).max(20), - otherText: multilineTextSchema.pipe(z.string().trim().max(4000)).nullable().optional(), + questionId: z.string().trim().min(1).max(160), + optionIds: z.array(z.string().trim().min(1).max(160)).max(129), + otherText: multilineTextSchema.pipe(z.string().trim().max(100000)).nullable().optional(), }); export const askUserQuestionsResultSchema = z.object({ version: z.literal(1), outcome: z.enum(["withdrawn", "issue_closed", "addressee_deleted"]).optional(), reason: z.string().trim().max(4000).nullable().optional(), - answers: z.array(askUserQuestionsAnswerSchema).max(20), + answers: z.array(askUserQuestionsAnswerSchema).max(64), cancelled: z.literal(true).optional(), cancellationReason: z.string().trim().max(4000).nullable().optional(), expirationReason: z.enum(["superseded_by_comment", "superseded_by_newer_interaction"]).optional(), diff --git a/server/src/__tests__/issue-agent-mutation-ownership-routes.test.ts b/server/src/__tests__/issue-agent-mutation-ownership-routes.test.ts index 1db628e3f9..2a19a6fc65 100644 --- a/server/src/__tests__/issue-agent-mutation-ownership-routes.test.ts +++ b/server/src/__tests__/issue-agent-mutation-ownership-routes.test.ts @@ -1003,6 +1003,8 @@ describe("agent issue mutation checkout ownership", () => { issueId, expect.objectContaining({ status: "done" }), expect.anything(), + undefined, + expect.any(Array), ); }); diff --git a/server/src/__tests__/issue-comment-reopen-routes.test.ts b/server/src/__tests__/issue-comment-reopen-routes.test.ts index b63c7056a6..078a1ffda3 100644 --- a/server/src/__tests__/issue-comment-reopen-routes.test.ts +++ b/server/src/__tests__/issue-comment-reopen-routes.test.ts @@ -2450,6 +2450,7 @@ describe.sequential("issue comment reopen routes", () => { }), mockTx, expect.any(Array), + expect.any(Array), ); const updatePatch = mockIssueService.update.mock.calls[0]?.[1] as Record; const decisionId = updatePatch.executionState.lastDecisionId; @@ -2545,6 +2546,8 @@ describe.sequential("issue comment reopen routes", () => { }), }), mockTx, + undefined, + expect.any(Array), ); }); @@ -2630,6 +2633,8 @@ describe.sequential("issue comment reopen routes", () => { }), }), mockTx, + undefined, + expect.any(Array), ); }); @@ -3239,6 +3244,8 @@ describe.sequential("issue comment reopen routes", () => { "11111111-1111-4111-8111-111111111111", expect.objectContaining({ status: "done" }), mockTx, + undefined, + expect.any(Array), ); expect(mockLogActivity).toHaveBeenCalledWith( expect.anything(), diff --git a/server/src/__tests__/issue-execution-policy-routes.test.ts b/server/src/__tests__/issue-execution-policy-routes.test.ts index bd5d9a72c5..cfd98a9ac6 100644 --- a/server/src/__tests__/issue-execution-policy-routes.test.ts +++ b/server/src/__tests__/issue-execution-policy-routes.test.ts @@ -792,6 +792,8 @@ describe("issue execution policy routes", () => { actorUserId: "local-board", }), expect.anything(), + undefined, + expect.any(Array), ); expect(mockHeartbeatService.cancelRun).not.toHaveBeenCalled(); }); @@ -849,6 +851,8 @@ describe("issue execution policy routes", () => { actorUserId: "local-board", }), expect.anything(), + undefined, + expect.any(Array), ); const updatePatch = mockIssueService.update.mock.calls[0]?.[1] as Record; expect(updatePatch.status).toBe("cancelled"); diff --git a/server/src/__tests__/issue-thread-interaction-routes.test.ts b/server/src/__tests__/issue-thread-interaction-routes.test.ts index d8bce7b3e0..e13e5a4f78 100644 --- a/server/src/__tests__/issue-thread-interaction-routes.test.ts +++ b/server/src/__tests__/issue-thread-interaction-routes.test.ts @@ -46,6 +46,16 @@ const mockInteractionService = vi.hoisted(() => ({ const mockHeartbeatService = vi.hoisted(() => ({ wakeup: vi.fn(async () => undefined), + cancelRun: vi.fn(async () => null), +})); +const mockRequestNativeQuestionRunCancellation = vi.hoisted(() => + vi.fn(async () => null as string | null) +); + +vi.mock("../services/native-runtime/native-question-bridge.js", () => ({ + deliverNativeQuestionResponse: vi.fn(async () => "not_native"), + requestNativeQuestionRunCancellation: mockRequestNativeQuestionRunCancellation, + validateNativeQuestionResponseInput: vi.fn(), })); const mockQuestionResponseDeliveries = vi.hoisted(() => ({ deliver: vi.fn(async () => null), @@ -279,6 +289,20 @@ async function createApp(actor: Record = { return app; } +async function resolveMockInteraction( + args: unknown[], + interaction: Record, +) { + const mutationOptions = args[4] as { + afterResolveInTransaction?: ( + tx: Record, + resolved: Record, + ) => Promise; + } | undefined; + await mutationOptions?.afterResolveInTransaction?.({}, interaction); + return interaction; +} + describe.sequential("issue thread interaction routes", () => { beforeEach(() => { vi.resetModules(); @@ -290,6 +314,7 @@ describe.sequential("issue thread interaction routes", () => { vi.clearAllMocks(); mockInteractionService.getForIssue.mockReset(); mockQuestionResponseDeliveries.deliver.mockResolvedValue(null); + mockRequestNativeQuestionRunCancellation.mockResolvedValue(null); mockResolveTaskWatchdogMutationScope.mockReset(); mockResolveCoreTrustPreset.mockReset(); mockAccessDecide.mockReset(); @@ -317,7 +342,7 @@ describe.sequential("issue thread interaction routes", () => { status: "pending", payload: { version: 1, questions: [] }, }); - mockInteractionService.withdrawInteraction.mockResolvedValue({ + mockInteractionService.withdrawInteraction.mockImplementation((...args) => resolveMockInteraction(args, { id: "interaction-withdraw", companyId: "company-1", issueId: "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", @@ -327,7 +352,7 @@ describe.sequential("issue thread interaction routes", () => { continuationPolicy: "wake_assignee", payload: { version: 1, prompt: "Proceed?" }, result: { version: 1, outcome: "withdrawn", reason: "Replanning" }, - }); + })); mockInteractionService.recordSecretProposalExecutionResult.mockImplementation( async (_issue, _interactionId, _proposalId, execution) => ({ ...(await mockInteractionService.acceptInteraction.mock.results.at(-1)?.value)?.interaction, @@ -490,7 +515,7 @@ describe.sequential("issue thread interaction routes", () => { }, newlyResolvedItemIds: ["docs"], }); - mockInteractionService.cancelQuestions.mockResolvedValue({ + mockInteractionService.cancelQuestions.mockImplementation((...args) => resolveMockInteraction(args, { id: "interaction-2", companyId: "company-1", issueId: "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", @@ -519,7 +544,7 @@ describe.sequential("issue thread interaction routes", () => { createdAt: "2026-04-20T12:00:00.000Z", updatedAt: "2026-04-20T12:05:00.000Z", resolvedAt: "2026-04-20T12:05:00.000Z", - }); + })); mockDbSelect.mockImplementation(() => ({ from: mockDbSelectFrom })); mockDbSelectFrom.mockImplementation(() => ({ where: mockDbSelectWhere })); mockDbSelectWhere.mockImplementation(() => ({ @@ -853,6 +878,7 @@ describe.sequential("issue thread interaction routes", () => { "interaction-withdraw", { reason: "Replanning" }, expect.objectContaining({ userId: "local-board" }), + expect.objectContaining({ afterResolveInTransaction: expect.any(Function) }), ); expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith(ASSIGNEE_AGENT_ID, expect.objectContaining({ payload: expect.objectContaining({ interactionStatus: "cancelled" }), @@ -862,6 +888,71 @@ describe.sequential("issue thread interaction routes", () => { })); }); + it("cancels the bound native run when its question is withdrawn", async () => { + mockInteractionService.withdrawInteraction.mockImplementationOnce((...args) => resolveMockInteraction(args, { + id: "interaction-withdraw", + companyId: "company-1", + issueId: ISSUE_ID, + kind: "ask_user_questions", + createdByAgentId: CREATED_AGENT_ID, + sourceRunId: RUN_1, + status: "cancelled", + continuationPolicy: "none", + payload: { version: 1, questions: [] }, + result: { version: 1, answers: [], cancelled: true }, + })); + mockRequestNativeQuestionRunCancellation.mockResolvedValueOnce(RUN_1); + + const res = await request(await createApp()) + .post(`/api/issues/${ISSUE_ID}/interactions/interaction-withdraw/withdraw`) + .send({ reason: "No longer needed" }); + + expect(res.status).toBe(200); + expect(mockRequestNativeQuestionRunCancellation).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ id: "interaction-withdraw", sourceRunId: RUN_1 }), + { kind: "interaction_withdrawn", interactionId: "interaction-withdraw" }, + ); + expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith( + RUN_1, + "Question withdrawn while waiting for operator input", + expect.objectContaining({ + resultJson: expect.objectContaining({ + withdrawnInteractionId: "interaction-withdraw", + withdrawnByActorType: "user", + }), + }), + ); + }); + + it("keeps a durable withdrawal intent when immediate native cancellation fails", async () => { + mockInteractionService.withdrawInteraction.mockImplementationOnce((...args) => resolveMockInteraction(args, { + id: "interaction-withdraw", + companyId: "company-1", + issueId: ISSUE_ID, + kind: "ask_user_questions", + createdByAgentId: CREATED_AGENT_ID, + sourceRunId: RUN_1, + status: "cancelled", + continuationPolicy: "none", + payload: { version: 1, questions: [] }, + result: { version: 1, answers: [], cancelled: true }, + })); + mockRequestNativeQuestionRunCancellation.mockResolvedValueOnce(RUN_1); + mockHeartbeatService.cancelRun.mockRejectedValueOnce(new Error("process unavailable")); + + const res = await request(await createApp()) + .post(`/api/issues/${ISSUE_ID}/interactions/interaction-withdraw/withdraw`) + .send({ reason: "No longer needed" }); + + expect(res.status).toBe(200); + expect(mockRequestNativeQuestionRunCancellation).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ id: "interaction-withdraw" }), + { kind: "interaction_withdrawn", interactionId: "interaction-withdraw" }, + ); + }); + it("allows the creator agent to withdraw and wakes a different assignee", async () => { mockIssueService.getById.mockResolvedValueOnce(createIssue({ status: "in_review", reviewPolicy: null })); mockInteractionService.getForIssue.mockResolvedValueOnce({ @@ -919,6 +1010,7 @@ describe.sequential("issue thread interaction routes", () => { "interaction-withdraw", {}, expect.objectContaining({ agentId: ASSIGNEE_AGENT_ID, runId: RUN_WATCHDOG }), + expect.objectContaining({ afterResolveInTransaction: expect.any(Function) }), ); expect(res.status).toBe(200); }); @@ -948,6 +1040,7 @@ describe.sequential("issue thread interaction routes", () => { "interaction-2", {}, expect.objectContaining({ userId: "local-board" }), + expect.objectContaining({ afterResolveInTransaction: expect.any(Function) }), ); expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith( ASSIGNEE_AGENT_ID, @@ -970,6 +1063,28 @@ describe.sequential("issue thread interaction routes", () => { ); }); + it("durably marks a board-cancelled native question before cancelling its run", async () => { + mockRequestNativeQuestionRunCancellation.mockResolvedValueOnce(RUN_2); + + const res = await request(await createApp()) + .post(`/api/issues/${ISSUE_ID}/interactions/interaction-2/cancel`) + .send({}); + + expect(res.status).toBe(200); + expect(mockRequestNativeQuestionRunCancellation).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ id: "interaction-2", sourceRunId: RUN_2 }), + { kind: "interaction_cancelled", interactionId: "interaction-2" }, + ); + expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith( + RUN_2, + "Cancelled while waiting for operator input", + expect.objectContaining({ + resultJson: expect.objectContaining({ cancelledInteractionId: "interaction-2" }), + }), + ); + }); + it("accepts request confirmations and wakes the current assignee when configured for accept-only wakeups", async () => { mockInteractionService.acceptInteraction.mockResolvedValueOnce({ interaction: { diff --git a/server/src/__tests__/issue-tree-control-service.test.ts b/server/src/__tests__/issue-tree-control-service.test.ts index afb23add9e..43c620053a 100644 --- a/server/src/__tests__/issue-tree-control-service.test.ts +++ b/server/src/__tests__/issue-tree-control-service.test.ts @@ -2,12 +2,14 @@ import { randomUUID } from "node:crypto"; import { eq, inArray } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import { + activityLog, agents, agentWakeupRequests, companies, createDb, heartbeatRuns, issueComments, + issueThreadInteractions, issueTreeHoldMembers, issueTreeHolds, issues, @@ -38,6 +40,8 @@ describeEmbeddedPostgres("issueTreeControlService", () => { }, 20_000); afterEach(async () => { + await db.delete(issueThreadInteractions); + await db.delete(activityLog); await db.delete(issueTreeHoldMembers); await db.delete(issueTreeHolds); await db.delete(issueComments); @@ -137,7 +141,6 @@ describeEmbeddedPostgres("issueTreeControlService", () => { createdAt: new Date("2026-04-21T10:03:00.000Z"), }, ]); - const svc = issueTreeControlService(db); const preview = await svc.preview(companyId, rootIssueId, { mode: "pause" }); @@ -314,6 +317,14 @@ describeEmbeddedPostgres("issueTreeControlService", () => { createdAt: new Date("2026-04-21T10:03:00.000Z"), }, ]); + const [pendingInteraction] = await db.insert(issueThreadInteractions).values({ + companyId, + issueId: runningChildId, + kind: "request_confirmation", + status: "pending", + continuationPolicy: "none", + payload: { version: 1, prompt: "Continue?" }, + }).returning(); const svc = issueTreeControlService(db); const cancel = await svc.createHold(companyId, rootIssueId, { @@ -340,6 +351,11 @@ describeEmbeddedPostgres("issueTreeControlService", () => { [todoChildId]: "cancelled", [doneChildId]: "done", }); + const [expiredInteraction] = await db + .select({ status: issueThreadInteractions.status }) + .from(issueThreadInteractions) + .where(eq(issueThreadInteractions.id, pendingInteraction!.id)); + expect(expiredInteraction?.status).toBe("expired"); await db .update(issues) diff --git a/server/src/__tests__/issue-update-comment-wakeup-routes.test.ts b/server/src/__tests__/issue-update-comment-wakeup-routes.test.ts index b6336d02e9..5e716f2aa0 100644 --- a/server/src/__tests__/issue-update-comment-wakeup-routes.test.ts +++ b/server/src/__tests__/issue-update-comment-wakeup-routes.test.ts @@ -32,6 +32,12 @@ const mockIssueThreadInteractionService = vi.hoisted(() => ({ expireStaleRequestConfirmationsForIssueDocument: vi.fn(async () => []), })); +vi.mock("../services/native-runtime/native-question-bridge.js", () => ({ + deliverNativeQuestionResponse: vi.fn(async () => "not_native"), + nativeQuestionRunToCancel: vi.fn(async () => null), + validateNativeQuestionResponseInput: vi.fn(), +})); + vi.mock("../services/index.js", () => ({ companyService: () => ({ getById: vi.fn(async () => ({ id: "company-1" })), @@ -484,41 +490,6 @@ describe("issue update comment wakeups", () => { ); }); - it("does not wake the assignee when a closure comment marks the issue done", async () => { - const existing = makeIssue({ - assigneeAgentId: ASSIGNEE_AGENT_ID, - assigneeUserId: null, - status: "in_progress", - }); - const updated = { - ...existing, - status: "done", - completedAt: new Date("2026-06-26T16:30:00.000Z"), - }; - mockIssueService.getById.mockResolvedValue(existing); - mockIssueService.update.mockResolvedValue(updated); - mockIssueService.addComment.mockResolvedValue({ - id: "comment-close-1", - issueId: existing.id, - companyId: existing.companyId, - body: "Closing this out.", - }); - - const res = await request(await createApp()) - .patch(`/api/issues/${existing.id}`) - .send({ - status: "done", - comment: "Closing this out.", - }); - - expect(res.status).toBe(200); - await new Promise((resolve) => setImmediate(resolve)); - const issueCommentedWakeCalls = mockHeartbeatService.wakeup.mock.calls.filter( - ([, wakeup]: [string, { reason?: string }]) => wakeup?.reason === "issue_commented", - ); - expect(issueCommentedWakeCalls).toEqual([]); - }); - it("wakes the assignee on top-level board issue comments", async () => { const existing = makeIssue({ assigneeAgentId: ASSIGNEE_AGENT_ID, diff --git a/server/src/__tests__/pipelines-service.test.ts b/server/src/__tests__/pipelines-service.test.ts index bd63341402..3ffda9d205 100644 --- a/server/src/__tests__/pipelines-service.test.ts +++ b/server/src/__tests__/pipelines-service.test.ts @@ -10,6 +10,7 @@ import { heartbeatRuns, instanceSettings, issueComments, + issueThreadInteractions, issues, pipelineAutomationExecutions, pipelineCaseBlockers, @@ -68,6 +69,7 @@ describeEmbeddedPostgres("pipelineService", () => { await db.delete(pipelineTransitions); await db.delete(pipelineStages); await db.delete(pipelines); + await db.delete(issueThreadInteractions); await db.delete(issueComments); await db.delete(activityLog); await db.delete(routineRuns); @@ -1707,6 +1709,27 @@ describeEmbeddedPostgres("pipelineService", () => { status: "failed", error: "boom", }).returning(); + const [automationIssue] = await db.insert(issues).values({ + companyId: company.id, + title: "Retry-owned automation issue", + status: "in_progress", + priority: "medium", + }).returning(); + await db.insert(pipelineCaseIssueLinks).values({ + companyId: company.id, + caseId: parent.case.id, + issueId: automationIssue!.id, + role: "automation", + automationAttemptId: attempt!.id, + }); + const [pendingInteraction] = await db.insert(issueThreadInteractions).values({ + companyId: company.id, + issueId: automationIssue!.id, + kind: "request_confirmation", + status: "pending", + continuationPolicy: "none", + payload: { version: 1, prompt: "Continue automation?" }, + }).returning(); const child = await svc.ingestCase({ companyId: company.id, pipelineId: pipeline.id, @@ -1777,6 +1800,16 @@ describeEmbeddedPostgres("pipelineService", () => { expect(freshParent!.stageKey).toBe("review"); expect(freshChild!.terminalKind).toBe("cancelled"); expect(freshChild!.retiredReason).toBe("automation_retry"); + const [cancelledAutomationIssue] = await db + .select({ status: issues.status }) + .from(issues) + .where(eq(issues.id, automationIssue!.id)); + expect(cancelledAutomationIssue?.status).toBe("cancelled"); + const [expiredInteraction] = await db + .select({ status: issueThreadInteractions.status }) + .from(issueThreadInteractions) + .where(eq(issueThreadInteractions.id, pendingInteraction!.id)); + expect(expiredInteraction?.status).toBe("expired"); const events = await svc.listCaseEvents(company.id, parent.case.id); expect(events.filter((pipelineEvent) => pipelineEvent.type === "children_terminal")).toHaveLength(2); }); diff --git a/server/src/__tests__/question-response-delivery.test.ts b/server/src/__tests__/question-response-delivery.test.ts index 9619c0bfcd..dbeddfa395 100644 --- a/server/src/__tests__/question-response-delivery.test.ts +++ b/server/src/__tests__/question-response-delivery.test.ts @@ -241,6 +241,56 @@ describeEmbeddedPostgres("question response delivery", () => { expect(JSON.stringify(deliveryEvents[0]?.details)).not.toContain("Node.js"); }); + it("resolves an in-flight native input request before creating a continuation", async () => { + const seeded = await seed({ + adapterType: "paperclip_runner", + runtimeMode: "native", + sourceStatus: "running", + }); + const wakeup = vi.fn(); + const resolveNativeQuestion = vi.fn().mockResolvedValue("queued" as const); + + const outcome = await questionResponseDeliveryService(db, { + heartbeat: { wakeup } as never, + resolveNativeQuestion, + }).deliver(seeded.interaction.id); + + expect(outcome).toMatchObject({ + status: "delivered", + mode: "steered", + targetRunId: seeded.sourceRunId, + }); + expect(resolveNativeQuestion).toHaveBeenCalledWith(expect.objectContaining({ + id: seeded.interaction.id, + status: "answered", + })); + expect(wakeup).not.toHaveBeenCalled(); + }); + + it("keeps native input delivery pending while its PRP session is unavailable", async () => { + const seeded = await seed({ + adapterType: "paperclip_runner", + runtimeMode: "native", + sourceStatus: "running", + }); + const wakeup = vi.fn(); + + const outcome = await questionResponseDeliveryService(db, { + heartbeat: { wakeup } as never, + resolveNativeQuestion: vi.fn().mockResolvedValue("pending" as const), + }).deliver(seeded.interaction.id); + + expect(outcome).toBeNull(); + expect(wakeup).not.toHaveBeenCalled(); + const [delivery] = await db.select().from(issueQuestionResponseDeliveries); + expect(delivery).toMatchObject({ + status: "pending", + attemptCount: 1, + errorCount: 0, + lastErrorCode: "native_question_session_unavailable", + }); + }); + it("coalesces into a queued successor without creating another wake", async () => { const seeded = await seed({ successorStatus: "queued" }); const successor = await db.select().from(heartbeatRuns) diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 791df7154b..015faec1c3 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -326,6 +326,13 @@ vi.mock("../services/question-response-delivery.js", () => ({ })), })); +vi.mock("../services/native-runtime/native-question-bridge.js", () => ({ + deliverNativeQuestionResponse: vi.fn(async () => "not_native"), + nativeQuestionCancellationIdentity: vi.fn(() => null), + nativeQuestionRunToCancel: vi.fn(async () => null), + validateNativeQuestionResponseInput: vi.fn(), +})); + vi.mock("../services/secret-proposals.js", () => ({ createSecretProposalsService: vi.fn(() => ({ sweepExpired: vi.fn(async () => 0), diff --git a/server/src/index.ts b/server/src/index.ts index 09a66a0d91..6d85140dbd 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -69,6 +69,7 @@ import { workspaceOperationService, } from "./services/index.js"; import { questionResponseDeliveryService } from "./services/question-response-delivery.js"; +import { deliverNativeQuestionResponse } from "./services/native-runtime/native-question-bridge.js"; import { queueIssueAssignmentWakeup } from "./services/issue-assignment-wakeup.js"; import { createSecretProposalsService } from "./services/secret-proposals.js"; import { environmentRuntimeService } from "./services/environment-runtime.js"; @@ -1084,6 +1085,7 @@ export async function startServer(): Promise { heartbeat ?? heartbeatService(db as any, { pluginWorkerManager }); const questionResponseDeliveries = questionResponseDeliveryService(db as any, { heartbeat: environmentLeaseCleanupHeartbeat, + resolveNativeQuestion: (interaction) => deliverNativeQuestionResponse(db as any, interaction), }); const runEnvironmentLeaseCleanupSweep = (backoffMs: number) => environmentLeaseCleanupHeartbeat diff --git a/server/src/routes/agents.ts b/server/src/routes/agents.ts index 89960672bc..028e868e54 100644 --- a/server/src/routes/agents.ts +++ b/server/src/routes/agents.ts @@ -5298,6 +5298,7 @@ export function agentRoutes( const columns = { id: heartbeatRuns.id, + runtimeMode: heartbeatRuns.runtimeMode, companyId: heartbeatRuns.companyId, status: heartbeatRuns.status, invocationSource: heartbeatRuns.invocationSource, @@ -5523,6 +5524,7 @@ export function agentRoutes( const liveRuns = await db .select({ id: heartbeatRuns.id, + runtimeMode: heartbeatRuns.runtimeMode, status: heartbeatRuns.status, invocationSource: heartbeatRuns.invocationSource, triggerDetail: heartbeatRuns.triggerDetail, diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index ab07d8c789..1bff0fbc14 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -202,6 +202,7 @@ import { ISSUE_WAKE_DIAGNOSTICS_MAX_ACTIVITY_RECORDS, ISSUE_WAKE_DIAGNOSTICS_MAX_WAKE_REQUESTS, readAcceptedPlanConfirmationTarget, + type IssuePostCommitAction, } from "../services/issues.js"; import { authorizationDeniedDetails } from "../services/authorization.js"; import { stalledReviewDecisionService } from "../services/stalled-review-decisions.js"; @@ -209,6 +210,11 @@ import { environmentService } from "../services/environments.js"; import { environmentRuntimeService } from "../services/environment-runtime.js"; import { redactSensitiveText } from "../redaction.js"; import { createRunSecretRedactionRegistry } from "../services/run-secret-redaction.js"; +import { + deliverNativeQuestionResponse, + requestNativeQuestionRunCancellation, + validateNativeQuestionResponseInput, +} from "../services/native-runtime/native-question-bridge.js"; import { createCompanySearchRateLimiter, type CompanySearchRateLimiter, @@ -2855,7 +2861,14 @@ export function issueRoutes( const issueThreadInteractionsSvc = issueThreadInteractionService(db); const questionResponseDeliveries = questionResponseDeliveryService(db, { heartbeat, + resolveNativeQuestion: (interaction) => deliverNativeQuestionResponse(db, interaction), }); + const flushIssuePostCommitActions = async (actions: readonly IssuePostCommitAction[]) => { + if (actions.length === 0) return; + const { executeIssuePostCommitActions } = await import("../services/issues.js"); + await executeIssuePostCommitActions(db, actions); + }; + const memoizeIssueRead = createRequestPromiseMemo>>({ shouldCache: (issue) => issue !== null, }); @@ -6876,6 +6889,7 @@ export function issueRoutes( const actor = getActorInfo(req); const actionStatus = outcome === "cancelled" ? "cancelled" : "resolved"; const postCommitActivityPublications: ActivityPublication[] = []; + const postCommitIssueActions: IssuePostCommitAction[] = []; const result = await db.transaction(async (tx) => { const lockedIssue = await tx .select() @@ -6996,16 +7010,20 @@ export function issueRoutes( } } - const updatedIssue = await svc.update( - id, - { - ...updateFields, - actorAgentId: actor.agentId ?? null, - actorUserId: actor.actorType === "user" ? actor.actorId : null, - }, - tx, - postCommitActivityPublications, - ); + const issueUpdate = { + ...updateFields, + actorAgentId: actor.agentId ?? null, + actorUserId: actor.actorType === "user" ? actor.actorId : null, + }; + const updatedIssue = sourceIssueStatus === "done" || sourceIssueStatus === "cancelled" + ? await svc.update( + id, + issueUpdate, + tx, + postCommitActivityPublications, + postCommitIssueActions, + ) + : await svc.update(id, issueUpdate, tx, postCommitActivityPublications); if (!updatedIssue) throw notFound("Issue not found"); issue = updatedIssue; } @@ -7035,6 +7053,7 @@ export function issueRoutes( return { issue, recoveryAction }; }); for (const publication of postCommitActivityPublications) publishActivity(publication); + await flushIssuePostCommitActions(postCommitIssueActions); await routinesSvc.syncRunStatusForIssue(result.issue.id); @@ -9852,6 +9871,7 @@ export function issueRoutes( value: Awaited>; } = { value: null }; const postCommitActivityPublications: ActivityPublication[] = []; + const postCommitIssueActions: IssuePostCommitAction[] = []; const issueUpdateData = { ...updateFields, actorAgentId: actor.agentId ?? null, @@ -9859,10 +9879,15 @@ export function issueRoutes( }; const shouldCollectCompletionPublication = actor.actorType === "user" && existing.status !== "done" && updateFields.status === "done"; + const shouldCollectTerminalIssueActions = + updateFields.status === "done" || updateFields.status === "cancelled"; const updateIssue = (tx?: Parameters[2]) => { if (tx) { - return shouldCollectCompletionPublication - ? svc.update(id, issueUpdateData, tx, postCommitActivityPublications) + if (shouldCollectCompletionPublication) { + return svc.update(id, issueUpdateData, tx, postCommitActivityPublications, postCommitIssueActions); + } + return shouldCollectTerminalIssueActions + ? svc.update(id, issueUpdateData, tx, undefined, postCommitIssueActions) : svc.update(id, issueUpdateData, tx); } return shouldCollectCompletionPublication @@ -10033,6 +10058,7 @@ export function issueRoutes( return; } for (const publication of postCommitActivityPublications) publishActivity(publication); + await flushIssuePostCommitActions(postCommitIssueActions); if (enteringBlocked) { const blockedIssue = issue; @@ -11682,9 +11708,12 @@ export function issueRoutes( interactionId, ); if (!authorizedResolution) return; - const { interactionSvc, resolutionAuthorization } = authorizedResolution; + const { interactionSvc, current, resolutionAuthorization } = authorizedResolution; const actor = getActorInfo(req); + if (current.kind === "ask_user_questions") { + validateNativeQuestionResponseInput(current, req.body); + } const interaction = await interactionSvc.answerQuestions(issue, interactionId, req.body, { agentId: actor.agentId, runId: actor.runId, @@ -11829,11 +11858,42 @@ export function issueRoutes( await assertPendingReviewInteractionVerdictAllowed(req, issue, current); const actor = getActorInfo(req); - const interaction = await interactionSvc.withdrawInteraction(issue, interactionId, req.body, { - agentId: actor.agentId, - runId: actor.runId, - userId: actor.actorType === "user" ? actor.actorId : null, - }); + let nativeRunId: string | null = null; + const interaction = await interactionSvc.withdrawInteraction( + issue, + interactionId, + req.body, + { + agentId: actor.agentId, + runId: actor.runId, + userId: actor.actorType === "user" ? actor.actorId : null, + }, + { + afterResolveInTransaction: async (tx, resolved) => { + if (resolved.kind !== "ask_user_questions") return; + nativeRunId = await requestNativeQuestionRunCancellation(tx, resolved, { + kind: "interaction_withdrawn", + interactionId: resolved.id, + }); + }, + }, + ); + if (nativeRunId) { + try { + await heartbeat.cancelRun(nativeRunId, "Question withdrawn while waiting for operator input", { + resultJson: { + withdrawnInteractionId: interaction.id, + withdrawnByActorType: actor.actorType, + withdrawnByActorId: actor.actorId, + }, + }); + } catch (err) { + logger.warn( + { err, runId: nativeRunId, interactionId: interaction.id }, + "native question withdrawal cancellation deferred to recovery sweep", + ); + } + } await logActivity(db, { companyId: issue.companyId, actorType: actor.actorType, @@ -11881,10 +11941,25 @@ export function issueRoutes( assertBoard(req); const actor = getActorInfo(req); - const interaction = await issueThreadInteractionService(db).cancelQuestions(issue, interactionId, req.body, { - agentId: actor.agentId, - userId: actor.actorType === "user" ? actor.actorId : null, - }); + let nativeRunId: string | null = null; + const interaction = await issueThreadInteractionService(db).cancelQuestions( + issue, + interactionId, + req.body, + { + agentId: actor.agentId, + userId: actor.actorType === "user" ? actor.actorId : null, + }, + { + afterResolveInTransaction: async (tx, resolved) => { + if (resolved.kind !== "ask_user_questions") return; + nativeRunId = await requestNativeQuestionRunCancellation(tx, resolved, { + kind: "interaction_cancelled", + interactionId: resolved.id, + }); + }, + }, + ); await logActivity(db, { companyId: issue.companyId, @@ -11907,6 +11982,23 @@ export function issueRoutes( }, }); + if (nativeRunId) { + try { + await heartbeat.cancelRun(nativeRunId, "Cancelled while waiting for operator input", { + resultJson: { + cancelledByActorType: "user", + cancelledByUserId: req.actor.userId ?? null, + cancelledInteractionId: interaction.id, + }, + }); + } catch (err) { + logger.warn( + { err, runId: nativeRunId, interactionId: interaction.id }, + "native question board cancellation deferred to recovery sweep", + ); + } + } + await queueResolvedInteractionContinuationWakeup({ db, heartbeat, @@ -12444,6 +12536,7 @@ export function issueRoutes( }; let txResult: { comment: Awaited>; issue: NonNullable>> }; const postCommitActivityPublications: ActivityPublication[] = []; + const postCommitIssueActions: IssuePostCommitAction[] = []; try { txResult = await db.transaction(async (tx) => { const insertedComment = await svc.addComment( @@ -12459,8 +12552,8 @@ export function issueRoutes( tx, ); const updated = actor.actorType === "user" && currentIssue.status !== "done" - ? await svc.update(id, updatePatch, tx, postCommitActivityPublications) - : await svc.update(id, updatePatch, tx); + ? await svc.update(id, updatePatch, tx, postCommitActivityPublications, postCommitIssueActions) + : await svc.update(id, updatePatch, tx, undefined, postCommitIssueActions); // Throw (not return null) so drizzle rolls back the inserted comment when the issue // has been concurrently deleted between the initial fetch and the in-transaction update. if (!updated) throw new AutoApprovalIssueMissingError(); @@ -12490,6 +12583,7 @@ export function issueRoutes( throw err; } for (const publication of postCommitActivityPublications) publishActivity(publication); + await flushIssuePostCommitActions(postCommitIssueActions); comment = txResult.comment; currentIssue = txResult.issue; // Mirror the normal status-change audit trail: every other in_review -> done path diff --git a/server/src/services/activity.ts b/server/src/services/activity.ts index a9c6f592dd..5cc2cc625a 100644 --- a/server/src/services/activity.ts +++ b/server/src/services/activity.ts @@ -381,6 +381,7 @@ export function activityService(db: Db) { const runs = await db .select({ runId: heartbeatRuns.id, + runtimeMode: heartbeatRuns.runtimeMode, status: heartbeatRuns.status, agentId: heartbeatRuns.agentId, adapterType: agents.adapterType, diff --git a/server/src/services/decisions.ts b/server/src/services/decisions.ts index 704a713ae4..c9a5706354 100644 --- a/server/src/services/decisions.ts +++ b/server/src/services/decisions.ts @@ -8,7 +8,10 @@ import { conflict, forbidden, notFound, tooManyRequests, unprocessable } from ". import { authorizationService, type AuthorizationActor } from "./authorization.js"; import { logActivity, publishActivity, type ActivityPublication } from "./activity-log.js"; import { signDecisionSpec, verifyDecisionSpec } from "./decision-signing.js"; -import { issueService } from "./issues.js"; +import { + issueService, + type IssuePostCommitAction, +} from "./issues.js"; import { decisionRetentionService, hashAttentionArchiveManifest } from "./decision-retention.js"; type Snapshot = { status: string; assigneeAgentId: string | null; assigneeUserId: string | null; updatedAt: string; @@ -406,6 +409,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { try { const postCommitActivityPublications: ActivityPublication[] = []; + const postCommitIssueActions: IssuePostCommitAction[] = []; const executionResult = await db.transaction(async (tx) => { await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${lockKey}, 0))`); let execution = await tx.select().from(decisionEffectExecutions).where(and(eq(decisionEffectExecutions.decisionId, decision.id), eq(decisionEffectExecutions.effectIndex, effectIndex))) @@ -472,6 +476,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { { status: effect.status, actorUserId: decidedByUserId }, tx, postCommitActivityPublications, + postCommitIssueActions, ); if (effect.comment) await svc.addComment(target.id, interpolate(effect.comment, values), { userId: decidedByUserId }, undefined, tx); result = { issueId: updated?.id, status: updated?.status }; @@ -485,6 +490,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { }, tx, postCommitActivityPublications, + postCommitIssueActions, ); if (effect.comment) await svc.addComment(target.id, interpolate(effect.comment, values), { userId: decidedByUserId }, undefined, tx); result = { issueId: updated?.id }; @@ -498,6 +504,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { }, tx, postCommitActivityPublications, + postCommitIssueActions, ); result = { removedBlockedByIssueIds: effect.removeBlockedByIssueIds }; } else if (effect.type === "create_issue") { @@ -515,6 +522,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { { status: "cancelled", actorUserId: decidedByUserId }, tx, postCommitActivityPublications, + postCommitIssueActions, ); } await svc.addComment(target.id, interpolate(effect.reasonComment, values), { userId: decidedByUserId }, undefined, tx); @@ -525,6 +533,10 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { return row; }); for (const publication of postCommitActivityPublications) publishActivity(publication); + if (postCommitIssueActions.length > 0) { + const { executeIssuePostCommitActions } = await import("./issues.js"); + await executeIssuePostCommitActions(db, postCommitIssueActions); + } return executionResult; } catch (error) { const message = error instanceof Error ? error.message : "Decision effect execution failed"; diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 5a8ace8b63..e2da98277c 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -461,6 +461,7 @@ const MAX_AGENT_SESSION_MESSAGE_CHARS = 12_000; const execFile = promisify(execFileCallback); const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const; const CANCELLABLE_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const; +const NATIVE_QUESTION_CANCELLATION_CONTEXT_KEY = "nativeQuestionCancellation"; const HEARTBEAT_RUN_TERMINAL_STATUSES = ["succeeded", "interrupted", "failed", "cancelled", "timed_out"] as const; const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = ["failed", "cancelled", "timed_out"] as const; const TIMER_ACTIONABLE_ISSUE_STATUSES = ["todo", "in_progress"] as const; @@ -2569,6 +2570,7 @@ const heartbeatRunLogAccessColumns = { const heartbeatRunIssueSummaryColumns = { id: heartbeatRuns.id, + runtimeMode: heartbeatRuns.runtimeMode, status: heartbeatRuns.status, invocationSource: heartbeatRuns.invocationSource, triggerDetail: heartbeatRuns.triggerDetail, @@ -13940,6 +13942,56 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); + // A terminal issue transition writes this intent in the same transaction + // that expires the native question. Consume it before generic orphan + // recovery so a restart preserves the requested cancellation outcome. + const cancellationRequests = await db + .select({ + id: heartbeatRuns.id, + contextSnapshot: heartbeatRuns.contextSnapshot, + }) + .from(heartbeatRuns) + .where(and( + inArray(heartbeatRuns.status, [...CANCELLABLE_HEARTBEAT_RUN_STATUSES]), + sql`${heartbeatRuns.contextSnapshot} -> ${NATIVE_QUESTION_CANCELLATION_CONTEXT_KEY} is not null`, + )); + for (const request of cancellationRequests) { + const marker = parseObject( + parseObject(request.contextSnapshot)[NATIVE_QUESTION_CANCELLATION_CONTEXT_KEY], + ); + const issueId = readNonEmptyString(marker.issueId); + const issueStatus = readNonEmptyString(marker.issueStatus); + const interactionId = readNonEmptyString(marker.interactionId); + const kind = readNonEmptyString(marker.kind); + const reason = kind === "interaction_withdrawn" + ? "Question withdrawn while waiting for operator input" + : kind === "interaction_cancelled" + ? "Cancelled while waiting for operator input" + : "Task closed while waiting for operator input"; + try { + await cancelRunInternal(request.id, reason, { + resultJson: { + ...(kind === "interaction_withdrawn" && interactionId + ? { withdrawnInteractionId: interactionId } + : {}), + ...(kind === "interaction_cancelled" && interactionId + ? { cancelledInteractionId: interactionId } + : {}), + ...((!kind || kind === "issue_terminal") && issueStatus + ? { cancelledByIssueStatus: issueStatus } + : {}), + ...(issueId ? { cancelledIssueId: issueId } : {}), + }, + }); + } catch (err) { + // Keep the marker intact for the next startup/periodic sweep. + logger.warn( + { err, runId: request.id }, + "native question cancellation recovery attempt failed", + ); + } + } + // Find all runs stuck in "running" state (queued runs are legitimately waiting; resumeQueuedRuns handles them) const activeRuns = await db .select({ diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index 118e075e85..ccca8310de 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -107,6 +107,11 @@ type InteractionActor = { resolutionDetails?: Record; }; +type CreateInteractionOptions = { + /** Keep independently owned pending cards actionable. Internal runtime bridges use this. */ + supersedePendingSiblingInteractions?: boolean; +}; + type InteractionWakeup = (agentId: string, options: { source: "automation"; triggerDetail: "system"; @@ -128,6 +133,14 @@ export type IssueThreadInteractionServiceOptions = { now?: () => Date; }; +type DbTransaction = Parameters[0]>[0]; +type InteractionResolutionMutationOptions = { + afterResolveInTransaction?: ( + tx: DbTransaction, + interaction: IssueThreadInteraction, + ) => Promise; +}; + const GITHUB_PULL_REQUEST_URL_PATTERN = /https:\/\/(?:www\.)?github\.com\/([A-Za-z0-9_.-]+)\/([A-Za-z0-9_.-]+)\/pull\/([1-9][0-9]*)/gi; const GITHUB_PULL_REQUEST_SHORTHAND_PATTERN = /(^|[^A-Za-z0-9_.-])([A-Za-z0-9_.-]+)\/([A-Za-z0-9_.-]+)#([1-9][0-9]*)\b/g; const MERGE_CONFIRMATION_INTENT_PATTERN = /^(?:please\s+)?(?:confirm(?:\s+that)?\s+.{0,80}\s+)?(?:merge|merged)\b|\bready\s+to\s+merge\b/i; @@ -2469,6 +2482,7 @@ export function issueThreadInteractionService(db: Db, opts: IssueThreadInteracti issue: { id: string; companyId: string }, input: CreateIssueThreadInteraction, actor: InteractionActor, + options: CreateInteractionOptions = {}, ) => { const data = normalizeCreateInteractionInput(createIssueThreadInteractionSchema.parse(input)); const usedDeprecatedResolverPolicyAlias = @@ -2634,10 +2648,13 @@ export function issueThreadInteractionService(db: Db, opts: IssueThreadInteracti // result shape. Scoped strictly to the same agent + issue + kind, so // other agents' or other kinds' pending cards are untouched. const canSupersedeSiblingCards = - (data.kind === "request_confirmation" - && data.payload.toolAction === undefined - && data.payload.secretProposal === undefined) - || data.kind === "ask_user_questions"; + options.supersedePendingSiblingInteractions !== false + && ( + (data.kind === "request_confirmation" + && data.payload.toolAction === undefined + && data.payload.secretProposal === undefined) + || data.kind === "ask_user_questions" + ); if (!actor.agentId || !canSupersedeSiblingCards) { return { row, supersededRows: [] }; } @@ -3481,6 +3498,7 @@ export function issueThreadInteractionService(db: Db, opts: IssueThreadInteracti interactionId: string, input: WithdrawIssueThreadInteraction, actor: InteractionActor, + mutationOptions: InteractionResolutionMutationOptions = {}, ) => { assertIssueOpenForInteractionResolution(issue); const data = withdrawIssueThreadInteractionSchema.parse(input); @@ -3546,6 +3564,7 @@ export function issueThreadInteractionService(db: Db, opts: IssueThreadInteracti )) .returning(); if (!row) throw interactionAlreadyResolvedError(); + await mutationOptions.afterResolveInTransaction?.(tx, hydrateInteraction(row)); return row; }); @@ -3628,6 +3647,7 @@ export function issueThreadInteractionService(db: Db, opts: IssueThreadInteracti interactionId: string, input: CancelIssueThreadInteraction, actor: InteractionActor, + mutationOptions: InteractionResolutionMutationOptions = {}, ) => { assertIssueOpenForInteractionResolution(issue); const data = cancelIssueThreadInteractionSchema.parse(input); @@ -3649,32 +3669,35 @@ export function issueThreadInteractionService(db: Db, opts: IssueThreadInteracti } const reason = data.reason?.trim() || null; - const [updated] = await db - .update(issueThreadInteractions) - .set({ - status: "cancelled", - result: { - version: 1, - answers: [], - cancelled: true, - cancellationReason: reason, - summaryMarkdown: null, - }, - resolvedByAgentId: actor.agentId ?? null, - resolvedByRunId: actor.runId ?? null, - resolvedByUserId: actor.userId ?? null, - resolvedAt: new Date(), - updatedAt: new Date(), - }) - .where(and( - eq(issueThreadInteractions.id, interactionId), - eq(issueThreadInteractions.status, "pending"), - )) - .returning(); + const updated = await db.transaction(async (tx) => { + const resolvedAt = new Date(); + const [row] = await tx + .update(issueThreadInteractions) + .set({ + status: "cancelled", + result: { + version: 1, + answers: [], + cancelled: true, + cancellationReason: reason, + summaryMarkdown: null, + }, + resolvedByAgentId: actor.agentId ?? null, + resolvedByRunId: actor.runId ?? null, + resolvedByUserId: actor.userId ?? null, + resolvedAt, + updatedAt: resolvedAt, + }) + .where(and( + eq(issueThreadInteractions.id, interactionId), + eq(issueThreadInteractions.status, "pending"), + )) + .returning(); - if (!updated) { - throw interactionAlreadyResolvedError(); - } + if (!row) throw interactionAlreadyResolvedError(); + await mutationOptions.afterResolveInTransaction?.(tx, hydrateInteraction(row)); + return row; + }); await touchIssue(db, issue.id); const cancelled = hydrateInteraction(updated); diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index f0b3ac1e88..c3b20dbdda 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -22,7 +22,7 @@ import { type IssueTreePreviewWarning, } from "@paperclipai/shared"; import { conflict, notFound, unprocessable } from "../errors.js"; -import { finalizeSummarySlotsForTerminalIssue } from "./summary-slot-finalization.js"; +import type { IssuePostCommitAction } from "./issues.js"; type IssueRow = typeof issues.$inferSelect; type HoldRow = typeof issueTreeHolds.$inferSelect; @@ -870,43 +870,40 @@ export function issueTreeControlService(db: Db) { if (issueIds.length === 0) return { updatedIssueIds: [], updatedIssues: [] }; const now = new Date(); + const postCommitIssueActions: IssuePostCommitAction[] = []; + const { executeIssuePostCommitActions, issueService } = await import("./issues.js"); + const svc = issueService(db); const updated = await db.transaction(async (tx) => { - const rows = await tx - .update(issues) - .set({ - status: "cancelled", - cancelledAt: now, - completedAt: null, - checkoutRunId: null, - executionRunId: null, - executionAgentNameKey: null, - executionLockedAt: null, - updatedAt: now, - }) + const eligibleIssues = await tx + .select({ id: issues.id }) + .from(issues) .where( and( eq(issues.companyId, companyId), inArray(issues.id, issueIds), notInArray(issues.status, ["done", "cancelled"]), ), - ) - .returning({ - id: issues.id, - companyId: issues.companyId, - identifier: issues.identifier, - title: issues.title, - status: issues.status, - assigneeAgentId: issues.assigneeAgentId, - }); + ); - for (const issue of rows) { - await finalizeSummarySlotsForTerminalIssue(tx, { - ...issue, - status: coerceIssueStatus(issue.status), - }); + const rows = []; + for (const issue of eligibleIssues) { + const updatedIssue = await svc.update( + issue.id, + { + status: "cancelled", + cancelledAt: now, + actorAgentId: hold.createdByAgentId, + actorUserId: hold.createdByUserId, + }, + tx, + undefined, + postCommitIssueActions, + ); + if (updatedIssue) rows.push(updatedIssue); } return rows; }); + await executeIssuePostCommitActions(db, postCommitIssueActions); return { updatedIssueIds: updated.map((issue) => issue.id), diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 1d88d3f987..5d64eebeb7 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -164,6 +164,44 @@ const ISSUE_CREATE_IDEMPOTENCY_KEY_CLEANUP_BATCH_SIZE = 500; const DELETED_ISSUE_COMMENT_BODY = ""; const ISSUE_WAKE_DIAGNOSTICS_ACTIVITY_ACTIONS = ["issue.tree_hold_wakeup_deferred"] as const; +export type IssuePostCommitAction = { + type: "cancel_native_question_run"; + runId: string; + issueId: string; + issueStatus: string; +}; + +/** Execute side effects that must never run before the issue transaction commits. */ +export async function executeIssuePostCommitActions( + db: Db, + actions: readonly IssuePostCommitAction[], +): Promise { + if (actions.length === 0) return; + const { heartbeatService } = await import("./heartbeat.js"); + const heartbeat = heartbeatService(db); + const cancelledRunIds = new Set(); + for (const action of actions) { + if (cancelledRunIds.has(action.runId)) continue; + cancelledRunIds.add(action.runId); + try { + await heartbeat.cancelRun(action.runId, "Task closed while waiting for operator input", { + resultJson: { + cancelledByIssueStatus: action.issueStatus, + cancelledIssueId: action.issueId, + }, + }); + } catch (err) { + // The durable marker written by the issue transaction remains available + // to startup and periodic recovery. Do not report a post-commit failure + // as though the already-committed issue transition had rolled back. + logger.warn( + { err, runId: action.runId, issueId: action.issueId }, + "native question cancellation deferred to recovery sweep", + ); + } + } +} + function wakeRequestTargetsIssue(issueId: string) { return sql`( ${agentWakeupRequests.payload} ->> 'issueId' = ${issueId} @@ -7661,9 +7699,12 @@ export function issueService(db: Db) { }, dbOrTx: any = db, postCommitActivityPublications?: ActivityPublication[], + postCommitActions?: IssuePostCommitAction[], ) => { const ownedActivityPublications: ActivityPublication[] = []; const activityPublications = postCommitActivityPublications ?? ownedActivityPublications; + const ownedPostCommitActions: IssuePostCommitAction[] = []; + const queuedPostCommitActions = postCommitActions ?? ownedPostCommitActions; const existing = await dbOrTx .select() .from(issues) @@ -7910,7 +7951,36 @@ export function issueService(db: Db) { updated, { agentId: actorAgentId ?? null, userId: actorUserId ?? null }, ); + const { + nativeQuestionCancellationIdentity, + requestNativeQuestionRunCancellation, + } = await import( + "./native-runtime/native-question-bridge.js" + ); for (const interaction of expiredInteractions) { + if (interaction.kind === "ask_user_questions") { + const nativeQuestion = nativeQuestionCancellationIdentity(interaction); + if (nativeQuestion) { + if (dbOrTx !== db && !postCommitActions) { + throw new Error( + "Terminal native question updates in an external transaction require a post-commit action queue", + ); + } + const runId = await requestNativeQuestionRunCancellation( + tx, + nativeQuestion, + { kind: "issue_terminal", issueStatus: updated.status }, + ); + if (runId) { + queuedPostCommitActions.push({ + type: "cancel_native_question_run", + runId, + issueId: updated.id, + issueStatus: updated.status, + }); + } + } + } await logActivity(tx as unknown as Db, { companyId: updated.companyId, actorType: actorAgentId ? "agent" : actorUserId ? "user" : "system", @@ -8072,6 +8142,9 @@ export function issueService(db: Db) { if (dbOrTx === db && !postCommitActivityPublications) { for (const publication of ownedActivityPublications) publishActivity(publication); } + if (dbOrTx === db && !postCommitActions) { + await executeIssuePostCommitActions(db, ownedPostCommitActions); + } return result; }, diff --git a/server/src/services/native-runtime/native-question-bridge.test.ts b/server/src/services/native-runtime/native-question-bridge.test.ts new file mode 100644 index 0000000000..3c04925ebf --- /dev/null +++ b/server/src/services/native-runtime/native-question-bridge.test.ts @@ -0,0 +1,490 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { eq, sql } from "drizzle-orm"; + +import { + activityLog, + agents, + companies, + createDb, + heartbeatRuns, + issueQuestionResponseDeliveries, + issueThreadInteractions, + issues, +} from "@paperclipai/db"; +import type { PrpEvent } from "@paperclipai/paperclip-runner"; + +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "../../__tests__/helpers/embedded-postgres.js"; +import { issueThreadInteractionService } from "../issue-thread-interactions.js"; +import { + deliverNativeQuestionResponse, + flushNativeQuestionResponses, + nativeQuestionBridgeInternals, + nativeQuestionRunToCancel, + projectNativeRuntimeRequest, + registerNativeQuestionCommandTarget, + requestNativeQuestionRunCancellation, + validateNativeQuestionResponseInput, +} from "./native-question-bridge.js"; +import { + executeIssuePostCommitActions, + issueService, + type IssuePostCommitAction, +} from "../issues.js"; +import { heartbeatService } from "../heartbeat.js"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping native question bridge tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} + +describeEmbeddedPostgres("native question bridge", () => { + let temporary: Awaited> | null = null; + let db: ReturnType; + let companyId: string; + let issueId: string; + let agentId: string; + let runId: string; + let sessionId: string; + let runnerInstanceId: string; + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-question-"); + db = createDb(temporary.connectionString); + }, 20_000); + + afterEach(async () => { + nativeQuestionBridgeInternals.resetForTests(); + await db.execute(sql.raw(` + TRUNCATE TABLE + "activity_log", + "issue_thread_interactions", + "heartbeat_runs", + "agent_wakeup_requests", + "issues", + "agents", + "companies" + RESTART IDENTITY CASCADE + `)); + }); + + afterAll(async () => temporary?.cleanup()); + + async function seed() { + companyId = randomUUID(); + issueId = randomUUID(); + agentId = randomUUID(); + runId = randomUUID(); + sessionId = randomUUID(); + runnerInstanceId = randomUUID(); + await db.insert(companies).values({ + id: companyId, + name: "Native questions", + issuePrefix: `NQ${companyId.replaceAll("-", "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Native Codex", + adapterType: "paperclip_runner", + status: "running", + adapterConfig: { provider: "codex" }, + runtimeConfig: {}, + }); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Answer a native question", + status: "in_progress", + assigneeAgentId: agentId, + }); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + status: "running", + runtimeMode: "native", + runtimeModeResolvedAt: new Date(), + nativeIssueId: issueId, + nativeSessionId: sessionId, + runnerInstanceId, + driverKind: "codex", + contextSnapshot: { issueId }, + }); + } + + function runtimeRequestEvent(): PrpEvent { + return { + schema: "paperclip.prp.event.v1", + sourceEventId: "runtime-question-1", + sourceSeq: 1, + sourceInstanceId: runnerInstanceId, + sourceKind: "runner", + runId, + normalizedSessionId: sessionId, + turnId: "turn-1", + itemId: "item-1", + eventType: "runtime_request.created", + schemaVersion: 1, + priority: 0, + emittedAt: "2026-08-25T18:00:00.000Z", + payload: { + request: { + schema: "paperclip.runtime_request.v2", + requestKind: "runtime", + requestId: "request-1", + type: "input", + status: "pending", + prompt: "Choose a deployment color", + input: { + schema: "paperclip.question_set.v1", + title: "Deployment", + questions: [{ + id: "color", + prompt: "Which color?", + required: true, + answerMode: "single_select", + options: [ + { id: "blue", label: "Blue" }, + { id: "green", label: "Green" }, + ], + }], + }, + }, + }, + }; + } + + function binding() { + return { + companyId, + issueId, + runId, + agentId, + normalizedSessionId: sessionId, + runnerSourceInstanceId: runnerInstanceId, + completionContractId: randomUUID(), + completionContractSha256: `sha256:${"a".repeat(64)}`, + completionContractRevision: "1", + completionContractCriterionIds: [], + }; + } + + it("materializes, validates, and durably resumes a provider-neutral question response", async () => { + await seed(); + const interaction = await projectNativeRuntimeRequest({ + db, + binding: binding(), + event: runtimeRequestEvent(), + }); + + expect(interaction).toMatchObject({ + kind: "ask_user_questions", + status: "pending", + sourceRunId: runId, + continuationPolicy: "none", + effectiveResolverPolicy: "human_only", + payload: { + runtimeRequestId: "request-1", + supersedeOnUserComment: false, + questionSet: { schema: "paperclip.question_set.v1" }, + questions: [{ + id: "color", + selectionMode: "single", + allowOther: false, + options: [{ id: "blue", label: "Blue" }, { id: "green", label: "Green" }], + }], + }, + }); + expect(await db.select().from(activityLog)).toHaveLength(1); + + const answer = { answers: [{ questionId: "color", optionIds: ["blue"] }] }; + validateNativeQuestionResponseInput(interaction!, answer); + expect(() => validateNativeQuestionResponseInput(interaction!, { + answers: [{ questionId: "color", optionIds: ["red"] }], + })).toThrow(/unknown option red/); + + const answered = await issueThreadInteractionService(db).answerQuestions( + { id: issueId, companyId, status: "in_progress" }, + interaction!.id, + answer, + { userId: "operator-1" }, + ); + const queueCommand = vi.fn(() => ({ commandId: "question", controllerSeq: 1 })); + const release = registerNativeQuestionCommandTarget({ + binding: { companyId, issueId, runId, agentId }, + queueCommand, + }); + + await flushNativeQuestionResponses(db, runId); + expect(queueCommand).toHaveBeenCalledWith( + "request.resolve", + { + requestId: "request-1", + response: { + schema: "paperclip.question_response.v1", + answers: { color: { selectedOptionIds: ["blue"] } }, + }, + }, + `question_${interaction!.id}`, + ); + expect(queueCommand).toHaveBeenCalledTimes(1); + const [delivery] = await db.select().from(issueQuestionResponseDeliveries); + expect(delivery).toMatchObject({ + interactionId: interaction!.id, + status: "delivered", + deliveryMode: "steered", + targetRunId: runId, + }); + expect(answered.kind).toBe("ask_user_questions"); + if (answered.kind !== "ask_user_questions") throw new Error("expected question interaction"); + await expect(nativeQuestionRunToCancel(db, answered)).resolves.toBe(runId); + release(); + }); + + it("binds projection to the persisted native run and ignores legacy delivery", async () => { + await seed(); + const mismatched = runtimeRequestEvent(); + mismatched.runId = randomUUID(); + await expect(projectNativeRuntimeRequest({ db, binding: binding(), event: mismatched })) + .rejects.toThrow("native_runtime_request_binding_mismatch"); + + const interaction = await projectNativeRuntimeRequest({ + db, + binding: binding(), + event: runtimeRequestEvent(), + }); + const answered = await issueThreadInteractionService(db).answerQuestions( + { id: issueId, companyId, status: "in_progress" }, + interaction!.id, + { answers: [{ questionId: "color", optionIds: ["green"] }] }, + { userId: "operator-1" }, + ); + await db.update(heartbeatRuns).set({ runtimeMode: "legacy" }).where(eq(heartbeatRuns.id, runId)); + expect(answered.kind).toBe("ask_user_questions"); + if (answered.kind !== "ask_user_questions") throw new Error("expected question interaction"); + await expect(deliverNativeQuestionResponse(db, answered)).resolves.toBe("not_native"); + await expect(nativeQuestionRunToCancel(db, answered)).resolves.toBeNull(); + }); + + it("does not duplicate the task card when the runner replays a request", async () => { + await seed(); + const first = await projectNativeRuntimeRequest({ db, binding: binding(), event: runtimeRequestEvent() }); + const second = await projectNativeRuntimeRequest({ db, binding: binding(), event: runtimeRequestEvent() }); + expect(second?.id).toBe(first?.id); + expect(await db.select().from(issueThreadInteractions)).toHaveLength(1); + expect(await db.select().from(activityLog)).toHaveLength(1); + }); + + it("cancels the active native run when the shared issue service expires its question", async () => { + await seed(); + const interaction = await projectNativeRuntimeRequest({ + db, + binding: binding(), + event: runtimeRequestEvent(), + }); + await issueService(db).update(issueId, { status: "cancelled" }); + + const [persistedInteraction] = await db.select({ status: issueThreadInteractions.status }) + .from(issueThreadInteractions) + .where(eq(issueThreadInteractions.id, interaction!.id)); + const [persistedRun] = await db.select({ + status: heartbeatRuns.status, + resultJson: heartbeatRuns.resultJson, + }).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(persistedInteraction?.status).toBe("expired"); + expect(persistedRun).toMatchObject({ + status: "cancelled", + resultJson: { + cancelledByIssueStatus: "cancelled", + cancelledIssueId: issueId, + }, + }); + }); + + it("defers native cancellation until an external issue transaction commits", async () => { + await seed(); + await projectNativeRuntimeRequest({ + db, + binding: binding(), + event: runtimeRequestEvent(), + }); + const postCommitActions: IssuePostCommitAction[] = []; + await db.transaction(async (tx) => { + await issueService(db).update( + issueId, + { status: "done" }, + tx, + undefined, + postCommitActions, + ); + const [runInsideTransaction] = await tx.select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)); + expect(runInsideTransaction?.status).toBe("running"); + }); + + expect(postCommitActions).toHaveLength(1); + await executeIssuePostCommitActions(db, postCommitActions); + const [persistedRun] = await db.select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)); + expect(persistedRun?.status).toBe("cancelled"); + }); + + it("recovers a durable native cancellation when the post-commit process exits", async () => { + await seed(); + await projectNativeRuntimeRequest({ + db, + binding: binding(), + event: runtimeRequestEvent(), + }); + const postCommitActions: IssuePostCommitAction[] = []; + await db.transaction(async (tx) => { + await issueService(db).update( + issueId, + { status: "done" }, + tx, + undefined, + postCommitActions, + ); + }); + + const [markedRun] = await db.select({ + status: heartbeatRuns.status, + contextSnapshot: heartbeatRuns.contextSnapshot, + }).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(markedRun).toMatchObject({ + status: "running", + contextSnapshot: { + nativeQuestionCancellation: { + version: 1, + issueId, + issueStatus: "done", + }, + }, + }); + + // Simulate process exit before executeIssuePostCommitActions can run. + await heartbeatService(db).reapOrphanedRuns(); + + const [persistedRun] = await db.select({ + status: heartbeatRuns.status, + resultJson: heartbeatRuns.resultJson, + }).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(persistedRun).toMatchObject({ + status: "cancelled", + resultJson: { + cancelledByIssueStatus: "done", + cancelledIssueId: issueId, + }, + }); + }); + + it("recovers an explicit question withdrawal committed with its cancellation intent", async () => { + await seed(); + const interaction = await projectNativeRuntimeRequest({ + db, + binding: binding(), + event: runtimeRequestEvent(), + }); + await issueThreadInteractionService(db).withdrawInteraction( + { id: issueId, companyId }, + interaction!.id, + { reason: "No longer needed" }, + { userId: "operator-1" }, + { + afterResolveInTransaction: async (tx, withdrawn) => { + await expect(requestNativeQuestionRunCancellation(tx, withdrawn, { + kind: "interaction_withdrawn", + interactionId: withdrawn.id, + })).resolves.toBe(runId); + }, + }, + ); + + const [markedRun] = await db.select({ contextSnapshot: heartbeatRuns.contextSnapshot }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)); + expect(markedRun?.contextSnapshot).toMatchObject({ + nativeQuestionCancellation: { + version: 1, + kind: "interaction_withdrawn", + interactionId: interaction!.id, + issueId, + }, + }); + + await heartbeatService(db).reapOrphanedRuns(); + + const [cancelledRun] = await db.select({ + status: heartbeatRuns.status, + resultJson: heartbeatRuns.resultJson, + }).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(cancelledRun).toMatchObject({ + status: "cancelled", + resultJson: { + withdrawnInteractionId: interaction!.id, + cancelledIssueId: issueId, + }, + }); + }); + + it("removes the UI-only marker from a canonical custom response", async () => { + await seed(); + const event = runtimeRequestEvent(); + const request = event.payload.request as Record; + const input = request.input as Record; + input.questions = [{ + id: "color", + prompt: "Which color?", + required: true, + answerMode: "single_select", + options: [{ id: "blue", label: "Blue" }], + customAnswer: { enabled: true, label: "Another color" }, + }]; + const interaction = await projectNativeRuntimeRequest({ db, binding: binding(), event }); + const answer = { + answers: [{ + questionId: "color", + optionIds: ["paperclip_custom_answer"], + otherText: "purple", + }], + }; + validateNativeQuestionResponseInput(interaction!, answer); + const answered = await issueThreadInteractionService(db).answerQuestions( + { id: issueId, companyId, status: "in_progress" }, + interaction!.id, + answer, + { userId: "operator-1" }, + ); + expect(answered.kind).toBe("ask_user_questions"); + if (answered.kind !== "ask_user_questions") throw new Error("expected question interaction"); + + const queueCommand = vi.fn(() => ({ commandId: "question", controllerSeq: 1 })); + registerNativeQuestionCommandTarget({ + binding: { companyId, issueId, runId, agentId }, + queueCommand, + }); + await expect(deliverNativeQuestionResponse(db, answered)).resolves.toBe("queued"); + expect(queueCommand).toHaveBeenCalledWith( + "request.resolve", + { + requestId: "request-1", + response: { + schema: "paperclip.question_response.v1", + answers: { color: { selectedOptionIds: [], customText: "purple" } }, + }, + }, + `question_${interaction!.id}`, + ); + }); +}); diff --git a/server/src/services/native-runtime/native-question-bridge.ts b/server/src/services/native-runtime/native-question-bridge.ts new file mode 100644 index 0000000000..391320d4e8 --- /dev/null +++ b/server/src/services/native-runtime/native-question-bridge.ts @@ -0,0 +1,444 @@ +import { and, eq, inArray, sql } from "drizzle-orm"; + +import type { Db } from "@paperclipai/db"; +import { heartbeatRuns, issueThreadInteractions } from "@paperclipai/db"; +import type { + AskUserQuestionsAnswer, + AskUserQuestionsInteraction, + AskUserQuestionsQuestionOption, + PaperclipQuestionSetPayload, + RespondIssueThreadInteraction, +} from "@paperclipai/shared"; + +import type { PrpEvent } from "../../vendor/paperclip-runner/index.js"; +import { + parsePaperclipQuestionResponse, + parsePaperclipQuestionSet, + type PaperclipQuestionResponse, + type PaperclipQuestionSet, +} from "../../vendor/paperclip-runner/index.js"; +import { logger } from "../../middleware/logger.js"; +import { unprocessable } from "../../errors.js"; +import { logActivity } from "../activity-log.js"; +import { issueThreadInteractionService } from "../issue-thread-interactions.js"; +import { questionResponseDeliveryService } from "../question-response-delivery.js"; +import type { NativeRunStoreBinding } from "./native-run-coordinator-store.js"; + +const QUESTION_KEY_PREFIX = "paperclip-runner-question:"; +const REQUEST_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,159}$/; +const TEXT_ANSWER_OPTION_ID = "paperclip_text_answer"; +const CUSTOM_ANSWER_OPTION_ID = "paperclip_custom_answer"; +export const NATIVE_QUESTION_CANCELLATION_CONTEXT_KEY = "nativeQuestionCancellation"; + +type QueueCommand = ( + type: string, + payload?: Record, + commandId?: string, +) => { readonly commandId: string; readonly controllerSeq: number }; + +interface NativeQuestionCommandTarget { + binding: Pick; + queueCommand: QueueCommand; +} + +const activeTargets = new Map(); + +interface NativeQuestionIdentity { + idempotencyKey?: string | null; + sourceRunId?: string | null; + payload: unknown; +} + +export interface NativeQuestionAuthorizationIdentity extends NativeQuestionIdentity { + companyId: string; + issueId: string; +} + +export type NativeQuestionCancellationCause = + | { kind: "issue_terminal"; issueStatus: string } + | { kind: "interaction_withdrawn"; interactionId: string } + | { kind: "interaction_cancelled"; interactionId: string }; + +type DbTransaction = Parameters[0]>[0]; +type NativeQuestionMutationDb = Pick; + +function record(value: unknown): Record | null { + return value && typeof value === "object" && !Array.isArray(value) + ? value as Record + : null; +} + +function requestIdForInteraction( + interaction: NativeQuestionIdentity, +): string | null { + const payload = record(interaction.payload); + if (!interaction.sourceRunId || !payload?.questionSet) return null; + const key = interaction.idempotencyKey; + const expectedPrefix = `${QUESTION_KEY_PREFIX}${interaction.sourceRunId}:`; + if (!key?.startsWith(expectedPrefix)) return null; + const requestId = key.slice(expectedPrefix.length); + if (!REQUEST_ID_PATTERN.test(requestId)) return null; + return typeof payload.runtimeRequestId === "string" && payload.runtimeRequestId !== requestId + ? null + : requestId; +} + +function uniqueSyntheticOptionId(existing: readonly string[], preferred: string): string { + const ids = new Set(existing); + if (!ids.has(preferred)) return preferred; + for (let suffix = 2; suffix < 10_000; suffix += 1) { + const candidate = `${preferred}_${suffix}`; + if (!ids.has(candidate)) return candidate; + } + throw new Error("native_question_synthetic_option_exhausted"); +} + +function toInteractionPayload(questionSet: PaperclipQuestionSet, runtimeRequestId: string) { + return { + version: 1 as const, + ...(questionSet.title ? { title: questionSet.title.slice(0, 240) } : {}), + ...(questionSet.submitLabel ? { submitLabel: questionSet.submitLabel.slice(0, 120) } : {}), + questions: questionSet.questions.map((question) => { + const canonicalOptions = question.options ?? []; + const options: AskUserQuestionsQuestionOption[] = canonicalOptions.map((option) => ({ + id: option.id, + label: option.label, + ...(option.description ? { description: option.description } : {}), + })); + if (question.answerMode === "text") { + options.push({ + id: uniqueSyntheticOptionId([], TEXT_ANSWER_OPTION_ID), + label: question.header ?? "Type an answer", + ...(question.textValidation?.inputType + ? { description: `Expected ${question.textValidation.inputType} input` } + : {}), + freeText: true, + }); + } else if (question.customAnswer?.enabled) { + options.push({ + id: uniqueSyntheticOptionId(canonicalOptions.map((option) => option.id), CUSTOM_ANSWER_OPTION_ID), + label: question.customAnswer.label ?? "Other", + ...(question.customAnswer.placeholder ? { description: question.customAnswer.placeholder } : {}), + freeText: true, + }); + } + return { + id: question.id, + prompt: question.prompt, + ...((question.helpText || question.header) + ? { helpText: question.helpText ?? question.header } + : {}), + selectionMode: question.answerMode === "multi_select" ? "multi" as const : "single" as const, + required: question.required, + allowOther: question.answerMode === "text" || question.customAnswer?.enabled === true, + options, + }; + }), + questionSet: questionSet as PaperclipQuestionSetPayload, + runtimeRequestId, + // A generic task comment cannot satisfy this provider request. Keep the + // card actionable until a validated answer or an explicit terminal action. + supersedeOnUserComment: false, + }; +} + +function canonicalResponse( + questionSet: PaperclipQuestionSetPayload, + answers: readonly AskUserQuestionsAnswer[], +): PaperclipQuestionResponse { + const answerByQuestionId = new Map(answers.map((answer) => [answer.questionId, answer])); + const response: PaperclipQuestionResponse = { + schema: "paperclip.question_response.v1", + answers: {}, + }; + for (const question of questionSet.questions) { + const answer = answerByQuestionId.get(question.id); + if (!answer) continue; + if (question.answerMode === "text") { + response.answers[question.id] = { + ...(answer.otherText !== undefined && answer.otherText !== null + ? { text: answer.otherText } + : {}), + }; + } else { + const customOptionId = question.customAnswer?.enabled + ? uniqueSyntheticOptionId( + (question.options ?? []).map((option) => option.id), + CUSTOM_ANSWER_OPTION_ID, + ) + : null; + response.answers[question.id] = { + selectedOptionIds: answer.optionIds.filter((optionId) => optionId !== customOptionId), + ...(answer.otherText !== undefined && answer.otherText !== null + ? { customText: answer.otherText } + : {}), + }; + } + } + return parsePaperclipQuestionResponse(questionSet, response); +} + +async function authorizedNativeRun( + db: Pick, + interaction: NativeQuestionAuthorizationIdentity, +) { + const requestId = requestIdForInteraction(interaction); + if (!requestId || !interaction.sourceRunId) return null; + const run = await db.select({ + id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + issueId: heartbeatRuns.nativeIssueId, + agentId: heartbeatRuns.agentId, + runtimeMode: heartbeatRuns.runtimeMode, + status: heartbeatRuns.status, + }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, interaction.sourceRunId), + eq(heartbeatRuns.companyId, interaction.companyId), + eq(heartbeatRuns.nativeIssueId, interaction.issueId), + eq(heartbeatRuns.runtimeMode, "native"), + )).limit(1).then((rows) => rows[0] ?? null); + return run ? { ...run, requestId } : null; +} + +/** Materialize a canonical runtime input request as the existing task-thread card. */ +export async function projectNativeRuntimeRequest(input: { + db: Db; + binding: NativeRunStoreBinding; + event: PrpEvent; +}): Promise { + if (input.event.eventType !== "runtime_request.created") return null; + if ( + input.event.runId !== input.binding.runId + || input.event.normalizedSessionId !== input.binding.normalizedSessionId + || input.event.sourceInstanceId !== input.binding.runnerSourceInstanceId + ) { + throw new Error("native_runtime_request_binding_mismatch"); + } + const request = record(record(input.event.payload)?.request); + if ( + !request + || request.schema !== "paperclip.runtime_request.v2" + || request.requestKind !== "runtime" + || request.type !== "input" + || request.status !== "pending" + || typeof request.requestId !== "string" + || !REQUEST_ID_PATTERN.test(request.requestId) + ) { + throw new Error("native_runtime_request_invalid"); + } + const questionSet = parsePaperclipQuestionSet(request.input); + if (questionSet.questions.some((question) => question.textValidation?.pattern !== undefined)) { + // JavaScript regular expressions have no execution budget. Provider-authored + // patterns therefore stay fail-closed until the runner contract supplies a + // bounded regex dialect rather than exposing the server to catastrophic backtracking. + throw new Error("native_runtime_question_pattern_unsupported"); + } + const idempotencyKey = `${QUESTION_KEY_PREFIX}${input.binding.runId}:${request.requestId}`; + const existing = await input.db.select({ id: issueThreadInteractions.id }) + .from(issueThreadInteractions) + .where(and( + eq(issueThreadInteractions.companyId, input.binding.companyId), + eq(issueThreadInteractions.issueId, input.binding.issueId), + eq(issueThreadInteractions.idempotencyKey, idempotencyKey), + )) + .limit(1) + .then((rows) => rows[0] ?? null); + const interaction = await issueThreadInteractionService(input.db).create( + { id: input.binding.issueId, companyId: input.binding.companyId }, + { + kind: "ask_user_questions", + idempotencyKey, + sourceRunId: input.binding.runId, + resolverPolicy: "human_only", + continuationPolicy: "none", + ...(questionSet.title ? { title: questionSet.title.slice(0, 240) } : {}), + ...(typeof request.prompt === "string" ? { summary: request.prompt.slice(0, 1000) } : {}), + payload: toInteractionPayload(questionSet, request.requestId), + }, + { agentId: input.binding.agentId, runId: input.binding.runId }, + { supersedePendingSiblingInteractions: false }, + ) as AskUserQuestionsInteraction; + if (!existing) { + await logActivity(input.db, { + companyId: input.binding.companyId, + actorType: "agent", + actorId: input.binding.agentId, + agentId: input.binding.agentId, + runId: input.binding.runId, + action: "issue.thread_interaction_created", + entityType: "issue", + entityId: input.binding.issueId, + details: { + interactionId: interaction.id, + interactionKind: interaction.kind, + interactionStatus: interaction.status, + runtimeMode: "native", + }, + }); + } + if (interaction.status === "answered") { + await deliverNativeQuestionResponseDurably(input.db, interaction); + } + return interaction; +} + +/** Validate untrusted board input before the existing interaction service persists it. */ +export function validateNativeQuestionResponseInput( + interaction: AskUserQuestionsInteraction, + input: RespondIssueThreadInteraction, +): void { + if (!requestIdForInteraction(interaction) || !interaction.payload.questionSet) return; + try { + canonicalResponse(interaction.payload.questionSet, input.answers); + } catch (error) { + throw unprocessable( + error instanceof Error ? error.message : "Invalid native question response", + { code: "invalid_question_response" }, + ); + } +} + +/** Queue an answered interaction into the active durable PRP command stream. */ +export async function deliverNativeQuestionResponse( + db: Db, + interaction: AskUserQuestionsInteraction, +): Promise<"not_native" | "pending" | "queued"> { + if (interaction.status !== "answered" || !interaction.result || !interaction.payload.questionSet) { + return "not_native"; + } + const run = await authorizedNativeRun(db, interaction); + if (!run) return "not_native"; + const response = canonicalResponse(interaction.payload.questionSet, interaction.result.answers); + const target = activeTargets.get(run.id); + if ( + !target + || target.binding.companyId !== run.companyId + || target.binding.issueId !== run.issueId + || target.binding.agentId !== run.agentId + ) { + return "pending"; + } + try { + target.queueCommand( + "request.resolve", + { requestId: run.requestId, response: response as unknown as Record }, + `question_${interaction.id}`, + ); + return "queued"; + } catch (error) { + logger.warn( + { err: error, runId: run.id, interactionId: interaction.id }, + "native question response remains durable for session recovery", + ); + return "pending"; + } +} + +async function deliverNativeQuestionResponseDurably( + db: Db, + interaction: AskUserQuestionsInteraction, +): Promise { + await questionResponseDeliveryService(db, { + heartbeat: { + wakeup: async () => { + throw new Error("native_question_wake_unreachable"); + }, + } as never, + resolveNativeQuestion: (candidate) => deliverNativeQuestionResponse(db, candidate), + }).deliver(interaction.id); +} + +export async function flushNativeQuestionResponses( + db: Db, + runId: string, +): Promise { + const target = activeTargets.get(runId); + if (!target) return; + const interactions = await issueThreadInteractionService(db).listForIssue(target.binding.issueId); + for (const interaction of interactions) { + if ( + interaction.kind === "ask_user_questions" + && interaction.sourceRunId === runId + && interaction.status === "answered" + ) { + await deliverNativeQuestionResponseDurably(db, interaction); + } + } +} + +export function registerNativeQuestionCommandTarget(target: NativeQuestionCommandTarget): () => void { + const existing = activeTargets.get(target.binding.runId); + if (existing) throw new Error("native_question_command_target_conflict"); + activeTargets.set(target.binding.runId, target); + return () => { + if (activeTargets.get(target.binding.runId) === target) { + activeTargets.delete(target.binding.runId); + } + }; +} + +export async function nativeQuestionRunToCancel( + db: Db, + interaction: NativeQuestionAuthorizationIdentity, +): Promise { + const run = await authorizedNativeRun(db, interaction); + return run && ["queued", "running"].includes(run.status) ? run.id : null; +} + +/** + * Persist cancellation intent in the same transaction that closes the issue. + * The post-commit fast path and the heartbeat recovery sweep both consume this + * marker, so process exit or a transient process-termination failure cannot + * strand a native run after its question has expired. + */ +export async function requestNativeQuestionRunCancellation( + db: NativeQuestionMutationDb, + interaction: NativeQuestionAuthorizationIdentity, + cause: NativeQuestionCancellationCause, +): Promise { + const run = await authorizedNativeRun(db, interaction); + if (!run || !["queued", "running"].includes(run.status)) return null; + const marker = JSON.stringify({ + version: 1, + issueId: interaction.issueId, + ...cause, + requestedAt: new Date().toISOString(), + }); + return db.update(heartbeatRuns).set({ + contextSnapshot: sql`jsonb_set( + case + when jsonb_typeof(${heartbeatRuns.contextSnapshot}) = 'object' + then ${heartbeatRuns.contextSnapshot} + else '{}'::jsonb + end, + array[${NATIVE_QUESTION_CANCELLATION_CONTEXT_KEY}], + ${marker}::jsonb, + true + )`, + updatedAt: new Date(), + }).where(and( + eq(heartbeatRuns.id, run.id), + eq(heartbeatRuns.companyId, interaction.companyId), + eq(heartbeatRuns.nativeIssueId, interaction.issueId), + eq(heartbeatRuns.runtimeMode, "native"), + inArray(heartbeatRuns.status, ["queued", "running"]), + )).returning({ id: heartbeatRuns.id }).then((rows) => rows[0]?.id ?? null); +} + +/** Capture the minimum bound identity needed to cancel after the issue transaction commits. */ +export function nativeQuestionCancellationIdentity( + interaction: NativeQuestionAuthorizationIdentity, +): NativeQuestionAuthorizationIdentity | null { + if (!requestIdForInteraction(interaction)) return null; + return { + companyId: interaction.companyId, + issueId: interaction.issueId, + sourceRunId: interaction.sourceRunId, + payload: interaction.payload, + idempotencyKey: interaction.idempotencyKey, + }; +} + +export const nativeQuestionBridgeInternals = { + resetForTests: () => activeTargets.clear(), +}; diff --git a/server/src/services/native-runtime/runner-prp-coordinator.ts b/server/src/services/native-runtime/runner-prp-coordinator.ts index 9d9d9f2f91..067e77903f 100644 --- a/server/src/services/native-runtime/runner-prp-coordinator.ts +++ b/server/src/services/native-runtime/runner-prp-coordinator.ts @@ -13,6 +13,11 @@ import { import { registerRunnerPrpAuthority } from "../../realtime/runner-prp-ws.js"; import { NativeRunCoordinatorStore } from "./native-run-coordinator-store.js"; +import { + flushNativeQuestionResponses, + projectNativeRuntimeRequest, + registerNativeQuestionCommandTarget, +} from "./native-question-bridge.js"; import { PaperclipRunnerSemanticAuthority } from "./runner-semantic-authority.js"; const UUID_PATTERN = @@ -214,7 +219,7 @@ export function runnerPrpCoordinator( agentId: input.agentId, }); const semanticTools = await semanticAuthority.listAlwaysAvailableTools(); - const nativeStore = new NativeRunCoordinatorStore(db, { + const storeBinding = { companyId: input.companyId, issueId: input.issueId, runId: input.runId, @@ -225,7 +230,8 @@ export function runnerPrpCoordinator( completionContractSha256: binding.run.completionContractSha256, completionContractRevision: String(binding.completionContract.revision), completionContractCriterionIds: criterionIds, - }); + } as const; + const nativeStore = new NativeRunCoordinatorStore(db, storeBinding); type StoredCompletedRun = NonNullable>>; type CompletedRun = StoredCompletedRun & { readonly providerSessionId?: string }; const withProviderSession = async (stored: StoredCompletedRun): Promise => { @@ -255,6 +261,9 @@ export function runnerPrpCoordinator( connectionLeaseTtlMs, onCommittedEvent: async (event) => { await nativeStore.appendEvent(event); + if (event.eventType === "runtime_request.created") { + await projectNativeRuntimeRequest({ db, binding: storeBinding, event }); + } await nativeStore.reconcileTerminalEvent(event); if (event.eventType === "run.terminal") { const stored = await nativeStore.readCompletedRun(); @@ -274,15 +283,38 @@ export function runnerPrpCoordinator( }, }); - const registration = await registerRunnerPrpAuthority({ - companyId: input.companyId, - runId: input.runId, - authority, - }); + let registration: Awaited>; + try { + registration = await registerRunnerPrpAuthority({ + companyId: input.companyId, + runId: input.runId, + authority, + }); + } catch (error) { + authority.disconnectActiveRunner(); + throw error; + } + let releaseQuestionTarget = () => {}; + try { + releaseQuestionTarget = registerNativeQuestionCommandTarget({ + binding: storeBinding, + queueCommand: (type, payload = {}, commandId) => { + const command = authority.queueCommand(type, payload, commandId, true); + return { commandId: command.commandId, controllerSeq: command.controllerSeq }; + }, + }); + await flushNativeQuestionResponses(db, input.runId); + } catch (error) { + releaseQuestionTarget(); + authority.disconnectActiveRunner(); + await registration.release(); + throw error; + } let bootstrapTicket: string; try { bootstrapTicket = authority.issueBootstrapTicket(bootstrapTtlMs); } catch (error) { + releaseQuestionTarget(); authority.disconnectActiveRunner(); await registration.release(); throw error; @@ -339,6 +371,7 @@ export function runnerPrpCoordinator( release: async () => { if (released) return; released = true; + releaseQuestionTarget(); authority.disconnectActiveRunner(); await registration.release(); }, diff --git a/server/src/services/pipelines.ts b/server/src/services/pipelines.ts index dfe2bb43f4..90e55dbaf5 100644 --- a/server/src/services/pipelines.ts +++ b/server/src/services/pipelines.ts @@ -50,7 +50,7 @@ import { logActivity } from "./activity-log.js"; import { assertAssignableAgent } from "./agent-assignability.js"; import { authorizationService } from "./authorization.js"; import { visibleIssueCondition } from "./issue-visibility.js"; -import { finalizeSummarySlotsForTerminalIssue } from "./summary-slot-finalization.js"; +import type { IssuePostCommitAction } from "./issues.js"; import { formatPipelineCaseOutputContextMarkdown, pipelineCaseOutputsService, @@ -4503,6 +4503,9 @@ export function pipelineService(db: Db, deps: { heartbeat?: IssueAssignmentWakeu cleanup: PipelineAutomationRetryCleanupOptions; actor: PipelineActor; }) { + const postCommitIssueActions: IssuePostCommitAction[] = []; + const { executeIssuePostCommitActions, issueService } = await import("./issues.js"); + const issueSvc = issueService(db); const result = await db.transaction(async (tx) => { const detail = await getCaseWithStageForUpdateOrThrow(tx, input.companyId, input.caseId); if (detail.case.version !== input.expectedVersion) { @@ -4613,26 +4616,26 @@ export function pipelineService(db: Db, deps: { heartbeat?: IssueAssignmentWakeu ? effects.linkedAutomationIssueIds : []; if (issueIdsToCancel.length > 0) { - const cancelledIssues = await tx - .update(issues) - .set({ status: "cancelled", updatedAt: now }) + const cancellableIssues = await tx + .select({ id: issues.id }) + .from(issues) .where(and( eq(issues.companyId, input.companyId), inArray(issues.id, issueIdsToCancel), ne(issues.status, "done"), - )) - .returning({ - id: issues.id, - companyId: issues.companyId, - identifier: issues.identifier, - title: issues.title, - status: issues.status, - }); - for (const issue of cancelledIssues) { - await finalizeSummarySlotsForTerminalIssue(tx, { - ...issue, - status: "cancelled", - }); + )); + for (const issue of cancellableIssues) { + await issueSvc.update( + issue.id, + { + status: "cancelled", + actorAgentId: input.actor.type === "agent" ? input.actor.agentId : null, + actorUserId: input.actor.type === "user" ? input.actor.userId : null, + }, + tx, + undefined, + postCommitIssueActions, + ); } await tx .update(pipelineCaseIssueLinks) @@ -4717,6 +4720,7 @@ export function pipelineService(db: Db, deps: { heartbeat?: IssueAssignmentWakeu }, }; }); + await executeIssuePostCommitActions(db, postCommitIssueActions); const automationExecution = await executeAutomationLedger(result.ledger.id, input.actor); const { targetStageRow: _targetStageRow, automationRoutineId: _automationRoutineId, ...plan } = result.plan; return { diff --git a/server/src/services/question-response-delivery.ts b/server/src/services/question-response-delivery.ts index 0820b82faa..aaca08029d 100644 --- a/server/src/services/question-response-delivery.ts +++ b/server/src/services/question-response-delivery.ts @@ -55,6 +55,9 @@ type QuestionResponseSteer = (input: { message: string; correlationId: string; }) => Promise<{ turnId?: string | null }>; +type NativeQuestionResponseResolver = ( + interaction: AskUserQuestionsInteraction, +) => Promise<"not_native" | "pending" | "queued">; export interface QuestionResponseDeliveryEnvelope { schema: "paperclip.question_response_delivery.v1"; @@ -77,6 +80,8 @@ export interface QuestionResponseDeliveryServiceOptions { heartbeat: Heartbeat; /** Optional native steering seam. Direct adapters use the durable wake fallback. */ steer?: QuestionResponseSteer; + /** Resolve the original in-flight native input request before considering a continuation run. */ + resolveNativeQuestion?: NativeQuestionResponseResolver; now?: () => Date; /** Test-only lease timings. Production callers use the bounded defaults. */ claimStaleMs?: number; @@ -260,6 +265,7 @@ export function questionResponseDeliveryService( options: QuestionResponseDeliveryServiceOptions, ) { const steer = options.steer; + const resolveNativeQuestion = options.resolveNativeQuestion; const now = options.now ?? (() => new Date()); const claimStaleMs = Math.max(2, options.claimStaleMs ?? DELIVERY_CLAIM_STALE_MS); const claimRefreshMs = Math.max( @@ -568,7 +574,8 @@ export function questionResponseDeliveryService( const queuedSuccessor = issueRuns.find((run) => (run.status === "queued" || run.status === "scheduled_retry") && run.id !== interaction.sourceRunId, ) ?? null; - const envelope = buildQuestionResponseDeliveryEnvelope(hydrateQuestionInteraction(interaction)); + const hydratedInteraction = hydrateQuestionInteraction(interaction); + const envelope = buildQuestionResponseDeliveryEnvelope(hydratedInteraction); if (nativeSha256(envelope) !== claimed.payloadSha256) { return recordTerminal({ delivery: claimed, @@ -581,6 +588,53 @@ export function questionResponseDeliveryService( }); } + if (resolveNativeQuestion) { + try { + const nativeDisposition = await withClaimLease( + claimed, + () => resolveNativeQuestion(hydratedInteraction), + ); + if (nativeDisposition === "queued") { + return recordTerminal({ + delivery: claimed, + interaction, + status: "delivered", + mode: "steered", + targetRunId: interaction.sourceRunId, + adapter, + }); + } + if (nativeDisposition === "pending") { + await releaseForRetry(claimed, "native_question_session_unavailable", { bounded: false }); + return null; + } + } catch (error) { + if (error instanceof DeliveryClaimUnavailableError) return terminalOutcome(interactionId); + const errorCode = error instanceof Error && compactLine(error.message) + ? compactLine(error.message)!.slice(0, 160) + : "native_question_delivery_failed"; + const exhausted = await releaseForRetry(claimed, errorCode); + logger.warn({ + err: error, + deliveryId: claimed.id, + interactionId, + attemptCount: claimed.attemptCount, + errorCount: claimed.errorCount + 1, + exhausted, + }, "native question response delivery will retry"); + if (!exhausted) return null; + return recordTerminal({ + delivery: claimed, + interaction, + status: "failed", + mode: null, + targetRunId: interaction.sourceRunId, + adapter, + errorCode, + }); + } + } + let steeringErrorCode: string | null = null; if (successorRunning?.runtimeMode === "native" && steer) { try { diff --git a/server/src/services/stalled-review-decisions.ts b/server/src/services/stalled-review-decisions.ts index 86847864b6..acaf324b6c 100644 --- a/server/src/services/stalled-review-decisions.ts +++ b/server/src/services/stalled-review-decisions.ts @@ -2,9 +2,17 @@ import { and, eq } from "drizzle-orm"; import { issues, type Db } from "@paperclipai/db"; import type { StalledReviewDecisionAction } from "@paperclipai/shared"; import { conflict, notFound } from "../errors.js"; -import { logActivity } from "./activity-log.js"; +import { + logActivity, + publishActivity, + type ActivityPublication, +} from "./activity-log.js"; import { visibleIssueCondition } from "./issue-visibility.js"; -import { issueService } from "./issues.js"; +import { + executeIssuePostCommitActions, + issueService, + type IssuePostCommitAction, +} from "./issues.js"; export interface StalledReviewDecisionActor { userId: string; @@ -20,92 +28,105 @@ export interface DecideStalledReviewInput { } export function stalledReviewDecisionService(db: Db) { + const svc = issueService(db); return { - decide: async (input: DecideStalledReviewInput) => db.transaction(async (tx) => { - const txDb = tx as unknown as Db; - const lockedIssue = await tx - .select() - .from(issues) - .where(and( - eq(issues.id, input.issueId), - eq(issues.companyId, input.companyId), - visibleIssueCondition(), - )) - .for("update") - .then((rows) => rows[0] ?? null); + decide: async (input: DecideStalledReviewInput) => { + const postCommitActivityPublications: ActivityPublication[] = []; + const postCommitIssueActions: IssuePostCommitAction[] = []; + const result = await db.transaction(async (tx) => { + const txDb = tx as unknown as Db; + const lockedIssue = await tx + .select() + .from(issues) + .where(and( + eq(issues.id, input.issueId), + eq(issues.companyId, input.companyId), + visibleIssueCondition(), + )) + .for("update") + .then((rows) => rows[0] ?? null); - if (!lockedIssue) throw notFound("Issue not found"); - if (lockedIssue.status !== "in_review") { - throw conflict("Issue is no longer a stalled review", { - issueId: lockedIssue.id, - currentStatus: lockedIssue.status, - }); - } + if (!lockedIssue) throw notFound("Issue not found"); + if (lockedIssue.status !== "in_review") { + throw conflict("Issue is no longer a stalled review", { + issueId: lockedIssue.id, + currentStatus: lockedIssue.status, + }); + } - const svc = issueService(txDb); - const reviewAttention = await svc - .listReviewAttention(lockedIssue.companyId, [lockedIssue]) - .then((rows) => rows.get(lockedIssue.id)); - if (reviewAttention?.state !== "stalled") { - throw conflict("Issue is no longer a stalled review", { - issueId: lockedIssue.id, - reviewAttentionState: reviewAttention?.state ?? "none", - }); - } + const reviewAttention = await svc + .listReviewAttention(lockedIssue.companyId, [lockedIssue], tx) + .then((rows) => rows.get(lockedIssue.id)); + if (reviewAttention?.state !== "stalled") { + throw conflict("Issue is no longer a stalled review", { + issueId: lockedIssue.id, + reviewAttentionState: reviewAttention?.state ?? "none", + }); + } - const comment = input.note - ? await svc.addComment( - lockedIssue.id, - input.note, - { userId: input.actor.userId, runId: input.actor.runId ?? null }, - { authorType: "user" }, - tx, - ) - : null; - const status = input.action === "approve" ? "done" : "todo"; - const updated = await svc.update(lockedIssue.id, { - status, - actorUserId: input.actor.userId, - }, tx); - if (!updated) throw notFound("Issue not found"); + const comment = input.note + ? await svc.addComment( + lockedIssue.id, + input.note, + { userId: input.actor.userId, runId: input.actor.runId ?? null }, + { authorType: "user" }, + tx, + ) + : null; + const status = input.action === "approve" ? "done" : "todo"; + const updated = await svc.update( + lockedIssue.id, + { + status, + actorUserId: input.actor.userId, + }, + tx, + postCommitActivityPublications, + postCommitIssueActions, + ); + if (!updated) throw notFound("Issue not found"); - if (comment) { + if (comment) { + await logActivity(txDb, { + companyId: updated.companyId, + actorType: "user", + actorId: input.actor.userId, + runId: input.actor.runId ?? null, + action: "issue.comment_added", + entityType: "issue", + entityId: updated.id, + issueId: updated.id, + details: { + commentId: comment.id, + authorUserId: input.actor.userId, + source: "stalled_review_decision", + }, + }); + } await logActivity(txDb, { companyId: updated.companyId, actorType: "user", actorId: input.actor.userId, runId: input.actor.runId ?? null, - action: "issue.comment_added", + action: "issue.stalled_review_decided", entityType: "issue", entityId: updated.id, issueId: updated.id, details: { - commentId: comment.id, - authorUserId: input.actor.userId, - source: "stalled_review_decision", + action: input.action, + status, + identifier: updated.identifier, + commentId: comment?.id ?? null, + authorUserId: comment ? input.actor.userId : null, + _previous: { status: lockedIssue.status }, }, }); - } - await logActivity(txDb, { - companyId: updated.companyId, - actorType: "user", - actorId: input.actor.userId, - runId: input.actor.runId ?? null, - action: "issue.stalled_review_decided", - entityType: "issue", - entityId: updated.id, - issueId: updated.id, - details: { - action: input.action, - status, - identifier: updated.identifier, - commentId: comment?.id ?? null, - authorUserId: comment ? input.actor.userId : null, - _previous: { status: lockedIssue.status }, - }, - }); - return { issue: updated, comment }; - }), + return { issue: updated, comment }; + }); + for (const publication of postCommitActivityPublications) publishActivity(publication); + await executeIssuePostCommitActions(db, postCommitIssueActions); + return result; + }, }; } diff --git a/server/src/vendor/paperclip-runner/index.ts b/server/src/vendor/paperclip-runner/index.ts index 03c4346e8c..3fb08b0758 100644 --- a/server/src/vendor/paperclip-runner/index.ts +++ b/server/src/vendor/paperclip-runner/index.ts @@ -19,6 +19,8 @@ export type { PaperclipSemanticToolCall, PaperclipSemanticToolDefinition, PaperclipSemanticToolResult, + PaperclipQuestionSet, + PaperclipRuntimeInputRequest, PrpEvent, PrpStructuredRunResult, PrpTerminalState, @@ -36,6 +38,8 @@ const runner = await import(sourceUrl.href) as RunnerModule; export const DurablePrpControlPlane = runner.DurablePrpControlPlane; export const PaperclipSemanticDispatcher = runner.PaperclipSemanticDispatcher; +export const parsePaperclipQuestionSet = runner.parsePaperclipQuestionSet; +export const parsePaperclipQuestionResponse = runner.parsePaperclipQuestionResponse; export const validatePrpEvent = runner.validatePrpEvent; export const validatePrpStructuredRunResult = runner.validatePrpStructuredRunResult; diff --git a/ui/src/api/activity.ts b/ui/src/api/activity.ts index 30698b12ff..b4d9d2baf8 100644 --- a/ui/src/api/activity.ts +++ b/ui/src/api/activity.ts @@ -5,6 +5,7 @@ export type { RunLivenessState } from "@paperclipai/shared"; export interface RunForIssue { runId: string; + runtimeMode?: "legacy" | "native"; status: string; agentId: string; adapterType: string; diff --git a/ui/src/api/heartbeats.ts b/ui/src/api/heartbeats.ts index c9f4312c5c..80b4fcb429 100644 --- a/ui/src/api/heartbeats.ts +++ b/ui/src/api/heartbeats.ts @@ -15,6 +15,7 @@ export interface RunLivenessFields { export interface ActiveRunForIssue { id: string; + runtimeMode?: "legacy" | "native"; status: string; invocationSource: string; triggerDetail: string | null; @@ -44,6 +45,7 @@ export interface ActiveRunForIssue { export interface LiveRunForIssue { id: string; + runtimeMode?: "legacy" | "native"; status: string; invocationSource: string; triggerDetail: string | null; diff --git a/ui/src/components/IssueThreadInteractionCard.test.tsx b/ui/src/components/IssueThreadInteractionCard.test.tsx index fff0a67dd3..9ccde6b047 100644 --- a/ui/src/components/IssueThreadInteractionCard.test.tsx +++ b/ui/src/components/IssueThreadInteractionCard.test.tsx @@ -354,6 +354,32 @@ describe("IssueThreadInteractionCard", () => { expect(host.querySelectorAll('[role="radio"]').length).toBeGreaterThan(0); }); + it("keeps a closed native select set closed while preserving direct-question defaults", () => { + const closed = { + ...pendingAskUserQuestionsInteraction, + payload: { + ...pendingAskUserQuestionsInteraction.payload, + questions: pendingAskUserQuestionsInteraction.payload.questions.map((question) => ({ + ...question, + allowOther: false, + })), + }, + }; + const host = renderCard({ interaction: closed, onSubmitInteractionAnswers: vi.fn() }); + expect(Array.from(host.querySelectorAll("button")).some((button) => button.textContent === "Other")) + .toBe(false); + + act(() => root?.unmount()); + host.remove(); + root = null; + const legacy = renderCard({ + interaction: pendingAskUserQuestionsInteraction, + onSubmitInteractionAnswers: vi.fn(), + }); + expect(Array.from(legacy.querySelectorAll("button")).some((button) => button.textContent === "Other")) + .toBe(true); + }); + it("only shows question cancellation when a cancel handler is wired", () => { const withoutHandler = renderCard({ interaction: pendingAskUserQuestionsInteraction, diff --git a/ui/src/components/IssueThreadInteractionCard.tsx b/ui/src/components/IssueThreadInteractionCard.tsx index 1578f29a67..44c6a57db1 100644 --- a/ui/src/components/IssueThreadInteractionCard.tsx +++ b/ui/src/components/IssueThreadInteractionCard.tsx @@ -1310,7 +1310,7 @@ function AskUserQuestionsCard({ * free-text option so the card never shows two ways to type an * answer (PAP-419). */} - {hasFreeTextOption ? null : ( + {hasFreeTextOption || question.allowOther === false ? null : ( <>