fix(interactions): separate claim and error retry counts

packages/db/src/migrations/0228_violet_scorpion.sql

@@ -11,6 +11,7 @@ CREATE TABLE IF NOT EXISTS "issue_question_response_deliveries" (
 	"status" text DEFAULT 'pending' NOT NULL,
 	"delivery_mode" text,
 	"attempt_count" integer DEFAULT 0 NOT NULL,
+	"error_count" integer DEFAULT 0 NOT NULL,
 	"last_attempt_at" timestamp with time zone,
 	"acknowledged_at" timestamp with time zone,
 	"last_error_code" text,

packages/db/src/migrations/meta/0228_snapshot.json

@@ -9449,6 +9449,13 @@
           "notNull": true,
           "default": 0
         },
+        "error_count": {
+          "name": "error_count",
+          "type": "integer",
+          "primaryKey": false,
+          "notNull": true,
+          "default": 0
+        },
         "last_attempt_at": {
           "name": "last_attempt_at",
           "type": "timestamp with time zone",
@@ -42163,4 +42170,4 @@
     "schemas": {},
     "tables": {}
   }
-}
+}

packages/db/src/schema/issue_question_response_deliveries.ts

@@ -34,7 +34,10 @@ export const issueQuestionResponseDeliveries = pgTable(
     payloadSha256: text("payload_sha256").notNull(),
     status: text("status").notNull().default("pending"),
     deliveryMode: text("delivery_mode"),
+    /** Monotonic claim generation used to fence stale workers. */
     attemptCount: integer("attempt_count").notNull().default(0),
+    /** Actual side-effect failures; scheduling suppression does not consume this budget. */
+    errorCount: integer("error_count").notNull().default(0),
     lastAttemptAt: timestamp("last_attempt_at", { withTimezone: true }),
     acknowledgedAt: timestamp("acknowledged_at", { withTimezone: true }),
     lastErrorCode: text("last_error_code"),

server/src/__tests__/question-response-delivery.test.ts

@@ -311,6 +311,56 @@ describeEmbeddedPostgres("question response delivery", () => {
     expect(delivery).toMatchObject({
       status: "fallback_queued",
       attemptCount: 6,
+      errorCount: 0,
+      targetRunId: fallbackRunId,
+    });
+  });
+
+  it("preserves the actual-error retry budget after prolonged wake suppression", async () => {
+    const seeded = await seed({ sourceStatus: "running" });
+    const fallbackRunId = randomUUID();
+    let wakeAttempts = 0;
+    const wakeup = vi.fn().mockImplementation(async () => {
+      wakeAttempts += 1;
+      if (wakeAttempts <= 5) return null;
+      if (wakeAttempts === 6) throw new Error("scheduler temporarily unavailable");
+      return db.insert(heartbeatRuns).values({
+        id: fallbackRunId,
+        companyId: seeded.companyId,
+        agentId: seeded.agentId,
+        invocationSource: "automation",
+        status: "queued",
+        runtimeMode: "legacy",
+        driverKind: "codex",
+        contextSnapshot: { issueId: seeded.issueId },
+      }).returning().then((rows) => rows[0]!);
+    });
+    const service = questionResponseDeliveryService(db, {
+      heartbeat: { wakeup } as never,
+      steer: vi.fn(),
+    });
+
+    for (let attempt = 0; attempt < 5; attempt += 1) {
+      await expect(service.deliver(seeded.interaction.id)).resolves.toBeNull();
+    }
+    await expect(service.deliver(seeded.interaction.id)).resolves.toBeNull();
+    const [afterError] = await db.select().from(issueQuestionResponseDeliveries);
+    expect(afterError).toMatchObject({
+      status: "pending",
+      attemptCount: 6,
+      errorCount: 1,
+      lastErrorCode: "scheduler temporarily unavailable",
+    });
+
+    await expect(service.deliver(seeded.interaction.id)).resolves.toMatchObject({
+      status: "fallback_queued",
+      targetRunId: fallbackRunId,
+    });
+    const [delivered] = await db.select().from(issueQuestionResponseDeliveries);
+    expect(delivered).toMatchObject({
+      status: "fallback_queued",
+      attemptCount: 7,
+      errorCount: 1,
       targetRunId: fallbackRunId,
     });
   });

server/src/services/question-response-delivery.ts

@@ -371,18 +371,19 @@ export function questionResponseDeliveryService(
     options: { bounded: boolean } = { bounded: true },
   ) {
     const at = now();
-    const exhausted = options.bounded && delivery.attemptCount >= MAX_DELIVERY_ATTEMPTS;
-    if (!exhausted) {
-      await db.update(issueQuestionResponseDeliveries).set({
-        status: "pending",
-        lastErrorCode: errorCode,
-        updatedAt: at,
-      }).where(and(
-        eq(issueQuestionResponseDeliveries.id, delivery.id),
-        eq(issueQuestionResponseDeliveries.status, "delivering"),
-        eq(issueQuestionResponseDeliveries.attemptCount, delivery.attemptCount),
-      ));
-    }
+    const nextErrorCount = delivery.errorCount + (options.bounded ? 1 : 0);
+    const exhausted = options.bounded && nextErrorCount >= MAX_DELIVERY_ATTEMPTS;
+    await db.update(issueQuestionResponseDeliveries).set({
+      // Keep an exhausted claim owned until recordTerminal commits its outcome.
+      status: exhausted ? "delivering" : "pending",
+      ...(options.bounded ? { errorCount: nextErrorCount } : {}),
+      lastErrorCode: errorCode,
+      updatedAt: at,
+    }).where(and(
+      eq(issueQuestionResponseDeliveries.id, delivery.id),
+      eq(issueQuestionResponseDeliveries.status, "delivering"),
+      eq(issueQuestionResponseDeliveries.attemptCount, delivery.attemptCount),
+    ));
     return exhausted;
   }
 
@@ -662,6 +663,7 @@ export function questionResponseDeliveryService(
         deliveryId: claimed.id,
         interactionId,
         attemptCount: claimed.attemptCount,
+        errorCount: claimed.errorCount + 1,
         exhausted,
       }, "question response delivery will retry after wake failure");
       if (!exhausted) return null;