diff --git a/doc/idle-sleep-safety.md b/doc/idle-sleep-safety.md index ec87d70985e..82f3652fbc7 100644 --- a/doc/idle-sleep-safety.md +++ b/doc/idle-sleep-safety.md @@ -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 diff --git a/server/src/__tests__/idle-issue-schedule.test.ts b/server/src/__tests__/idle-issue-schedule.test.ts new file mode 100644 index 00000000000..6406a4b9624 --- /dev/null +++ b/server/src/__tests__/idle-issue-schedule.test.ts @@ -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; + let database: Awaited>; + 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 = {}; + 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); + }); + +}); diff --git a/server/src/__tests__/instance-settings-routes.test.ts b/server/src/__tests__/instance-settings-routes.test.ts index 23d8c32dd10..14abf47810d 100644 --- a/server/src/__tests__/instance-settings-routes.test.ts +++ b/server/src/__tests__/instance-settings-routes.test.ts @@ -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; } diff --git a/server/src/routes/instance-settings.ts b/server/src/routes/instance-settings.ts index 0df320ace06..800d2261822 100644 --- a/server/src/routes/instance-settings.ts +++ b/server/src/routes/instance-settings.ts @@ -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; } diff --git a/server/src/routes/openapi.ts b/server/src/routes/openapi.ts index 9895e95065d..8dd5210e07f 100644 --- a/server/src/routes/openapi.ts +++ b/server/src/routes/openapi.ts @@ -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 }, }); diff --git a/server/src/services/heartbeat/scheduling.test.ts b/server/src/services/heartbeat/scheduling.test.ts index 8a5659de786..8c1bb544da6 100644 --- a/server/src/services/heartbeat/scheduling.test.ts +++ b/server/src/services/heartbeat/scheduling.test.ts @@ -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); diff --git a/server/src/services/idle-issue-schedule.ts b/server/src/services/idle-issue-schedule.ts new file mode 100644 index 00000000000..431ea72f2fb --- /dev/null +++ b/server/src/services/idle-issue-schedule.ts @@ -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(AGENT_STATUSES); +const record = (value: unknown): Record | null => + value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record : null; + +export async function readIdleIssueSchedule(tx: Pick, 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() }; +} diff --git a/server/src/services/idle-scheduled-work.ts b/server/src/services/idle-scheduled-work.ts index b909f798fd6..6e9ed7f670b 100644 --- a/server/src/services/idle-scheduled-work.ts +++ b/server/src/services/idle-scheduled-work.ts @@ -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; } diff --git a/server/src/services/idle-sleep-safety.ts b/server/src/services/idle-sleep-safety.ts index fe9be54ffd2..c5dfb630490 100644 --- a/server/src/services/idle-sleep-safety.ts +++ b/server/src/services/idle-sleep-safety.ts @@ -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, scheduledRoutines = false, + scheduledIssueMonitors = false, ): Promise { 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; diff --git a/server/src/services/recovery/onboarding-first-task-wait.ts b/server/src/services/recovery/onboarding-first-task-wait.ts new file mode 100644 index 00000000000..9fa3e7a2802 --- /dev/null +++ b/server/src/services/recovery/onboarding-first-task-wait.ts @@ -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)`; +} diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 5d43e26ae05..23a092420c1 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -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(