From bc9ba7cd26b104abb65720fb753cda26435ccad8 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 29 Aug 2026 19:26:20 -0500 Subject: [PATCH] feat(runner): project native runs into task threads (#12321) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The experimental Paperclip Runner can execute a guarded Codex run and persist provider-neutral events. > - The task page still reads direct-adapter transcripts and cannot present those native events. > - Structured runner questions must also use the existing task interaction experience. > - Runtime selection must use the persisted run mode, not an adapter name or a current feature flag. > - This pull request projects native events and questions into the existing task thread. > - Direct adapters keep their existing transcript, composer, interaction, and finalization paths. > - The benefit is a complete native Codex task thread without a behavior change for existing adapters. ## Linked Issues or Issue Description Refs #12202. This pull request replaces that stale implementation on current `master`. **What happened?** The server persists native runner events and structured input requests. The task page only consumes direct-adapter transcripts. A native run therefore cannot present a complete transcript, usage, or question flow through the normal task experience. **Expected behavior** Native runs project persisted provider-neutral events into the existing task thread. Native structured questions use the existing interaction card. Direct adapters retain their current behavior. **Steps to reproduce** 1. Enable the experimental runner. 2. Start a native Codex run that emits progress, usage, a structured question, and a final reply. 3. Open the task page. 4. Observe that the direct-adapter transcript path cannot project the native event records. **Paperclip version or commit** `master` at `67f9867bc`. ## What Changed - Add the canonical structured-question validator and shared contract exports. - Materialize native input requests as existing task interactions. - Validate native answers and deliver them through the durable question-response receipt. - Resume the original PRP request with an idempotent `request.resolve` command. - Project native messages, tool activity, cumulative usage, and final replies into the existing transcript model. - Propagate persisted `runtimeMode` to the task page and select native handling only for `runtimeMode: "native"`. - Expire pending interactions through the shared issue service on every terminal transition, including decisions, stalled reviews, tree control, and pipeline retry cleanup. - Queue native run cancellation while a transaction is open and execute it only after the owning transaction commits. - Keep nonterminal and non-runner issue paths on their existing service call shapes and behavior. ## Verification - `pnpm --filter @paperclipai/server typecheck` — passed, including the Rust runner release build and protocol/catalog drift gates. - Focused native-thread and lifecycle suites — 18 files and 481 tests passed during review. - `issue-execution-policy-routes.test.ts` — 19/19 passed after the final transactional-queue expectation update. - `issue-agent-mutation-ownership-routes.test.ts` — 87/87 passed in the final isolated compatibility rerun. - GitHub Actions — policy, build, canary, typecheck/release registry, 5 serialized server shards, 8 general-test shards, 3 browser shards, and both aggregate gates passed on `7793f3193`. - Security — Snyk, Socket Project Report, Socket PR Alerts, and Superagent passed. - Greptile — 5/5 on `7793f3193`; all actionable review threads resolved. - `git diff --check` — passed. - Diff against `master`: 44 files. ## Compatibility Boundary - Native transcript polling only runs when the persisted run reports `runtimeMode: "native"`. - Missing or legacy runtime modes continue through `useLiveRunTranscripts`. - Legacy questions keep the existing optional free-text choice. - Native closed select sets can suppress that legacy fallback. - Terminal cleanup uses the same issue service for native and legacy interactions; only a bound native question schedules a native run cancellation. - Native cancellation happens after transaction commit, so failed or rolled-back writes do not cancel a still-valid run. - The durable delivery service checks the original native request before it considers a continuation run. - This pull request adds no migration, dependency, workflow, manifest, or lockfile change. ## Risks The main risk is routing a direct-adapter task through native handling or changing terminal issue behavior. The implementation selects the native path only from persisted runtime facts, retains the existing nonterminal call shape, and schedules native cancellation only for a validated bound native question after commit. Focused and repository-wide tests cover both paths. Native requests remain bound to the company, issue, run, and agent; answers are validated, durable, and idempotent across reconnects. ## Model Used OpenAI Codex, GPT-5 family. The client does not expose the exact deployment ID or context window. Agentic reasoning, tool use, and code execution were enabled. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change and contains no internal Paperclip ticket id or instance-derived details - [x] I have run the affected local tests and they pass - [x] I have added or updated tests where applicable - [x] I have updated the compatibility notes for this change - [x] I have considered and documented risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I addressed all Greptile and reviewer comments before requesting merge --- .../paperclip-runner/protocol/manifest.json | 2 +- .../protocol/schemas/usage.schema.json | 1 + .../crates/runner-core/src/provider_events.rs | 31 +- .../src/protocol/generated/schema-bundle.ts | 3 + packages/shared/src/index.ts | 1 + .../src/issue-thread-interactions.test.ts | 39 ++ packages/shared/src/types/issue.ts | 2 + packages/shared/src/validators/index.ts | 1 + packages/shared/src/validators/issue.ts | 69 ++- ...ue-agent-mutation-ownership-routes.test.ts | 2 + .../issue-comment-reopen-routes.test.ts | 7 + .../issue-execution-policy-routes.test.ts | 4 + .../issue-thread-interaction-routes.test.ts | 123 ++++- .../issue-tree-control-service.test.ts | 18 +- ...issue-update-comment-wakeup-routes.test.ts | 41 +- .../src/__tests__/pipelines-service.test.ts | 33 ++ .../question-response-delivery.test.ts | 50 ++ .../server-startup-feedback-export.test.ts | 7 + server/src/index.ts | 2 + server/src/routes/agents.ts | 2 + server/src/routes/issues.ts | 142 ++++- server/src/services/activity.ts | 1 + server/src/services/decisions.ts | 14 +- server/src/services/heartbeat.ts | 52 ++ .../src/services/issue-thread-interactions.ts | 81 +-- server/src/services/issue-tree-control.ts | 51 +- server/src/services/issues.ts | 73 +++ .../native-question-bridge.test.ts | 490 ++++++++++++++++++ .../native-runtime/native-question-bridge.ts | 444 ++++++++++++++++ .../native-runtime/runner-prp-coordinator.ts | 47 +- server/src/services/pipelines.ts | 38 +- .../services/question-response-delivery.ts | 56 +- .../src/services/stalled-review-decisions.ts | 165 +++--- server/src/vendor/paperclip-runner/index.ts | 4 + ui/src/api/activity.ts | 1 + ui/src/api/heartbeats.ts | 2 + .../IssueThreadInteractionCard.test.tsx | 26 + .../components/IssueThreadInteractionCard.tsx | 2 +- ui/src/components/TaskChatThread.test.tsx | 52 +- ui/src/components/TaskChatThread.tsx | 22 +- .../task-chat/TaskChatUsageReadout.tsx | 27 +- .../components/task-chat/task-chat-model.ts | 3 + .../task-chat/transcript-adapter.test.ts | 94 ++++ .../task-chat/transcript-adapter.ts | 41 +- .../transcript/native-run-events.test.ts | 253 +++++++++ .../transcript/native-run-events.ts | 217 ++++++++ .../transcript/useLiveRunTranscripts.ts | 1 + .../transcript/useNativeRunTranscripts.ts | 100 ++++ ui/src/lib/issue-chat-messages.ts | 1 + 49 files changed, 2686 insertions(+), 252 deletions(-) create mode 100644 server/src/services/native-runtime/native-question-bridge.test.ts create mode 100644 server/src/services/native-runtime/native-question-bridge.ts create mode 100644 ui/src/components/transcript/native-run-events.test.ts create mode 100644 ui/src/components/transcript/native-run-events.ts create mode 100644 ui/src/components/transcript/useNativeRunTranscripts.ts 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 : ( <>