fix(worker): prevent manual dispatch run leakage

apps/worker/src/db/queries/runs-read.test.ts

@@ -19,6 +19,7 @@ import {
   isRunRecordedFailed,
   hasDurableRunPublication,
   fetchRunModels,
+  findLiveRunClaimByRunId,
 } from "./runs-read.js";
 
 /** Minimal fixture: attributeRunModel only reads `.nodeId` / `.manifest.model.id`. */
@@ -537,6 +538,26 @@ describe("isRunRecordedFailed", () => {
   });
 });
 
+describe("findLiveRunClaimByRunId", () => {
+  it("returns the manual ticket metadata needed for atomic cancellation withdrawal", async () => {
+    await db.insert(activeRuns).values({
+      subjectKey: "ticket:jira:AIW-274",
+      ticketKey: "AIW-274",
+      ownerToken: "owner-manual",
+      runId: "wrun_manual",
+      state: "bound",
+      runKind: "manual_ticket",
+    });
+
+    await expect(findLiveRunClaimByRunId(db, "wrun_manual")).resolves.toEqual({
+      subjectKey: "ticket:jira:AIW-274",
+      ticketKey: "AIW-274",
+      ownerToken: "owner-manual",
+      kind: "manual_ticket",
+    });
+  });
+});
+
 describe("hasDurableRunPublication", () => {
   it("requires publication evidence on the exact run", async () => {
     await seed({ runId: "wrun_empty", status: "running" });

apps/worker/src/db/queries/runs-read.ts

@@ -11,6 +11,7 @@ import type {
   WorkflowRow,
 } from "@shared/contracts";
 import type { Db } from "../client.js";
+import type { RunKind } from "../../adapters/run-registry/types.js";
 import { activeRuns, workflowOwnedBranches, workflowRuns } from "../schema.js";
 import { attributeRunModel } from "../../lib/overview/attribute-run-model.js";
 
@@ -216,20 +217,31 @@ export async function hasDurableRunPublication(db: Db, runId: string): Promise<b
  * claims (one row per active subject) and has no index on run_id, but the table
  * is tiny, so the scan is cheap. Returns the subject and its exact owner token
  * so an operator cancel can drive cancelSubjectRunDetailed for a run that has no
- * ticket at all (a webhook or schedule trigger). Null when no live claim carries
- * this run id: the run is already terminal or unknown, and the caller falls back
- * to workflow_runs.
+ * ticket at all (a webhook or schedule trigger). The ticket and kind let manual
+ * ticket cancellation withdraw its AI-column enrolment before release. Null
+ * when no live claim carries this run id: the run is already terminal or
+ * unknown, and the caller falls back to workflow_runs.
  */
 export async function findLiveRunClaimByRunId(
   db: Db,
   runId: string,
-): Promise<{ subjectKey: string; ownerToken: string } | null> {
+): Promise<{
+  subjectKey: string;
+  ticketKey: string | null;
+  ownerToken: string;
+  kind: RunKind;
+} | null> {
   const [row] = await db
-    .select({ subjectKey: activeRuns.subjectKey, ownerToken: activeRuns.ownerToken })
+    .select({
+      subjectKey: activeRuns.subjectKey,
+      ticketKey: activeRuns.ticketKey,
+      ownerToken: activeRuns.ownerToken,
+      kind: activeRuns.runKind,
+    })
     .from(activeRuns)
     .where(eq(activeRuns.runId, runId))
     .limit(1);
-  return row ?? null;
+  return row ? { ...row, kind: row.kind as RunKind } : null;
 }
 
 /**

apps/worker/src/lib/cancel-run.test.ts

@@ -19,6 +19,14 @@ const state = vi.hoisted(() => ({
   warn: vi.fn(),
 }));
 
+vi.mock("../../env.js", () => ({
+  env: {
+    COLUMN_AI: "AI",
+    COLUMN_BACKLOG: "Backlog",
+    JIRA_BACKLOG_TRANSITION_ID: undefined,
+  },
+}));
+
 vi.mock("workflow/api", () => ({ getRun: state.getRun }));
 vi.mock("workflow/runtime", () => ({
   getWorld: () => ({ steps: { list: state.listSteps } }),
@@ -33,7 +41,10 @@ vi.mock("../clarifications/store.js", () => ({
 vi.mock("../approvals/store.js", () => ({
   retireApprovalCancellation: state.retireApproval,
 }));
-vi.mock("./ticket-transition.js", () => ({ moveTicketForRun: state.moveTicket }));
+vi.mock("./ticket-transition.js", () => ({
+  moveTicketForRun: state.moveTicket,
+  withdrawTicketFromAiForRun: state.moveTicket,
+}));
 vi.mock("./telemetry/run-telemetry.js", () => ({
   recordRunStatusReason: state.recordStatusReason,
   markRunBlockedOnCancel: state.markBlockedOnCancel,
@@ -344,6 +355,44 @@ describe("cancelRunById", () => {
     );
   });
 
+  it("withdraws a cancelled manual ticket before releasing its claim", async () => {
+    state.findLiveClaim.mockResolvedValue({
+      subjectKey: "ticket:jira:PROJ-1",
+      ticketKey: "PROJ-1",
+      ownerToken: "owner-a",
+      kind: "manual_ticket",
+    });
+    const runRegistry = registry(active({ kind: "manual_ticket" }));
+    const issueTracker = {} as IssueTrackerAdapter;
+
+    await expect(
+      cancelRunById(outerDb, "run-1", {
+        actorLabel: "operator kate",
+        runRegistry,
+        issueTracker,
+      }),
+    ).resolves.toEqual({
+      outcome: "cancelled",
+      subjectKey: "ticket:jira:PROJ-1",
+    });
+    expect(state.moveTicket).toHaveBeenCalledWith({
+      db: outerDb,
+      issueTracker,
+      ticketKey: "PROJ-1",
+      aiColumn: expect.any(String),
+      target: expect.any(String),
+      owner: expect.objectContaining({
+        subjectKey: "ticket:jira:PROJ-1",
+        ownerToken: "owner-a",
+        runId: "run-1",
+      }),
+      requiredOwnerState: "cancelling",
+    });
+    expect(state.moveTicket.mock.invocationCallOrder[0]).toBeLessThan(
+      vi.mocked(runRegistry.releaseCancellation).mock.invocationCallOrder[0]!,
+    );
+  });
+
   // The irreversible cancel already landed and the claim is released, so a
   // transient settle failure must never surface as a throw (E4 would map it to
   // 500) or flip the outcome; the cron backstops the row, like the park sibling.

apps/worker/src/lib/cancel-run.ts

@@ -188,6 +188,7 @@ export interface CancelRunByIdResult {
 export interface CancelRunByIdDeps {
   actorLabel: string;
   runRegistry: RunRegistryAdapter;
+  issueTracker?: IssueTrackerAdapter;
 }
 
 /**
@@ -217,13 +218,48 @@ export async function cancelRunById(
   const claim = await findLiveRunClaimByRunId(db, runId);
   if (claim) {
     const reason = `cancelled by ${actorLabel}`;
-    const result = await cancelSubjectRunDetailed(
-      claim.subjectKey,
-      { ownerToken: claim.ownerToken, runId },
-      runRegistry,
-      undefined,
-      reason,
-    );
+    if (claim.kind === "manual_ticket" && (!claim.ticketKey || !opts.issueTracker)) {
+      logger.warn(
+        { subjectKey: claim.subjectKey, runId },
+        "cancel_manual_ticket_withdrawal_unavailable",
+      );
+      return { outcome: "unconfirmed", subjectKey: claim.subjectKey };
+    }
+    const result = claim.kind === "manual_ticket"
+      ? await cancelOwnedSubject(
+          claim.subjectKey,
+          { ownerToken: claim.ownerToken, runId },
+          runRegistry,
+          undefined,
+          async (owner) => {
+            const [{ env }, { withdrawTicketFromAiForRun }] = await Promise.all([
+              import("../../env.js"),
+              import("./ticket-transition.js"),
+            ]);
+            await withdrawTicketFromAiForRun({
+              db,
+              issueTracker: opts.issueTracker!,
+              ticketKey: claim.ticketKey!,
+              aiColumn: env.COLUMN_AI,
+              target: env.JIRA_BACKLOG_TRANSITION_ID
+                ? {
+                    name: env.COLUMN_BACKLOG,
+                    transitionId: env.JIRA_BACKLOG_TRANSITION_ID,
+                  }
+                : env.COLUMN_BACKLOG,
+              owner,
+              requiredOwnerState: "cancelling",
+            });
+          },
+          reason,
+        )
+      : await cancelSubjectRunDetailed(
+          claim.subjectKey,
+          { ownerToken: claim.ownerToken, runId },
+          runRegistry,
+          undefined,
+          reason,
+        );
     // alreadyTerminal implies cancelled, so it must be checked first: the run
     // reached a terminal Workflow status on its own and keeps that outcome, so
     // no status is written (only the lingering claim was released).

apps/worker/src/lib/dispatch.test.ts

@@ -295,9 +295,19 @@ describe("dispatchTicket owner reservation", () => {
     expect(result).toEqual({ started: false, reason: "wrong_project_key" });
   });
 
-  it("returns already_claimed when the subject reservation loses", async () => {
-    const result = await dispatchTicket("PROJ-42", adapters(registry({ reserveResult: false })), 3);
+  it("does not auto-enrol a subject already claimed by a manual dispatch", async () => {
+    const manualClaim = entry({
+      subjectKey: "ticket:jira:PROJ-42",
+      ticketKey: "PROJ-42",
+      ownerToken: "owner:manual",
+      runId: "run-manual",
+      kind: "manual_ticket",
+    });
+    const connected = adapters(registry({ initial: [manualClaim] }));
+    const result = await dispatchTicket("PROJ-42", connected, 3);
     expect(result).toEqual({ started: false, reason: "already_claimed" });
+    expect(connected.issueTracker.fetchTicket).not.toHaveBeenCalled();
+    expect(mockStart).not.toHaveBeenCalled();
   });
 
   it("returns at_capacity without reserving when bound capacity is full", async () => {

apps/worker/src/lib/reconcile.test.ts

@@ -26,6 +26,7 @@ const mockHasDurableRunPublication = vi.fn();
 const mockDb = {} as Db;
 const mockStopSandboxesByIds = vi.fn();
 const mockListWorkflowSteps = vi.fn();
+const mockAssertActiveRunOwnerState = vi.hoisted(() => vi.fn());
 vi.mock("workflow/api", () => ({ getRun: (...args: any[]) => mockGetRun(...args) }));
 vi.mock("workflow/runtime", () => ({
   getWorld: () => ({
@@ -51,6 +52,9 @@ vi.mock("./run-start-lifecycle.js", () => ({
 vi.mock("../sandbox/stop-ticket-sandboxes.js", () => ({
   stopSandboxesByIds: (...args: any[]) => mockStopSandboxesByIds(...args),
 }));
+vi.mock("./active-run-owner.js", () => ({
+  assertActiveRunOwnerState: (...args: any[]) => mockAssertActiveRunOwnerState(...args),
+}));
 
 function entry(overrides: Partial<ActiveRunEntry> = {}): ActiveRunEntry {
   return {
@@ -135,6 +139,7 @@ describe("reconcileRuns owner-CAS recovery", () => {
       cursor: null,
       hasMore: false,
     });
+    mockAssertActiveRunOwnerState.mockResolvedValue(undefined);
   });
 
   it("leaves a fresh unbound reservation for its candidate", async () => {
@@ -349,6 +354,72 @@ describe("reconcileRuns owner-CAS recovery", () => {
     expect(onTicketCancelled).not.toHaveBeenCalled();
   });
 
+  it.each(["Review", "Done"])(
+    "releases a terminal manual ticket without overwriting live Jira %s from a stale AI snapshot",
+    async (liveStatus) => {
+      const manual = entry({ kind: "manual_ticket" });
+      const runRegistry = registry([manual]);
+      const tracker = issueTracker(liveStatus);
+      mockGetRun.mockReturnValue({ status: Promise.resolve("completed") });
+      const onReleased = vi.fn();
+      const { reconcileRuns } = await import("./reconcile.js");
+
+      await expect(
+        reconcileRuns(
+          new Set(["PROJ-1"]),
+          runRegistry,
+          tracker,
+          undefined,
+          onReleased,
+          undefined,
+          mockDb,
+        ),
+      ).resolves.toEqual({ cancelled: 0, cleaned: 1 });
+      expect(mockCancelRunDetailed).not.toHaveBeenCalled();
+      expect(tracker.fetchTicket).toHaveBeenCalledOnce();
+      expect(tracker.moveTicket).not.toHaveBeenCalled();
+      expect(mockAssertActiveRunOwnerState).toHaveBeenCalledWith(
+        mockDb,
+        manual,
+        "bound",
+      );
+      expect(runRegistry.release).toHaveBeenCalledWith(
+        manual.subjectKey,
+        manual.ownerToken,
+        manual.runId,
+      );
+      expect(onReleased).toHaveBeenCalledWith(manual.subjectKey);
+    },
+  );
+
+  it("retains a manual claim when stale-snapshot withdrawal cannot be confirmed", async () => {
+    const manual = entry({ kind: "manual_ticket" });
+    const runRegistry = registry([manual]);
+    const tracker = issueTracker("AI");
+    const moveError = new Error("response lost");
+    vi.mocked(tracker.fetchTicket)
+      .mockResolvedValueOnce({ trackerStatus: "AI" } as never)
+      .mockResolvedValueOnce({ trackerStatus: "AI" } as never);
+    vi.mocked(tracker.moveTicket).mockRejectedValue(moveError);
+    mockGetRun.mockReturnValue({ status: Promise.resolve("completed") });
+    const { reconcileRuns } = await import("./reconcile.js");
+
+    await expect(
+      reconcileRuns(
+        new Set(["PROJ-1"]),
+        runRegistry,
+        tracker,
+        undefined,
+        undefined,
+        undefined,
+        mockDb,
+      ),
+    ).resolves.toEqual({ cancelled: 0, cleaned: 0 });
+    expect(tracker.moveTicket).toHaveBeenCalledWith("PROJ-1", "Backlog");
+    expect(tracker.fetchTicket).toHaveBeenCalledTimes(2);
+    expect(runRegistry.release).not.toHaveBeenCalled();
+  });
+
   it("does not evict a ticket-triggered run that is still executing", async () => {
     const bound = entry();
     const runRegistry = registry([bound]);