Merge commit 'eb798009058f9d761ebea63491c711d0c60e3f1e' into codex/sleep-live-run-resource-recovery

* commit 'eb798009058f9d761ebea63491c711d0c60e3f1e':
  Remove trailing blank line in refresh notification module
  Remove unused aggregate watchdog diagnostics
  test: settle delegated chat completion before plan assertions
  test(github): isolate scheduler time from postgres housekeeping
  test: settle saved retry cleanup before idle assertion
  fix: wake retry scheduling only when retry state changes
  fix(external-objects): scope issue deletion activation
  fix: retain sandbox attachment run scope for backup cleanup

# Conflicts:
#	server/src/services/run-dispatch-notifications.ts
This commit is contained in:
Dotta committed 2026-10-30 22:14:39 -05:00
commit de500e03ec
21 files changed
+262 -82

No files matched your search

+1 -1
View File
@@ -68,7 +68,7 @@ export async function attachSandboxCreation(db: Db, id: string, scope: { company
}
if (!operation || operation.state !== "pending") throw new Error("Sandbox creation ownership has already transferred");
const lease = await attach(tx as unknown as Db);
await notifyEnvironmentCleanup(tx, { companyId: scope.companyId });
await notifyEnvironmentCleanup(tx, { companyId: scope.companyId, runId: operation.runId });
await tx.update(sandboxCreationOperations).set({ state: "attached", leaseId: lease.id, updatedAt: new Date() })
.where(eq(sandboxCreationOperations.id, id));
return lease;
+4 -4
View File
@@ -1,6 +1,6 @@
import { notifyRunDispatch } from "../run-dispatch-notifications.js";
import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
import { notifyAttentionRetentionWork } from "../retention-work-notifications.js";
import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js";
import { deriveTaskKeyWithHeartbeatFallback } from "./run-state.js";
import { infrastructureRetryDelayMs } from "../system-recovery-state.js";
import {
@@ -1000,8 +1000,8 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
retryContextSnapshot.previousRunId = continuation.previousRunId;
}
await notifyRunDispatch(tx, { companyId: run.companyId });
await notifyAttentionRetentionWork(tx, { companyId: run.companyId });
await notifyScheduledRetryDispatch(tx, { companyId: run.companyId });
const wakeupRequest = await tx
.insert(agentWakeupRequests)
.values({
@@ -1760,7 +1760,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
};
const updated = await db.transaction(async (tx) => {
await notifyRunDispatch(tx, { companyId: scheduled.run.companyId });
await notifyScheduledRetryDispatch(tx, { companyId: scheduled.run.companyId });
const row = await tx
.update(heartbeatRuns)
.set({
+9 -1
View File
@@ -100,6 +100,7 @@ import {
heartbeatRuns,
routineRuns,
executionWorkspaces,
externalObjectMentions,
issueApprovals,
issueAccessGrants,
issueAttachments,
@@ -12044,7 +12045,14 @@ export function issueService(db: Db) {
if (owner) {
await notifyIssueAttentionRetentionWork(tx, { companyId: owner.companyId, issueId: id });
await notifyExternalConversationSettlement(tx, { companyId: owner.companyId });
await notifyExternalObjectRefresh(tx, owner.companyId);
const mentionedObjects = await tx.selectDistinct({ objectId: externalObjectMentions.objectId })
.from(externalObjectMentions)
.where(and(
eq(externalObjectMentions.companyId, owner.companyId),
eq(externalObjectMentions.sourceIssueId, id),
isNotNull(externalObjectMentions.objectId),
));
for (const mention of mentionedObjects) await notifyExternalObjectRefresh(tx, owner.companyId, mention.objectId!);
await notifyApprovalWork(tx, owner.companyId);
await notifyStatusCardIssueChange(tx, owner.companyId);
}
@@ -1,5 +1,5 @@
import { notifyRunDispatch } from "../run-dispatch-notifications.js";
import { notifyAttentionRetentionWork } from "../retention-work-notifications.js";
import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js";
import { randomUUID } from "node:crypto";
import { and, eq } from "drizzle-orm";
import { agentWakeupRequests, heartbeatRuns, type Db } from "@paperclipai/db";
@@ -51,8 +51,8 @@ export async function materializeNativeProviderCapacityRetry(input: {
scheduledRetryAt: dueAt.toISOString(),
executionRetryAccounting: accountingForScheduledRetry(run, NATIVE_PROVIDER_OVERLOADED_CODE, attempt),
};
await notifyRunDispatch(input.tx, { companyId: run.companyId });
await notifyAttentionRetentionWork(input.tx, { companyId: input.companyId });
await notifyScheduledRetryDispatch(input.tx, { companyId: run.companyId });
const [wake] = await input.tx.insert(agentWakeupRequests).values({
companyId: run.companyId, agentId: run.agentId, source: "automation", triggerDetail: "system",
reason: NATIVE_PROVIDER_OVERLOADED_CODE, status: "queued", runId: retryId,
@@ -1,3 +1,4 @@
import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
import { getExecutionBlocker } from "../execution-blocker.js";
import { createRunDispatch, deriveCommentId } from "../../modules/run-dispatch/index.js";
import { buildExecutionContinuation } from "../execution-continuation.js";
@@ -577,7 +578,12 @@ const support = externalDatabaseUrl
});
it("persists exactly one linked successor under competing sweepers and restarts", async () => {
const source = await seed(2);
const owner = heartbeatService(db);
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } });
const worker = owner.registerRunDispatchWork(coordinator, () => true);
await worker.ready;
const now = new Date();
try {
await Promise.all([
reconcileSafeNativeReplacements(db, now),
reconcileSafeNativeReplacements(db, now),
@@ -608,6 +614,10 @@ const support = externalDatabaseUrl
.where(eq(nativeRunFinalizations.runId, source.runId))
)[0]?.failureDetail?.successorRunId,
).toBe(children[0]!.id);
await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(children[0]!.scheduledRetryAt!.getTime()), { timeout: 5_000 });
} finally {
await worker.stop(); await coordinator.stop(); await owner.drainActiveRunExecutions();
}
});
it("retries status publication after a crash without dispatching provider work", async () => {
await deliverExecutionStatuses(db);
@@ -1,3 +1,4 @@
import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js";
import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js";
import { notifyExecutionRunWork } from "../execution-control-notifications.js";
@@ -410,6 +411,7 @@ export async function reconcileSafeNativeReplacements(
})
.returning();
await notifyExecutionRunWork(tx, { companyId: run.companyId });
await notifyScheduledRetryDispatch(tx, { companyId: run.companyId });
await tx.insert(heartbeatRuns).values({
id: successorRunId,
companyId: run.companyId,
+3 -3
View File
@@ -1,4 +1,4 @@
import { notifyRunDispatch } from "../run-dispatch-notifications.js";
import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js";
import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js";
import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js";
import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js";
@@ -2716,7 +2716,7 @@ export function recoveryService(
const now = new Date();
const retryAt = readProviderQuotaRetryAt(input.latestRun, now);
return database.transaction(async (tx) => {
await notifyRunDispatch(tx, { companyId: input.issue.companyId });
await notifyScheduledRetryDispatch(tx, { companyId: input.issue.companyId });
const wakeup = await tx
.insert(agentWakeupRequests)
.values({
@@ -3317,7 +3317,7 @@ export function recoveryService(
created = Boolean(enqueuedRun);
} else {
scheduledRun = await db.transaction(async (tx) => {
await notifyRunDispatch(tx, { companyId: input.issue.companyId });
await notifyScheduledRetryDispatch(tx, { companyId: input.issue.companyId });
const wakeup = await tx
.insert(agentWakeupRequests)
.values({
@@ -7,6 +7,11 @@ export async function notifyRunDispatch(transaction: object, scope?: RunDispatch
await notifyIssueRecovery(transaction, scope ? { companyId: scope.companyId } : undefined);
const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*";
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.nativeProcessSessions, hint);
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch, hint);
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, hint);
}
/** Only a retry row or deadline change needs to re-read the saved retry schedule. */
export async function notifyScheduledRetryDispatch(transaction: object, scope?: RunDispatchScope) {
await notifyRunDispatch(transaction, scope);
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch);
}
@@ -1,9 +1,9 @@
import { notifyRunDispatch } from "./run-dispatch-notifications.js";
import { notifyScheduledRetryDispatch } from "./run-dispatch-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
export type ScheduledRetryCancellationScope = { companyId: string; runId?: string };
export async function notifyScheduledRetryCancellation(transaction: object, scope?: ScheduledRetryCancellationScope) {
await notifyRunDispatch(transaction, scope);
await notifyScheduledRetryDispatch(transaction, scope);
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryCancellation, scope ? JSON.stringify(scope) : "*");
}
@@ -1,16 +0,0 @@
export function createIncompleteTaskWatchdogScanReporter(
warn: (details: { incomplete: number; newIncompleteIssueIds: string[] }, message: string) => void,
) {
let previousIncompleteIssueIds = new Set<string>();
return (result: { incomplete: number; incompleteIssueIds: string[] }, phase: "startup" | "periodic") => {
const currentIssueIds = new Set(result.incompleteIssueIds);
const newIncompleteIssueIds = [...currentIssueIds].filter((id) => !previousIncompleteIssueIds.has(id));
previousIncompleteIssueIds = currentIssueIds;
if (newIncompleteIssueIds.length === 0) return;
warn(
{ incomplete: result.incomplete, newIncompleteIssueIds },
`${phase} task-watchdog reconciliation found incomplete subtree scans`,
);
};
}