fix(server): wake chat milestones after recovery classification

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-25 11:02:21 -05:00
1 parent 49e988e2e4
commit 0e5591054a
5 files changed
+96 -18

No files matched your search

@@ -1668,3 +1668,12 @@ all 40 existing stale-lock ownership/race cases and server types/build pass.
The conditional update and provider/process authority retain their existing
owners. Adapter-failure classification and remaining producer audits still gate
production registration; this does not migrate the stale-lock polling duty itself.
Slice 28 now covers the late adapter-failure classification write, which can
return an old failed run to the existing publication window after the issue's
earlier wake. The missing notice reproduced before repair; all three recovery
milestone cases, 89 recovery-action cases and server types/build pass. Four
classification callers reuse their issue identity, with no new queue table or
provider policy in shared code. Authority/configuration audit and production
wiring remain next; the original full stack and remote gates remain incomplete.
@@ -1317,3 +1317,36 @@ Adapter-failure classification remains a separate producer audit: it changes
errorCode/resultJson/updatedAt after issue-monitor or escalation writes, so an
earlier issue notification cannot by itself prove coverage of the later run
commit. Production registration and removal of the lifecycle poll remain gated.
## Late adapter-failure classification
The owning recovery path updates the issue monitor/escalation before persisting
its run classification. That later run write refreshes updatedAt and can bring
an old failed run into the projector's existing 24-hour window. The earlier
issue wake is insufficient: a real quota-recovery case established the monitor
and classification but produced no notice after the worker exhausted startup.
Classification now registers the issue intent in its own short write transaction.
All four original-assignee/review-participant classification paths pass their
already-owned issue identity. The update is company scoped and preserves its
JSON merge so concurrent workspace receipts remain intact. Provider execution,
classification policy, monitor deadlines, escalation and existing recovery
ownership are unchanged. Repeated recovery with an existing quota monitor is
quiet and does not dispatch another run.
Evidence: /tmp/sleep28-classification-before.log fails on the missing notice;
/tmp/sleep28-classification-tests.log passes all three recovery milestone cases.
All 89 existing recovery-action cases pass in
/tmp/sleep28-recovery-actions-acceptance.log. Their guarded external DB harness
uses paperclip_sleep28_recovery_actions_20261025_test, separately from worker
fixtures. Server typecheck/build pass in
/tmp/sleep28-classification-{typecheck,build}.log. The earlier 40 stale-lock and
43 identity/credential cases passed at their recorded checkpoints and were not
blindly repeated for this private classification write.
Still required: finish the authority/configuration producer dispositions,
register the lifecycle worker, retain the separate native-progress duty, remove
only the lifecycle poll, and complete current-head CI/review. Fresh master
34409a92c4 is fetched but not yet reconciled into the stack. Older parent CI and
voice security/deletion findings plus every original remaining reason stay open.
@@ -13,7 +13,7 @@ import { idleWorkSnapshot } from "../services/task-admission.js";
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_MILESTONE_RECOVERY_TEST_DATABASE_URL");
const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
(support.supported ? describe : describe.skip)("chat milestones after orphan recovery", () => {
(support.supported ? describe : describe.skip)("chat milestones after run recovery", () => {
let db: ReturnType<typeof createDb>;
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
let fixtures: ReturnType<typeof createChatMilestoneFixtures>;
@@ -75,4 +75,29 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
expect(await recovery.sweepStaleIssueLocks()).toEqual({ cleared: 0, issueIds: [], terminalizedRunIds: [] });
await flush(); await quiet(); expect(await publications()).toHaveLength(2);
});
it("wakes an older failed run when recovery refreshes its quota classification", async () => {
const runId = randomUUID(), old = new Date(Date.now() - 48 * 60 * 60_000);
await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: f.agentId,
status: "failed", runtimeMode: "legacy", responsibleUserId: f.userId,
createdAt: old, startedAt: old, finishedAt: old, updatedAt: old,
error: "You've hit your usage limit for GPT-5. Try again at 12:00 AM (UTC).", errorCode: "adapter_failed",
resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } },
contextSnapshot: { issueId: f.issueId, source: "chat:slack", endpointId: f.endpointId, wakeCommentId: f.commentId },
});
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: error => errors.push(error) });
await chatRunMilestoneWork(db).register(coordinator, () => true).ready;
await flush(); expect(await publications()).toEqual([]); await quiet();
const enqueueWakeup = vi.fn(async () => null);
const recovery = recoveryService(db, { enqueueWakeup });
expect(await recovery.reconcileStrandedAssignedIssues()).toMatchObject({ providerQuotaMonitored: 1 });
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0].errorCode).toBe("provider_quota");
expect(enqueueWakeup).not.toHaveBeenCalled();
await flush();
expect(await publications()).toEqual([expect.objectContaining({ idempotencyKey: `run:${runId}:failed:${f.endpointId}` })]);
await quiet();
expect(await recovery.reconcileStrandedAssignedIssues()).toMatchObject({ providerQuotaMonitored: 0, skipped: 1 });
await flush(); await quiet(); expect(await publications()).toHaveLength(1);
});
});
@@ -1,3 +1,4 @@
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { randomUUID } from "node:crypto";
import fs from "node:fs/promises";
import os from "node:os";
@@ -48,7 +49,8 @@ import { instanceSettingsService } from "../services/instance-settings.js";
import { withNativeWorkspaceFinalizationOwnership } from "../services/native-runtime/native-workspace-finalization-ownership.js";
import { noticeMetadataReferencesRecoveryAction } from "../services/recovery/successful-run-handoff.js";
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_RECOVERY_ACTION_TEST_DATABASE_URL");
const embeddedPostgresSupport = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
function makeRecoveryActionRow(overrides: Record<string, unknown> = {}) {
@@ -154,8 +156,8 @@ describeEmbeddedPostgres("issue recovery actions", () => {
beforeAll(async () => {
operationLogRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-recovery-operation-logs-"));
process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = operationLogRoot;
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-recovery-actions-");
db = createDb(tempDb.connectionString);
if (!external) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-recovery-actions-");
db = createDb(external ?? tempDb!.connectionString);
}, 30_000);
afterEach(async () => {
@@ -182,6 +184,7 @@ describeEmbeddedPostgres("issue recovery actions", () => {
});
afterAll(async () => {
await db?.$client.end({ timeout: 1 });
await tempDb?.cleanup();
if (priorOperationLogRoot === undefined) delete process.env.WORKSPACE_OPERATION_LOG_BASE_PATH;
else process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = priorOperationLogRoot;
@@ -616,8 +619,8 @@ describeEmbeddedPostgres("issue recovery actions", () => {
await db.update(agents).set({ status: "paused" }).where(eq(agents.id, coderId));
const enqueueWakeup = vi.fn(async () => null);
const recovery = recoveryService(db, { enqueueWakeup });
const operatorDb = createDb(tempDb!.connectionString, { maxConnections: 1 });
const probeDb = createDb(tempDb!.connectionString, { maxConnections: 1 });
const operatorDb = createDb(external ?? tempDb!.connectionString, { maxConnections: 1 });
const probeDb = createDb(external ?? tempDb!.connectionString, { maxConnections: 1 });
let release!: () => void;
let ready!: () => void;
let holderPid = 0;
@@ -766,7 +769,7 @@ describeEmbeddedPostgres("issue recovery actions", () => {
.where(eq(issues.id, sourceIssue.id));
}
const [observed] = await db.select().from(issues).where(eq(issues.id, sourceIssue.id));
const singleConnectionDb = createDb(tempDb!.connectionString, { maxConnections: 1 });
const singleConnectionDb = createDb(external ?? tempDb!.connectionString, { maxConnections: 1 });
let timer: ReturnType<typeof setTimeout> | undefined;
try {
const work = recoveryService(singleConnectionDb, { enqueueWakeup: async () => null })
+19 -11
View File
@@ -4310,6 +4310,7 @@ export function recoveryService(
}
async function persistAdapterFailureRecoveryClassification(
issue: Pick<typeof issues.$inferSelect, "id" | "companyId">,
latestRun: NonNullable<LatestIssueRun>,
classification: NonNullable<AdapterFailureRecoveryClassification>,
): Promise<NonNullable<LatestIssueRun>> {
@@ -4321,17 +4322,20 @@ export function recoveryService(
const classificationMetadata = withAdapterFailureRecoveryClassification(
{ ...latestRun, resultJson: {} }, classification,
).resultJson;
const [updated] = await db
.update(heartbeatRuns)
.set({
errorCode: classifiedRun.errorCode,
// Classification owns retry metadata, not workspace repair receipts
// that may have committed after this run was selected.
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify(classificationMetadata)}::jsonb`,
updatedAt: new Date(),
})
.where(eq(heartbeatRuns.id, latestRun.id))
.returning({ resultJson: heartbeatRuns.resultJson, errorCode: heartbeatRuns.errorCode });
const [updated] = await db.transaction(async tx => {
await notifyChatRunMilestones(tx, { companyId: issue.companyId, issueId: issue.id });
return tx
.update(heartbeatRuns)
.set({
errorCode: classifiedRun.errorCode,
// Classification owns retry metadata, not workspace repair receipts
// that may have committed after this run was selected.
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify(classificationMetadata)}::jsonb`,
updatedAt: new Date(),
})
.where(and(eq(heartbeatRuns.id, latestRun.id), eq(heartbeatRuns.companyId, issue.companyId)))
.returning({ resultJson: heartbeatRuns.resultJson, errorCode: heartbeatRuns.errorCode });
});
return updated ? { ...classifiedRun, ...updated } : classifiedRun;
}
@@ -4930,6 +4934,7 @@ export function recoveryService(
});
if (monitored) {
latestRun = await persistAdapterFailureRecoveryClassification(
issue,
latestRun,
adapterFailureClassification,
);
@@ -4951,6 +4956,7 @@ export function recoveryService(
});
if (updated) {
latestRun = await persistAdapterFailureRecoveryClassification(
issue,
latestRun,
adapterFailureClassification,
);
@@ -5154,6 +5160,7 @@ export function recoveryService(
});
if (monitored) {
latestRun = await persistAdapterFailureRecoveryClassification(
issue,
participantLatestRun,
participantAdapterFailureClassification,
);
@@ -5180,6 +5187,7 @@ export function recoveryService(
});
if (updated) {
latestRun = await persistAdapterFailureRecoveryClassification(
issue,
participantLatestRun,
participantAdapterFailureClassification,
);