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

fix(server): schedule issue monitors without heartbeat polling (#17863)

Browse Source 
<!-- entire-trail-link-start -->
https://entire.com/gh/paperclipai/-/trails/56/changes/paperclip/56
<!-- entire-trail-link-end -->

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Issue monitors save a future check on an existing issue.
> - The heartbeat timer still scans monitors on every tick, even when
none are due.
> - The saved deadline and claim already provide durable scheduling
state.
> - This pull request gives monitors their own deadline worker and
connects state changes to it.
> - Idle monitor work stops querying. Existing admission and recovery
rules remain in force.

## Linked Issues or Issue Description

Refs #17840, #16325. Related: #17171 addresses rearming a monitor after
its admitted run fails. This PR changes when monitor inspection runs and
follows the merged routine prerequisite.

**What existing behavior does this improve?**

Issue-monitor scheduling and idle database use.

**Current behavior**

Heartbeat ticks call the combined agent and monitor sweep. The existing
next-deadline query only retains future monitors. Polling covers overdue
leftovers, stale claims, and changes that make a saved monitor eligible.

**Proposed behavior**

Use the shared deadline worker. Retain overdue work with bounded
retries. Wake after committed changes to monitor state, issue
eligibility, company state, or native ownership. Keep legacy agent ticks
separate for the next slice.

## What Changed

- Add monitor-specific sweep and deadline methods to the existing
scheduling owner. Keep the manual combined method.
- Include overdue work and the strict five-minute claim fence. Exclude
active native owners until their state changes.
- Register monitors with the shared deadline worker. Heartbeat ticks
wake only the legacy agent timer worker.
- Connect direct issue-state writers and existing execution-control
notifications to monitor work. Checkout uses a short transaction around
its existing CAS. Provider work stays outside it.
- Add real database lifecycle tests and retain routine, startup,
admission, execution-control, and native-projection coverage.
- Record the obligation, producer inventory, recovery behavior, and
limits in `doc/plans/2026-10-27-issue-monitor-sleep-audit.md`.

## Verification

- **1,233 passing cases in 35 complete files**, no skips, on accepted
source `c24aadfc91e9394d73c799f9e05fc67202160557`. Coverage includes
real monitor scheduling/admission, native settlement, committed
producers, retained routines and all shared deadline consumer
integration suites.
- Database build, production server typechecking/build, module
boundaries, target migration order and Runner profile identity pass.
Twenty-four fresh PostgreSQL 17 schemas plus per-file fixtures applied
all 412 journal entries through index 413; every selected source hash
and temporary fixture import was restored exactly.
- Normal parent reconciliation preserves master's admitted-wake result
and scoped issue deletion while keeping committed monitor notifications.
All other paths match the automatic merge; packages remain byte-exact to
the qualified parent. The final routine-parent publication changes only
its audit.
- Historical failures and corrections remain in [the monitor
audit](doc/plans/2026-10-27-issue-monitor-sleep-audit.md). Old
delegated-Slack rate and Runner drain CI failures are retained; the
current inherited clock fixture and package source are byte-identical to
merged #17835. Fresh final-head CI remains required.
- No full local repository-suite, live provider, deployed suspend/wake,
external-alarm, production-load or fleet-savings claim. Final-head
CI/security, exact-head Greptile 5/5 and actionable Ponytail findings
remain separate merge gates.

## Risks

- A missing producer notification can strand a monitor after polling is
removed. The audit covers ordinary writes, direct status/assignment
writers, recovery, and native ownership changes.
- The writer model is process-local. Independent SQL writers and other
replicas need a supported notification or reconciliation contract.
- Overdue work uses the configured heartbeat interval as a retry floor
for the monitor worker. New notifications during that backoff wait for
its next pass. Normal state-change bursts coalesce for one second.
- Notification fanout can inspect monitors during other active work. An
unchanged idle subsystem has no recurring monitor scan.
- The legacy agent timer still polls. Cloud sleep proofs and external
alarm coverage are unchanged.

## Model Used

OpenAI Codex (GPT-6-based coding agent), using repository inspection,
ordinary Git merges, composition proofs, PostgreSQL fixtures and
complete local integration suites. No subagents were used in this
reconciliation.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
DottaPaperclipCodex
 authored and GitHub committed 2026-10-29 08:40:56 -05:00
1 parent 929e840595
commit 0f6f0a3fee
31 files changed
+865 -155
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/__tests__/issue-monitor-scheduler.test.ts
+2 -2
View File
@@ -261,9 +261,9 @@ describeEmbeddedPostgres("issue monitor scheduler", () => {
const unrelatedPolicy = { commentRequired: false, futureAuthorization: { preserve: true } };
await db.update(issues).set({ executionPolicy: { ...before.executionPolicy, ...unrelatedPolicy } }).where(eq(issues.id, issueId));
const result = await heartbeat.tickTimers(tickAt);
const result = await heartbeat.tickIssueMonitors(tickAt);
expect(result.enqueued).toBe(1);
expect(result.triggered).toBe(1);
const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0]!);
expect(issue.monitorNextCheckAt).toBeNull();
server/src/__tests__/routine-schedule-work.test.ts
+10 -7
View File
@@ -7,7 +7,8 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } fro
import { routineService } from "../services/routines.js";
import { budgetService } from "../services/budgets.js";
import { companyService } from "../services/companies.js";
import { registerScheduledWork } from "../services/scheduled-work.js";
import { registerDeadlineWork } from "../services/deadline-work.js";
import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
import { idleWorkSnapshot } from "../services/task-admission.js";
@@ -54,14 +55,16 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
const svc = service();
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true,
onError: error => errors.push(error) });
const workers = registerScheduledWork(coordinator, db, { retryMs: 30_000,
const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.routineSchedules, {
retryMs: 30_000,
canRun: () => !draining,
tickTimers: async () => {},
tickRoutines: options.run ?? (signal => svc.tickScheduledTriggers(new Date(), () => !draining && !signal.aborted)),
nextRoutineRunAt: options.next ?? (() => svc.nextScheduledTriggerAt()),
run: options.run ?? (signal => svc.tickScheduledTriggers(new Date(), () => !draining && !signal.aborted)),
next: options.next ?? (() => svc.nextScheduledTriggerAt()),
});
await Promise.all([workers.routines.ready, workers.timers.ready]);
return workers;
await worker.ready;
return worker;
}
async function settled() {
await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
server/src/__tests__/scheduled-work-postgres.test.ts
+5 -2
View File
@@ -5,6 +5,7 @@ import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
import type { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
import { DELIVERY_QUEUES, type DeliveryQueue } from "../services/delivery-work-notifications.js";
import { heartbeatService } from "../services/heartbeat.js";
import { routineService } from "../services/routines.js";
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { registerScheduledWork } from "../services/scheduled-work.js";
@@ -60,12 +61,14 @@ describe.skipIf(!support.supported)("scheduled deadline database queries", () =>
registerScheduledWork({ register: (queue, task) => {
tasks.set(queue, task);
return { ready: Promise.resolve(), wake: () => {} };
} }, db, { retryMs: 60_000, tickTimers: async () => {}, tickRoutines: async () => {}, canRun: () => true,
} }, { retryMs: 60_000, tickAgentTimers: async () => {}, tickIssueMonitors: async () => {},
nextIssueMonitorAt: () => heartbeatService(db).nextIssueMonitorAt(), tickRoutines: async () => {}, canRun: () => true,
nextRoutineRunAt: () => routineService(db).nextScheduledTriggerAt() });
for (const task of tasks.values()) await task.run(new AbortController().signal);
expect(await routineService(db).nextScheduledTriggerAt()).toBe(at(-1).getTime());
expect(tasks.get(DELIVERY_QUEUES.routineSchedules)!.nextRunAt!()).toBeGreaterThan(now);
expect(tasks.get(DELIVERY_QUEUES.issueMonitors)!.nextRunAt!()).toBe(at(15).getTime());
expect(await heartbeatService(db).nextIssueMonitorAt()).toBe(at(-1).getTime());
expect(tasks.get(DELIVERY_QUEUES.issueMonitors)!.nextRunAt!()).toBe(now + 60_000);
await db.execute(sql`truncate table companies restart identity cascade`);
vi.setSystemTime(at(2));
for (const task of tasks.values()) await task.run(new AbortController().signal);
server/src/__tests__/scheduled-work.test.ts
+77 -25
View File
@@ -1,19 +1,19 @@
import { PgDialect } from "drizzle-orm/pg-core";
import type { Db } from "@paperclipai/db";
import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
import { afterEach, describe, expect, it, vi } from "vitest";
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../services/delivery-work-notifications.js";
import { registerScheduledWork } from "../services/scheduled-work.js";
import { idleWorkSnapshot } from "../services/task-admission.js";
import { idleWorkSnapshot, startTaskDrain, stopTaskDrain } from "../services/task-admission.js";
const coordinators: ReturnType<typeof createDeliveryWorkCoordinator>[] = [];
afterEach(async () => {
for (const coordinator of coordinators.splice(0)) await coordinator.stop();
stopTaskDrain();
vi.useRealTimers();
});
function fixture(input: { routine?: number | null; monitor?: number | null; tickTimers?: () => Promise<void>; tickRoutines?: () => Promise<void> } = {}) {
function fixture(input: { routine?: number | null; monitor?: number | null; tickIssueMonitors?: () => Promise<void>; tickRoutines?: () => Promise<void> } = {}) {
vi.useFakeTimers();
vi.setSystemTime(new Date("2020-01-01T00:00:00.000Z"));
const dialect = new PgDialect();
@@ -26,8 +26,7 @@ function fixture(input: { routine?: number | null; monitor?: number | null; tick
const text = compiled.sql;
if (text.includes("pg_current_xact_id")) return [{ xid: "42" }];
if (failRead) throw new Error("database unavailable");
const at = deadlines[text.includes("routine_triggers") ? "routine" : "monitor"];
return [{ next_at: at !== null && at > Date.parse(String(compiled.params[0])) ? new Date(at) : null }];
throw new Error(`Unexpected database query: ${text}`);
});
const owner = installDatabaseWorkSignals({ execute, async transaction(callback: (tx: any) => Promise<any>) {
return callback({ execute, transaction: owner.transaction });
@@ -35,7 +34,14 @@ function fixture(input: { routine?: number | null; monitor?: number | null; tick
const errors = vi.fn();
const coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => canRun, canReconcile: () => canReconcile, onError: errors });
coordinators.push(coordinator);
const tickTimers = vi.fn(input.tickTimers ?? (async () => {}));
const tickIssueMonitors = vi.fn(input.tickIssueMonitors ?? (async () => {
if (deadlines.monitor !== null && deadlines.monitor <= Date.now()) deadlines.monitor = null;
}));
const tickAgentTimers = vi.fn(async () => {});
const nextIssueMonitorAt = vi.fn(async () => {
if (failRead) throw new Error("database unavailable");
return deadlines.monitor;
});
const tickRoutines = vi.fn(input.tickRoutines ?? (async () => {
if (deadlines.routine !== null && deadlines.routine <= Date.now()) deadlines.routine = null;
}));
@@ -43,14 +49,60 @@ function fixture(input: { routine?: number | null; monitor?: number | null; tick
if (failRead) throw new Error("database unavailable");
return deadlines.routine;
});
const workers = registerScheduledWork(coordinator, owner as unknown as Db,
{ retryMs: 60_000, tickTimers, tickRoutines, nextRoutineRunAt, canRun: () => canRun });
return { coordinator, workers, owner, deadlines, execute, nextRoutineRunAt, tickTimers, tickRoutines, errors,
ready: () => Promise.all([workers.routines.ready, workers.timers.ready]),
const workers = registerScheduledWork(coordinator,
{ retryMs: 60_000, tickAgentTimers, tickIssueMonitors, nextIssueMonitorAt, tickRoutines, nextRoutineRunAt, canRun: () => canRun });
return { coordinator, workers, owner, deadlines, execute, nextRoutineRunAt, nextIssueMonitorAt, tickAgentTimers, tickIssueMonitors, tickRoutines, errors,
ready: () => Promise.all([workers.routines.ready, workers.monitors.ready, workers.timers.ready]),
canRun: (value: boolean) => { canRun = value; }, canReconcile: (value: boolean) => { canReconcile = value; }, failRead: (value: boolean) => { failRead = value; } };
}
describe("scheduled routine and monitor deadlines", () => {
it.each(["run", "deadline read"])("retains a monitor pass interrupted during %s by a maintenance drain", async boundary => {
const f = fixture({ tickIssueMonitors: async () => { if (boundary === "run") startTaskDrain(); } });
if (boundary === "deadline read") f.nextIssueMonitorAt.mockImplementationOnce(async () => { startTaskDrain(); return null; });
await f.ready();
expect(f.coordinator.nextWakeAt()).not.toBeNull();
expect(f.nextIssueMonitorAt).toHaveBeenCalledTimes(boundary === "run" ? 0 : 1);
await vi.advanceTimersByTimeAsync(120_000);
expect(f.tickIssueMonitors).toHaveBeenCalledOnce();
f.tickIssueMonitors.mockImplementation(async () => {});
stopTaskDrain();
await vi.advanceTimersByTimeAsync(60_000);
expect(f.tickIssueMonitors).toHaveBeenCalledTimes(2);
expect(f.coordinator.nextWakeAt()).toBeNull();
});
it("bounds notification scans to the configured cadence while preserving known deadlines", async () => {
const origin = Date.UTC(2020, 0, 1);
const f = fixture({ monitor: origin + 150_000 });
await f.ready();
for (let second = 0; second < 120; second++) {
await f.owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors));
await vi.advanceTimersByTimeAsync(1000);
}
expect(f.nextIssueMonitorAt).toHaveBeenCalledTimes(3);
expect(f.tickIssueMonitors).toHaveBeenCalledTimes(3);
await vi.advanceTimersByTimeAsync(30_000);
expect(f.tickIssueMonitors).toHaveBeenCalledTimes(4);
expect(f.coordinator.nextWakeAt()).toBeNull();
await vi.advanceTimersByTimeAsync(120_000);
expect(f.nextIssueMonitorAt).toHaveBeenCalledTimes(4);
});
it("legacy agent timer wakes do not query or dispatch issue monitors", async () => {
const f = fixture();
await f.ready();
const before = f.nextIssueMonitorAt.mock.calls.length;
for (let i = 0; i < 5; i++) {
f.workers.timers.wake();
await vi.advanceTimersByTimeAsync(60_000);
}
expect(f.tickAgentTimers).toHaveBeenCalledTimes(6);
expect(f.tickIssueMonitors).toHaveBeenCalledTimes(1);
expect(f.nextIssueMonitorAt).toHaveBeenCalledTimes(before);
expect(f.coordinator.nextWakeAt()).toBeNull();
});
it("dispatches at each saved deadline without waiting for a periodic tick", async () => {
const now = Date.UTC(2020, 0, 1);
const f = fixture({ routine: now + 10_000, monitor: now + 20_000 });
@@ -61,16 +113,16 @@ describe("scheduled routine and monitor deadlines", () => {
expect(f.tickRoutines).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(1);
expect(f.tickRoutines).toHaveBeenCalledTimes(2);
expect(f.tickTimers).toHaveBeenCalledTimes(1);
expect(f.tickIssueMonitors).toHaveBeenCalledTimes(1);
expect(f.coordinator.nextWakeAt()).toBe(now + 20_000);
await vi.advanceTimersByTimeAsync(10_000);
expect(f.tickTimers).toHaveBeenCalledTimes(2);
expect(f.tickIssueMonitors).toHaveBeenCalledTimes(2);
expect(f.coordinator.nextWakeAt()).toBeNull();
expect(f.execute).toHaveBeenCalledTimes(2);
expect(f.nextIssueMonitorAt).toHaveBeenCalledTimes(2);
expect(f.nextRoutineRunAt).toHaveBeenCalledTimes(2);
expect(idleWorkSnapshot().active).toBe(0);
await vi.advanceTimersByTimeAsync(3_600_000);
expect(f.execute).toHaveBeenCalledTimes(2);
expect(f.nextIssueMonitorAt).toHaveBeenCalledTimes(2);
expect(f.nextRoutineRunAt).toHaveBeenCalledTimes(2);
});
@@ -78,16 +130,16 @@ describe("scheduled routine and monitor deadlines", () => {
let release!: () => void;
const held = new Promise<void>(resolve => { release = resolve; });
const tick = vi.fn().mockImplementationOnce(() => held).mockImplementation(async () => {
if (kind === "routine") f.deadlines.routine = null;
f.deadlines[kind] = null;
});
const now = Date.UTC(2020, 0, 1);
const f = fixture({ [kind]: now + 10_000,
...(kind === "routine" ? { tickRoutines: tick } : { tickTimers: tick }) });
...(kind === "routine" ? { tickRoutines: tick } : { tickIssueMonitors: tick }) });
try {
await (kind === "routine" ? f.workers.timers.ready : f.workers.routines.ready);
await (kind === "routine" ? f.workers.monitors.ready : f.workers.routines.ready);
await vi.advanceTimersByTimeAsync(12_000);
} finally { release(); await f.ready(); }
const expected = kind === "routine" ? now + 60_000 : Date.now();
const expected = now + 60_000;
expect(f.coordinator.nextWakeAt()).toBe(expected);
await vi.advanceTimersByTimeAsync(expected - Date.now() + 1);
expect(tick).toHaveBeenCalledTimes(2);
@@ -95,15 +147,15 @@ describe("scheduled routine and monitor deadlines", () => {
expect(idleWorkSnapshot().active).toBe(0);
});
it("retains capped overdue work with bounded spacing, then disarms after completion", async () => {
it.each(["routine", "monitor"] as const)("retains capped overdue %s work with bounded spacing, then disarms after completion", async kind => {
const tick = vi.fn(async () => {});
const now = Date.UTC(2020, 0, 1);
const f = fixture({ routine: now - 10_000, tickRoutines: tick });
const f = fixture({ [kind]: now - 10_000, ...(kind === "routine" ? { tickRoutines: tick } : { tickIssueMonitors: tick }) });
await f.ready();
expect(f.coordinator.nextWakeAt()).toBe(now + 60_000);
await vi.advanceTimersByTimeAsync(59_999);
expect(tick).toHaveBeenCalledTimes(1);
f.deadlines.routine = null;
f.deadlines[kind] = null;
await vi.advanceTimersByTimeAsync(1);
expect(tick).toHaveBeenCalledTimes(2);
expect(f.coordinator.nextWakeAt()).toBeNull();
@@ -133,7 +185,7 @@ describe("scheduled routine and monitor deadlines", () => {
expect(idleWorkSnapshot().active).toBe(0);
await vi.advanceTimersByTimeAsync(8999);
expect(f.tickRoutines).toHaveBeenCalledTimes(3);
expect(f.tickTimers).toHaveBeenCalledTimes(1);
expect(f.tickIssueMonitors).toHaveBeenCalledTimes(1);
});
it("keeps monitor dispatch responsive while routine work is slow", async () => {
@@ -141,9 +193,9 @@ describe("scheduled routine and monitor deadlines", () => {
const held = new Promise<void>(resolve => { release = resolve; });
const f = fixture({ monitor: Date.UTC(2020, 0, 1) + 10_000, tickRoutines: () => held });
try {
await f.workers.timers.ready;
await f.workers.monitors.ready;
await vi.advanceTimersByTimeAsync(10_000);
expect(f.tickTimers).toHaveBeenCalledTimes(2);
expect(f.tickIssueMonitors).toHaveBeenCalledTimes(2);
expect(f.tickRoutines).toHaveBeenCalledTimes(1);
} finally { release(); await f.workers.routines.ready; }
});
@@ -157,7 +209,7 @@ describe("scheduled routine and monitor deadlines", () => {
expect(idleWorkSnapshot().active).toBe(0);
f.canReconcile(false);
await vi.advanceTimersByTimeAsync(60_000);
expect(f.execute).toHaveBeenCalledTimes(1);
expect(f.nextIssueMonitorAt).toHaveBeenCalledTimes(1);
expect(f.nextRoutineRunAt).toHaveBeenCalledTimes(1);
await f.coordinator.stop();
f.canRun(true);
server/src/__tests__/server-startup-feedback-export.test.ts
+5 -3
View File
@@ -154,7 +154,9 @@ const {
sweepStaleIssueLocks: vi.fn(async () => ({ cleared: 0 })),
sweepPendingCleanupLeases: vi.fn(async () => ({ swept: 0, destroyed: 0, capped: 0 })),
sweepExpiredRuntimeStatuses: vi.fn(() => 0),
tickTimers: vi.fn(async () => ({ checked: 0, enqueued: 0, skipped: 0 })),
tickAgentTimers: vi.fn(async () => ({ checked: 0, enqueued: 0, skipped: 0 })),
tickIssueMonitors: vi.fn(async () => ({ checked: 0, triggered: 0, skipped: 0 })),
nextIssueMonitorAt: vi.fn(async () => null),
};
const heartbeatServiceFactoryMock = vi.fn(() => heartbeatServiceMock);
const issueThreadInteractionServiceMock = {
@@ -1168,7 +1170,7 @@ describe("startServer feedback export wiring", () => {
await settleStartupHousekeeping();
expect(heartbeatServiceMock.reapOrphanedRuns).not.toHaveBeenCalled();
expect(heartbeatServiceMock.tickTimers).not.toHaveBeenCalled();
expect(heartbeatServiceMock.tickAgentTimers).not.toHaveBeenCalled();
expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).toHaveBeenCalledTimes(1);
expect(intervalCallback).not.toBeNull();
@@ -1176,7 +1178,7 @@ describe("startServer feedback export wiring", () => {
await Promise.resolve();
await Promise.resolve();
expect(heartbeatServiceMock.tickTimers).not.toHaveBeenCalled();
expect(heartbeatServiceMock.tickAgentTimers).not.toHaveBeenCalled();
expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).toHaveBeenCalledTimes(1);
expect(issueThreadInteractionServiceMock.sweepMergedPullRequestConfirmations).toHaveBeenCalledTimes(1);
expect(executionWorkspaceServiceMock.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1);
server/src/index.ts
+7 -4
View File
@@ -2078,21 +2078,24 @@ async function startServerWithDatabaseTeardown(
}));
};
const startHeartbeatScheduler = () => {
const scheduled = registerScheduledWork(deliveryWork, db, {
const scheduled = registerScheduledWork(deliveryWork, {
retryMs: config.heartbeatSchedulerIntervalMs,
tickTimers: async () => {
tickAgentTimers: async () => {
if ((await heartbeat.resolveSchedulingSuppression()).suppressed) return;
const result = await heartbeat.tickTimers(new Date());
const result = await heartbeat.tickAgentTimers(new Date());
if (result.enqueued > 0) logger.info({ ...result }, "heartbeat timer tick enqueued runs");
},
canRun: () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped,
nextIssueMonitorAt: () => heartbeat.nextIssueMonitorAt(),
tickIssueMonitors: signal => heartbeat.tickIssueMonitors(new Date(),
() => !signal.aborted && !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped),
nextRoutineRunAt: () => routines.nextScheduledTriggerAt(),
tickRoutines: signal => routines.tickScheduledTriggers(new Date(),
() => !signal.aborted && !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).then(result => {
if (result.triggered > 0) logger.info({ ...result }, "routine scheduler tick enqueued runs");
}),
});
trackHeartbeatSchedulerWork(Promise.all([scheduled.timers.ready, scheduled.routines.ready]));
trackHeartbeatSchedulerWork(Promise.all([scheduled.timers.ready, scheduled.monitors.ready, scheduled.routines.ready]));
startHeartbeatSchedulerInterval(createHeartbeatSchedulerTick(scheduled));
};
trackHeartbeatSchedulerWork(runStartupHousekeeping().then(() => {
server/src/modules/active-run-watchdog/adapters/postgres.ts
+2
View File
@@ -1,3 +1,4 @@
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js";
import { notifyExternalAgentSettlement } from "../../external-agents/index.js";
import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js";
@@ -347,6 +348,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch
await notifyCancelledWorkspacePreservation(tx, { companyId, runId: input.run.id });
await notifyUnsafeWorkspaceExportRecovery(tx, { companyId, issueId: input.sourceIssue.id });
await notifyNativeStatusReassessment(tx, { companyId, issueId: input.sourceIssue.id });
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
const [updatedRun] = await tx
.update(heartbeatRuns)
.set({
server/src/services/access.ts
+2
View File
@@ -1,3 +1,4 @@
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
import { notifyChatCompanyAuthorityChange } from "./chat-run-milestone-notifications.js";
@@ -742,6 +743,7 @@ export function accessService(db: Db) {
sql`${issues.status} not in ('done', 'cancelled')`,
);
await notifyNativeStatusReassessment(tx, { companyId });
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
const resetInProgress = await tx
.update(issues)
.set({
server/src/services/agent-conversations.ts
+3
View File
@@ -1,3 +1,4 @@
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
import { FAST_RESPONSE_AGENT_GUIDANCE, fastResponseHistoryBody } from "@paperclipai/shared";
import {
@@ -268,6 +269,7 @@ export async function prepareConversationTurn(
context.conversationReplayThroughCommentId = replayThrough?.id ?? commentId;
}
}
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await tx
.update(issues)
.set({
@@ -386,6 +388,7 @@ export async function settleConversationTurn(
issue.conversationState === conversationState
)
return true;
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
if (status === "in_review") await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: issue.companyId, issueId: issue.id });
await tx
.update(issues)
server/src/services/agent-deletion.ts
+2
View File
@@ -1,3 +1,4 @@
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js";
@@ -35,6 +36,7 @@ export async function deleteAgentDependencies(tx: Db, companyId: string, id: str
});
await notifyExternalConversationSettlement(tx, { companyId });
await issueThreadInteractionService(tx).cancelPendingForDeletedAddressee(companyId, id);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await tx.update(agents).set({ reportsTo: null }).where(eq(agents.reportsTo, id));
await tx
.update(issues)
server/src/services/budgets.ts
+2
View File
@@ -302,6 +302,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
if (scope.paused) return;
await notifyCompanyRetentionWork(db, policy.companyId);
await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules);
await notifyDeliveryWork(db, DELIVERY_QUEUES.issueMonitors);
await notifyDeliveryWork(db, DELIVERY_QUEUES.agentLifecycle, policy.companyId);
await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync);
await db
@@ -360,6 +361,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
if (scope.pauseReason !== "budget") return;
await notifyCompanyRetentionWork(db, policy.companyId);
await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules);
await notifyDeliveryWork(db, DELIVERY_QUEUES.issueMonitors);
await notifyDeliveryWork(db, DELIVERY_QUEUES.agentLifecycle, policy.companyId);
await db
.update(companies)
server/src/services/companies.ts
+2
View File
@@ -334,6 +334,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
if (companyPatch.status !== undefined && companyPatch.status !== existing.status) {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyCompanyRetentionWork(tx, id);
await notifyChatInboundWakeNotices(tx, { companyId: id });
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id);
@@ -488,6 +489,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
if (!wasAlreadyArchived) {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyCompanyRetentionWork(tx, id);
await notifyChatInboundWakeNotices(tx, { companyId: id });
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id);
server/src/services/company-deletion.ts
+1
View File
@@ -75,6 +75,7 @@ export async function deleteCompany(db: Db, id: string) {
await chatCompanyDeletion.deleteCompanyData(tx, id);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.secretProposalExpiry, id);
await notifyToolReviewChange(tx, id);
await agentObservabilityService(tx as unknown as Db).removeForCompanyDeletion(id);
server/src/services/deadline-work.ts
+6 -2
View File
@@ -23,7 +23,7 @@ export function registerDeadlineWork(
pausePassiveMaintenance: true,
onSettled: () => { requested++; },
run: async signal => {
if (!task.canRun()) { requested++; return; }
if (!task.canRun()) { requested++; next = null; return; }
if (Date.now() < earliest) { next = earliest; return; }
if (Date.now() < coalesceUntil && (next === null || next > Date.now())) {
next = next === null ? coalesceUntil : Math.min(next, coalesceUntil);
@@ -33,8 +33,12 @@ export function registerDeadlineWork(
const startedAt = Date.now();
earliest = startedAt + task.retryMs;
await task.run(signal);
if (signal.aborted) return;
if (signal.aborted || !task.canRun()) { next = null; return; }
const deadline = await task.next();
if (!task.canRun()) { next = null; return; }
// Back off failed or still-overdue work, not a successfully read future deadline.
earliest = deadline === null || deadline > Date.now()
? 0 : startedAt + (task.continuationMs ?? task.retryMs);
server/src/services/delivery-work-notifications.ts
+1
View File
@@ -110,6 +110,7 @@ export const DELIVERY_QUEUES = {
terminalWorkspaces: "terminal-workspaces",
routineSchedules: "routine-schedules",
issueMonitors: "issue-monitors",
agentTimers: "agent-timers",
caseHandoffs: "case-handoffs",
nativeCheckpointExpiration: "native-checkpoint-expiration",
} as const;
server/src/services/execution-control-notifications.ts
+2 -1
View File
@@ -3,7 +3,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork, type DeliveryQueue } from "./deliv
export async function notifyExecutionRunWork(transaction: object) {
for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization,
DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition]) {
DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors]) {
await notifyDeliveryWork(transaction, queue);
}
}
@@ -14,6 +14,7 @@ export function writeExecutionControl<T>(
write: (tx: Db) => PromiseLike<T>,
): Promise<T> {
return db.transaction(async transaction => {
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.issueMonitors);
for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic);
return write(transaction as unknown as Db);
});
server/src/services/execution-recovery-resolution.ts
+1
View File
@@ -594,6 +594,7 @@ export async function settleUnrecoverableExecutions(
if (current) {
await notifyNativeStatusReassessment(tx, { companyId: task.companyId, issueId: task.id });
if (task.status !== "blocked") await notifyChatResponseAvailability(tx, { companyId: task.companyId, issueId: task.id });
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
const [projected] = await tx
.update(issues)
.set({
server/src/services/heartbeat.ts
+6
View File
@@ -1560,6 +1560,9 @@ export function heartbeatService(
const {
triggerIssueMonitor,
tickTimers,
tickAgentTimers,
tickIssueMonitors,
nextIssueMonitorAt,
recoverActiveSessionGoals,
recoverPendingSessionGoalActions,
} = createHeartbeatScheduling(db, {
@@ -4264,6 +4267,9 @@ export function heartbeatService(
buildRunOutputSilence,
tickTimers,
tickAgentTimers,
tickIssueMonitors,
nextIssueMonitorAt,
cancelRun: (runId: string, reason?: string, options?: CancelRunOptions) =>
cancelRunInternal(runId, reason, options),
server/src/services/heartbeat/queue.ts
+2
View File
@@ -1,3 +1,4 @@
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js";
import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery, notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js";
@@ -3904,6 +3905,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
].join("\n");
await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: issue.companyId, issueId: issue.id });
await notifyNativeStatusReassessment(tx, { companyId: issue.companyId, issueId: issue.id });
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
if (issue.status !== "blocked") await notifyChatResponseAvailability(tx, { companyId: issue.companyId, issueId: issue.id });
await tx
.update(issues)
server/src/services/heartbeat/scheduling.test.ts
+334 -8
View File
@@ -1,12 +1,22 @@
import { randomUUID } from "node:crypto";
import { setTimeout as realDelay } from "node:timers/promises";
import type { Db } from "@paperclipai/db";
import {
activityLog, agents, agentSessionGoalActions, agentTaskSessions, companies, createDb,
activityLog, applyPendingMigrations, reconcileDatabaseWork, agents, agentSessionGoalActions, agentTaskSessions, companies, createDb,
completionContracts, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issueComments, issues, nativeRunFinalizations, nativeRunResults,
} from "@paperclipai/db";
import { eq, sql } from "drizzle-orm";
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
import { heartbeatService } from "../heartbeat.js";
import { instanceSettingsService } from "../instance-settings.js";
import { companyService } from "../companies.js";
import { issueService } from "../issues.js";
import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
import { registerScheduledWork } from "../scheduled-work.js";
import { idleWorkSnapshot, readTaskDrain, startTaskDrain, stopTaskDrain } from "../task-admission.js";
import { conflict } from "../../errors.js";
import { createHeartbeatRunPreparation } from "./run-preparation.js";
import { createHeartbeatRunState } from "./run-state.js";
@@ -63,7 +73,7 @@ describe("heartbeat scheduling module boundary", () => {
for (const callback of Object.values(deps)) if (vi.isMockFunction(callback)) expect(callback).not.toHaveBeenCalled();
});
it.each(["tickTimers", "recoverActiveSessionGoals", "recoverPendingSessionGoalActions"] as const)(
it.each(["tickTimers", "tickAgentTimers", "tickIssueMonitors", "recoverActiveSessionGoals", "recoverPendingSessionGoalActions"] as const)(
"honors suppression before %s reads or admits work", async (operation) => {
const database = guardedDatabase();
const deps = callbacks(database.db);
@@ -77,19 +87,24 @@ describe("heartbeat scheduling module boundary", () => {
);
});
const postgres = await getEmbeddedPostgresTestSupport();
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_HEARTBEAT_SCHEDULING_TEST_DATABASE_URL");
const postgres = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
describe.skipIf(!postgres.supported)("heartbeat scheduling database boundary", () => {
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
let db: Db;
beforeAll(async () => {
database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-scheduling-module");
db = createDb(database.connectionString);
if (external) await applyPendingMigrations(external);
else database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-scheduling-module");
db = createDb(external ?? database.connectionString, { idleTimeoutSeconds: 0, maxLifetimeSeconds: 0, connectTimeoutSeconds: 0 });
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
afterAll(async () => {
await db?.$client.end();
await database?.cleanup();
});
beforeEach(async () => {
await reconcileDatabaseWork(db, DELIVERY_QUEUES.issueMonitors);
await db.execute(sql`truncate table companies restart identity cascade`);
});
@@ -319,7 +334,7 @@ describe.skipIf(!postgres.supported)("heartbeat scheduling database boundary", (
expect(f.deps.enqueueWakeup).toHaveBeenCalledTimes(1);
});
it.each(["scheduled", "manual"] as const)("applies expired monitor recovery after workspace settlement without another turn (%s)", async (dispatch) => {
it.each(["scheduled", "manual", "worker"] as const)("applies expired monitor recovery after workspace settlement without another turn (%s)", async (dispatch) => {
const f = await fixture();
const start = new Date();
const nextCheckAt = new Date(start.getTime() + 60_000).toISOString();
@@ -364,17 +379,31 @@ describe.skipIf(!postgres.supported)("heartbeat scheduling database boundary", (
const expiredAt = new Date(start.getTime() + 180_000);
vi.useFakeTimers({ toFake: ["Date"] });
vi.setSystemTime(expiredAt);
let monitorWorker: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
try {
await expect(nativeCompletionFeedback(db, run!.id, result)).rejects.toThrow("persisted, eligible monitor");
expect(await f.scheduler.tickTimers(expiredAt)).toEqual({ checked: 0, enqueued: 0, skipped: 0 });
await expect(f.scheduler.triggerIssueMonitor(f.issue.id, { now: expiredAt })).rejects.toThrow("waiting for the previous run to settle");
if (dispatch === "worker") {
monitorWorker = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } });
const workers = registerScheduledWork(monitorWorker, { retryMs: 1000, canRun: () => true,
tickAgentTimers: async () => {}, tickRoutines: async () => {}, nextRoutineRunAt: async () => null,
tickIssueMonitors: signal => f.scheduler.tickIssueMonitors(new Date(), () => !signal.aborted),
nextIssueMonitorAt: () => f.scheduler.nextIssueMonitorAt(),
});
await Promise.all([workers.monitors.ready, workers.routines.ready, workers.timers.ready]);
expect(monitorWorker.nextWakeAt()).toBeNull();
}
await finalizeNativeRun({ db, runId: run!.id, workspaceFinalizeStatus: "succeeded" });
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run!.id)))[0])
.toMatchObject({ phase: "committed", failureCode: null });
expect(await readIssue(f.issue.id)).toMatchObject({ executionRunId: null, checkoutRunId: null,
monitorNextCheckAt: new Date(nextCheckAt) });
if (dispatch === "manual") await f.scheduler.triggerIssueMonitor(f.issue.id, { now: expiredAt });
else await f.scheduler.tickTimers(expiredAt);
else if (dispatch === "worker") {
vi.setSystemTime(new Date(expiredAt.getTime() + 2000));
await vi.waitFor(async () => expect((await readIssue(f.issue.id)).monitorNextCheckAt).toBeNull(), { timeout: 5000 });
} else await f.scheduler.tickTimers(expiredAt);
expect(await readIssue(f.issue.id)).toMatchObject({ monitorNextCheckAt: null,
executionState: { monitor: { status: "cleared", clearReason: "timeout_exceeded" } } });
expect(await db.select().from(issueComments).where(eq(issueComments.issueId, f.issue.id))).toHaveLength(1);
@@ -384,6 +413,7 @@ describe.skipIf(!postgres.supported)("heartbeat scheduling database boundary", (
expect(f.deps.enqueueWakeup).not.toHaveBeenCalled();
expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, saved!.id)))[0]!.resultJson).toEqual(saved!.resultJson);
} finally {
await monitorWorker?.stop();
vi.useRealTimers();
}
});
@@ -482,4 +512,300 @@ describe.skipIf(!postgres.supported)("heartbeat scheduling database boundary", (
const failed = await db.select().from(agentSessionGoalActions).where(eq(agentSessionGoalActions.id, action!.id));
expect(failed[0]).toMatchObject({ status: "failed", error: "session_goal_control_payload_invalid" });
});
describe("issue monitor deadline lifecycle", () => {
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
let draining = false;
const errors: unknown[] = [];
beforeEach(() => {
draining = false; errors.length = 0;
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
vi.setSystemTime(now);
});
afterEach(async () => {
await coordinator?.stop(); coordinator = undefined;
stopTaskDrain();
vi.useRealTimers(); vi.restoreAllMocks();
expect(errors).toEqual([]);
});
async function start(f: { scheduler: Pick<ReturnType<typeof createHeartbeatScheduling>, "tickIssueMonitors" | "nextIssueMonitorAt"> }) {
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining,
canReconcile: () => true, onError: error => errors.push(error) });
const workers = registerScheduledWork(coordinator, { retryMs: 30_000, canRun: () => !draining,
tickAgentTimers: async () => {}, tickRoutines: async () => {}, nextRoutineRunAt: async () => null,
tickIssueMonitors: signal => f.scheduler.tickIssueMonitors(new Date(), () => !draining && !signal.aborted),
nextIssueMonitorAt: () => f.scheduler.nextIssueMonitorAt(),
});
await Promise.all([workers.monitors.ready, workers.routines.ready, workers.timers.ready]);
return workers;
}
async function advance(milliseconds: number) {
for (let remaining = milliseconds; remaining > 0; remaining -= 1000) {
await vi.advanceTimersByTimeAsync(Math.min(1000, remaining));
await realDelay(5);
}
}
async function quiet() {
await advance(31_000);
if (!draining) await vi.waitFor(() => {
expect(idleWorkSnapshot().active).toBe(0);
expect(coordinator!.nextWakeAt()).toBeNull();
}, { timeout: 5000 });
const select = vi.spyOn(db, "select"), execute = vi.spyOn(db, "execute"), transaction = vi.spyOn(db, "transaction");
try {
await advance(120_000);
expect(select).not.toHaveBeenCalled(); expect(execute).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled();
} finally { select.mockRestore(); execute.mockRestore(); transaction.mockRestore(); }
}
async function schedule(f: Awaited<ReturnType<typeof fixture>>, at: Date, owner = db) {
return issueService(owner).update(f.issue.id, { monitorNextCheckAt: at, executionPolicy: {
mode: "normal", commentRequired: true, stages: [], monitor: { nextCheckAt: at.toISOString() },
} });
}
it.each(["commit", "rollback"] as const)("reconciles worktree activation only after its outer transaction %s", async outcome => {
const f = await fixture();
const deadline = now.getTime() + 3_600_000;
await db.update(issues).set({ monitorNextCheckAt: new Date(deadline),
executionPolicy: { monitor: { nextCheckAt: new Date(deadline).toISOString() } } }).where(eq(issues.id, f.issue.id));
const runtimeEnv = { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "monitor-activation-fixture" };
const settings = instanceSettingsService(db, { runtimeEnv });
await settings.updateExperimental({ enableWorktreeRunExecution: false });
const heartbeat = heartbeatService(db, { runtimeEnv });
await start({ scheduler: heartbeat });
await quiet();
let release!: () => void, entered!: () => void;
const held = new Promise<void>(resolve => { release = resolve; });
const ready = new Promise<void>(resolve => { entered = resolve; });
const writing = db.transaction(async tx => {
await instanceSettingsService(tx as unknown as Db, { runtimeEnv }).updateExperimental({ enableWorktreeRunExecution: true });
entered();
await held;
if (outcome === "rollback") throw new Error("rollback activation");
}).then(() => undefined, error => error as Error);
try {
await ready;
await advance(31_000);
expect(await heartbeat.nextIssueMonitorAt()).toBeNull();
expect((await readIssue(f.issue.id)).monitorAttemptCount).toBe(0);
} finally { release(); }
const error = await writing;
if (outcome === "rollback") {
expect(error?.message).toBe("rollback activation");
await quiet();
} else {
expect(error).toBeUndefined();
await advance(31_000);
await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(deadline), { timeout: 5000 });
await settings.updateExperimental({ enableWorktreeRunExecution: false });
await quiet();
}
expect(await readIssue(f.issue.id)).toMatchObject({ monitorNextCheckAt: new Date(deadline), monitorAttemptCount: 0 });
});
it("uses fresh worktree activation for immediate monitor deadline inspection", async () => {
const f = await fixture();
const runtimeEnv = { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "monitor-fresh-fixture" };
const settings = instanceSettingsService(db, { runtimeEnv });
await settings.updateExperimental({ enableWorktreeRunExecution: false });
const heartbeat = heartbeatService(db, { runtimeEnv });
expect(await heartbeat.nextIssueMonitorAt()).toBeNull();
await settings.updateExperimental({ enableWorktreeRunExecution: true });
expect(await heartbeat.nextIssueMonitorAt()).toBe(dueAt.getTime());
await settings.updateExperimental({ enableWorktreeRunExecution: false });
expect(await heartbeat.nextIssueMonitorAt()).toBeNull();
expect(await heartbeat.tickIssueMonitors()).toEqual({ checked: 0, triggered: 0, skipped: 0 });
expect((await readIssue(f.issue.id)).monitorNextCheckAt).toEqual(dueAt);
});
it("stays quiet with retained history, then a real producer schedules and consumes its deadline", async () => {
const f = await fixture();
await db.update(issues).set({ monitorNextCheckAt: null, executionPolicy: null }).where(eq(issues.id, f.issue.id));
await db.insert(issues).values({ companyId: f.company.id, title: "Old monitor", status: "done",
assigneeAgentId: f.agent.id, monitorNextCheckAt: dueAt });
await start(f); await quiet();
expect(coordinator!.nextWakeAt()).toBeNull();
const at = new Date(Date.now() + 10_000);
await schedule(f, at);
await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(at.getTime()));
await advance(10_000);
await vi.waitFor(async () => expect((await readIssue(f.issue.id)).monitorNextCheckAt).toBeNull());
expect(f.deps.enqueueWakeup).toHaveBeenCalledOnce();
await quiet(); expect(coordinator!.nextWakeAt()).toBeNull();
});
it.each([false, true])("waits for the outer producer transaction and handles rollback=%s", async rollback => {
const f = await fixture();
await db.update(issues).set({ monitorNextCheckAt: null, executionPolicy: null }).where(eq(issues.id, f.issue.id));
await start(f);
let release!: () => void, entered!: () => void;
const hold = new Promise<void>(r => { release = r; });
const written = new Promise<void>(r => { entered = r; });
const at = new Date(Date.now() + 60_000);
const writing = db.transaction(async tx => {
await schedule(f, at, tx as unknown as Db);
entered(); await hold;
if (rollback) throw new Error("rollback monitor");
}).catch(error => { if (!rollback) throw error; expect(error.message).toBe("rollback monitor"); });
try {
await written;
await advance(2000);
expect(f.deps.enqueueWakeup).not.toHaveBeenCalled();
expect((await readIssue(f.issue.id)).monitorNextCheckAt).toBeNull();
} finally { release(); await writing; }
if (rollback) {
await advance(30_000);
await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeNull());
await quiet();
}
else {
await advance(30_000);
await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(at.getTime()));
}
});
it.each(["stop", "expiry"] as const)("resumes a saved monitor after maintenance drain %s", async end => {
const f = await fixture();
const at = new Date(Date.now() + 10_000);
await db.update(issues).set({ monitorNextCheckAt: at }).where(eq(issues.id, f.issue.id));
f.deps.getSchedulingSuppression.mockImplementation(async () => ({ suppressed: readTaskDrain(new Date()) !== null }));
await start(f);
startTaskDrain({ ttlMs: end === "expiry" ? 90_000 : null });
const select = vi.spyOn(db, "select");
try {
await advance(60_000);
expect(f.deps.enqueueWakeup).not.toHaveBeenCalled();
expect(select).not.toHaveBeenCalled();
expect(coordinator!.nextWakeAt()).not.toBeNull();
} finally { select.mockRestore(); }
if (end === "stop") stopTaskDrain();
await advance(60_000);
await vi.waitFor(async () => expect((await readIssue(f.issue.id)).monitorNextCheckAt).toBeNull());
expect(f.deps.enqueueWakeup).toHaveBeenCalledOnce();
await quiet();
});
it("retains a first producer during drain without passive SQL, then resumes", async () => {
const f = await fixture();
await db.update(issues).set({ monitorNextCheckAt: null, executionPolicy: null }).where(eq(issues.id, f.issue.id));
await start(f); draining = true;
await schedule(f, new Date(Date.now() + 10_000));
await quiet(); expect(f.deps.enqueueWakeup).not.toHaveBeenCalled();
draining = false;
await advance(30_000);
await vi.waitFor(async () => expect((await readIssue(f.issue.id)).monitorNextCheckAt).toBeNull());
expect(f.deps.enqueueWakeup).toHaveBeenCalledOnce();
await quiet();
});
it("backs off a deferred admission and disarms after the retry succeeds", async () => {
const f = await fixture(); f.deps.enqueueWakeup.mockResolvedValueOnce(null);
await start(f);
expect(f.deps.enqueueWakeup).toHaveBeenCalledOnce();
await advance(5000);
expect(f.deps.enqueueWakeup).toHaveBeenCalledOnce();
expect(coordinator!.nextWakeAt()).not.toBeNull();
await advance(30_000);
await vi.waitFor(async () => expect((await readIssue(f.issue.id)).monitorNextCheckAt).toBeNull());
expect(f.deps.enqueueWakeup).toHaveBeenCalledTimes(2);
await quiet();
});
it("retains the overdue remainder of a capped pass until the next bounded drain", async () => {
const f = await fixture();
await db.insert(issues).values(Array.from({ length: 50 }, (_, index) => ({
companyId: f.company.id, title: `Due monitor ${index}`, status: "in_progress", assigneeAgentId: f.agent.id,
monitorNextCheckAt: dueAt, executionPolicy: f.issue.executionPolicy,
})));
await start(f);
expect(f.deps.enqueueWakeup).toHaveBeenCalledTimes(50);
expect(await f.scheduler.nextIssueMonitorAt()).toBe(dueAt.getTime());
await advance(30_000);
await vi.waitFor(() => expect(f.deps.enqueueWakeup).toHaveBeenCalledTimes(51));
await quiet();
expect(await f.scheduler.nextIssueMonitorAt()).toBeNull();
}, 30_000);
it("replaces an earlier deadline and cancels it through the ordinary issue producer", async () => {
const f = await fixture();
await db.update(issues).set({ monitorNextCheckAt: null, executionPolicy: null }).where(eq(issues.id, f.issue.id));
await start(f);
const later = new Date(Date.now() + 600_000);
await schedule(f, later);
await advance(30_000);
await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(later.getTime()), { timeout: 5000 });
const earlier = new Date(Date.now() + 60_000);
await schedule(f, earlier);
await advance(30_000);
await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(earlier.getTime()), { timeout: 5000 });
await issueService(db).update(f.issue.id, { monitorNextCheckAt: null, executionPolicy: null });
await quiet();
await advance(600_000);
expect(f.deps.enqueueWakeup).not.toHaveBeenCalled();
});
it("reconstructs a claim deadline and respects its strict five-minute fence", async () => {
const f = await fixture();
const claimedAt = new Date();
await db.update(issues).set({ monitorWakeRequestedAt: claimedAt }).where(eq(issues.id, f.issue.id));
expect(await f.scheduler.nextIssueMonitorAt()).toBe(claimedAt.getTime() + 300_001);
expect(await f.scheduler.tickIssueMonitors(new Date(claimedAt.getTime() + 300_000))).toMatchObject({ triggered: 0 });
await start(f);
expect(coordinator!.nextWakeAt()).toBe(claimedAt.getTime() + 300_001);
await advance(300_001);
await vi.waitFor(async () => expect((await readIssue(f.issue.id)).monitorNextCheckAt).toBeNull());
expect(f.deps.enqueueWakeup).toHaveBeenCalledOnce();
await quiet();
});
it("disarms while the company is paused and reactivates on its committed resume", async () => {
const f = await fixture();
await companyService(db).update(f.company.id, { status: "paused" });
await start(f); await quiet();
expect(coordinator!.nextWakeAt()).toBeNull();
await companyService(db).update(f.company.id, { status: "active" });
await vi.waitFor(async () => expect((await readIssue(f.issue.id)).monitorNextCheckAt).toBeNull());
expect(f.deps.enqueueWakeup).toHaveBeenCalledOnce();
await quiet();
});
it("bounds global scans during repeated ordinary checkout and release writes", async () => {
const f = await fixture();
const at = new Date(Date.now() + 300_000);
await db.update(issues).set({ status: "todo", monitorNextCheckAt: at }).where(eq(issues.id, f.issue.id));
const next = vi.spyOn(f.scheduler, "nextIssueMonitorAt");
const tick = vi.spyOn(f.scheduler, "tickIssueMonitors");
await start(f);
const service = issueService(db);
for (let i = 0; i < 12; i++) {
await service.checkout(f.issue.id, f.agent.id, ["todo"], null);
await service.release(f.issue.id, f.agent.id, null);
await advance(2000);
}
expect(next).toHaveBeenCalledOnce();
expect(tick).toHaveBeenCalledOnce();
await service.checkout(f.issue.id, f.agent.id, ["todo"], null);
await advance(6000);
await vi.waitFor(() => expect(next).toHaveBeenCalledTimes(2));
expect(tick).toHaveBeenCalledTimes(2);
await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(at.getTime()));
expect(f.deps.enqueueWakeup).not.toHaveBeenCalled();
});
it("checkout reactivates a retained monitor on a formerly ineligible task", async () => {
const f = await fixture();
await db.update(issues).set({ status: "todo" }).where(eq(issues.id, f.issue.id));
await start(f); await quiet();
expect(coordinator!.nextWakeAt()).toBeNull();
await issueService(db).checkout(f.issue.id, f.agent.id, ["todo"], null);
await vi.waitFor(async () => expect((await readIssue(f.issue.id)).monitorNextCheckAt).toBeNull());
expect(f.deps.enqueueWakeup).toHaveBeenCalledOnce();
await quiet();
});
});
});
server/src/services/heartbeat/scheduling.ts
+51 -16
View File
@@ -1,3 +1,4 @@
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
import { notifyIssueReviewState } from "../issue-review-state-notifications.js";
import { insertIssueComment } from "../issue-comment-writes.js";
import {
@@ -72,7 +73,7 @@ export interface HeartbeatSchedulingDependencies
Pick<RunPreparation, "toAgentOrgRow" | "groupAgentOrgRowsByCompany"> {
issuesSvc: Pick<ReturnType<typeof issueService>, "listReviewAttention" | "create">;
scheduleBoundedRetryForRun: ReturnType<typeof createHeartbeatRetries>["scheduleBoundedRetryForRun"];
getSchedulingSuppression: () => Promise<{ suppressed: boolean }>;
getSchedulingSuppression: (fresh?: boolean) => Promise<{ suppressed: boolean }>;
}
function readNonEmptyString(value: unknown): string | null {
@@ -591,6 +592,7 @@ export function createHeartbeatScheduling(
const publications: ActivityPublication[] = [];
try {
await db.transaction(async (tx) => {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
const released = await tx.update(issues)
.set({ monitorWakeRequestedAt: null, updatedAt: input.now })
.where(issueMonitorClaimCondition(claimed)).returning({ id: issues.id });
@@ -622,8 +624,11 @@ export function createHeartbeatScheduling(
} catch (err) {
// Diagnostics are optional. Release the original claim even if audit storage
// failed, and never clobber a monitor that was replaced in the meantime.
await db.update(issues).set({ monitorWakeRequestedAt: null, updatedAt: input.now })
.where(issueMonitorClaimCondition(claimed));
await db.transaction(async tx => {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await tx.update(issues).set({ monitorWakeRequestedAt: null, updatedAt: input.now })
.where(issueMonitorClaimCondition(claimed));
});
logger.warn({ err, issueId: claimed.id }, "issue monitor diagnostic could not be recorded");
}
}
@@ -819,8 +824,9 @@ export function createHeartbeatScheduling(
}
}
const consumed = await db
.update(issues)
const consumed = await db.transaction(async tx => {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
return tx.update(issues)
.set({
...monitorOnlyDispatchPatch(buildIssueMonitorTriggeredPatch({
issue: claimed,
@@ -831,8 +837,7 @@ export function createHeartbeatScheduling(
})
.where(issueMonitorClaimCondition(claimed))
.returning({ id: issues.id });
// Admission already accepted the check. A replacement schedule must stay
// intact, but losing its consumption fence does not undo that accepted wake.
});
if (consumed.length === 0) return { outcome: "triggered" as const };
await logActivity(db, {
@@ -970,6 +975,7 @@ export function createHeartbeatScheduling(
const staleClaimThreshold = new Date(now.getTime() - 5 * 60 * 1000);
const claimed = await db.transaction(async (tx) => {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
const [updated] = await tx
.update(issues)
.set({
@@ -1012,7 +1018,7 @@ export function createHeartbeatScheduling(
});
}
async function tickDueIssueMonitors(now = new Date()) {
async function tickDueIssueMonitors(now: Date, canContinue: () => boolean) {
const staleClaimThreshold = new Date(now.getTime() - 5 * 60 * 1000);
const dueMonitors = await db
.select(issueMonitorDispatchColumns)
@@ -1040,7 +1046,9 @@ export function createHeartbeatScheduling(
let skipped = 0;
for (const due of dueMonitors) {
if (!canContinue()) break;
const claimed = await db.transaction(async (tx) => {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
const [updated] = await tx
.update(issues)
.set({
@@ -1067,6 +1075,7 @@ export function createHeartbeatScheduling(
});
if (!claimed) continue;
if (!canContinue()) break;
try {
const result = await dispatchClaimedIssueMonitor(claimed, {
@@ -1095,6 +1104,26 @@ export function createHeartbeatScheduling(
};
}
async function tickIssueMonitors(now = new Date(), canContinue: () => boolean = () => true) {
if (!canContinue() || (await getSchedulingSuppression(true)).suppressed) {
return { checked: 0, triggered: 0, skipped: 0 };
}
return tickDueIssueMonitors(now, canContinue);
}
async function nextIssueMonitorAt(): Promise<number | null> {
if ((await getSchedulingSuppression(true)).suppressed) return null;
const [row] = await db.select({ at: sql<string | null>`min(greatest(${issues.monitorNextCheckAt},
${issues.monitorWakeRequestedAt} + interval '5 minutes' + interval '1 millisecond'))` })
.from(issues).innerJoin(companies, eq(companies.id, issues.companyId))
.where(and(eq(companies.status, "active"), noActiveNativeMonitorRun(),
sql`${issues.monitorNextCheckAt} is not null`, isNull(issues.assigneeUserId),
sql`${issues.assigneeAgentId} is not null`, inArray(issues.status, ["in_progress", "in_review"])));
return row?.at == null ? null : new Date(row.at).getTime();
}
async function recoverActiveSessionGoals() {
if ((await getSchedulingSuppression()).suppressed) {
return { scanned: 0, enqueued: 0 };
@@ -1256,7 +1285,7 @@ export function createHeartbeatScheduling(
return { scanned: pending.length, enqueued, alreadyQueued, invalid };
}
async function tickTimers(now = new Date()) {
async function tickAgentTimers(now = new Date()) {
if ((await getSchedulingSuppression()).suppressed) {
return {
checked: 0,
@@ -1334,18 +1363,24 @@ export function createHeartbeatScheduling(
else skipped += 1;
}
const issueMonitors = await tickDueIssueMonitors(now);
return {
checked: checked + issueMonitors.checked,
enqueued: enqueued + issueMonitors.triggered,
skipped: skipped + issueMonitors.skipped,
};
return { checked, enqueued, skipped };
}
async function tickTimers(now = new Date()) {
const timers = await tickAgentTimers(now);
const monitors = await tickIssueMonitors(now);
return { checked: timers.checked + monitors.checked,
enqueued: timers.enqueued + monitors.triggered, skipped: timers.skipped + monitors.skipped };
}
return {
triggerIssueMonitor,
tickTimers,
tickAgentTimers,
tickIssueMonitors,
nextIssueMonitorAt,
recoverActiveSessionGoals,
recoverPendingSessionGoalActions,
};
server/src/services/issue-review-state-notifications.ts
+2
View File
@@ -1,8 +1,10 @@
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
import { notifyNativeReviewResponse } from "./native-review-response-notifications.js";
import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
export async function notifyIssueReviewState(tx: object, scope: { companyId: string; issueId?: string; runId?: string; agentId?: string }) {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyExternalConversationSettlement(tx, scope);
await notifyNativeReviewResponse(tx, scope);
await notifyNativeStatusReassessment(tx, scope);
server/src/services/issue-tree-control.ts
+2
View File
@@ -1,3 +1,4 @@
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
import { notifyChatResponseAvailability } from "./chat-response-availability.js";
@@ -999,6 +1000,7 @@ export function issueTreeControlService(db: Db) {
const now = new Date();
const releasedCancelHoldIds = activeCancelHolds.map((hold) => hold.id);
const updatedIssues = await db.transaction(async (tx) => {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
const restored: TreeStatusUpdateResult["updatedIssues"] = [];
for (const [status, issueIdsForStatus] of issueIdsByStatus) {
if (issueIdsForStatus.length === 0) continue;
server/src/services/issues.ts
+46 -36
View File
@@ -11937,6 +11937,7 @@ export function issueService(db: Db) {
await notifyApprovalWork(tx, owner.companyId);
}
await notifyDecisionTargetIssueRemoval(tx, id);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
removedIssue = await tx
.delete(issues)
.where(eq(issues.id, id))
@@ -12062,27 +12063,30 @@ export function issueService(db: Db) {
eq(issues.executionRunId, checkoutRunId),
)
: isNull(issues.executionRunId);
const updated = await db
.update(issues)
.set({
assigneeAgentId: agentId,
assigneeUserId: null,
checkoutRunId,
executionRunId: checkoutRunId,
status: "in_progress",
startedAt: now,
updatedAt: now,
})
.where(
and(
eq(issues.id, id),
inArray(issues.status, expectedStatuses),
or(isNull(issues.assigneeAgentId), sameRunAssigneeCondition),
executionLockCondition,
),
)
.returning()
.then((rows) => rows[0] ?? null);
const updated = await db.transaction(async tx => {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
return tx
.update(issues)
.set({
assigneeAgentId: agentId,
assigneeUserId: null,
checkoutRunId,
executionRunId: checkoutRunId,
status: "in_progress",
startedAt: now,
updatedAt: now,
})
.where(
and(
eq(issues.id, id),
inArray(issues.status, expectedStatuses),
or(isNull(issues.assigneeAgentId), sameRunAssigneeCondition),
executionLockCondition,
),
)
.returning()
.then((rows) => rows[0] ?? null);
});
if (updated) {
const [enriched] = await withIssueLabels(db, [updated]);
@@ -12186,22 +12190,26 @@ export function issueService(db: Db) {
if (current.status !== "in_progress") {
adoptionSet.startedAt = now;
}
const adopted = await db
.update(issues)
.set(adoptionSet)
.where(
and(
eq(issues.id, id),
inArray(issues.status, expectedStatuses),
eq(issues.executionRunId, current.executionRunId),
or(
isNull(issues.assigneeAgentId),
eq(issues.assigneeAgentId, agentId),
const staleExecutionRunId = current.executionRunId;
const adopted = await db.transaction(async tx => {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
return tx
.update(issues)
.set(adoptionSet)
.where(
and(
eq(issues.id, id),
inArray(issues.status, expectedStatuses),
eq(issues.executionRunId, staleExecutionRunId),
or(
isNull(issues.assigneeAgentId),
eq(issues.assigneeAgentId, agentId),
),
),
),
)
.returning()
.then((rows) => rows[0] ?? null);
)
.returning()
.then((rows) => rows[0] ?? null);
});
if (adopted) {
const [enriched] = await withIssueLabels(db, [adopted]);
return enriched;
@@ -12438,6 +12446,7 @@ export function issueService(db: Db) {
existing.status === "in_progress" ? "todo" : existing.status;
await notifyTerminalWorkspaceWork(tx, `issue:${id}`);
await notifyNativeStatusReassessment(tx, { companyId: existing.companyId, issueId: id });
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
const updated = await tx
.update(issues)
.set({
@@ -12496,6 +12505,7 @@ export function issueService(db: Db) {
if (existing.checkoutRunId || existing.executionRunId) {
await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: existing.companyId, issueId: id });
}
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
const updated = await tx
.update(issues)
.set(patch)
server/src/services/native-run-projection-notifications.ts
+2 -1
View File
@@ -2,7 +2,8 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio
export type NativeRunProjectionScope = { companyId: string; issueId?: string; runId?: string };
export function notifyNativeRunProjection(tx: object, scope?: NativeRunProjectionScope) {
export async function notifyNativeRunProjection(tx: object, scope?: NativeRunProjectionScope) {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeRunProjection,
scope ? JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId }) : "*");
}
server/src/services/native-runtime/native-preserved-monitor-wait.test.ts
+37 -18
View File
@@ -1,26 +1,31 @@
import { randomUUID } from "node:crypto";
import { and, eq } from "drizzle-orm";
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
import { agents, agentWakeupRequests, companies, completionContracts, createDb, heartbeatRuns,
import { and, eq, sql } from "drizzle-orm";
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
import { agents, agentWakeupRequests, companies, completionContracts, createDb, applyPendingMigrations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns,
issueComments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, subscribeDatabaseWork } from "@paperclipai/db";
import { PROVIDER_QUOTA_MONITOR_SERVICE_NAME } from "@paperclipai/shared";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
import { PaperclipRunnerToolAuthority } from "./paperclip-runner-tool-authority.js";
import { finalizeNativeRun, repairCommittedNativeChatResponse } from "./native-run-finalizer.js";
import { nativeSha256 } from "./canonical.js";
import * as monitorDelivery from "./native-monitor-delivery.js";
import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
import { heartbeatService } from "../heartbeat.js";
import { hasPendingRunCleanup } from "../execution-blocker.js";
const support = await getEmbeddedPostgresTestSupport();
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_PRESERVED_MONITOR_TEST_DATABASE_URL");
const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
describe.skipIf(!support.supported)("released native monitor results", () => {
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
let db: ReturnType<typeof createDb>;
beforeAll(async () => {
temporary = await startEmbeddedPostgresTestDatabase("native-preserved-monitor-");
db = createDb(temporary.connectionString);
});
if (external) await applyPendingMigrations(external);
else temporary = await startEmbeddedPostgresTestDatabase("native-preserved-monitor-");
db = createDb(external ?? temporary.connectionString);
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
afterAll(async () => { await db?.$client.end(); await temporary?.cleanup(); });
beforeEach(async () => { await db.execute(sql`truncate table companies restart identity cascade`); });
async function fixture() {
const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID();
@@ -160,18 +165,32 @@ describe.skipIf(!support.supported)("released native monitor results", () => {
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].finishedAt).toEqual(f.finishedAt);
});
it("notifies the existing monitor scheduler once when finalization stops blocking it", async () => {
const f = await fixture(), events: string[] = [];
it("makes the saved deadline eligible after finalization commits without replaying the result", async () => {
const f = await fixture();
const scheduler = heartbeatService(db, { runtimeEnv: {} });
const observedDeadlines: Promise<number | null>[] = [];
expect(await hasPendingRunCleanup(db, f.companyId, f.runId)).toBe(true);
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.issueMonitors, event => { events.push(event); });
expect(await scheduler.nextIssueMonitorAt()).toBeNull();
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.issueMonitors, event => {
if (event === "settled") observedDeadlines.push(scheduler.nextIssueMonitorAt());
});
try {
await f.finalize();
expect(events).toEqual(["intent", "settled"]);
const deadline = new Date(f.nextCheckAt).getTime();
expect(await Promise.all(observedDeadlines)).toContain(deadline);
expect(await scheduler.nextIssueMonitorAt()).toBe(deadline);
expect(await hasPendingRunCleanup(db, f.companyId, f.runId)).toBe(false);
expect((await f.readIssue()).monitorNextCheckAt?.toISOString()).toBe(f.nextCheckAt);
const decisions = await db.select().from(statusDecisions).where(eq(statusDecisions.runId, f.runId));
await f.finalize();
expect(events).toEqual(["intent", "settled"]);
} finally { unsubscribe(); }
expect(await scheduler.nextIssueMonitorAt()).toBe(deadline);
expect(await db.select().from(statusDecisions).where(eq(statusDecisions.runId, f.runId))).toEqual(decisions);
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toEqual([]);
} finally {
unsubscribe();
await Promise.all(observedDeadlines);
}
});
it.each(["live", "recovered"])("publishes the accepted monitor explanation once (%s)", async mode => {
@@ -207,8 +226,8 @@ describe.skipIf(!support.supported)("released native monitor results", () => {
const [owner] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }).returning();
const nextCheckAt = new Date(Date.now() + 7_200_000).toISOString();
const original = monitorDelivery.readNativePreservedMonitorWait;
const events: string[] = [];
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.issueMonitors, event => { events.push(event); });
const scheduler = heartbeatService(db, { runtimeEnv: {} });
expect(await scheduler.nextIssueMonitorAt()).toBeNull();
const race = vi.spyOn(monitorDelivery, "readNativePreservedMonitorWait").mockImplementationOnce(async (...args) => {
const proof = await original(...args);
expect(proof).not.toBeNull();
@@ -220,10 +239,10 @@ describe.skipIf(!support.supported)("released native monitor results", () => {
} }).where(eq(issues.id, f.issueId));
return proof;
});
try { await f.finalize(); } finally { race.mockRestore(); unsubscribe(); }
try { await f.finalize(); } finally { race.mockRestore(); }
const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
expect(coordinator.phase).toBe(state === "rescheduled" ? "committed" : "retryable_failure");
expect(events).toEqual(state === "rescheduled" ? ["intent", "settled"] : []);
expect(await scheduler.nextIssueMonitorAt()).toBe(state === "rescheduled" ? new Date(nextCheckAt).getTime() : null);
const task = await f.readIssue();
expect(task.executionRunId).toBe(state === "owner" ? owner!.id : null);
expect(task.monitorNextCheckAt?.toISOString() ?? null).toBe(state === "cleared" ? null : state === "rescheduled" ? nextCheckAt : f.nextCheckAt);
server/src/services/native-runtime/native-workspace-reconstruction.ts
+2
View File
@@ -1,3 +1,4 @@
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js";
import { and, eq, gt, inArray, ne, notInArray, or, sql } from "drizzle-orm";
import { agentWakeupRequests, assets, environmentLeases, heartbeatRuns, issueAttachments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db";
@@ -184,6 +185,7 @@ export async function reconstructNativeWorkspace(input: {
evidence: sql`${issueRecoveryActions.evidence} || ${JSON.stringify({ workspaceReconstructionRequest: requestSha256 })}::jsonb`,
wakePolicy: null, monitorPolicy: null,
}).where(eq(issueRecoveryActions.id, source.action.id));
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await tx.update(issues).set({ status: "todo", unblockDescriptor: null, blockedTransitionAt: null,
blockedOwnerNotifiedAt: null, updatedAt: now }).where(eq(issues.id, issue.id));
const publication = await persistActivity(db, { companyId: input.companyId, actorType: "user", actorId: input.actorId,
server/src/services/scheduled-work.ts
+21 -30
View File
@@ -1,49 +1,40 @@
import { readTaskDrain } from "./task-admission.js";
import { registerDeadlineWork } from "./deadline-work.js";
import type { Db } from "@paperclipai/db";
import { sql, type SQL } from "drizzle-orm";
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
import { DELIVERY_QUEUES, type DeliveryQueue } from "./delivery-work-notifications.js";
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
export function registerScheduledWork(
coordinator: Pick<ReturnType<typeof createDeliveryWorkCoordinator>, "register">,
db: Pick<Db, "execute">,
options: {
retryMs: number;
tickTimers: () => Promise<unknown>;
tickAgentTimers: () => Promise<unknown>;
tickIssueMonitors: (signal: AbortSignal) => Promise<unknown>;
nextIssueMonitorAt: () => Promise<number | null>;
tickRoutines: (signal: AbortSignal) => Promise<unknown>;
nextRoutineRunAt: () => Promise<number | null>;
canRun: () => boolean;
},
) {
function register(queue: DeliveryQueue, work: () => Promise<unknown>, query: (now: Date) => SQL) {
let next: number | null = null;
return coordinator.register(queue, {
retryMs: options.retryMs,
run: async () => {
const cutoff = new Date();
await work();
const [row] = await db.execute<{ next_at: Date | string | null }>(query(cutoff));
if (!row || row.next_at === undefined) throw new Error("Scheduled deadline query returned no result");
next = row.next_at === null ? null
: row.next_at instanceof Date ? row.next_at.getTime() : Date.parse(row.next_at);
if (next !== null && !Number.isFinite(next)) throw new Error("Scheduled deadline is invalid");
},
hasPending: async () => false,
nextRunAt: () => next,
});
}
const routines = registerDeadlineWork(coordinator, DELIVERY_QUEUES.routineSchedules, {
run: options.tickRoutines,
next: options.nextRoutineRunAt,
canRun: options.canRun,
retryMs: options.retryMs,
});
const timers = register(DELIVERY_QUEUES.issueMonitors, options.tickTimers, now => sql`
SELECT min(i.monitor_next_check_at) AS next_at
FROM issues i JOIN companies c ON c.id = i.company_id
WHERE c.status = 'active' AND i.status IN ('in_progress', 'in_review')
AND i.assignee_agent_id IS NOT NULL AND i.assignee_user_id IS NULL
AND i.monitor_next_check_at > ${now.toISOString()}::timestamptz`);
return { routines, timers };
const monitors = registerDeadlineWork(coordinator, DELIVERY_QUEUES.issueMonitors, {
run: options.tickIssueMonitors,
next: options.nextIssueMonitorAt,
canRun: () => options.canRun() && readTaskDrain(new Date()) === null,
retryMs: options.retryMs,
notificationCoalesceMs: options.retryMs,
});
const timers = coordinator.register(DELIVERY_QUEUES.agentTimers, {
retryMs: options.retryMs,
run: options.tickAgentTimers,
hasPending: async () => false,
nextRunAt: () => null,
});
return { routines, monitors, timers };
}
server/src/services/slack-conversation-lifecycle.ts
+2
View File
@@ -1,3 +1,4 @@
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
import { and, asc, desc, eq, gt, inArray, isNull, sql } from "drizzle-orm";
import { chatConversations, chatEndpoints, heartbeatRuns, issueComments, issues, type Db } from "@paperclipai/db";
@@ -130,6 +131,7 @@ export async function settleExternalConversation(db: Db, companyId: string, issu
.returning({ id: chatConversations.id });
if (!changed.length) return false;
await notifyUnsafeWorkspaceExportRecovery(tx, { companyId, issueId });
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await tx.update(issues).set({ status: "in_review", updatedAt: new Date() })
.where(and(eq(issues.id, issueId), eq(issues.companyId, companyId)));
activity = (await persistActivity(tx as unknown as Db, {