fix(runner): settle stopped native questions without stranded recovery

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/__tests__/issue-recovery-actions.test.ts

@@ -820,8 +820,9 @@ describeEmbeddedPostgres("issue recovery actions", () => {
     },
   );
 
-  it("stands down while the latest run was cancelled by a board operator", async () => {
+  it.each(["idle", "paused", "error"] as const)("stands down after a board Stop when the assignee is %s", async (agentStatus) => {
     const { companyId, coderId, sourceIssueId } = await seedCompany();
+    await db.update(agents).set({ status: agentStatus }).where(eq(agents.id, coderId));
     await db.insert(heartbeatRuns).values({
       id: randomUUID(),
       companyId,
@@ -843,6 +844,9 @@ describeEmbeddedPostgres("issue recovery actions", () => {
     expect(result.operatorCancelExempted).toBe(1);
     expect(await db.select().from(issueRecoveryActions)).toHaveLength(0);
     expect(enqueueWakeup).not.toHaveBeenCalled();
+    const [issue] = await db.select().from(issues).where(eq(issues.id, sourceIssueId));
+    expect(issue.status).toBe("in_progress");
+    expect(issue.assigneeAgentId).toBe(coderId);
   });
 
   it("stands down after an operator interrupt cancellation", async () => {

server/src/services/issue-thread-interactions.ts

@@ -4723,6 +4723,50 @@ export function issueThreadInteractionService(
       return expired;
     },
 
+    // A native callback is no longer answerable after its owned cancellation.
+    // Historical questions and answers committed concurrently retain their
+    // normal lifecycle. The bridge supplies the validated native identity.
+    expireCancelledNativeQuestion: async (input: {
+      companyId: string; issueId: string; sourceRunId: string; requestId: string;
+      idempotencyKey: string; agentId: string;
+    }) => {
+      const expired = await db.transaction(async (tx) => {
+        // Answering locks the task before the card. Use the same order so
+        // cancellation cannot deadlock a concurrent human response.
+        const [issue] = await tx.select({ id: issues.id }).from(issues).where(and(
+          eq(issues.id, input.issueId), eq(issues.companyId, input.companyId),
+        )).for("no key update").limit(1);
+        if (!issue) throw notFound("Issue not found");
+        const [row] = await tx.update(issueThreadInteractions).set({
+          status: "expired",
+          result: { version: 1, cancelled: true, cancellationReason: "Native question cancelled", answers: [], summaryMarkdown: null },
+          resolvedByRunId: input.sourceRunId, resolvedAt: new Date(), updatedAt: new Date(),
+        }).where(and(
+          eq(issueThreadInteractions.companyId, input.companyId),
+          eq(issueThreadInteractions.issueId, input.issueId),
+          eq(issueThreadInteractions.sourceRunId, input.sourceRunId),
+          eq(issueThreadInteractions.idempotencyKey, input.idempotencyKey),
+          eq(issueThreadInteractions.kind, "ask_user_questions"),
+          eq(issueThreadInteractions.status, "pending"),
+          sql`${issueThreadInteractions.payload}->>'runtimeRequestId' = ${input.requestId}`,
+          sql`${issueThreadInteractions.payload}->'questionSet'->>'schema' = 'paperclip.question_set.v1'`,
+        )).returning();
+        if (!row) return null;
+        const interaction = hydrateInteraction(row) as AskUserQuestionsInteraction;
+        await enqueueTerminalIssueInteractionChatPublications(tx as unknown as Db, interaction);
+        await touchIssue(tx, input.issueId);
+        await logActivity(tx as unknown as Db, {
+          companyId: input.companyId, actorType: "system", actorId: "native-runtime",
+          agentId: input.agentId, runId: input.sourceRunId, action: "issue.thread_interaction_expired",
+          entityType: "issue", entityId: input.issueId,
+          details: { interactionId: row.id, interactionKind: row.kind, interactionStatus: "expired", reason: "native_question_cancelled" },
+        });
+        return interaction;
+      });
+      if (expired) await emitInteractionResolvedTelemetry(db, expired);
+      return expired;
+    },
+
     expireConnectionIntentsForOwnershipChange: async (issue: { id: string; companyId: string }) => {
       const expired = await db.update(issueThreadInteractions).set({
         status: "expired", result: { version: 1, outcome: "expired", reason: "The task assignment changed" },

server/src/services/native-runtime/local-native-question-bridge.ts

@@ -46,8 +46,6 @@ export function createLocalNativeQuestionBridge(input: {
       await flushNativeQuestionResponses(input.db, input.binding.runId);
     },
     async observe(event: PrpEvent) {
-      const request = event.payload.request as Record<string, unknown> | undefined;
-      if (event.eventType !== "runtime_request.created" || request?.type !== "input") return;
       await projectNativeRuntimeRequest({ db: input.db, binding: input.binding, event });
     },
   };

server/src/services/native-runtime/native-question-bridge.test.ts

@@ -236,6 +236,88 @@ describeEmbeddedPostgres("native question bridge", () => {
     } } };
   }
 
+  async function savedRuntimeQuestion() {
+    const event = runtimeRequestEvent();
+    await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq: 1,
+      eventType: event.eventType, sourceSeq: event.sourceSeq, sourceInstanceId: runnerInstanceId,
+      sourceEventId: event.sourceEventId, payload: { prpEvent: event } });
+    const interaction = await projectNativeRuntimeRequest({ db, binding: binding(), event });
+    const cancelled: PrpEvent = { ...event, sourceEventId: "runtime-question-cancelled", sourceSeq: 2,
+      eventType: "runtime_request.cancelled", payload: { requestId: "request-1", requestKind: "runtime",
+        requestType: "input", turnId: event.turnId, itemId: event.itemId, action: "cancel", reason: "turn_terminal" } };
+    return { interaction: interaction!, cancelled };
+  }
+
+  it("expires only the cancelled native callback, idempotently, and rejects a late answer", async () => {
+    await seed();
+    const { interaction, cancelled } = await savedRuntimeQuestion();
+    const unrelated = await issueThreadInteractionService(db).create({ id: issueId, companyId }, {
+      kind: "ask_user_questions", payload: interaction.payload, sourceRunId: runId,
+      idempotencyKey: "ordinary-historical-question", continuationPolicy: "none",
+    }, { agentId, runId }, { supersedePendingSiblingInteractions: false });
+    const bridge = createLocalNativeQuestionBridge({ db, binding: binding(), resolve: vi.fn() });
+    await bridge.observe(cancelled);
+    await bridge.observe(cancelled);
+    const [expired] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id));
+    expect(expired).toMatchObject({ status: "expired", result: { cancelled: true, answers: [] } });
+    const [historical] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, unrelated.id));
+    expect(historical.status).toBe("pending");
+    const [issue] = await db.select().from(issues).where(eq(issues.id, issueId));
+    expect(issue.status).toBe("in_progress");
+    const activity = await db.select().from(activityLog).where(eq(activityLog.action, "issue.thread_interaction_expired"));
+    expect(activity).toHaveLength(1);
+    await expect(issueThreadInteractionService(db).answerQuestions({ id: issueId, companyId }, interaction.id,
+      { answers: [{ questionId: "color", optionIds: ["blue"] }] }, { userId: "operator-1" })).rejects.toMatchObject({ status: 409 });
+    expect(await db.select().from(issueQuestionResponseDeliveries)).toHaveLength(0);
+  });
+
+  it.each(["company", "issue", "agent", "run", "session", "runner", "turn", "item", "request"])("rejects cancellation with foreign %s identity", async (field) => {
+    await seed();
+    const { interaction, cancelled } = await savedRuntimeQuestion();
+    const owner = binding();
+    if (field === "company") owner.companyId = randomUUID();
+    if (field === "issue") owner.issueId = randomUUID();
+    if (field === "agent") owner.agentId = randomUUID();
+    if (field === "run") cancelled.runId = randomUUID();
+    if (field === "session") cancelled.normalizedSessionId = randomUUID();
+    if (field === "runner") cancelled.sourceInstanceId = randomUUID();
+    if (field === "turn") cancelled.turnId = "foreign-turn";
+    if (field === "item") cancelled.itemId = "foreign-item";
+    if (field === "request") cancelled.payload.requestId = "foreign-request";
+    await expect(projectNativeRuntimeRequest({ db, binding: owner, event: cancelled })).rejects.toThrow(/native_runtime_/);
+    const [current] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id));
+    expect(current.status).toBe("pending");
+  });
+
+  it("preserves a human answer that committed before native cancellation", async () => {
+    await seed();
+    const { interaction, cancelled } = await savedRuntimeQuestion();
+    await issueThreadInteractionService(db).answerQuestions({ id: issueId, companyId }, interaction.id,
+      { answers: [{ questionId: "color", optionIds: ["blue"] }] }, { userId: "operator-1" });
+    await projectNativeRuntimeRequest({ db, binding: binding(), event: cancelled });
+    const [current] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id));
+    expect(current).toMatchObject({ status: "answered", result: { answers: [{ questionId: "color", optionIds: ["blue"] }] } });
+  });
+
+  it("settles concurrent native cancellation and human answer without replay or a deadlock", async () => {
+    await seed();
+    const { interaction, cancelled } = await savedRuntimeQuestion();
+    const [stop, answer] = await Promise.allSettled([
+      projectNativeRuntimeRequest({ db, binding: binding(), event: cancelled }),
+      issueThreadInteractionService(db).answerQuestions({ id: issueId, companyId }, interaction.id,
+        { answers: [{ questionId: "color", optionIds: ["blue"] }] }, { userId: "operator-1" }),
+    ]);
+    expect(stop.status).toBe("fulfilled");
+    const [current] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id));
+    if (answer.status === "fulfilled") {
+      expect(current.status).toBe("answered");
+      expect(current.result).toMatchObject({ answers: [{ questionId: "color", optionIds: ["blue"] }] });
+    } else {
+      expect(answer.reason).toMatchObject({ status: 409 });
+      expect(current).toMatchObject({ status: "expired", result: { cancelled: true, answers: [] } });
+    }
+  });
+
   it("commits and acknowledges an ACP permission, then queues only an admin's exact turn-bound decision", async () => {
     await seed();
     const event = permissionRequestEvent();

server/src/services/native-runtime/native-question-bridge.ts

@@ -1,7 +1,7 @@
 import { and, eq, inArray, sql } from "drizzle-orm";
 
 import type { Db } from "@paperclipai/db";
-import { heartbeatRuns, issueThreadInteractions } from "@paperclipai/db";
+import { heartbeatRuns, heartbeatRunEvents, issueThreadInteractions } from "@paperclipai/db";
 import type {
   AskUserQuestionsInteraction,
   RespondIssueThreadInteraction,
@@ -118,7 +118,7 @@ export async function projectNativeRuntimeRequest(input: {
   binding: Pick<NativeRunStoreBinding, "companyId" | "issueId" | "runId" | "agentId" | "normalizedSessionId" | "runnerSourceInstanceId">;
   event: PrpEvent;
 }): Promise<AskUserQuestionsInteraction | null> {
-  if (input.event.eventType !== "runtime_request.created") return null;
+  if (!["runtime_request.created", "runtime_request.cancelled"].includes(input.event.eventType)) return null;
   if (
     input.event.runId !== input.binding.runId
     || input.event.normalizedSessionId !== input.binding.normalizedSessionId
@@ -126,6 +126,47 @@ export async function projectNativeRuntimeRequest(input: {
   ) {
     throw new Error("native_runtime_request_binding_mismatch");
   }
+  if (input.event.eventType === "runtime_request.cancelled") {
+    const outcome = record(input.event.payload)!;
+    // Permission cards have their own privileged resolver. A provider-loss
+    // expiry is a durable handoff, not cancellation of a historical question.
+    if (outcome.requestKind !== "runtime" || outcome.requestType !== "input") return null;
+    if (outcome.action !== "cancel" || typeof outcome.requestId !== "string"
+      || !REQUEST_ID_PATTERN.test(outcome.requestId) || typeof input.event.turnId !== "string"
+      || outcome.turnId !== input.event.turnId || (outcome.itemId ?? null) !== (input.event.itemId ?? null)
+      || ["response", "answers", "answer", "replay"].some((key) => Object.hasOwn(outcome, key))) {
+      throw new Error("native_runtime_cancellation_invalid");
+    }
+    const [run] = await input.db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
+      eq(heartbeatRuns.id, input.binding.runId), eq(heartbeatRuns.companyId, input.binding.companyId),
+      eq(heartbeatRuns.nativeIssueId, input.binding.issueId), eq(heartbeatRuns.agentId, input.binding.agentId),
+      eq(heartbeatRuns.nativeSessionId, input.binding.normalizedSessionId),
+      eq(heartbeatRuns.runtimeMode, "native"),
+    )).limit(1);
+    if (!run) throw new Error("native_runtime_cancellation_run_mismatch");
+    // Match the committed creation, including its turn and item. A run can
+    // contain several callbacks; its request ID alone must not retire another.
+    const [created] = await input.db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and(
+      eq(heartbeatRunEvents.companyId, input.binding.companyId),
+      eq(heartbeatRunEvents.runId, input.binding.runId),
+      eq(heartbeatRunEvents.eventType, "runtime_request.created"),
+      eq(heartbeatRunEvents.sourceInstanceId, input.binding.runnerSourceInstanceId),
+      sql`${heartbeatRunEvents.payload}->'prpEvent'->'payload'->'request'->>'requestId' = ${outcome.requestId}`,
+      sql`${heartbeatRunEvents.sourceSeq} < ${input.event.sourceSeq}`,
+    )).limit(1);
+    const original = record(record(created?.payload)?.prpEvent);
+    const originalRequest = record(record(original?.payload)?.request);
+    if (!original || original.normalizedSessionId !== input.event.normalizedSessionId
+      || original.turnId !== input.event.turnId || (original.itemId ?? null) !== (input.event.itemId ?? null)
+      || originalRequest?.type !== "input" || originalRequest.requestKind !== "runtime") {
+      throw new Error("native_runtime_cancellation_creation_mismatch");
+    }
+    return issueThreadInteractionService(input.db).expireCancelledNativeQuestion({
+      companyId: input.binding.companyId, issueId: input.binding.issueId, agentId: input.binding.agentId,
+      sourceRunId: input.binding.runId, requestId: outcome.requestId,
+      idempotencyKey: `${QUESTION_KEY_PREFIX}${input.binding.runId}:${outcome.requestId}`,
+    });
+  }
   const request = record(record(input.event.payload)?.request);
   if (
     !request

server/src/services/native-runtime/runner-prp-coordinator.ts

@@ -282,9 +282,7 @@ export function runnerPrpCoordinator(
         connectionLeaseTtlMs,
         onCommittedEvent: async (event) => {
           await nativeStore.appendEvent(event);
-          if (event.eventType === "runtime_request.created") {
-            await projectNativeRuntimeRequest({ db, binding: storeBinding, event });
-          }
+          await projectNativeRuntimeRequest({ db, binding: storeBinding, event });
           await nativeStore.reconcileTerminalEvent(event);
           if (event.eventType === "run.terminal") {
             const stored = await nativeStore.readCompletedRun();

server/src/services/recovery/service.ts

@@ -4493,6 +4493,14 @@ export function recoveryService(
         }
       }
 
+      // Stop remains authoritative even when cancellation accounting or an
+      // unrelated operator action has since paused the assignee. Check it
+      // before an uninvokable agent can be escalated as stranded work.
+      if (issue.status !== "in_review" && isOperatorCancelledRun(latestRun, agentId)) {
+        result.operatorCancelExempted += 1;
+        continue;
+      }
+
       const agent = await getAgent(agentId);
       const agentInvokable =
         agent && agent.companyId === issue.companyId