fix(server): eagerly cancel stale queued runs on issue reassignment (RENA-55610)

server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts

@@ -1120,6 +1120,159 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => {
     expect(countExecuteCallsForRun(runId)).toBe(0);
   });
 
+  it("cancelStaleQueuedRunsForIssue eagerly cancels the previous assignee's queued run right after a reassignment and clears the execution lock it held (RENA-55610)", async () => {
+    const { companyId, agentId } = await seedCompanyAndAgent({ agentName: "OriginalCoder" });
+    const replacementAgentId = randomUUID();
+    await db.insert(agents).values({
+      id: replacementAgentId,
+      companyId,
+      name: "ReplacementCoder",
+      role: "engineer",
+      status: "active",
+      adapterType: "codex_local",
+      adapterConfig: {},
+      runtimeConfig: {
+        heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 },
+      },
+      permissions: {},
+    });
+
+    const issueId = randomUUID();
+    const { runId, wakeupRequestId } = await seedQueuedRun({
+      companyId,
+      agentId,
+      issueId,
+      wakeReason: "issue_assigned",
+    });
+
+    // Mirrors what the reassign PATCH leaves behind before this fix: the
+    // assignee already flipped, but the original assignee's queued run still
+    // holds the issue's execution lock (RENA-55610 Beleg 2).
+    await db.insert(issues).values({
+      id: issueId,
+      companyId,
+      title: "Reassigned mid-queue",
+      status: "in_progress",
+      priority: "high",
+      assigneeAgentId: replacementAgentId,
+      executionRunId: runId,
+      executionAgentNameKey: "originalcoder",
+      executionLockedAt: new Date(),
+    });
+
+    const cancelledRunIds = await heartbeat.cancelStaleQueuedRunsForIssue(companyId, issueId);
+    expect(cancelledRunIds).toEqual([runId]);
+
+    const [run, wakeup, issue] = await Promise.all([
+      db
+        .select({ status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode })
+        .from(heartbeatRuns)
+        .where(eq(heartbeatRuns.id, runId))
+        .then((rows) => rows[0] ?? null),
+      db
+        .select({ status: agentWakeupRequests.status })
+        .from(agentWakeupRequests)
+        .where(eq(agentWakeupRequests.id, wakeupRequestId))
+        .then((rows) => rows[0] ?? null),
+      db
+        .select({ executionRunId: issues.executionRunId })
+        .from(issues)
+        .where(eq(issues.id, issueId))
+        .then((rows) => rows[0] ?? null),
+    ]);
+
+    expect(run?.status).toBe("cancelled");
+    expect(run?.errorCode).toBe("issue_assignee_changed");
+    expect(wakeup?.status).toBe("skipped");
+    expect(issue?.executionRunId).toBeNull();
+    expect(countExecuteCallsForRun(runId)).toBe(0);
+  });
+
+  it("sweeps and cancels obviously-stale queued runs even when the agent has no free concurrency slots (RENA-55610)", async () => {
+    const { companyId, agentId } = await seedCompanyAndAgent({ agentName: "BusyAgent", maxConcurrentRuns: 1 });
+    const replacementAgentId = randomUUID();
+    await db.insert(agents).values({
+      id: replacementAgentId,
+      companyId,
+      name: "ReplacementCoder",
+      role: "engineer",
+      status: "active",
+      adapterType: "codex_local",
+      adapterConfig: {},
+      runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } },
+      permissions: {},
+    });
+
+    // Occupy the agent's only concurrency slot so startNextQueuedRunForAgent's
+    // availableSlots is 0 — the exact condition under which the pre-fix lazy
+    // check never looked at any queued run (RENA-55610 Beleg 3).
+    const busyRunId = randomUUID();
+    await db.insert(heartbeatRuns).values({
+      id: busyRunId,
+      companyId,
+      agentId,
+      invocationSource: "on_demand",
+      triggerDetail: "manual",
+      status: "running",
+      startedAt: new Date(),
+      contextSnapshot: {},
+    });
+
+    const issueId = randomUUID();
+    await db.insert(issues).values({
+      id: issueId,
+      companyId,
+      title: "Reassigned while agent is fully busy",
+      status: "in_progress",
+      priority: "high",
+      assigneeAgentId: replacementAgentId,
+    });
+    const { runId, wakeupRequestId } = await seedQueuedRun({
+      companyId,
+      agentId,
+      issueId,
+      wakeReason: "issue_assigned",
+    });
+
+    await heartbeat.resumeQueuedRuns();
+
+    await waitForCondition(async () => {
+      const run = await db
+        .select({ status: heartbeatRuns.status })
+        .from(heartbeatRuns)
+        .where(eq(heartbeatRuns.id, runId))
+        .then((rows) => rows[0] ?? null);
+      return run?.status === "cancelled";
+    });
+
+    const [run, wakeup, busyRun] = await Promise.all([
+      db
+        .select({ status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode })
+        .from(heartbeatRuns)
+        .where(eq(heartbeatRuns.id, runId))
+        .then((rows) => rows[0] ?? null),
+      db
+        .select({ status: agentWakeupRequests.status })
+        .from(agentWakeupRequests)
+        .where(eq(agentWakeupRequests.id, wakeupRequestId))
+        .then((rows) => rows[0] ?? null),
+      db
+        .select({ status: heartbeatRuns.status })
+        .from(heartbeatRuns)
+        .where(eq(heartbeatRuns.id, busyRunId))
+        .then((rows) => rows[0] ?? null),
+    ]);
+
+    expect(run?.status).toBe("cancelled");
+    expect(run?.errorCode).toBe("issue_assignee_changed");
+    expect(wakeup?.status).toBe("skipped");
+    expect(countExecuteCallsForRun(runId)).toBe(0);
+    // The genuinely running run must be left untouched by the sweep.
+    expect(busyRun?.status).toBe("running");
+
+    await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(eq(heartbeatRuns.id, busyRunId));
+  });
+
   it("cancels queued runs when the issue reaches a terminal status before the run starts", async () => {
     const { companyId, agentId } = await seedCompanyAndAgent();
     const issueId = randomUUID();

server/src/routes/issues.ts

@@ -8999,6 +8999,20 @@ export function issueRoutes(
     }
     for (const publication of postCommitActivityPublications) publishActivity(publication);
 
+    if (existing.assigneeAgentId && nextAssigneeAgentId !== existing.assigneeAgentId) {
+      // The previous assignee may still have a queued run for this issue.
+      // Cancel it now instead of leaving it for the lazy staleness check in
+      // claimQueuedRun, which only runs once that run reaches the queue head
+      // — behind a busy queue that can take hours and leaves the issue's
+      // execution lock held by a run nobody will ever start.
+      await heartbeat.cancelStaleQueuedRunsForIssue(existing.companyId, existing.id).catch((err) => {
+        logger.warn(
+          { err, issueId: existing.id, previousAssigneeAgentId: existing.assigneeAgentId },
+          "failed to eager-cancel stale queued runs after issue reassignment",
+        );
+      });
+    }
+
     if (enteringBlocked) {
       const blockedIssue = issue;
       let ownerNotifiedAt: Date | null = null;

server/src/services/heartbeat.ts

@@ -12806,6 +12806,38 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
     return cancelled;
   }
 
+  // Eagerly cancels queued runs left behind by the previous assignee right
+  // after a reassignment, instead of waiting for the lazy staleness check in
+  // claimQueuedRun to reach them (which only happens once the run reaches the
+  // queue head and a concurrency slot frees up — it can take hours behind a
+  // busy queue). Reuses evaluateQueuedRunStaleness/cancelQueuedRunForStaleIssue
+  // so the interaction-wake and review-participant exceptions still apply, and
+  // so the issue's executionRunId lock is cleared when a cancelled queued run
+  // was the one holding it.
+  async function cancelStaleQueuedRunsForIssue(companyId: string, issueId: string) {
+    const candidates = await db
+      .select()
+      .from(heartbeatRuns)
+      .where(
+        and(
+          eq(heartbeatRuns.companyId, companyId),
+          eq(heartbeatRuns.status, "queued"),
+          sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`,
+        ),
+      );
+    if (candidates.length === 0) return [];
+
+    const cancelledRunIds: string[] = [];
+    for (const run of candidates) {
+      const context = parseObject(run.contextSnapshot);
+      const staleness = await evaluateQueuedRunStaleness(run, issueId, context);
+      if (!staleness.stale) continue;
+      const cancelled = await cancelQueuedRunForStaleIssue(run, issueId, staleness);
+      if (cancelled) cancelledRunIds.push(cancelled.id);
+    }
+    return cancelledRunIds;
+  }
+
   function truncateAgentErrorReason(reason: string | null | undefined): string | null {
     if (!reason) return null;
     const trimmed = reason.trim();
@@ -13443,7 +13475,6 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
       const policy = parseHeartbeatPolicy(agent);
       const runningCount = await countRunningRunsForAgent(agentId);
       const availableSlots = Math.max(0, policy.maxConcurrentRuns - runningCount);
-      if (availableSlots <= 0) return [];
 
       const queuedRuns = await db
         .select()
@@ -13456,7 +13487,6 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
         .orderBy(asc(heartbeatRuns.createdAt));
       if (queuedRuns.length === 0) return [];
 
-      const dependencyReadiness = await listQueuedRunDependencyReadiness(agent.companyId, queuedRuns);
       const queuedIssueIds = [...new Set(
         queuedRuns
           .map((run) => readNonEmptyString(parseObject(run.contextSnapshot).issueId))
@@ -13467,6 +13497,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
           id: issues.id,
           status: issues.status,
           priority: issues.priority,
+          assigneeAgentId: issues.assigneeAgentId,
         })
         .from(issues)
         .where(
@@ -13475,8 +13506,46 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
             : sql`false`,
         );
       const issueById = new Map(issueRows.map((row) => [row.id, row]));
+
+      // Cheap in-memory prefilter (no extra queries beyond the batch issue
+      // fetch above) so a backed-up queue — e.g. dozens of runs stuck behind a
+      // maxed-out concurrency cap — still gets its obviously-stale entries
+      // (reassigned/terminal/deleted issue) cancelled on every scheduling
+      // pass, not just once a run reaches the queue head with a free slot.
+      // evaluateQueuedRunStaleness still makes the final call so interaction-
+      // wake and review-participant exceptions keep applying.
+      let liveRuns = queuedRuns;
+      const obviouslyStaleCandidates = queuedRuns.filter((run) => {
+        const runIssueId = readNonEmptyString(parseObject(run.contextSnapshot).issueId);
+        if (!runIssueId) return false;
+        const issueRow = issueById.get(runIssueId);
+        return (
+          !issueRow ||
+          issueRow.status === "done" ||
+          issueRow.status === "cancelled" ||
+          issueRow.assigneeAgentId !== agentId
+        );
+      });
+      if (obviouslyStaleCandidates.length > 0) {
+        const cancelledRunIds = new Set<string>();
+        for (const run of obviouslyStaleCandidates) {
+          const context = parseObject(run.contextSnapshot);
+          const runIssueId = readNonEmptyString(context.issueId);
+          if (!runIssueId) continue;
+          const staleness = await evaluateQueuedRunStaleness(run, runIssueId, context);
+          if (!staleness.stale) continue;
+          const cancelled = await cancelQueuedRunForStaleIssue(run, runIssueId, staleness);
+          if (cancelled) cancelledRunIds.add(run.id);
+        }
+        if (cancelledRunIds.size > 0) {
+          liveRuns = liveRuns.filter((run) => !cancelledRunIds.has(run.id));
+        }
+      }
+      if (liveRuns.length === 0 || availableSlots <= 0) return [];
+
+      const dependencyReadiness = await listQueuedRunDependencyReadiness(agent.companyId, liveRuns);
       const companyAgents = await listCompanyAgentOrgRows(agent.companyId);
-      const prioritizedRuns = [...queuedRuns].sort((left, right) => {
+      const prioritizedRuns = [...liveRuns].sort((left, right) => {
         const leftIssueId = readNonEmptyString(parseObject(left.contextSnapshot).issueId);
         const rightIssueId = readNonEmptyString(parseObject(right.contextSnapshot).issueId);
         const leftReadiness = leftIssueId ? dependencyReadiness.get(leftIssueId) : null;
@@ -19061,6 +19130,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
 
     cancelRun: (runId: string, reason?: string, options?: CancelRunOptions) => cancelRunInternal(runId, reason, options),
 
+    cancelStaleQueuedRunsForIssue,
+
     /**
      * Pause-only. Emits errorCode "agent_paused" unconditionally; its sole caller is the
      * agent pause route. For non-pause cancellations use cancelRun, or call the internal