diff --git a/packages/db/src/migrations/0228_violet_scorpion.sql b/packages/db/src/migrations/0228_violet_scorpion.sql index 0278a22fe09..59c19802dff 100644 --- a/packages/db/src/migrations/0228_violet_scorpion.sql +++ b/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, diff --git a/packages/db/src/migrations/meta/0228_snapshot.json b/packages/db/src/migrations/meta/0228_snapshot.json index 77645f1276b..aea6b30b74b 100644 --- a/packages/db/src/migrations/meta/0228_snapshot.json +++ b/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": {} } -} \ No newline at end of file +} diff --git a/packages/db/src/schema/issue_question_response_deliveries.ts b/packages/db/src/schema/issue_question_response_deliveries.ts index c5f4c2c056d..6059ee4befe 100644 --- a/packages/db/src/schema/issue_question_response_deliveries.ts +++ b/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"), diff --git a/server/src/__tests__/question-response-delivery.test.ts b/server/src/__tests__/question-response-delivery.test.ts index 2083847faff..ff893a932d4 100644 --- a/server/src/__tests__/question-response-delivery.test.ts +++ b/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, }); }); diff --git a/server/src/services/question-response-delivery.ts b/server/src/services/question-response-delivery.ts index d43c128944e..3608415ec82 100644 --- a/server/src/services/question-response-delivery.ts +++ b/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;