Jannik/PaperClipAI
1
0 Fork 0
mirror of https://github.com/paperclipai/paperclip.git synced 2026-11-02 16:58:52 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge verified sleep prerequisites into 03

Browse Source 
Co-Authored-By: Paperclip <noreply@paperclip.ing>

* commit '98a68b492f933890da59e2ec2f9ddd31f997d5b5':
  Remove unused aggregate watchdog diagnostics
  test: settle delegated chat completion before plan assertions
  test: settle saved retry cleanup before idle assertion
  fix: wake retry scheduling only when retry state changes
This commit is contained in:
DottaandPaperclip committed 2026-10-30 22:12:53 -05:00
2 parents 06c3c6e9d1 98a68b492f
commit 5bc8bef9f4
15 files changed
+112 -75
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/__tests__/deferred-execution-receipt-work.test.ts
+10 -2
View File
@@ -78,7 +78,15 @@ describe("deferred receipt durable dispatch", () => {
return { companyId, agent, issue, source, actorId };
}
async function quiet() {
await vi.waitFor(async () => { expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 40_000 });
await vi.waitFor(async () => {
if (vi.isFakeTimers()) {
// Fake-clock cases have no live execution. Settle finite cleanup
// deadlines explicitly instead of advancing at waitFor's wall speed.
await vi.advanceTimersByTimeAsync(1000);
await Promise.allSettled([...passes]); await immediate();
}
expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0);
}, { timeout: 40_000 });
const reads = vi.spyOn(db, "select"), executes = vi.spyOn(db, "execute");
const selected = reads.mock.calls.length, executed = executes.mock.calls.length;
if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
@@ -455,8 +463,8 @@ describe("deferred receipt durable dispatch", () => {
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!));
expect(wake.runId, JSON.stringify(wake)).toBeTruthy(); runId = wake.runId;
}, { timeout: 40_000 });
// Observe this Retry before closing the task and draining its follow-ups.
await vi.waitFor(async () => {
await heartbeat.drainActiveRunExecutions();
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!));
expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode })).toBe("succeeded");
expect(run.stdoutExcerpt).toContain("receipt-proof");
server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts
+10 -9
View File
@@ -3,7 +3,7 @@ import { PgDialect } from "drizzle-orm/pg-core";
import { afterEach, describe, expect, it, vi } from "vitest";
import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
import { notifyRunDispatch } from "../services/run-dispatch-notifications.js";
import { notifyRunDispatch, notifyScheduledRetryDispatch } from "../services/run-dispatch-notifications.js";
import { registerRunDispatchWork } from "../services/heartbeat/dispatch-work.js";
const cleanup: Array<() => Promise<void>> = [];
@@ -38,6 +38,7 @@ function fixture(input: { due?: number; suppressed?: boolean; draining?: boolean
}, () => !state.draining && !state.standby);
cleanup.push(() => coordinator.stop());
return { db, state, promote, next, dispatch, coordinator, worker,
notifyRetry: () => db.transaction(tx => notifyScheduledRetryDispatch(tx)),
notify: (scope?: { companyId: string }) => db.transaction(tx => notifyRunDispatch(tx, scope)),
settle: () => vi.advanceTimersByTimeAsync(1_000) };
}
@@ -61,9 +62,9 @@ describe("run dispatch obligations", () => {
});
it("rearms earlier work and cancels a future deadline on a committed producer change", async () => {
const f = fixture({ due: 1_600_000 }); await f.worker.ready;
f.state.due = 1_005_000; await f.notify(); await f.settle();
f.state.due = 1_005_000; await f.notifyRetry(); await f.settle();
expect(f.coordinator.nextWakeAt()).toBe(1_005_000);
f.state.due = null; await f.notify(); await f.settle();
f.state.due = null; await f.notifyRetry(); await f.settle();
expect(f.coordinator.nextWakeAt()).toBeNull();
const before = f.next.mock.calls.length;
await vi.advanceTimersByTimeAsync(20 * 60_000); expect(f.next).toHaveBeenCalledTimes(before);
@@ -76,7 +77,7 @@ describe("run dispatch obligations", () => {
await f.notify(source);
await vi.advanceTimersByTimeAsync(10);
}
expect(f.promote.mock.calls.length).toBeLessThanOrEqual(2);
expect(f.promote).toHaveBeenCalledOnce();
expect(f.dispatch.mock.calls.length).toBeLessThanOrEqual(2);
await f.settle();
expect(f.dispatch).toHaveBeenLastCalledWith({ companyId: "company-a" });
@@ -87,7 +88,7 @@ describe("run dispatch obligations", () => {
});
it("remembers the first producer arriving during drain without owner SQL", async () => {
const f = fixture(); await f.worker.ready; f.state.draining = true;
await f.notify(); await f.settle(); await vi.advanceTimersByTimeAsync(90_000);
await f.notifyRetry(); await f.settle(); await vi.advanceTimersByTimeAsync(90_000);
expect(f.promote).toHaveBeenCalledOnce(); expect(f.next).toHaveBeenCalledOnce();
expect(idleWorkSnapshot().active).toBe(0);
f.state.draining = false; await vi.advanceTimersByTimeAsync(30_000);
@@ -96,7 +97,7 @@ describe("run dispatch obligations", () => {
});
it("releases a failed producer hold during drain and retains recovery for resume", async () => {
const f = fixture(); await f.worker.ready;
f.state.fail = true; await f.notify(); await f.settle();
f.state.fail = true; await f.notifyRetry(); await f.settle();
expect(f.promote).toHaveBeenCalledTimes(2);
f.state.draining = true; await vi.advanceTimersByTimeAsync(90_000);
expect(f.promote).toHaveBeenCalledTimes(2); expect(f.next).toHaveBeenCalledOnce();
@@ -109,8 +110,8 @@ describe("run dispatch obligations", () => {
let release!: () => void;
const held = new Promise<void>(resolve => { release = resolve; });
f.promote.mockImplementationOnce(() => held);
await f.notify(); await f.settle();
await f.notify(); await f.settle();
await f.notifyRetry(); await f.settle();
await f.notifyRetry(); await f.settle();
expect(f.promote).toHaveBeenCalledTimes(2);
release(); await f.settle();
await vi.advanceTimersByTimeAsync(30_000);
@@ -121,7 +122,7 @@ describe("run dispatch obligations", () => {
await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).not.toHaveBeenCalled();
f.state.standby = false; await vi.advanceTimersByTimeAsync(30_000);
expect(f.promote).toHaveBeenCalledOnce(); expect(f.state.due).toBeNull();
f.state.due = Date.now() + 60_000; await f.notify(); await f.settle(); await f.worker.stop();
f.state.due = Date.now() + 60_000; await f.notifyRetry(); await f.settle(); await f.worker.stop();
await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).toHaveBeenCalledTimes(2);
});
});
server/src/__tests__/heartbeat-run-dispatch-work.test.ts
+47 -1
View File
@@ -2,9 +2,11 @@ import { idleWorkSnapshot } from "../services/task-admission.js";
import { randomUUID } from "node:crypto";
import { and, eq, sql } from "drizzle-orm";
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
import { agents, companies, costEvents, createDb, databaseWorkPending, heartbeatRuns } from "@paperclipai/db";
import { agents, companies, costEvents, createDb, databaseWorkPending, heartbeatRuns, issues } from "@paperclipai/db";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { issueService } from "../services/issues.js";
import { materializeNativeProviderCapacityRetry } from "../services/native-runtime/native-provider-capacity-retry.js";
import { agentService } from "../services/agents.js";
import { budgetService } from "../services/budgets.js";
import { heartbeatService } from "../services/heartbeat.js";
@@ -24,9 +26,14 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled
let reads: ReturnType<typeof vi.spyOn>;
let draining = false;
const errors: unknown[] = [];
const retryQueries: string[] = [];
beforeAll(async () => {
if (!external) database = await startEmbeddedPostgresTestDatabase("run-dispatch-work-");
db = createDb(external ?? database!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
db.$client.options.debug = (_connection, query) => {
if (query.startsWith("select") && query.includes('"heartbeat_runs"') &&
query.includes("startupCancellation") && query.includes('"scheduled_retry_at"')) retryQueries.push(query);
};
await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
}, 90_000);
beforeEach(async () => {
@@ -88,6 +95,45 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled
await fixture(); await worker!.stop(); await coordinator!.stop();
await start(); await quiet();
});
it.each(["agent", "issue"] as const)("does not scan global scheduled retries for repeated ordinary %s edits", async kind => {
const f = await fixture();
const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Edit source", status: "backlog" }).returning();
await start(); await quiet();
const before = retryQueries.length;
expect(before).toBeGreaterThan(0); // The passive observer saw real startup retry SQL.
for (let index = 0; index < 12; index++) {
if (kind === "agent") await agentService(db).update(f.agentId, { name: `Edited agent ${index}` });
else await issueService(db).update(issue.id, { title: `Edited issue ${index}` });
await new Promise(resolve => setTimeout(resolve, 100));
}
await quiet();
expect(retryQueries).toHaveLength(before);
});
it.each([false, true])("observes a native capacity retry only after its transaction settles (rollback=%s)", async rollback => {
const f = await fixture();
const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Capacity retry", status: "in_progress",
assigneeAgentId: f.agentId }).returning();
await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issue.id }).where(eq(heartbeatRuns.id, f.runId));
await start(); await quiet();
let retryId = "";
const writing = db.transaction(async tx => {
retryId = await materializeNativeProviderCapacityRetry({ tx: tx as unknown as typeof db, ...f, issueId: issue.id });
expect(databaseWorkPending(db, DELIVERY_QUEUES.scheduledRetryDispatch)).toBe(true);
expect(await status(retryId)).toBeUndefined();
if (rollback) throw new Error("native capacity rollback");
});
if (rollback) {
await expect(writing).rejects.toThrow("native capacity rollback");
await quiet(); expect(await status(retryId)).toBeUndefined();
} else {
await writing;
const retry = await status(retryId);
expect(retry.status).toBe("scheduled_retry");
await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(retry.scheduledRetryAt!.getTime()), { timeout: 5_000 });
await heartbeat.cancelRun(retryId, "fixture teardown", { suppressImmediateRecovery: true });
await quiet();
}
});
it("ordinary retry creation wakes the worker, respects the exact deadline, dispatches and settles", async () => {
const f = await fixture(); await start(); await quiet();
const result = await heartbeat.scheduleBoundedRetry(f.runId, { delayMs: 4_000, random: () => 0 });
server/src/__tests__/task-watchdogs-classifier.test.ts
+1 -36
View File
@@ -1,8 +1,6 @@
import { describe, expect, it, vi } from "vitest";
import { describe, expect, it } from "vitest";
import { classifyTaskWatchdogSubtree, type TaskWatchdogClassifierIssue } from "../services/task-watchdogs.ts";
import { createIncompleteTaskWatchdogScanReporter } from "../services/task-watchdog-diagnostics.ts";
const companyId = "company-1";
const sourceId = "source-1";
const childId = "child-1";
@@ -37,39 +35,6 @@ function classify(overrides: Partial<Parameters<typeof classifyTaskWatchdogSubtr
}
describe("task watchdog subtree classifier", () => {
it("reports an unchanged incomplete scan once across startup and a day of periodic ticks", () => {
const warn = vi.fn();
const report = createIncompleteTaskWatchdogScanReporter(warn);
report({ incomplete: 2, incompleteIssueIds: [sourceId, childId] }, "startup");
for (let tick = 0; tick < 2880; tick += 1) {
report({ incomplete: 2, incompleteIssueIds: [childId, sourceId] }, "periodic");
}
expect(warn).toHaveBeenCalledTimes(1);
expect(warn).toHaveBeenCalledWith(
{ incomplete: 2, newIncompleteIssueIds: [sourceId, childId] },
"startup task-watchdog reconciliation found incomplete subtree scans",
);
});
it("reports new incomplete issues and reports a resolved issue if it becomes incomplete again", () => {
const warn = vi.fn();
const report = createIncompleteTaskWatchdogScanReporter(warn);
report({ incomplete: 1, incompleteIssueIds: [sourceId] }, "startup");
report({ incomplete: 2, incompleteIssueIds: [sourceId, childId] }, "periodic");
expect(warn).toHaveBeenLastCalledWith(
{ incomplete: 2, newIncompleteIssueIds: [childId] },
"periodic task-watchdog reconciliation found incomplete subtree scans",
);
report({ incomplete: 0, incompleteIssueIds: [] }, "periodic");
expect(warn).toHaveBeenCalledTimes(2);
report({ incomplete: 1, incompleteIssueIds: [sourceId] }, "periodic");
expect(warn).toHaveBeenCalledTimes(3);
expect(warn).toHaveBeenLastCalledWith(
{ incomplete: 1, newIncompleteIssueIds: [sourceId] },
"periodic task-watchdog reconciliation found incomplete subtree scans",
);
});
it("does not produce a stopped fingerprint from an incomplete scan", () => {
const result = classify({ subtreeIncomplete: true });
expect(result.state).toBe("incomplete");