mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-21 00:05:49 +02:00
fix(server): support issue monitor deadlines in idle sleep reports (#16325)
## Thinking Path > - Paperclip manages work for companies and AI agents. > - Hosted workspaces can sleep when their required work has a safe wake path. > - The current idle check treats every open issue as background work. > - Backlog, unassigned and human-waiting records can therefore keep an idle process running. > - Future automated monitors still need their saved check time and execution claims. > - This PR adds a signed host capability for passive issues and verified monitor deadlines. > - The host can save one alarm for the earliest required routine or monitor time. ## Linked Issues or Issue Description **What happened?** A quiet workspace with a backlog or unassigned todo issue reports background work as present. A future issue monitor also prevents sleep even when the host can save a durable wake alarm. **Expected behavior** Proven passive records can sleep. A validated future monitor can sleep only after the host saves its alarm and validates the same owned report. Active, unknown and unresolved work stays awake. **Steps to reproduce** 1. Create a quiet workspace with an unassigned todo issue and no other work. 2. Acquire a bounded, signed Cloud idle hold. 3. Request the routine-only safety report with `scheduledWake=1`. The issue blocks sleep. 4. Request the new capability with `scheduledWake=2`. A passive issue can now report no work. 5. Add a valid future monitor to an assigned in-progress issue. The second capability returns its saved deadline for host acknowledgement. **Paperclip version or commit** Base: `c30fb7d2b75ea30f72d63656b2a7cfff7fe0929d`. **Deployment mode** Cloud-managed authenticated instance. Local scheduling remains in the existing scheduler. Related protocol: [#16215](https://github.com/paperclipai/paperclip/pull/16215). Open scheduler and monitor PRs were checked for overlap. This change leaves monitor admission, dispatch and quota recovery with their existing owners. The roadmap already includes scheduled work; this is a focused idle-safety extension. ## What Changed - Add signed scheduling capability `2`; capability `1` retains the routine-only rules. - Allow untouched onboarding prompts with idle assignees to sleep using the same awaiting-user predicate as recovery; typing, answering the opening card or a prior run removes the exemption. - Inspect passive issue state and future monitors in the same bounded, read-only snapshot as other work. - Check ownership, company scope, known states, claims, policy/deadline agreement, attempts, timeout and dependency repair. - Combine routine and monitor deadlines into a known proof kind and catalog generation. - Preserve other sleep blockers, including task watchdogs, conversations, live runs, external credentials and recovery. - Update the protected route, OpenAPI contract, documentation and PostgreSQL regression tests. ## Verification - Full workspace typecheck and build passed. Final server typecheck and full build passed after the final guards. - Initial safety, scheduling, protected-route and control-authentication checks: 326 tests passed. - Final safety and heartbeat scheduling checks: 253 tests passed. Includes reporting a future deadline, restoring scheduling from PostgreSQL and one dispatch across competing schedulers. - Local general server checks: 20,260 passed; UI: 8,460 passed; CLI: 534 passed, including isolated reruns of two embedded-PostgreSQL startup failures. All remaining workspace groups and 158 serialized suites passed. The suite was completed in groups after the isolated infrastructure retries; the final onboarding change also passed the focused 674-test recovery/safety run. Earlier unchanged Git fixtures failed under Xcode's system `init.defaultBranch=main`, reproduced on the clean base and passed with a test-scoped fixture default; no source or machine configuration was changed. - Rebased issue, protected-route and scheduler checks: 115 tests passed; server typecheck passed. - The expanded change passed 674 idle-safety, scheduling and recovery regression tests, including six onboarding admission/invalidation cases. Final server typecheck and build passed. - Current-head Apex is 5/5 with no unresolved threads. All 54 current-head checks passed, with two expected skips. Four jobs were interrupted by runner shutdowns or lost communication on their first attempts; their targeted reruns passed. Live hosting qualification awaits the verified master release and matching host capability. ## Risks This intentionally covers a limited set of passive states and valid future monitors. Assigned active tasks without a proved waiting path, unknown states, claimed monitors and quota recovery still block sleep. It does not make every open issue safe to sleep. A matching host must recognize `routines_and_issue_monitors`, acknowledge the durable alarm and revalidate the exact owner, generation and deadline before stopping. The report alone grants no stop authority. Older hosts retain capability `1`. Independent database writers remain outside the admission contract. Existing scheduler claims and execution authority are unchanged. No schema migration is required. Roll back host negotiation to capability `1` to restore conservative admission, while retaining alarm delivery until outstanding alarms have been handled. ## Model Used - OpenAI GPT-6 (Codex), with repository editing, command execution and GitHub tools. The exact deployment ID and context-window metadata are not exposed in this session. ## 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 - [x] All Paperclip CI gates are green - [x] 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>
This commit is contained in:
Devin FoleyandPaperclip authored and GitHub committed
1 parent
aec80cbc8c
commit
35b3345033
11 files changed
+378
-48
No files matched your search
@@ -74,6 +74,41 @@ Verify a full scheduled sleep/wake/execution cycle on staging before enabling
|
||||
this host capability. The [tests](../server/src/__tests__/idle-sleep-safety.test.ts)
|
||||
cover ownership, future deadlines, changed generations and conservative fallback.
|
||||
|
||||
## Passive issues and scheduled monitors
|
||||
|
||||
Signed Cloud control callers can request `scheduledWake=2` to include issue
|
||||
monitoring alongside routines. Earlier hosts requesting `scheduledWake=1`
|
||||
retain the routine-only rules, including the conservative open-issue check.
|
||||
Browser administrators and tenant credentials cannot request either extension.
|
||||
|
||||
With the second capability, backlog tasks, unassigned todo tasks, explicit
|
||||
blocked tasks and human-owned tasks may be passive records. Their presence
|
||||
alone does not require compute. An assigned onboarding first task is also
|
||||
passive with an idle assignee before its first run, user-authored comment or user-resolved opening
|
||||
card. This uses the same awaiting-user predicate as recovery; a seeded greeting
|
||||
does not start work. User input still enters through the host wake path.
|
||||
Execution locks, automated pending review,
|
||||
conversation state, unknown state and dependency repair remain blockers. A
|
||||
blocked task whose blockers have all finished stays awake for reconciliation.
|
||||
Enabled task watchdogs and unsupported recovery keep their existing checks.
|
||||
Incoming changes still require the hosting admission and wake path.
|
||||
|
||||
An active issue monitor needs a saved future `monitor_next_check_at` matching
|
||||
its validated execution policy, a company-scoped agent, no outstanding dispatch
|
||||
claim, and remaining attempts and timeout. Due, malformed, quota-recovery,
|
||||
unowned and inconsistent monitors keep the workspace awake. Inspection does
|
||||
not clear schedules, advance attempts or dispatch an agent.
|
||||
|
||||
The combined report uses `version: 2` and a scheduled proof with `version: 1`
|
||||
and `kind: "routines_and_issue_monitors"`. Its deadline is the earliest required
|
||||
routine or monitor time across all companies. Its generation binds the current
|
||||
process, idle owner and inspected catalogs. Hosts must acknowledge this known
|
||||
proof kind, persist the alarm and validate the unchanged owned report before
|
||||
stopping compute. Existing hosts reject the new kind. Monitor dispatch and
|
||||
startup catch-up continue through the existing scheduler and atomic claims;
|
||||
self-hosted scheduling is unchanged. This capability does not cover all future
|
||||
work in the instance: every other safety check continues to apply.
|
||||
|
||||
## Retry history
|
||||
|
||||
Retry promotion keeps `scheduled_retry_at` on a run for inspection. That date
|
||||
|
||||
@@ -0,0 +1,187 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
|
||||
import { ONBOARDING_FIRST_TASK_ORIGIN_KIND, PROVIDER_QUOTA_MONITOR_SERVICE_NAME } from "@paperclipai/shared";
|
||||
import { agents, agentApiKeys, companies, createDb, heartbeatRuns, issues, issueComments, issueRelations, issueThreadInteractions, routines, routineTriggers } from "@paperclipai/db";
|
||||
import { beginIdleTrackedWork, idleWorkSnapshot } from "../services/task-admission.js";
|
||||
import { readIdleSleepSafety, type IdleSleepDrainStatus } from "../services/idle-sleep-safety.js";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
|
||||
const now = Date.UTC(2020, 0, 1);
|
||||
const ownerId = "d0b833f4-4098-42de-8420-1907f3aa4895";
|
||||
const owned = (): IdleSleepDrainStatus => ({ ownerId, draining: true, startedAt: new Date(now - 1000),
|
||||
expiresAt: new Date(now + 60_000), activeRuns: 0, pendingWakes: 0 });
|
||||
const emptyLocal = async () => "none" as const;
|
||||
const present = { version: 1, backgroundWork: "present" };
|
||||
const none = { version: 1, backgroundWork: "none" };
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
if (!support.supported) console.warn(`Skipping issue schedule Postgres tests: ${support.reason}`);
|
||||
(support.supported ? describe : describe.skip)("idle issue schedule safety", () => {
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("idle-issue-schedule-");
|
||||
db = createDb(database.connectionString);
|
||||
}, 60_000);
|
||||
afterEach(async () => { await db.execute(sql`TRUNCATE companies, plugins, environments CASCADE`); });
|
||||
afterAll(async () => { await db?.$client.end(); await database?.cleanup(); });
|
||||
const readScheduled = () => readIdleSleepSafety(db, owned, () => now, ownerId, emptyLocal, undefined, undefined, true);
|
||||
const readIssueSchedules = () => readIdleSleepSafety(db, owned, () => now, ownerId, emptyLocal, undefined, undefined, true, true);
|
||||
async function seed() {
|
||||
const companyId = randomUUID();
|
||||
const agentId = randomUUID();
|
||||
const runId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Idle test", issuePrefix: `T${companyId.slice(0, 6).toUpperCase()}` });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "On-demand agent", role: "engineer", status: "idle", adapterType: "process" });
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", status: "succeeded" });
|
||||
return { companyId, agentId, runId };
|
||||
}
|
||||
|
||||
async function scheduledRoutine(at = now + 600_000) {
|
||||
const fixture = await seed();
|
||||
const [routine] = await db.insert(routines).values({ companyId: fixture.companyId,
|
||||
title: "Scheduled fixture", status: "active", assigneeAgentId: fixture.agentId }).returning();
|
||||
const [trigger] = await db.insert(routineTriggers).values({ companyId: fixture.companyId,
|
||||
routineId: routine!.id, kind: "schedule", cronExpression: "*/15 * * * *", timezone: "UTC",
|
||||
nextRunAt: new Date(at), enabled: true, archived: false, setupPending: false }).returning();
|
||||
return { ...fixture, routine: routine!, trigger: trigger! };
|
||||
}
|
||||
|
||||
async function monitoredIssue(at = now + 600_000) {
|
||||
const fixture = await seed();
|
||||
const monitor = { nextCheckAt: new Date(at).toISOString(), notes: "Inspect fixture result" };
|
||||
const [issue] = await db.insert(issues).values({ companyId: fixture.companyId, title: "Monitor fixture",
|
||||
status: "in_progress", assigneeAgentId: fixture.agentId, monitorNextCheckAt: new Date(at),
|
||||
executionPolicy: { monitor } }).returning();
|
||||
return { ...fixture, issue: issue!, monitor };
|
||||
}
|
||||
|
||||
it("allows passive issue records only with the issue-scheduling host capability", async () => {
|
||||
const f = await seed();
|
||||
await db.insert(issues).values([
|
||||
{ companyId: f.companyId, title: "Planned task", status: "backlog", assigneeAgentId: f.agentId },
|
||||
{ companyId: f.companyId, title: "Unassigned task", status: "todo" },
|
||||
{ companyId: f.companyId, title: "Explicitly blocked task", status: "blocked", assigneeAgentId: f.agentId },
|
||||
{ companyId: f.companyId, title: "Human review", status: "in_review", assigneeUserId: "fixture-user" },
|
||||
]);
|
||||
expect(await readScheduled()).toEqual(present);
|
||||
expect(await readIssueSchedules()).toEqual(none);
|
||||
expect(await db.select().from(issues)).toHaveLength(4);
|
||||
});
|
||||
|
||||
it.each(["todo", "in_progress", "in_review"])("retains an assigned %s task without a proved waiting path", async status => {
|
||||
const f = await seed();
|
||||
await db.insert(issues).values({ companyId: f.companyId, title: "Unfinished automated task", status, assigneeAgentId: f.agentId });
|
||||
expect(await readIssueSchedules()).toEqual(present);
|
||||
});
|
||||
|
||||
it.each(["typed", "legacy-typed", "interaction", "run", "origin", "agent-state"])(
|
||||
"allows an untouched onboarding task, but retains it after %s activity", async activity => {
|
||||
const f = await seed();
|
||||
const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "First task",
|
||||
status: "todo", assigneeAgentId: f.agentId, originKind: ONBOARDING_FIRST_TASK_ORIGIN_KIND }).returning();
|
||||
await db.insert(issueComments).values({ companyId: f.companyId, issueId: issue!.id,
|
||||
authorType: "agent", authorAgentId: f.agentId, body: "Seeded greeting" });
|
||||
const [card] = await db.insert(issueThreadInteractions).values({ companyId: f.companyId,
|
||||
issueId: issue!.id, kind: "request_confirmation", payload: { version: 1, prompt: "Start?" } }).returning();
|
||||
expect(await readScheduled()).toEqual(present);
|
||||
expect(await readIssueSchedules()).toEqual(none);
|
||||
if (activity === "typed" || activity === "legacy-typed") {
|
||||
await db.insert(issueComments).values({ companyId: f.companyId, issueId: issue!.id,
|
||||
authorType: activity === "typed" ? "user" : null, authorUserId: "fixture-user", body: "Start work" });
|
||||
}
|
||||
if (activity === "interaction") await db.update(issueThreadInteractions).set({ status: "resolved",
|
||||
resolvedByUserId: "fixture-user", resolvedAt: new Date(now) }).where(eq(issueThreadInteractions.id, card!.id));
|
||||
if (activity === "run") await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: issue!.id } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (activity === "origin") await db.update(issues).set({ originKind: "manual" }).where(eq(issues.id, issue!.id));
|
||||
if (activity === "agent-state") await db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId));
|
||||
expect(await readIssueSchedules()).toEqual(present);
|
||||
},
|
||||
);
|
||||
|
||||
it("keeps resolved dependency repair awake while allowing an unresolved blocked task", async () => {
|
||||
const f = await seed();
|
||||
const [blocker] = await db.insert(issues).values({ companyId: f.companyId, title: "Unassigned blocker", status: "todo" }).returning();
|
||||
const [dependent] = await db.insert(issues).values({ companyId: f.companyId, title: "Waiting task", status: "blocked", assigneeAgentId: f.agentId }).returning();
|
||||
await db.insert(issueRelations).values({ companyId: f.companyId, issueId: blocker!.id, relatedIssueId: dependent!.id, type: "blocks" });
|
||||
expect(await readIssueSchedules()).toEqual(none);
|
||||
await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, blocker!.id));
|
||||
expect(await readIssueSchedules()).toEqual(none);
|
||||
await db.update(issues).set({ status: "done" }).where(eq(issues.id, blocker!.id));
|
||||
expect(await readIssueSchedules()).toEqual(present);
|
||||
});
|
||||
|
||||
it("combines routine and issue monitor deadlines across companies without changing dispatch state", async () => {
|
||||
await scheduledRoutine(now + 900_000);
|
||||
const f = await monitoredIssue();
|
||||
expect(await readScheduled()).toEqual(present);
|
||||
const first = await readIssueSchedules();
|
||||
expect(first).toMatchObject({ version: 2, backgroundWork: "none", scheduledWake: {
|
||||
version: 1, kind: "routines_and_issue_monitors", nextWakeAt: f.monitor.nextCheckAt } });
|
||||
expect((await readIssueSchedules()).scheduledWake).toEqual(first.scheduledWake);
|
||||
const earlier = new Date(now + 400_000);
|
||||
await db.update(issues).set({ monitorNextCheckAt: earlier, executionPolicy: { monitor: { ...f.monitor, nextCheckAt: earlier.toISOString() } } }).where(eq(issues.id, f.issue.id));
|
||||
const edited = await readIssueSchedules();
|
||||
expect(edited.scheduledWake?.nextWakeAt).toBe(earlier.toISOString());
|
||||
expect(edited.scheduledWake?.generation).not.toBe(first.scheduledWake?.generation);
|
||||
expect(await db.select().from(issues).where(eq(issues.id, f.issue.id))).toMatchObject([{ monitorNextCheckAt: earlier, monitorWakeRequestedAt: null, monitorAttemptCount: 0 }]);
|
||||
await db.update(issues).set({ status: "blocked", monitorNextCheckAt: null, executionPolicy: null }).where(eq(issues.id, f.issue.id));
|
||||
expect(await readIssueSchedules()).toMatchObject({ version: 2, backgroundWork: "none", scheduledWake: {
|
||||
kind: "routines", nextWakeAt: new Date(now + 900_000).toISOString() } });
|
||||
});
|
||||
|
||||
it.each(["due", "mismatch", "claimed", "timeout", "exhausted", "negative-attempts", "pending-review", "quota", "unowned"])(
|
||||
"retains %s monitor work", async kind => {
|
||||
const f = await monitoredIssue();
|
||||
const patch: Partial<typeof issues.$inferInsert> = {};
|
||||
if (kind === "due") patch.monitorNextCheckAt = new Date(now);
|
||||
if (kind === "mismatch") patch.monitorNextCheckAt = new Date(now + 700_000);
|
||||
if (kind === "claimed") patch.monitorWakeRequestedAt = new Date(now);
|
||||
if (kind === "pending-review") patch.executionState = { status: "pending" };
|
||||
if (kind === "unowned") patch.assigneeAgentId = null;
|
||||
if (kind === "timeout") patch.executionPolicy = { monitor: { ...f.monitor, timeoutAt: new Date(now + 500_000).toISOString() } };
|
||||
if (kind === "exhausted") { patch.monitorAttemptCount = 2; patch.executionPolicy = { monitor: { ...f.monitor, maxAttempts: 2 } }; }
|
||||
if (kind === "negative-attempts") patch.monitorAttemptCount = -1;
|
||||
if (kind === "quota") patch.executionPolicy = { monitor: { ...f.monitor, serviceName: PROVIDER_QUOTA_MONITOR_SERVICE_NAME } };
|
||||
await db.update(issues).set(patch).where(eq(issues.id, f.issue.id));
|
||||
expect(await readIssueSchedules()).toEqual(present);
|
||||
},
|
||||
);
|
||||
|
||||
it("retains a cross-company assignee and unknown issue states", async () => {
|
||||
const f = await monitoredIssue();
|
||||
const other = await seed();
|
||||
await db.update(issues).set({ assigneeAgentId: other.agentId }).where(eq(issues.id, f.issue.id));
|
||||
expect(await readIssueSchedules()).toEqual(present);
|
||||
await db.update(issues).set({ assigneeAgentId: f.agentId, status: "unknown" }).where(eq(issues.id, f.issue.id));
|
||||
expect(await readIssueSchedules()).toEqual(present);
|
||||
});
|
||||
|
||||
it("retains an unknown assignee state even for a blocked task", async () => {
|
||||
const f = await seed();
|
||||
await db.insert(issues).values({ companyId: f.companyId, title: "Blocked fixture", status: "blocked", assigneeAgentId: f.agentId });
|
||||
await db.execute(sql`UPDATE agents SET status = 'unknown' WHERE id = ${f.agentId}`);
|
||||
expect(await readIssueSchedules()).toEqual(present);
|
||||
});
|
||||
|
||||
it("refuses monitor sleep if a new run or external credential needs the instance", async () => {
|
||||
const f = await monitoredIssue();
|
||||
await db.update(heartbeatRuns).set({ status: "queued" }).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(await readIssueSchedules()).toEqual(present);
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await db.insert(agentApiKeys).values({ companyId: f.companyId, agentId: f.agentId, name: "Fixture external client", keyHash: "fixture-key-hash" });
|
||||
expect(await readIssueSchedules()).toEqual(present);
|
||||
});
|
||||
|
||||
it("retains a monitor which becomes due during the owned backup checkpoint", async () => {
|
||||
await monitoredIssue(now + 1000);
|
||||
let clock = now;
|
||||
const backup = async () => {
|
||||
const done = beginIdleTrackedWork();
|
||||
try { clock += 1000; return true; } finally { done(); }
|
||||
};
|
||||
expect(await readIdleSleepSafety(db, owned, () => clock, ownerId, emptyLocal, undefined, backup, true, true)).toEqual(present);
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
});
|
||||
|
||||
});
|
||||
@@ -907,7 +907,7 @@ describe("instance settings routes", () => {
|
||||
await request(createApp(adminActor)).post("/api/instance/task-drain").send({ purpose: "idle", ttlMs: 60_000 }).expect(200);
|
||||
expect(mockHeartbeatService.computeTaskDrain).toHaveBeenCalledWith({ purpose: "idle", ttlMs: 60_000 });
|
||||
await request(createApp(adminActor)).get(`/api/instance/task-drain?idleSleepSafety=1&ownerId=${ownerId}`).expect(200);
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenLastCalledWith(mockDb, expect.any(Function), Date.now, ownerId, expect.any(Function), undefined, undefined, false);
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenLastCalledWith(mockDb, expect.any(Function), Date.now, ownerId, expect.any(Function), undefined, undefined, false, false);
|
||||
});
|
||||
|
||||
it.each([{}, { ttlMs: null }, { ttlMs: 1 }, { ttlMs: 300_001 }])("rejects an unbounded or invalid idle TTL: %j", async body => {
|
||||
@@ -934,21 +934,21 @@ describe("instance settings routes", () => {
|
||||
const res = await request(createApp(adminActor)).get("/api/instance/task-drain?idleSleepSafety=1");
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.body).toEqual({ ...idleStatus, idleSleepSafety: report });
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, undefined, expect.any(Function), undefined, undefined, false);
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, undefined, expect.any(Function), undefined, undefined, false, false);
|
||||
expect(mockReadIdleSleepSafety.mock.calls[0][1]()).toEqual(idleStatus);
|
||||
});
|
||||
|
||||
it("reserves routine deadline reports for verified Cloud control callers", async () => {
|
||||
it.each(["1", "2"])("reserves scheduling capability %s for verified Cloud control callers", async version => {
|
||||
mockHeartbeatService.getTaskDrainStatus.mockReturnValue(idleStatus);
|
||||
for (const source of ["session", "cloud_tenant", "local_implicit"]) {
|
||||
await request(createApp({ ...adminActor, source }))
|
||||
.get("/api/instance/task-drain?idleSleepSafety=1&scheduledWake=1&ownerId=fixture-owner").expect(403);
|
||||
.get(`/api/instance/task-drain?idleSleepSafety=1&scheduledWake=${version}&ownerId=fixture-owner`).expect(403);
|
||||
}
|
||||
expect(mockReadIdleSleepSafety).not.toHaveBeenCalled();
|
||||
await request(createApp({ ...adminActor, source: "cloud_control" }))
|
||||
.get("/api/instance/task-drain?idleSleepSafety=1&scheduledWake=1&ownerId=fixture-owner").expect(200);
|
||||
.get(`/api/instance/task-drain?idleSleepSafety=1&scheduledWake=${version}&ownerId=fixture-owner`).expect(200);
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now,
|
||||
"fixture-owner", expect.any(Function), undefined, undefined, true);
|
||||
"fixture-owner", expect.any(Function), undefined, undefined, true, version === "2");
|
||||
});
|
||||
|
||||
it("uses the live worker manager for inspection and releases only the owned hold", async () => {
|
||||
@@ -983,7 +983,7 @@ describe("instance settings routes", () => {
|
||||
expect(mockReadIdleSleepSafety).not.toHaveBeenCalled();
|
||||
await request(app).get("/api/instance/task-drain?idleSleepSafety=1&ownerId=fixture-owner").expect(200);
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now,
|
||||
"fixture-owner", expect.any(Function), undefined, backup, false);
|
||||
"fixture-owner", expect.any(Function), undefined, backup, false, false);
|
||||
mockReadIdleSleepSafety.mockClear();
|
||||
await request(createApp(nonAdminActor, undefined, backup)).get("/api/instance/task-drain?idleSleepSafety=1").expect(403);
|
||||
expect(mockReadIdleSleepSafety).not.toHaveBeenCalled();
|
||||
@@ -1005,7 +1005,7 @@ describe("instance settings routes", () => {
|
||||
const control = createApp({ ...adminActor, source: "cloud_control" }, undefined, backup);
|
||||
await request(control).get("/api/instance/task-drain?idleSleepSafety=1&ownerId=fixture-owner").expect(200);
|
||||
expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now,
|
||||
"fixture-owner", expect.any(Function), undefined, backup, false);
|
||||
"fixture-owner", expect.any(Function), undefined, backup, false, false);
|
||||
} finally {
|
||||
delete process.env.PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN;
|
||||
}
|
||||
|
||||
@@ -305,7 +305,8 @@ export function instanceSettingsRoutes(db: Db, pluginWorkers?: PluginWorkerManag
|
||||
// The report covers every company in this process. Ordinary company
|
||||
// members may read process counters, but not instance-wide work state.
|
||||
assertCanManageInstanceSettings(req);
|
||||
const scheduledRoutines = req.query.scheduledWake === "1";
|
||||
const scheduledIssueMonitors = req.query.scheduledWake === "2";
|
||||
const scheduledRoutines = req.query.scheduledWake === "1" || scheduledIssueMonitors;
|
||||
if (scheduledRoutines && req.actor.source !== "cloud_control") {
|
||||
throw forbidden("Scheduled wake reports require Cloud control authority");
|
||||
}
|
||||
@@ -320,7 +321,7 @@ export function instanceSettingsRoutes(db: Db, pluginWorkers?: PluginWorkerManag
|
||||
const idleSleepSafety = await readIdleSleepSafety(db, () => heartbeat.getTaskDrainStatus(), Date.now,
|
||||
typeof req.query.ownerId === "string" ? req.query.ownerId : undefined,
|
||||
() => readIdleLocalWork({ backupCheckpoint: Boolean(prepareIdleDatabaseBackup) }),
|
||||
pluginWorkers?.inspectIdleSleep?.bind(pluginWorkers), prepareIdleDatabaseBackup, scheduledRoutines);
|
||||
pluginWorkers?.inspectIdleSleep?.bind(pluginWorkers), prepareIdleDatabaseBackup, scheduledRoutines, scheduledIssueMonitors);
|
||||
res.json({ ...heartbeat.getTaskDrainStatus(), idleSleepSafety });
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -6857,7 +6857,7 @@ registry.registerPath({
|
||||
tags: ["instance"],
|
||||
summary:
|
||||
"Get the task-drain status for this process only; quiescent counts in-process work, and a process restart clears it even when the database still holds running rows",
|
||||
request: { query: z.object({ ownerId: z.string().uuid().optional(), idleSleepSafety: z.literal("1").optional().describe("Instance admins can request a conservative durable-work report while admission and ingress are held. Unknown or present work must prevent automatic sleep."), scheduledWake: z.literal("1").optional().describe("Cloud control callers may request an owned routine deadline report. Cloud must acknowledge a durable alarm and revalidate this report before stopping the instance.") }) },
|
||||
request: { query: z.object({ ownerId: z.string().uuid().optional(), idleSleepSafety: z.literal("1").optional().describe("Instance admins can request a conservative durable-work report while admission and ingress are held. Unknown or present work must prevent automatic sleep."), scheduledWake: z.enum(["1", "2"]).optional().describe("Cloud control callers may request routine deadlines (1), or routines and issue monitors (2). Cloud must acknowledge a durable alarm and revalidate this report before stopping the instance.") }) },
|
||||
responses: { 200: r.ok(), 401: r.unauthorized, 403: r.forbidden },
|
||||
});
|
||||
|
||||
|
||||
@@ -11,6 +11,7 @@ import { conflict } from "../../errors.js";
|
||||
import { createHeartbeatRunPreparation } from "./run-preparation.js";
|
||||
import { createHeartbeatRunState } from "./run-state.js";
|
||||
import { createHeartbeatScheduling, type HeartbeatSchedulingDependencies } from "./scheduling.js";
|
||||
import { readIdleSleepSafety } from "../idle-sleep-safety.js";
|
||||
|
||||
vi.mock("../live-events.js", () => ({ publishLiveEvent: vi.fn() }));
|
||||
|
||||
@@ -134,6 +135,28 @@ describe.skipIf(!postgres.supported)("heartbeat scheduling database boundary", (
|
||||
expect(await readIssue(f.issue.id)).toMatchObject({ monitorNextCheckAt: null, monitorAttemptCount: 1 });
|
||||
});
|
||||
|
||||
it("restores the reported future deadline and claims monitor delivery exactly once", async () => {
|
||||
const f = await fixture();
|
||||
const at = new Date(now.getTime() + 600_000);
|
||||
await db.update(issues).set({ monitorNextCheckAt: at,
|
||||
executionPolicy: { monitor: { nextCheckAt: at.toISOString() } } }).where(eq(issues.id, f.issue.id));
|
||||
const ownerId = randomUUID();
|
||||
const hold = () => ({ ownerId, draining: true, activeRuns: 0, pendingWakes: 0,
|
||||
startedAt: now, expiresAt: new Date(now.getTime() + 120_000) });
|
||||
const report = await readIdleSleepSafety(db, hold, () => now.getTime(), ownerId,
|
||||
async () => "none", undefined, undefined, true, true);
|
||||
expect(report).toMatchObject({ version: 2, backgroundWork: "none", scheduledWake: {
|
||||
kind: "routines_and_issue_monitors", nextWakeAt: at.toISOString() } });
|
||||
expect(await f.scheduler.tickTimers(now)).toEqual({ checked: 0, enqueued: 0, skipped: 0 });
|
||||
expect(f.deps.enqueueWakeup).not.toHaveBeenCalled();
|
||||
const restored = createHeartbeatScheduling(db, f.deps);
|
||||
const deadline = new Date(report.scheduledWake!.nextWakeAt);
|
||||
await Promise.all([restored.tickTimers(deadline), f.scheduler.tickTimers(deadline)]);
|
||||
expect(f.deps.enqueueWakeup).toHaveBeenCalledOnce();
|
||||
expect(f.deps.enqueueWakeup).toHaveBeenCalledWith(f.agent.id, expect.objectContaining({ reason: "issue_monitor_due" }));
|
||||
expect(await readIssue(f.issue.id)).toMatchObject({ monitorNextCheckAt: null, monitorAttemptCount: 1 });
|
||||
});
|
||||
|
||||
it("preserves a replacement schedule written during wake admission", async () => {
|
||||
const f = await fixture();
|
||||
const replacement = new Date(now.getTime() + 60_000);
|
||||
|
||||
@@ -0,0 +1,94 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { AGENT_STATUSES, issueExecutionMonitorPolicySchema, PROVIDER_QUOTA_MONITOR_SERVICE_NAME } from "@paperclipai/shared";
|
||||
import type { ScheduledWake } from "./idle-scheduled-work.js";
|
||||
import { onboardingFirstTaskAwaitingUserSql } from "./recovery/onboarding-first-task-wait.js";
|
||||
|
||||
const processGeneration = randomUUID();
|
||||
const limit = 10_000;
|
||||
const statuses = new Set(["backlog", "todo", "in_progress", "in_review", "blocked"]);
|
||||
const agentStatuses = new Set<string>(AGENT_STATUSES);
|
||||
const record = (value: unknown): Record<string, unknown> | null =>
|
||||
value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : null;
|
||||
|
||||
export async function readIdleIssueSchedule(tx: Pick<Db, "execute">, ownerId: string, now: number): Promise<{
|
||||
blocked: boolean;
|
||||
scheduledWake?: ScheduledWake;
|
||||
}> {
|
||||
const rows = await tx.execute<{
|
||||
id: string; company_id: string; status: string; updated_at: Date | string;
|
||||
assignee_agent_id: string | null; assignee_user_id: string | null;
|
||||
checkout_run_id: string | null; execution_run_id: string | null;
|
||||
execution_policy: unknown; execution_state: unknown; unblock_descriptor: unknown;
|
||||
conversation_agent_id: string | null; monitor_next_check_at: Date | string | null;
|
||||
monitor_wake_requested_at: Date | string | null; monitor_attempt_count: number;
|
||||
company_status: string; company_updated: Date | string; agent_status: string | null;
|
||||
agent_updated: Date | string | null; scoped: boolean; unsafe_dependencies: boolean; dependency_ready: boolean;
|
||||
untouched_onboarding: boolean;
|
||||
}>(sql`SELECT i.id, i.company_id, i.status, i.updated_at, i.assignee_agent_id, i.assignee_user_id,
|
||||
i.checkout_run_id, i.execution_run_id, i.execution_policy, i.execution_state, i.unblock_descriptor,
|
||||
i.conversation_agent_id, i.monitor_next_check_at, i.monitor_wake_requested_at, i.monitor_attempt_count,
|
||||
c.status AS company_status, c.updated_at AS company_updated,
|
||||
a.status AS agent_status, a.updated_at AS agent_updated,
|
||||
(i.assignee_agent_id IS NULL OR a.id IS NOT NULL) AS scoped,
|
||||
(${onboardingFirstTaskAwaitingUserSql(sql`i.company_id`, sql`i.id`, sql`i.origin_kind`)}
|
||||
AND NOT EXISTS (SELECT 1 FROM heartbeat_runs h WHERE h.company_id = i.company_id
|
||||
AND (h.context_snapshot ->> 'issueId' = i.id::text OR h.context_snapshot ->> 'taskId' = i.id::text))) AS untouched_onboarding,
|
||||
EXISTS (SELECT 1 FROM issue_relations r LEFT JOIN issues b ON b.id = r.issue_id
|
||||
WHERE r.related_issue_id = i.id AND (r.company_id <> i.company_id OR b.company_id <> i.company_id
|
||||
OR b.id IS NULL OR r.type <> 'blocks' OR b.status NOT IN ('backlog', 'todo', 'in_progress', 'in_review', 'blocked', 'done', 'cancelled'))) AS unsafe_dependencies,
|
||||
(EXISTS (SELECT 1 FROM issue_relations r WHERE r.related_issue_id = i.id AND r.type = 'blocks')
|
||||
AND NOT EXISTS (SELECT 1 FROM issue_relations r JOIN issues b ON b.id = r.issue_id
|
||||
WHERE r.related_issue_id = i.id AND r.type = 'blocks' AND b.status <> 'done')) AS dependency_ready
|
||||
FROM issues i JOIN companies c ON c.id = i.company_id
|
||||
LEFT JOIN agents a ON a.id = i.assignee_agent_id AND a.company_id = i.company_id
|
||||
WHERE i.status NOT IN ('done', 'cancelled') ORDER BY i.id LIMIT ${limit + 1}`);
|
||||
if (rows.length > limit) return { blocked: true };
|
||||
let earliest = Infinity;
|
||||
for (const row of rows) {
|
||||
const policy = record(row.execution_policy);
|
||||
const state = record(row.execution_state);
|
||||
if (!statuses.has(row.status) || row.company_status !== "active" || row.scoped !== true
|
||||
|| (row.assignee_agent_id && !agentStatuses.has(row.agent_status ?? ""))
|
||||
|| row.unsafe_dependencies !== false || row.checkout_run_id || row.execution_run_id
|
||||
|| row.conversation_agent_id || row.monitor_wake_requested_at
|
||||
|| (row.execution_state !== null && state?.status !== "idle")
|
||||
|| (row.execution_policy !== null && !policy)) return { blocked: true };
|
||||
if (row.status === "blocked" && row.dependency_ready) return { blocked: true };
|
||||
if (row.monitor_next_check_at !== null) {
|
||||
const monitor = issueExecutionMonitorPolicySchema.safeParse(policy?.monitor);
|
||||
const at = new Date(row.monitor_next_check_at).getTime();
|
||||
if (!monitor.success || !row.assignee_agent_id || row.assignee_user_id
|
||||
|| !["in_progress", "in_review"].includes(row.status)
|
||||
|| !["idle", "active", "error"].includes(row.agent_status ?? "")
|
||||
|| !Number.isFinite(at) || at <= now || Date.parse(monitor.data.nextCheckAt) !== at
|
||||
|| monitor.data.serviceName === PROVIDER_QUOTA_MONITOR_SERVICE_NAME
|
||||
|| !Number.isSafeInteger(row.monitor_attempt_count) || row.monitor_attempt_count < 0
|
||||
|| (monitor.data.timeoutAt && Date.parse(monitor.data.timeoutAt) <= at)
|
||||
|| (monitor.data.maxAttempts !== null && row.monitor_attempt_count >= monitor.data.maxAttempts)) return { blocked: true };
|
||||
earliest = Math.min(earliest, at);
|
||||
continue;
|
||||
}
|
||||
if (policy?.monitor != null) return { blocked: true };
|
||||
const unblock = record(row.unblock_descriptor);
|
||||
const passive = row.status === "backlog"
|
||||
|| (row.status === "todo" && !row.assignee_agent_id)
|
||||
|| (row.status === "todo" && row.assignee_agent_id && row.agent_status === "idle" && row.untouched_onboarding === true)
|
||||
|| (row.status === "blocked" && (row.unblock_descriptor === null || unblock?.owner === "board"))
|
||||
|| Boolean(row.assignee_user_id && !row.assignee_agent_id);
|
||||
if (!passive) return { blocked: true };
|
||||
}
|
||||
if (!Number.isFinite(earliest)) return { blocked: false };
|
||||
return { blocked: false, scheduledWake: { version: 1, kind: "routines_and_issue_monitors",
|
||||
generation: createHash("sha256").update(JSON.stringify([processGeneration, ownerId, rows])).digest("hex"),
|
||||
nextWakeAt: new Date(earliest).toISOString() } };
|
||||
}
|
||||
|
||||
export function combineIdleSchedules(routine?: ScheduledWake, issue?: ScheduledWake): ScheduledWake | undefined {
|
||||
if (!issue) return routine;
|
||||
if (!routine) return issue;
|
||||
return { version: 1, kind: "routines_and_issue_monitors",
|
||||
generation: createHash("sha256").update(JSON.stringify([routine, issue])).digest("hex"),
|
||||
nextWakeAt: new Date(Math.min(Date.parse(routine.nextWakeAt), Date.parse(issue.nextWakeAt))).toISOString() };
|
||||
}
|
||||
@@ -8,7 +8,7 @@ const limit = 10_000;
|
||||
|
||||
export interface ScheduledWake {
|
||||
version: 1;
|
||||
kind: "routines";
|
||||
kind: "routines" | "routines_and_issue_monitors";
|
||||
generation: string;
|
||||
nextWakeAt: string;
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { idleWorkSnapshot } from "./task-admission.js";
|
||||
import { readIdleLocalWork, type IdleLocalWork } from "./idle-local-work.js";
|
||||
import { readIdleRoutineSchedule, type ScheduledWake } from "./idle-scheduled-work.js";
|
||||
import { combineIdleSchedules, readIdleIssueSchedule } from "./idle-issue-schedule.js";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { sql } from "drizzle-orm";
|
||||
|
||||
@@ -222,6 +223,7 @@ export async function readIdleSleepSafety(
|
||||
inspectPlugins?: InspectIdlePlugins,
|
||||
prepareBackup?: () => Promise<boolean>,
|
||||
scheduledRoutines = false,
|
||||
scheduledIssueMonitors = false,
|
||||
): Promise<IdleSleepSafety> {
|
||||
const unknown: IdleSleepSafety = { version: 1, backgroundWork: "unknown" };
|
||||
const before = getDrainStatus();
|
||||
@@ -242,8 +244,9 @@ export async function readIdleSleepSafety(
|
||||
let scheduledWake: ScheduledWake | undefined;
|
||||
const blocked = await db.transaction(async (tx) => {
|
||||
await tx.execute(sql`SET LOCAL statement_timeout = '5s'`);
|
||||
const checks = WORK_CHECKS.filter(query => !scheduledRoutines ||
|
||||
query !== `SELECT 1 FROM routines WHERE status NOT IN ('paused', 'archived')`)
|
||||
const checks = WORK_CHECKS.filter(query =>
|
||||
(!scheduledRoutines || query !== `SELECT 1 FROM routines WHERE status NOT IN ('paused', 'archived')`)
|
||||
&& (!scheduledIssueMonitors || query !== `SELECT 1 FROM issues WHERE status NOT IN ('done', 'cancelled')`))
|
||||
.map((query) => sql`EXISTS (${sql.raw(query)})`);
|
||||
// Only the current worker's owned runtime drain can exempt a plugin.
|
||||
// Missing workers, old SDKs, and newly installed plugins still block.
|
||||
@@ -259,6 +262,12 @@ export async function readIdleSleepSafety(
|
||||
if (schedule.blocked) return true;
|
||||
scheduledWake = schedule.scheduledWake;
|
||||
}
|
||||
if (scheduledIssueMonitors) {
|
||||
if (!scheduledRoutines || !ownerId || ownerId !== before.ownerId) return undefined;
|
||||
const schedule = await readIdleIssueSchedule(tx, ownerId, now());
|
||||
if (schedule.blocked) return true;
|
||||
scheduledWake = combineIdleSchedules(scheduledWake, schedule.scheduledWake);
|
||||
}
|
||||
return false;
|
||||
}, { isolationLevel: "repeatable read", accessMode: "read only" });
|
||||
if (blocked === undefined || !sameQuietHold(before, getDrainStatus(), now())) return unknown;
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
import { sql, type SQLWrapper } from "drizzle-orm";
|
||||
import { ONBOARDING_FIRST_TASK_ORIGIN_KIND } from "@paperclipai/shared";
|
||||
|
||||
export function onboardingFirstTaskAwaitingUserSql(companyId: SQLWrapper, issueId: SQLWrapper, originKind: SQLWrapper) {
|
||||
return sql`${originKind} = ${ONBOARDING_FIRST_TASK_ORIGIN_KIND}
|
||||
AND NOT EXISTS (SELECT 1 FROM issue_comments c
|
||||
WHERE c.company_id = ${companyId} AND c.issue_id = ${issueId}
|
||||
AND (c.author_type = 'user' OR (c.author_type IS NULL AND c.author_user_id IS NOT NULL)))
|
||||
AND NOT EXISTS (SELECT 1 FROM issue_thread_interactions t
|
||||
WHERE t.company_id = ${companyId} AND t.issue_id = ${issueId} AND t.resolved_by_user_id IS NOT NULL)`;
|
||||
}
|
||||
@@ -1,4 +1,5 @@
|
||||
import { isAgentAwaitingSetup } from "../../modules/agent-lifecycle/index.js";
|
||||
import { onboardingFirstTaskAwaitingUserSql } from "./onboarding-first-task-wait.js";
|
||||
import { isAiAuthenticationBlocked } from "../ai-auth-failure.js";
|
||||
import { hasCommittedNativePlanWait } from "../native-runtime/native-plan-wait.js";
|
||||
import { isNativeWorkspaceExportRepairCause } from "@paperclipai/shared";
|
||||
@@ -2077,40 +2078,9 @@ export function recoveryService(
|
||||
issue: typeof issues.$inferSelect,
|
||||
) {
|
||||
if (issue.originKind !== ONBOARDING_FIRST_TASK_ORIGIN_KIND) return false;
|
||||
const userComment = await db
|
||||
.select({ id: issueComments.id })
|
||||
.from(issueComments)
|
||||
.where(
|
||||
and(
|
||||
eq(issueComments.companyId, issue.companyId),
|
||||
eq(issueComments.issueId, issue.id),
|
||||
or(
|
||||
eq(issueComments.authorType, "user"),
|
||||
and(
|
||||
isNull(issueComments.authorType),
|
||||
sql`${issueComments.authorUserId} is not null`,
|
||||
),
|
||||
),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (userComment !== null) return false;
|
||||
// Answering the seeded opening card ("interview me" / "I have a task in
|
||||
// mind") is the user's first input too, even though it is not a comment.
|
||||
const userResolvedInteraction = await db
|
||||
.select({ id: issueThreadInteractions.id })
|
||||
.from(issueThreadInteractions)
|
||||
.where(
|
||||
and(
|
||||
eq(issueThreadInteractions.companyId, issue.companyId),
|
||||
eq(issueThreadInteractions.issueId, issue.id),
|
||||
sql`${issueThreadInteractions.resolvedByUserId} is not null`,
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
return userResolvedInteraction === null;
|
||||
const [result] = await db.execute<{ awaiting: boolean }>(sql`SELECT
|
||||
${onboardingFirstTaskAwaitingUserSql(sql`${issue.companyId}`, sql`${issue.id}`, sql`${issue.originKind}`)} AS awaiting`);
|
||||
return result?.awaiting === true;
|
||||
}
|
||||
|
||||
async function isInvocationBudgetBlocked(
|
||||
|
||||
Reference in new issue
Block a user