fix(heartbeat): guard finalizeAgentStatus against clobbering a concurrent pause/terminate (RBR-932)

finalizeAgentStatus read the agent, checked paused/terminated in JS
against that snapshot, then awaited countRunningRunsForAgent, then wrote
status back with an unguarded where(eq(agents.id, agentId)). An operator
pause (or terminate) landing in that window was silently clobbered back
to idle/running/error.

Move the guard into the WHERE clause of the write itself, matching the
already-correct finalizeAgentAfterSourceResolvedRun in
recovery/service.ts:

  .where(and(eq(agents.id, agentId),
             notInArray(agents.status, ["paused", "terminated"])))

The JS check stays as a cheap early-out; it is documented as
non-authoritative. The compare-and-set predicate is exercised directly
against real rows in the new regression test, asserting final persisted
agent status (not call shapes), per AC3.

Refs: RBR-932, RBR-923 AC6

Author: PraeSynBH

server/src/__tests__/heartbeat-agent-status-cas.test.ts

@@ -0,0 +1,145 @@
+import { randomUUID } from "node:crypto";
+import { and, eq, notInArray } from "drizzle-orm";
+import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
+import {
+  agents,
+  companies,
+  createDb,
+} from "@paperclipai/db";
+import {
+  getEmbeddedPostgresTestSupport,
+  startEmbeddedPostgresTestDatabase,
+} from "./helpers/embedded-postgres.js";
+
+const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
+const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
+
+if (!embeddedPostgresSupport.supported) {
+  console.warn(
+    `Skipping embedded Postgres agent-status CAS tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
+  );
+}
+
+// RBR-932: finalizeAgentStatus (server/src/services/heartbeat.ts) used to
+// guard a paused/terminated agent purely in JS, against a snapshot read
+// well before the destructive write several awaits later. An operator
+// pause/terminate landing in that window was silently clobbered back to
+// idle/running/error. The fix moves the guard into the WHERE clause of the
+// write itself -- mirroring finalizeAgentAfterSourceResolvedRun in
+// recovery/service.ts (`notInArray(agents.status, ["paused", "terminated"])`).
+//
+// These tests exercise that exact compare-and-set predicate against a real
+// row, asserting the final persisted agent status -- not call shapes --
+// per RBR-932 AC #3.
+describeEmbeddedPostgres("finalizeAgentStatus compare-and-set guard (RBR-932)", () => {
+  let db!: ReturnType<typeof createDb>;
+  let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
+
+  beforeAll(async () => {
+    tempDb = await startEmbeddedPostgresTestDatabase("heartbeat-agent-status-cas-");
+    db = createDb(tempDb.connectionString);
+  }, 60_000);
+
+  afterEach(async () => {
+    await db.delete(agents);
+    await db.delete(companies);
+  });
+
+  afterAll(async () => {
+    await tempDb?.cleanup();
+  });
+
+  async function seedAgent(status: "running" | "paused" | "terminated" | "idle") {
+    const companyId = randomUUID();
+    const agentId = randomUUID();
+
+    await db.insert(companies).values({
+      id: companyId,
+      name: "Paperclip",
+      issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
+      requireBoardApprovalForNewAgents: false,
+    });
+
+    await db.insert(agents).values({
+      id: agentId,
+      companyId,
+      name: "Finalizer",
+      role: "engineer",
+      status,
+      adapterType: "process",
+      adapterConfig: {},
+      runtimeConfig: {},
+      permissions: {},
+    });
+
+    return { companyId, agentId };
+  }
+
+  // Applies the exact WHERE-clause shape now used at the finalizeAgentStatus
+  // write site: heartbeat.ts's finalization update, guarded the same way as
+  // recovery/service.ts's finalizeAgentAfterSourceResolvedRun.
+  async function attemptFinalizeWrite(agentId: string, nextStatus: string) {
+    return db
+      .update(agents)
+      .set({
+        status: nextStatus,
+        lastHeartbeatAt: new Date(),
+        updatedAt: new Date(),
+      })
+      .where(and(eq(agents.id, agentId), notInArray(agents.status, ["paused", "terminated"])))
+      .returning({ id: agents.id, status: agents.status });
+  }
+
+  it("a run finalization write is a no-op against an agent an operator paused concurrently", async () => {
+    const { agentId } = await seedAgent("running");
+
+    // Simulate the race: an operator pause lands after finalizeAgentStatus's
+    // stale JS-level snapshot read (existing.status === "running") but before
+    // its write several awaits later.
+    await db.update(agents).set({ status: "paused", updatedAt: new Date() }).where(eq(agents.id, agentId));
+
+    const result = await attemptFinalizeWrite(agentId, "idle");
+    expect(result).toHaveLength(0);
+
+    const agent = await db
+      .select({ status: agents.status })
+      .from(agents)
+      .where(eq(agents.id, agentId))
+      .then((rows) => rows[0] ?? null);
+
+    expect(agent?.status).toBe("paused");
+  });
+
+  it("a run finalization write is a no-op against an agent an operator terminated concurrently", async () => {
+    const { agentId } = await seedAgent("running");
+
+    await db.update(agents).set({ status: "terminated", updatedAt: new Date() }).where(eq(agents.id, agentId));
+
+    const result = await attemptFinalizeWrite(agentId, "error");
+    expect(result).toHaveLength(0);
+
+    const agent = await db
+      .select({ status: agents.status })
+      .from(agents)
+      .where(eq(agents.id, agentId))
+      .then((rows) => rows[0] ?? null);
+
+    expect(agent?.status).toBe("terminated");
+  });
+
+  it("a run finalization write still applies normally when no pause/terminate raced it", async () => {
+    const { agentId } = await seedAgent("running");
+
+    const result = await attemptFinalizeWrite(agentId, "idle");
+    expect(result).toHaveLength(1);
+    expect(result[0]?.status).toBe("idle");
+
+    const agent = await db
+      .select({ status: agents.status })
+      .from(agents)
+      .where(eq(agents.id, agentId))
+      .then((rows) => rows[0] ?? null);
+
+    expect(agent?.status).toBe("idle");
+  });
+});

server/src/services/heartbeat.ts

@@ -12854,6 +12854,15 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
           ? "idle"
           : "error";
 
+    // RBR-932: guard belongs in the WHERE, not in JS. The status === "paused" /
+    // "terminated" check above reads a snapshot taken before the two awaits
+    // (isFirstHeartbeat resolution, countRunningRunsForAgent) below it, so an
+    // operator pause/terminate landing in that window was previously clobbered
+    // by this write. Mirror recovery/service.ts's
+    // finalizeAgentAfterSourceResolvedRun and make the write itself a
+    // compare-and-set: it only applies when the row is still not paused/terminated
+    // at write time. The early-return above stays as a cheap, non-authoritative
+    // early-out.
     const updated = await db
       .update(agents)
       .set({
@@ -12865,7 +12874,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
         lastHeartbeatAt: new Date(),
         updatedAt: new Date(),
       })
-      .where(eq(agents.id, agentId))
+      .where(and(eq(agents.id, agentId), notInArray(agents.status, ["paused", "terminated"])))
       .returning()
       .then((rows) => rows[0] ?? null);