diff --git a/server/src/services/heartbeat/idle-timer-schedule.ts b/server/src/services/heartbeat/idle-timer-schedule.ts new file mode 100644 index 00000000000..8627456de03 --- /dev/null +++ b/server/src/services/heartbeat/idle-timer-schedule.ts @@ -0,0 +1,53 @@ +import { agents, companies, type Db } from "@paperclipai/db"; +import { AGENT_STATUSES } from "@paperclipai/shared"; +import { eq, getTableColumns } from "drizzle-orm"; +import { evaluateAgentInvokability } from "../agent-invokability.js"; +import { supportsUnscopedTimerHeartbeat } from "../heartbeat-policy.js"; +import { createHeartbeatTimerPolicy } from "./timer-policy.js"; +import { idleDeadlineRowLimit, isFutureIdleDeadline, type IdleDeadlineSnapshot } from "../idle-deadline-types.js"; + +/** Normal Cloud scheduling only. Worktree overrides/restore are not a transferable execution authority. */ +export async function readIdleAgentTimerSchedule(tx: Pick, now: number, + env: Record = process.env, checkBudget: () => void = () => {}): Promise { + const rows = await tx.select({ agent: getTableColumns(agents), companyStatus: companies.status, + companyUpdatedAt: companies.updatedAt }).from(agents).leftJoin(companies, eq(companies.id, agents.companyId)) + .orderBy(agents.id).limit(idleDeadlineRowLimit + 1); + const state: unknown[] = [rows]; + const blocked = () => ({ blocked: true, state, deadlines: [] }); + if (rows.length > idleDeadlineRowLimit) return blocked(); + const byCompany = new Map(); + for (const { agent } of rows) { + const group = byCompany.get(agent.companyId) ?? []; + group.push(agent); + byCompany.set(agent.companyId, group); + } + const owner = createHeartbeatTimerPolicy(tx); + let inspectedTimers = 0; + const deadlines: number[] = []; + for (const { agent, companyStatus } of rows) { + checkBudget(); + if (!AGENT_STATUSES.includes(agent.status as typeof AGENT_STATUSES[number]) || !companyStatus || agent.status === "running") return blocked(); + if (companyStatus !== "active") continue; + const policy = owner.parseHeartbeatPolicy(agent); + if (!Number.isFinite(policy.intervalSec)) return blocked(); + if (!policy.enabled || policy.intervalSec <= 0 || !supportsUnscopedTimerHeartbeat(agent)) continue; + // Org-chain evaluation belongs to its owner. Bound its candidate count + // even when every timer needs no additional SQL (for example zero caps). + if (++inspectedTimers > 128) return blocked(); + const invokability = evaluateAgentInvokability(agent, byCompany.get(agent.companyId)!); + if (!invokability.invokable) { + if (invokability.invalidOrgChain || invokability.reason === "unknown_status") return blocked(); + continue; + } + const flags = [env.HEARTBEAT_SCHEDULER_ENABLED, env.PAPERCLIP_IN_WORKTREE, + env.PAPERCLIP_DATABASE_RESTORE_IN_PROGRESS, env.PAPERCLIP_RESTORE_IN_PROGRESS]; + state.push(flags); + if (flags[0] === "false" || flags.slice(1).some(v => ["true", "1", "yes", "on"].includes(v ?? ""))) return blocked(); + const snapshot = await owner.readTimerHeartbeatSchedule(agent, new Date(now)); + state.push([agent.id, snapshot]); + if (snapshot.deadline === null) continue; + if (!isFutureIdleDeadline(snapshot.deadline, now)) return blocked(); + deadlines.push(snapshot.deadline); + } + return { blocked: false, state, deadlines }; +} diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 18b5033ea39..6e03be0bf49 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,4 @@ +import { createHeartbeatTimerPolicy } from "./timer-policy.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import { PersistedExecutionWaitRefusal, writeDeferredExecutionWait } from "../deferred-execution-wait.js"; import { notifyDeferredExecutionWork } from "../deferred-execution-notifications.js"; @@ -35,10 +36,8 @@ import { isCompletedOnboardingHandoffWake, } from "../chat-completion-delivery.js"; import { - compareCents, hasWorkspaceRestoreFailure, } from "@paperclipai/shared"; -import { nonIdleSlackIssueCondition } from "../slack-conversation-state.js"; import { readQueuedInteractionResponse } from "../queued-interaction-response.js"; import { isConversation, @@ -85,15 +84,12 @@ import { gte, inArray, isNull, - lt, lte, ne, - notInArray, or, sql, } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; -import { AGENT_DEFAULT_MAX_CONCURRENT_RUNS } from "@paperclipai/shared"; import { agents, caseHandoffs, @@ -103,7 +99,6 @@ import { chatConversations, chatEndpoints, companies, - costEvents, executionWorkspaces, heartbeatRunEvents, heartbeatRuns, @@ -128,8 +123,6 @@ import { } from "../issue-queued-comment-queue.js"; import { parseObject, - asBoolean, - asNumber, } from "../../adapters/utils.js"; import { isExternalChatWaitAuthorizationContention } from "../native-runtime/chat-attachment-reuse.js"; import { emitAgentTaskRun } from "../agent-task-run-telemetry.js"; @@ -174,7 +167,7 @@ import { shouldCancelRunsForNonInvokableAgent, type AgentOrgRow, } from "../agent-invokability.js"; -import { isHeartbeatWakeOnDemandEnabled, supportsUnscopedTimerHeartbeat } from "../heartbeat-policy.js"; +import { supportsUnscopedTimerHeartbeat } from "../heartbeat-policy.js"; import { isUuidLike } from "@paperclipai/shared"; import { retryChatControlAdmission } from "../chat-control-admission-retry.js"; @@ -234,11 +227,8 @@ export interface HeartbeatQueueDependencies extends Pick string; } -const HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT = AGENT_DEFAULT_MAX_CONCURRENT_RUNS; -const HEARTBEAT_MAX_CONCURRENT_RUNS_MIN = 1; -const HEARTBEAT_MAX_CONCURRENT_RUNS_MAX = 50; export const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext"; @@ -248,7 +238,6 @@ export const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = [ "scheduled_retry", ] as const; -const TIMER_ACTIONABLE_ISSUE_STATUSES = ["todo", "in_progress"] as const; export class ChatControlRecoveryUnresolvedError extends Error { constructor() { @@ -265,16 +254,6 @@ const RUNNING_ISSUE_WAKE_REASONS_REQUIRING_FOLLOWUP = new Set([ "issue_recovery_action_restored", ]); -function normalizeMaxConcurrentRuns(value: unknown) { - const parsed = Math.floor( - asNumber(value, HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT), - ); - if (!Number.isFinite(parsed)) return HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT; - return Math.max( - HEARTBEAT_MAX_CONCURRENT_RUNS_MIN, - Math.min(HEARTBEAT_MAX_CONCURRENT_RUNS_MAX, parsed), - ); -} export function shouldRequireIssueCommentForWake( contextSnapshot: Record | null | undefined, @@ -415,137 +394,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende filterZombieCoalesceTarget, } = dependencies; - function parseHeartbeatPolicy(agent: typeof agents.$inferSelect) { - const runtimeConfig = parseObject(agent.runtimeConfig); - const heartbeat = parseObject(runtimeConfig.heartbeat); - - return { - enabled: asBoolean(heartbeat.enabled, false), - intervalSec: Math.max(0, asNumber(heartbeat.intervalSec, 0)), - wakeOnDemand: isHeartbeatWakeOnDemandEnabled(agent), - // A Dot binding has one external turn. Competing assignments must retain - // their queue position instead of claiming a second run that cannot bind. - maxConcurrentRuns: agent.adapterType === "paperclip_runner" && - ["openai_dot", "grok_bot", "muse"].includes(String(parseObject(agent.adapterConfig).provider)) - ? 1 : normalizeMaxConcurrentRuns(heartbeat.maxConcurrentRuns), - skipTimerWhenNoActionableWork: asBoolean( - heartbeat.skipTimerWhenNoActionableWork ?? - heartbeat.requireActionableTimerWork ?? - heartbeat.issueOnlyTimer, - false, - ), - maxDailyRuns: normalizeOptionalNonNegativeInteger( - heartbeat.maxDailyRuns ?? - heartbeat.dailyRunLimit ?? - heartbeat.dailyRunCap ?? - heartbeat.maxRunsPerDay, - ), - maxDailyCostCents: normalizeOptionalNonNegativeInteger( - heartbeat.maxDailyCostCents ?? - heartbeat.dailyCostCentsLimit ?? - heartbeat.dailySpendCentsLimit ?? - heartbeat.dailyBudgetCents, - ), - }; - } - - function normalizeOptionalNonNegativeInteger(value: unknown) { - if (value === null || value === undefined || value === "") return null; - const normalized = Math.floor(asNumber(value, 0)); - return normalized >= 0 ? normalized : null; - } - - function currentUtcDayWindow(now = new Date()) { - const start = new Date( - Date.UTC( - now.getUTCFullYear(), - now.getUTCMonth(), - now.getUTCDate(), - 0, - 0, - 0, - 0, - ), - ); - const end = new Date( - Date.UTC( - now.getUTCFullYear(), - now.getUTCMonth(), - now.getUTCDate() + 1, - 0, - 0, - 0, - 0, - ), - ); - return { start, end }; - } - - async function getHeartbeatDailyCapBlock( - agent: typeof agents.$inferSelect, - policy: ReturnType, - options: { - checkRunCap?: boolean; - checkCostCap?: boolean; - excludeRunId?: string | null; - now?: Date; - } = {}, - client: Pick = db, - ) { - const checkRunCap = options.checkRunCap ?? true; - const checkCostCap = options.checkCostCap ?? true; - const { start, end } = currentUtcDayWindow(options.now); - if (checkRunCap && policy.maxDailyRuns !== null) { - const conditions = [ - eq(heartbeatRuns.companyId, agent.companyId), - eq(heartbeatRuns.agentId, agent.id), - gte(heartbeatRuns.startedAt, start), - lt(heartbeatRuns.startedAt, end), - notInArray(heartbeatRuns.status, ["queued", "scheduled_retry"]), - ]; - if (options.excludeRunId) { - conditions.push(sql`${heartbeatRuns.id} <> ${options.excludeRunId}`); - } - const [row] = await client - .select({ total: sql`count(*)::integer` }) - .from(heartbeatRuns) - .where(and(...conditions)); - const observed = Number(row?.total ?? 0); - if (observed >= policy.maxDailyRuns) { - return { - reason: "heartbeat.daily_run_limit", - observed, - limit: policy.maxDailyRuns, - }; - } - } - - if (checkCostCap && policy.maxDailyCostCents !== null) { - const [row] = await client - .select({ - total: sql`coalesce(sum(${costEvents.costCents}), 0)::text`, - }) - .from(costEvents) - .where( - and( - eq(costEvents.companyId, agent.companyId), - eq(costEvents.agentId, agent.id), - gte(costEvents.occurredAt, start), - lt(costEvents.occurredAt, end), - ), - ); - const observed = Number(row?.total ?? 0); - if (compareCents(String(row?.total ?? 0), policy.maxDailyCostCents) >= 0) { - return { - reason: "heartbeat.daily_cost_limit", - observed, - limit: policy.maxDailyCostCents, - }; - } - } - - return null; - } + const { parseHeartbeatPolicy, getHeartbeatDailyCapBlock, hasActionableTimerWork, nextTimerHeartbeatAt } = createHeartbeatTimerPolicy(db); async function cancelQueuedRunForHeartbeatDailyCap( run: typeof heartbeatRuns.$inferSelect, @@ -597,38 +446,6 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende return cancelled; } - async function hasActionableTimerWork(agent: typeof agents.$inferSelect) { - const row = await db - .select({ id: issues.id }) - .from(issues) - .where( - and( - eq(issues.companyId, agent.companyId), - eq(issues.assigneeAgentId, agent.id), - isNull(issues.assigneeUserId), - isNull(issues.hiddenAt), - inArray(issues.status, [...TIMER_ACTIONABLE_ISSUE_STATUSES]), - isNull(issues.conversationAgentId), - nonIdleSlackIssueCondition(), - ), - ) - .limit(1) - .then((rows) => rows[0] ?? null); - return Boolean(row); - } - - async function nextTimerHeartbeatAt(agent: typeof agents.$inferSelect, now = new Date()): Promise { - const policy = parseHeartbeatPolicy(agent); - if (!policy.enabled || policy.intervalSec <= 0) return null; - - if (policy.maxDailyRuns === 0 || policy.maxDailyCostCents === 0) return null; - if (policy.skipTimerWhenNoActionableWork && !(await hasActionableTimerWork(agent))) return null; - const dueAt = new Date(agent.lastHeartbeatAt ?? agent.createdAt).getTime() + policy.intervalSec * 1000; - const cap = await getHeartbeatDailyCapBlock(agent, policy, { now }); - if (!cap) return dueAt; - return Math.max(dueAt, currentUtcDayWindow(now).end.getTime()); - } - async function markTimerHeartbeatChecked( agentId: string, source: WakeupOptions["source"], diff --git a/server/src/services/heartbeat/timer-policy.ts b/server/src/services/heartbeat/timer-policy.ts new file mode 100644 index 00000000000..5b4fdf8b4c6 --- /dev/null +++ b/server/src/services/heartbeat/timer-policy.ts @@ -0,0 +1,195 @@ +import { agents, heartbeatRuns, costEvents, issues, type Db } from "@paperclipai/db"; +import { AGENT_DEFAULT_MAX_CONCURRENT_RUNS, compareCents } from "@paperclipai/shared"; +import { and, eq, gte, lt, notInArray, sql, isNull, inArray } from "drizzle-orm"; +import { asBoolean, asNumber, parseObject } from "../../adapters/utils.js"; +import { isHeartbeatWakeOnDemandEnabled } from "../heartbeat-policy.js"; +import { nonIdleSlackIssueCondition } from "../slack-conversation-state.js"; + +const HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT = AGENT_DEFAULT_MAX_CONCURRENT_RUNS; +const HEARTBEAT_MAX_CONCURRENT_RUNS_MIN = 1; +const HEARTBEAT_MAX_CONCURRENT_RUNS_MAX = 50; +const TIMER_ACTIONABLE_ISSUE_STATUSES = ["todo", "in_progress"] as const; + +function normalizeMaxConcurrentRuns(value: unknown) { + const parsed = Math.floor( + asNumber(value, HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT), + ); + if (!Number.isFinite(parsed)) return HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT; + return Math.max( + HEARTBEAT_MAX_CONCURRENT_RUNS_MIN, + Math.min(HEARTBEAT_MAX_CONCURRENT_RUNS_MAX, parsed), + ); +} + +export function parseHeartbeatPolicy(agent: typeof agents.$inferSelect) { + const runtimeConfig = parseObject(agent.runtimeConfig); + const heartbeat = parseObject(runtimeConfig.heartbeat); + + return { + enabled: asBoolean(heartbeat.enabled, false), + intervalSec: Math.max(0, asNumber(heartbeat.intervalSec, 0)), + wakeOnDemand: isHeartbeatWakeOnDemandEnabled(agent), + // A Dot binding has one external turn. Competing assignments must retain + // their queue position instead of claiming a second run that cannot bind. + maxConcurrentRuns: agent.adapterType === "paperclip_runner" && + ["openai_dot", "grok_bot", "muse"].includes(String(parseObject(agent.adapterConfig).provider)) + ? 1 : normalizeMaxConcurrentRuns(heartbeat.maxConcurrentRuns), + skipTimerWhenNoActionableWork: asBoolean( + heartbeat.skipTimerWhenNoActionableWork ?? + heartbeat.requireActionableTimerWork ?? + heartbeat.issueOnlyTimer, + false, + ), + maxDailyRuns: normalizeOptionalNonNegativeInteger( + heartbeat.maxDailyRuns ?? + heartbeat.dailyRunLimit ?? + heartbeat.dailyRunCap ?? + heartbeat.maxRunsPerDay, + ), + maxDailyCostCents: normalizeOptionalNonNegativeInteger( + heartbeat.maxDailyCostCents ?? + heartbeat.dailyCostCentsLimit ?? + heartbeat.dailySpendCentsLimit ?? + heartbeat.dailyBudgetCents, + ), + }; + } + + function normalizeOptionalNonNegativeInteger(value: unknown) { + if (value === null || value === undefined || value === "") return null; + const normalized = Math.floor(asNumber(value, 0)); + return normalized >= 0 ? normalized : null; + } + + function currentUtcDayWindow(now = new Date()) { + const start = new Date( + Date.UTC( + now.getUTCFullYear(), + now.getUTCMonth(), + now.getUTCDate(), + 0, + 0, + 0, + 0, + ), + ); + const end = new Date( + Date.UTC( + now.getUTCFullYear(), + now.getUTCMonth(), + now.getUTCDate() + 1, + 0, + 0, + 0, + 0, + ), + ); + return { start, end }; + } + + +/** Shared by dispatch and owned, read-only sleep inspections. */ +export function createHeartbeatTimerPolicy(db: Pick) { + async function getHeartbeatDailyCapBlock( + agent: typeof agents.$inferSelect, + policy: ReturnType, + options: { + checkRunCap?: boolean; + checkCostCap?: boolean; + excludeRunId?: string | null; + now?: Date; + } = {}, + client: Pick = db, + ) { + const checkRunCap = options.checkRunCap ?? true; + const checkCostCap = options.checkCostCap ?? true; + const { start, end } = currentUtcDayWindow(options.now); + if (checkRunCap && policy.maxDailyRuns !== null) { + const conditions = [ + eq(heartbeatRuns.companyId, agent.companyId), + eq(heartbeatRuns.agentId, agent.id), + gte(heartbeatRuns.startedAt, start), + lt(heartbeatRuns.startedAt, end), + notInArray(heartbeatRuns.status, ["queued", "scheduled_retry"]), + ]; + if (options.excludeRunId) { + conditions.push(sql`${heartbeatRuns.id} <> ${options.excludeRunId}`); + } + const [row] = await client + .select({ total: sql`count(*)::integer` }) + .from(heartbeatRuns) + .where(and(...conditions)); + const observed = Number(row?.total ?? 0); + if (observed >= policy.maxDailyRuns) { + return { + reason: "heartbeat.daily_run_limit", + observed, + limit: policy.maxDailyRuns, + }; + } + } + + if (checkCostCap && policy.maxDailyCostCents !== null) { + const [row] = await client + .select({ + total: sql`coalesce(sum(${costEvents.costCents}), 0)::text`, + }) + .from(costEvents) + .where( + and( + eq(costEvents.companyId, agent.companyId), + eq(costEvents.agentId, agent.id), + gte(costEvents.occurredAt, start), + lt(costEvents.occurredAt, end), + ), + ); + const observed = Number(row?.total ?? 0); + if (compareCents(String(row?.total ?? 0), policy.maxDailyCostCents) >= 0) { + return { + reason: "heartbeat.daily_cost_limit", + observed, + limit: policy.maxDailyCostCents, + }; + } + } + + return null; + } + + async function hasActionableTimerWork(agent: typeof agents.$inferSelect) { + const row = await db + .select({ id: issues.id }) + .from(issues) + .where( + and( + eq(issues.companyId, agent.companyId), + eq(issues.assigneeAgentId, agent.id), + isNull(issues.assigneeUserId), + isNull(issues.hiddenAt), + inArray(issues.status, [...TIMER_ACTIONABLE_ISSUE_STATUSES]), + isNull(issues.conversationAgentId), + nonIdleSlackIssueCondition(), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null); + return Boolean(row); + } + + async function readTimerHeartbeatSchedule(agent: typeof agents.$inferSelect, now = new Date()) { + const policy = parseHeartbeatPolicy(agent); + if (!policy.enabled || policy.intervalSec <= 0 || policy.maxDailyRuns === 0 || policy.maxDailyCostCents === 0) + return { deadline: null, policy, actionable: null, cap: null }; + const actionable = policy.skipTimerWhenNoActionableWork ? await hasActionableTimerWork(agent) : null; + if (actionable === false) return { deadline: null, policy, actionable, cap: null }; + const dueAt = new Date(agent.lastHeartbeatAt ?? agent.createdAt).getTime() + policy.intervalSec * 1000; + const cap = await getHeartbeatDailyCapBlock(agent, policy, { now }); + return { deadline: cap ? Math.max(dueAt, currentUtcDayWindow(now).end.getTime()) : dueAt, + policy, actionable, cap }; + } + + async function nextTimerHeartbeatAt(agent: typeof agents.$inferSelect, now = new Date()): Promise { + return (await readTimerHeartbeatSchedule(agent, now)).deadline; + } + return { parseHeartbeatPolicy, getHeartbeatDailyCapBlock, hasActionableTimerWork, nextTimerHeartbeatAt, readTimerHeartbeatSchedule }; +} diff --git a/server/src/services/idle-deadline-proof.test.ts b/server/src/services/idle-deadline-proof.test.ts new file mode 100644 index 00000000000..b5235916cd1 --- /dev/null +++ b/server/src/services/idle-deadline-proof.test.ts @@ -0,0 +1,124 @@ +import type { Db } from "@paperclipai/db"; +import { describe, expect, it } from "vitest"; +import { composeIdleDeadlineProof, createIdleDeadlineInspectionBudget, deadlineWorkReplacement, readIdleDeadlineProof, + IdleDeadlineInspectionBudgetExceeded, type IdleDeadlineProofOptions } from "./idle-deadline-proof.js"; +import type { VerifiedIdleCapabilities } from "./idle-incoming-wake-capability.js"; +import { projectIdlePluginJobSchedule, type IdlePluginJobRow } from "./plugin-job-idle-schedule.js"; +import { projectIdleProviderTraceSchedule, type IdleProviderTraceRow } from "./provider-trace-idle-schedule.js"; +import { projectIdleGitHubGrantContinuity } from "./tool-access/github-grant-continuity.js"; + +const now = Date.parse("2099-01-01T00:00:00Z"); +// The caller's authentication boundary has its own real signed-capability suite. +const verifiedCapability = { deadlineCatalog: "domain_deadlines_v1" } as VerifiedIdleCapabilities; +const options: IdleDeadlineProofOptions = { now, verifiedCapability, quietPluginIds: [] }; +const hold = { ownerId: "owned", expiresAt: now + 300_000 }; +const snapshot = (state: unknown = [], deadlines: number[] = []) => ({ blocked: false, state, deadlines }); +const fixtures = () => ({ timers: snapshot(["agent"], [now + 600_000]), jobs: snapshot(), traces: snapshot(), + github: { ...snapshot(), connectionIds: [] as string[] } }); + +describe("domain deadline handoff", () => { + it("preserves earliest legacy and owner obligations without moving an inspected deadline", () => { + const all = fixtures(); all.traces = snapshot(["trace"], [now + 500_000]); + const routine = { version: 1 as const, kind: "routines_and_issue_monitors" as const, + generation: "b".repeat(64), nextWakeAt: new Date(now + 400_000).toISOString() }; + const first = composeIdleDeadlineProof(hold, { ...options, scheduledWake: routine }, all); + expect(first.scheduledWake).toMatchObject({ version: 2, kind: "domain_deadlines", catalog: "domain_deadlines_v1", nextWakeAt: routine.nextWakeAt }); + expect(composeIdleDeadlineProof(hold, { ...options, now: now + 1, scheduledWake: routine }, all)).toEqual(first); + }); + it("binds passive state, drain owner/expiry and exact plugin worker generation", () => { + const all = fixtures(), first = composeIdleDeadlineProof(hold, options, all); + const passive = fixtures(); passive.jobs.state = ["changed-paused-job"]; + expect(composeIdleDeadlineProof(hold, options, passive).generation).not.toBe(first.generation); + expect(composeIdleDeadlineProof({ ...hold, ownerId: "replacement" }, options, all).generation).not.toBe(first.generation); + expect(composeIdleDeadlineProof({ ...hold, expiresAt: hold.expiresAt + 1 }, options, all).generation).not.toBe(first.generation); + const plugin = { ...options, quietPluginIds: ["plugin"], quietPluginGeneration: "worker-1" }; + expect(composeIdleDeadlineProof(hold, plugin, all).generation).not.toBe(composeIdleDeadlineProof(hold, { ...plugin, quietPluginGeneration: "worker-2" }, all).generation); + }); + it("invalidates moved-earlier/cancelled obligations and refuses due/unknown work", () => { + const all = fixtures(), first = composeIdleDeadlineProof(hold, options, all); + all.timers = snapshot(["edited"], [now + 450_000]); + expect(composeIdleDeadlineProof(hold, options, all).generation).not.toBe(first.generation); + all.timers = snapshot(["cancelled"]); + expect(composeIdleDeadlineProof(hold, options, all).scheduledWake).toBeUndefined(); + all.timers = snapshot(["due"], [now]); + expect(composeIdleDeadlineProof(hold, options, all)).toEqual({ blocked: true }); + expect(composeIdleDeadlineProof(hold, { ...options, verifiedCapability: undefined }, fixtures())).toEqual({ blocked: true }); + expect(composeIdleDeadlineProof(hold, { ...options, quietPluginIds: ["missing-worker"] }, fixtures())).toEqual({ blocked: true }); + }); + it("keeps watchdogs, pending merge cards, and recovery predicates intact", () => { + expect(deadlineWorkReplacement("agents.execution_or_timer")).toBe("SELECT 1 FROM agents WHERE status = 'running'"); + expect(deadlineWorkReplacement("plugin_jobs")).toContain("NOT IN ('active', 'paused')"); + for (const reason of ["issue_watchdogs", "issue_thread_interactions", "heartbeat_runs", "unknown"]) expect(deadlineWorkReplacement(reason)).toBeUndefined(); + }); + it("bounds whole-inspection query count and monotonic elapsed time before final hold headroom", () => { + let elapsed = 0; + const budget = createIdleDeadlineInspectionBudget(now, hold.expiresAt, () => elapsed); + for (let i = 0; i < 64; i++) budget.beforeQuery(); + expect(budget.queries).toBe(64); + expect(() => budget.beforeQuery()).toThrow(IdleDeadlineInspectionBudgetExceeded); + const timed = createIdleDeadlineInspectionBudget(now, hold.expiresAt, () => elapsed); + elapsed = 2_000; + expect(() => timed.beforeQuery()).toThrow(IdleDeadlineInspectionBudgetExceeded); + const short = createIdleDeadlineInspectionBudget(now, now + 60_000, () => 0); + expect(() => short.beforeQuery()).toThrow(IdleDeadlineInspectionBudgetExceeded); + expect(short.queries).toBe(0); + }); +}); + +const pluginJob = (): IdlePluginJobRow => ({ job: { id: "job", pluginId: "plugin", jobKey: "daily", status: "active", schedule: "0 * * * *", nextRunAt: new Date(now + 600_000) } as IdlePluginJobRow["job"], plugin: { id: "plugin", status: "ready", manifestJson: { jobs: [{ jobKey: "daily", schedule: "0 * * * *" }] } } as IdlePluginJobRow["plugin"], unsettled: false }); +it("plugin jobs require exact quiet catalog/declaration and retain unsettled/missing/due work", () => { + const row = pluginJob(); + expect(projectIdlePluginJobSchedule([row], ["plugin"], now).deadlines).toEqual([now + 600_000]); + expect(projectIdlePluginJobSchedule([row], [], now).blocked).toBe(true); + for (const edit of [{ unsettled: true }, { job: { ...row.job, nextRunAt: null } }, { job: { ...row.job, nextRunAt: new Date(now) } }]) + expect(projectIdlePluginJobSchedule([{ ...row, ...edit }], ["plugin"], now).blocked).toBe(true); + row.plugin!.manifestJson.jobs![0].schedule = "*/5 * * * *"; + expect(projectIdlePluginJobSchedule([row], ["plugin"], now).blocked).toBe(true); + row.job.status = "paused"; + expect(projectIdlePluginJobSchedule([row], [], now)).toMatchObject({ blocked: false, deadlines: [] }); +}); +const trace = (): IdleProviderTraceRow => ({ finishedRun: true, trace: { id: "trace", companyId: "company", runId: "run", status: "complete", traceRef: "s3-segments-v1:company/run/11111111-1111-4111-8111-111111111111.ndjson", expiresAt: new Date(now + 86_400_000), deletedAt: null } as IdleProviderTraceRow["trace"] }); +it("trace expiry transfers actual retained data; capture/import/deletion/cross-company work stays blocked", () => { + const row = trace(), original = JSON.stringify(row); + expect(projectIdleProviderTraceSchedule([row], now)).toMatchObject({ blocked: false, deadlines: [now + 86_400_000] }); + expect(JSON.stringify(row)).toBe(original); + for (const patch of [{ status: "capturing" }, { traceRef: row.trace.traceRef.replace("s3-segments-v1:", "s3-import-v1:") }, { companyId: "foreign" }, { expiresAt: new Date(now) }, { status: "deleted", deletedAt: new Date(now) }]) + expect(projectIdleProviderTraceSchedule([{ ...trace(), trace: { ...trace().trace, ...patch } }], now).blocked).toBe(true); + expect(projectIdleProviderTraceSchedule([{ ...trace(), finishedRun: false }], now).blocked).toBe(true); + const expired = trace(); expired.trace.status = "expired"; expired.trace.deletedAt = new Date(now); expired.trace.traceRef = "00000000-0000-0000-0000-000000000000.ndjson"; + expect(projectIdleProviderTraceSchedule([expired], now)).toMatchObject({ blocked: false, deadlines: [] }); +}); +type ContinuityRow = Parameters[0][number]; +const github = (): ContinuityRow => ({ companyStatus: "active", connection: { id: "connection", companyId: "company", status: "active", enabled: true, config: { sourceTemplateKey: "github", oauth: { connectorProfile: "github.code" } } } as unknown as ContinuityRow["connection"], grant: { id: "grant", companyId: "company", credentialSecretRefs: [{ configPath: "oauth.access_token", secretId: "secret" }], providerTenant: { oauth: { accessTokenExpiresAt: new Date(now + 8 * 3_600_000).toISOString(), refreshedAt: new Date(now).toISOString() }, github: { userId: "user", installationIds: ["101"], lastAccessRefreshAt: new Date(now).toISOString() } } } as ContinuityRow["grant"] }); +it("GitHub enrollment alone never bypasses its credential continuity deadline or unresolved authority", () => { + expect(projectIdleGitHubGrantContinuity([github()], now)).toMatchObject({ blocked: false, deadlines: [now + 7 * 3_600_000], connectionIds: ["connection"] }); + const nonexpiring = github(); delete nonexpiring.grant.providerTenant!.oauth; + expect(projectIdleGitHubGrantContinuity([nonexpiring], now).deadlines).toEqual([now + 30 * 86_400_000]); + const missing = github(); missing.grant.credentialSecretRefs = []; + expect(projectIdleGitHubGrantContinuity([missing], now).blocked).toBe(true); + const foreign = github(); foreign.grant.companyId = "foreign"; + expect(projectIdleGitHubGrantContinuity([foreign], now).blocked).toBe(true); + const expired = github(); expired.grant.providerTenant!.oauth!.accessTokenExpiresAt = new Date(now).toISOString(); + expect(projectIdleGitHubGrantContinuity([expired], now).blocked).toBe(true); +}); + + +it("stops owner read orchestration at its query budget and rejects row overflow", async () => { + const agentRows = Array.from({ length: 80 }, (_, i) => ({ agent: { id: `agent-${i}`, companyId: "company", name: "Timer", + status: "idle", reportsTo: null, adapterType: "codex_local", runtimeConfig: { heartbeat: { enabled: true, intervalSec: 3600, maxDailyRuns: 1 } }, + createdAt: new Date(now), lastHeartbeatAt: null }, companyStatus: "active", companyUpdatedAt: new Date(now) })); + let reads = 0; + const tx = { select: (columns: object) => { + reads++; + const rows = "agent" in columns ? agentRows : [{ total: 0 }]; + const query = { from: () => query, leftJoin: () => query, orderBy: () => query, where: () => query, + limit: () => Promise.resolve(rows), then: (resolve: (rows: unknown[]) => unknown) => Promise.resolve(rows).then(resolve) }; + return query; + } } as unknown as Pick; + expect(await readIdleDeadlineProof(tx, hold, options)).toEqual({ blocked: true }); + expect(reads).toBe(64); + agentRows.push(...Array.from({ length: 10_001 }, () => agentRows[0])); + reads = 0; + expect(await readIdleDeadlineProof(tx, hold, options)).toEqual({ blocked: true }); + expect(reads).toBe(1); +}); diff --git a/server/src/services/idle-deadline-proof.ts b/server/src/services/idle-deadline-proof.ts new file mode 100644 index 00000000000..8ad02af86b8 --- /dev/null +++ b/server/src/services/idle-deadline-proof.ts @@ -0,0 +1,98 @@ +import { createHash, randomUUID } from "node:crypto"; +import type { Db } from "@paperclipai/db"; +import type { VerifiedIdleCapabilities } from "./idle-incoming-wake-capability.js"; +import type { ScheduledWake } from "./idle-scheduled-work.js"; +import { IDLE_DEADLINE_CATALOG, type DomainScheduledWake, type IdleDeadlineSnapshot } from "./idle-deadline-types.js"; +import { readIdleAgentTimerSchedule } from "./heartbeat/idle-timer-schedule.js"; +import { readIdlePluginJobSchedule } from "./plugin-job-idle-schedule.js"; +import { readIdleProviderTraceSchedule } from "./provider-trace-idle-schedule.js"; +import { readIdleGitHubGrantContinuity } from "./tool-access/github-grant-continuity.js"; + +const processGeneration = randomUUID(); +export class IdleDeadlineInspectionBudgetExceeded extends Error {} +/** Global bounds supplement the transaction's per-statement timeout. */ +export function createIdleDeadlineInspectionBudget(now: number, expiresAt: number, + monotonicNow: () => number = () => performance.now()) { + const started = monotonicNow(); + const elapsedLimit = Math.min(2_000, expiresAt - now - 60_000); + let queries = 0; + const check = () => { + if (elapsedLimit <= 0 || monotonicNow() - started >= elapsedLimit) + throw new IdleDeadlineInspectionBudgetExceeded("Idle deadline inspection exhausted its time budget"); + }; + return { check, beforeQuery() { + check(); + if (++queries > 64) throw new IdleDeadlineInspectionBudgetExceeded("Idle deadline inspection exhausted its query budget"); + }, get queries() { return queries; } }; +} +export type IdleDeadlineProof = { + blocked: boolean; + generation?: string; + scheduledWake?: DomainScheduledWake; + githubContinuity?: { connectionIds: string[]; generation: string }; +}; +export type IdleDeadlineProofOptions = { + now: number; + verifiedCapability?: VerifiedIdleCapabilities; + quietPluginIds: readonly string[]; + quietPluginGeneration?: string; + scheduledWake?: ScheduledWake; + passiveGeneration?: string; +}; +const generation = (state: unknown) => createHash("sha256").update(JSON.stringify(state)).digest("hex"); + +/** Proof composition only. Domain ownership, durable reads and execution stay with their services. */ +export function composeIdleDeadlineProof(hold: { ownerId: string; expiresAt: number }, options: IdleDeadlineProofOptions, + snapshots: { timers: IdleDeadlineSnapshot; jobs: IdleDeadlineSnapshot; traces: IdleDeadlineSnapshot; + github: IdleDeadlineSnapshot & { connectionIds: string[] } }): IdleDeadlineProof { + if (options.verifiedCapability?.deadlineCatalog !== IDLE_DEADLINE_CATALOG || !hold.ownerId + || !Number.isFinite(hold.expiresAt) || hold.expiresAt <= options.now + || Object.values(snapshots).some(snapshot => snapshot.blocked) + || (options.quietPluginIds.length > 0 && !options.quietPluginGeneration)) return { blocked: true }; + const deadlines = Object.values(snapshots).flatMap(snapshot => snapshot.deadlines); + if (options.scheduledWake) deadlines.push(Date.parse(options.scheduledWake.nextWakeAt)); + if (deadlines.some(at => !Number.isFinite(at) || at <= options.now)) return { blocked: true }; + const proofGeneration = generation([IDLE_DEADLINE_CATALOG, processGeneration, hold, + options.quietPluginIds, options.quietPluginGeneration, options.scheduledWake, options.passiveGeneration, + Object.entries(snapshots).map(([owner, snapshot]) => [owner, snapshot.state, snapshot.deadlines])]); + const githubContinuity = { connectionIds: snapshots.github.connectionIds, + generation: generation([proofGeneration, snapshots.github.connectionIds]) }; + return { blocked: false, generation: proofGeneration, githubContinuity, + ...(deadlines.length ? { scheduledWake: { version: 2, kind: "domain_deadlines", catalog: IDLE_DEADLINE_CATALOG, + generation: proofGeneration, nextWakeAt: new Date(Math.min(...deadlines)).toISOString() } } : {}) }; +} + +export async function readIdleDeadlineProof(tx: Pick, hold: { ownerId: string; expiresAt: number }, + options: IdleDeadlineProofOptions): Promise { + if (options.verifiedCapability?.deadlineCatalog !== IDLE_DEADLINE_CATALOG) return { blocked: true }; + const budget = createIdleDeadlineInspectionBudget(options.now, hold.expiresAt); + const guarded = { select: (...args: unknown[]) => { + budget.beforeQuery(); + return Reflect.apply(tx.select, tx, args); + } } as Pick; + try { + // Sequential reads share the caller's snapshot and statement timeout. + const timers = await readIdleAgentTimerSchedule(guarded, options.now, process.env, budget.check); + budget.check(); + if (timers.blocked) return { blocked: true }; + const jobs = await readIdlePluginJobSchedule(guarded, options.quietPluginIds, options.now); + budget.check(); + if (jobs.blocked) return { blocked: true }; + const traces = await readIdleProviderTraceSchedule(guarded, options.now); + budget.check(); + if (traces.blocked) return { blocked: true }; + const github = await readIdleGitHubGrantContinuity(guarded, options.now); + budget.check(); + return composeIdleDeadlineProof(hold, options, { timers, jobs, traces, github }); + } catch (error) { + if (error instanceof IdleDeadlineInspectionBudgetExceeded) return { blocked: true }; + throw error; + } +} + +/** Caller retains the original reason identity and applies this only after complete proof. */ +export function deadlineWorkReplacement(reason: string): string | undefined { + if (reason === "agents.execution_or_timer") return `SELECT 1 FROM agents WHERE status = 'running'`; + if (reason === "plugin_jobs") return `SELECT 1 FROM plugin_jobs WHERE status NOT IN ('active', 'paused')`; + return undefined; +} diff --git a/server/src/services/idle-deadline-types.ts b/server/src/services/idle-deadline-types.ts new file mode 100644 index 00000000000..fb72e7fa3e8 --- /dev/null +++ b/server/src/services/idle-deadline-types.ts @@ -0,0 +1,17 @@ +/** Closed negotiated catalog; in-process scheduler entries are not wake proofs. */ +export const IDLE_DEADLINE_CATALOG = "domain_deadlines_v1" as const; +export interface DomainScheduledWake { + version: 2; + kind: "domain_deadlines"; + catalog: typeof IDLE_DEADLINE_CATALOG; + generation: string; + nextWakeAt: string; +} +export interface IdleDeadlineSnapshot { + blocked: boolean; + /** Exact durable state inspected by the domain; hashed, never returned to callers. */ + state: unknown; + deadlines: number[]; +} +export const idleDeadlineRowLimit = 10_000; +export const isFutureIdleDeadline = (at: number, now: number) => Number.isFinite(at) && at > now; diff --git a/server/src/services/idle-local-work.ts b/server/src/services/idle-local-work.ts index 95b6549ebbc..40098e6bc9e 100644 --- a/server/src/services/idle-local-work.ts +++ b/server/src/services/idle-local-work.ts @@ -3,10 +3,13 @@ import path from "node:path"; import { resolvePaperclipInstanceRoot } from "../home-paths.js"; import { idleWorkSnapshot } from "./task-admission.js"; +import type { VerifiedIdleCapabilities } from "./idle-incoming-wake-capability.js"; + +export type IdleLocalWorkOptions = { backupCheckpoint?: boolean; deadlineCapability?: VerifiedIdleCapabilities }; export type IdleLocalWork = "none" | "present" | "unknown"; const spoolDirectories = new Set(); -const localWorkProbes = new Set<() => IdleLocalWork>(); -export function registerIdleLocalWorkProbe(probe: () => IdleLocalWork): () => void { +const localWorkProbes = new Set<(options: IdleLocalWorkOptions) => IdleLocalWork>(); +export function registerIdleLocalWorkProbe(probe: (options: IdleLocalWorkOptions) => IdleLocalWork): () => void { localWorkProbes.add(probe); return () => { localWorkProbes.delete(probe); }; } @@ -61,13 +64,13 @@ export async function inspectIdleQuotaRecovery(directory: string): Promise { +export async function readIdleLocalWork(options: IdleLocalWorkOptions = {}): Promise { if (!startupComplete || !ingressTracked) return "unknown"; // This exemption belongs only to a scan which will finish a new backup // under the same admission hold before it can return an empty report. if ((scheduledBackups && !options.backupCheckpoint) || idleWorkSnapshot().active !== 0) return "present"; try { - const local = [...localWorkProbes].map(probe => probe()); + const local = [...localWorkProbes].map(probe => probe(options)); if (local.includes("unknown")) return "unknown"; if (local.includes("present")) return "present"; } catch { return "unknown"; } diff --git a/server/src/services/plugin-job-idle-schedule.ts b/server/src/services/plugin-job-idle-schedule.ts new file mode 100644 index 00000000000..77f5711d53a --- /dev/null +++ b/server/src/services/plugin-job-idle-schedule.ts @@ -0,0 +1,37 @@ +import { plugins, pluginJobs, type Db } from "@paperclipai/db"; +import { eq, getTableColumns, sql } from "drizzle-orm"; +import { validateCron } from "./cron.js"; +import { idleDeadlineRowLimit, isFutureIdleDeadline, type IdleDeadlineSnapshot } from "./idle-deadline-types.js"; + +type Job = typeof pluginJobs.$inferSelect; +type Plugin = typeof plugins.$inferSelect; +export type IdlePluginJobRow = { job: Job; plugin: Plugin | null; unsettled: boolean }; +export function projectIdlePluginJobSchedule(rows: IdlePluginJobRow[], quietPluginIds: readonly string[], now: number): IdleDeadlineSnapshot { + const state = rows; + const deadlines: number[] = []; + const blocked = () => ({ blocked: true, state, deadlines: [] }); + if (rows.length > idleDeadlineRowLimit) return blocked(); + const quiet = new Set(quietPluginIds); + for (const { job, plugin, unsettled } of rows) { + if (!["active", "paused"].includes(job.status) || unsettled !== false) return blocked(); + if (job.status === "paused") continue; + if (!plugin || plugin.id !== job.pluginId || plugin.status !== "ready" || !quiet.has(plugin.id)) return blocked(); + const declarations = plugin.manifestJson?.jobs; + if (!Array.isArray(declarations) || declarations.filter(d => d.jobKey === job.jobKey).length !== 1) return blocked(); + const declaration = declarations.find(d => d.jobKey === job.jobKey)!; + if ((declaration.schedule ?? "") !== job.schedule) return blocked(); + // An explicitly unscheduled declaration is manual-only, not a missing timer. + if (!job.schedule && job.nextRunAt === null) continue; + const at = job.nextRunAt?.getTime() ?? NaN; + if (validateCron(job.schedule) || !isFutureIdleDeadline(at, now)) return blocked(); + deadlines.push(at); + } + return { blocked: false, state, deadlines }; +} +export async function readIdlePluginJobSchedule(tx: Pick, quietPluginIds: readonly string[], now: number) { + const rows = await tx.select({ job: getTableColumns(pluginJobs), plugin: getTableColumns(plugins), + unsettled: sql`EXISTS (SELECT 1 FROM plugin_job_runs r WHERE r.job_id = ${pluginJobs.id} + AND (r.plugin_id <> ${pluginJobs.pluginId} OR r.status NOT IN ('succeeded', 'failed', 'cancelled') OR r.finished_at IS NULL))`, + }).from(pluginJobs).leftJoin(plugins, eq(plugins.id, pluginJobs.pluginId)).orderBy(pluginJobs.id).limit(idleDeadlineRowLimit + 1); + return projectIdlePluginJobSchedule(rows, quietPluginIds, now); +} diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index 363d15c065c..0ac08f7b046 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -1,3 +1,4 @@ +import { createHash } from "node:crypto"; /** * PluginWorkerManager — spawns and manages out-of-process plugin worker child * processes, routes JSON-RPC 2.0 calls over stdio, and handles lifecycle @@ -811,6 +812,7 @@ export interface PluginWorkerManager { onConfigApplied(listener: () => void): () => void; inspectIdleSleep?(hold: { ownerId: string; expiresAt: number }): Promise<{ backgroundWork: "none" | "present" | "unknown"; pluginIds: string[]; + catalogGeneration?: string; }>; releaseIdleSleep?(): void; /** @@ -3886,6 +3888,8 @@ export function createPluginWorkerManager( managerOptions?.maxConcurrentDuplexRoutes, ); + const idleWorkerIdentities = new WeakMap(); + let nextIdleWorkerIdentity = 0; return { onConfigApplied(listener) { configApplied.add(listener); @@ -3900,7 +3904,12 @@ export function createPluginWorkerManager( return { backgroundWork: "unknown", pluginIds: [] }; } const backgroundWork = results.includes("present") ? "present" : results.includes("unknown") ? "unknown" : "none"; - return { backgroundWork, pluginIds: backgroundWork === "none" ? snapshot.map(([id]) => id) : [] }; + const catalogGeneration = createHash("sha256").update(JSON.stringify(snapshot.map(([id, worker]) => { + let identity = idleWorkerIdentities.get(worker); + if (!identity) { identity = String(++nextIdleWorkerIdentity); idleWorkerIdentities.set(worker, identity); } + return [id, identity]; + }))).digest("hex"); + return { backgroundWork, pluginIds: backgroundWork === "none" ? snapshot.map(([id]) => id) : [], catalogGeneration }; }, releaseIdleSleep() { for (const worker of workers.values()) worker.releaseIdleSleep?.(); diff --git a/server/src/services/provider-trace-idle-schedule.ts b/server/src/services/provider-trace-idle-schedule.ts new file mode 100644 index 00000000000..b00b37bce62 --- /dev/null +++ b/server/src/services/provider-trace-idle-schedule.ts @@ -0,0 +1,36 @@ +import { providerTraceRecords, type Db } from "@paperclipai/db"; +import { getTableColumns, sql } from "drizzle-orm"; +import { importingProviderTraceRef, objectProviderTraceRef } from "./provider-trace-object-store.js"; +import { finishedIdleRun } from "./idle-work-proof.js"; +import { idleDeadlineRowLimit, isFutureIdleDeadline, type IdleDeadlineSnapshot } from "./idle-deadline-types.js"; + +const expiredTraceRef = "00000000-0000-0000-0000-000000000000.ndjson"; +export type IdleProviderTraceRow = { trace: typeof providerTraceRecords.$inferSelect; finishedRun: boolean }; +export function projectIdleProviderTraceSchedule(rows: IdleProviderTraceRow[], now: number): IdleDeadlineSnapshot { + const state = rows; + const deadlines: number[] = []; + const blocked = () => ({ blocked: true, state, deadlines: [] }); + if (rows.length > idleDeadlineRowLimit) return blocked(); + for (const { trace, finishedRun } of rows) { + if (finishedRun !== true) return blocked(); + if (trace.traceRef === expiredTraceRef) { + if (!["expired", "deleted"].includes(trace.status) || !trace.deletedAt) return blocked(); + continue; + } + const objectRef = objectProviderTraceRef(trace.traceRef); + if (importingProviderTraceRef(trace.traceRef) || (objectRef && (objectRef[1] !== trace.companyId || objectRef[2] !== trace.runId)) + || (!objectRef && !/^[a-f0-9-]{36}\.ndjson$/.test(trace.traceRef)) + || trace.deletedAt || !["complete", "incomplete", "truncated"].includes(trace.status)) return blocked(); + const at = trace.expiresAt.getTime(); + if (!isFutureIdleDeadline(at, now)) return blocked(); + deadlines.push(at); + } + return { blocked: false, state, deadlines }; +} +export async function readIdleProviderTraceSchedule(tx: Pick, now: number) { + const rows = await tx.select({ trace: getTableColumns(providerTraceRecords), + finishedRun: sql`EXISTS (SELECT 1 FROM heartbeat_runs r WHERE r.id = ${providerTraceRecords.runId} + AND r.company_id = ${providerTraceRecords.companyId} AND ${sql.raw(finishedIdleRun("r"))})`, + }).from(providerTraceRecords).orderBy(providerTraceRecords.id).limit(idleDeadlineRowLimit + 1); + return projectIdleProviderTraceSchedule(rows, now); +} diff --git a/server/src/services/provider-trace-retention-work.ts b/server/src/services/provider-trace-retention-work.ts index 7aa20d65f5f..dba9c7d591c 100644 --- a/server/src/services/provider-trace-retention-work.ts +++ b/server/src/services/provider-trace-retention-work.ts @@ -13,9 +13,14 @@ export function registerProviderTraceRetentionWork( const traces = providerTraceStore(db); let deadline: number | null | undefined; - const unregister = registerIdleLocalWorkProbe(() => - deadline === undefined ? "unknown" : deadline === null ? "none" : "present", - ); + const unregister = registerIdleLocalWorkProbe(options => { + if (deadline === undefined) return "unknown"; + if (deadline === null) return "none"; + // Only a full negotiated durable scan may transfer a future retention + // obligation. Due cleanup and failed/unknown reconciliation remain work. + return options.deadlineCapability?.deadlineCatalog === "domain_deadlines_v1" && deadline > Date.now() + ? "none" : "present"; + }); const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.providerTraceRetention, { canRun, retryMs: 30_000, diff --git a/server/src/services/tool-access/github-grant-continuity.ts b/server/src/services/tool-access/github-grant-continuity.ts index 13aaec986e7..69a67e7171c 100644 --- a/server/src/services/tool-access/github-grant-continuity.ts +++ b/server/src/services/tool-access/github-grant-continuity.ts @@ -395,3 +395,36 @@ export function githubContinuityRevision(grant: Pick, now: number) { + const rows = await tx.select({ grant: connectionGrants, connection: toolConnections, companyStatus: companies.status }) + .from(connectionGrants).leftJoin(toolConnections, eq(toolConnections.id, connectionGrants.connectionId)) + .leftJoin(companies, eq(companies.id, connectionGrants.companyId)) + .where(and(eq(connectionGrants.status, "active"), sql`${connectionGrants.providerTenant}->'github' IS NOT NULL`)) + .orderBy(connectionGrants.id).limit(10_001); + return projectIdleGitHubGrantContinuity(rows, now); +} +export function projectIdleGitHubGrantContinuity(rows: { + grant: typeof connectionGrants.$inferSelect; + connection: typeof toolConnections.$inferSelect | null; + companyStatus: string | null; +}[], now: number) { + const deadlines: number[] = []; + const connectionIds = new Set(); + const blocked = () => ({ blocked: true, state: rows, deadlines: [], connectionIds: [] as string[] }); + if (rows.length > 10_000) return blocked(); + for (const { grant, connection, companyStatus } of rows) { + if (!connection || connection.companyId !== grant.companyId || !companyStatus) return blocked(); + if (companyStatus !== "active" || connection.status !== "active" || !connection.enabled) continue; + const oauth = asRecord(grant.providerTenant?.oauth); + const github = grant.providerTenant?.github; + if (!github || github.bindingSync || github.continuityRetry || oauth.refreshLease) return blocked(); + const deadline = githubGrantContinuityDeadline(connection, grant); + if (deadline === null || !Number.isFinite(deadline) || deadline <= now) return blocked(); + if (!grant.credentialSecretRefs.some(ref => ref.configPath === "oauth.access_token")) return blocked(); + deadlines.push(deadline); + connectionIds.add(connection.id); + } + return { blocked: false, state: rows, deadlines, connectionIds: [...connectionIds].sort() }; +}