From 65965f0d3a53e55ba93aa4b7f6373dbdb3b17dea Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 9 Oct 2026 15:09:11 -0500 Subject: [PATCH] refactor(heartbeat): extract scheduling and monitor recovery (#15717) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The heartbeat service schedules and runs agent work. > - Previous refactors moved queue admission, recovery, and run control into separate modules. > - Timer ticks, issue monitors, and session-goal recovery still occupy more than 1,100 lines in the main service. > - This pull request moves that scheduling code into one module with explicit dependencies. > - This makes scheduling easier to review and test before we split the execution pipeline. ## Linked Issues or Issue Description Refs #15685, the merged run-control extraction. A search found no duplicate open scheduling extraction. **What existing behavior does this improve?** Maintenance and testing of heartbeat scheduling. **Current behavior** The main heartbeat service contains 10,473 lines. Scheduling shares a file with the execution pipeline. **Proposed behavior** Move scheduling into `server/src/services/heartbeat/scheduling.ts`. Keep the same public methods, admission rules, database claims, and execution behavior. The main file falls to 9,330 lines. **Reason and benefit** Give timer ticks, issue-monitor recovery, and session-goal recovery one clear module boundary. **Breaking changes** None intended. Existing heartbeat exports and method signatures remain available. ## What Changed - Add `createHeartbeatScheduling` with 11 explicit dependencies. Construction does not query the database or start timers. - Move timer ticks, issue-monitor dispatch and recovery, and session-goal recovery into the new module. - Keep queue admission, heartbeat policy parsing, and atomic timer claims in `queue.ts`. - Share the unchanged goal-control parser and issue-link formatter with their existing callers. - Add 20 focused tests. Document the module boundary in `doc/DEVELOPING.md`. ## Verification - 122 tests pass across 11 files: 20 new scheduler tests, 52 existing scheduling integration tests, and 50 adjacent queue, goal, native-selection, and task-drain tests. - New database tests cover cross-instance monitor claims, replacement schedules, deferred and failed admission, native ownership, archived companies, timer claim loss, worktree cutoffs, agent eligibility, goal recovery keys, pending-action deduplication, and invalid-action persistence. - AST comparison confirms 20 unchanged function bodies, unchanged remaining service statements, unchanged `executeRun`, and all 140 public exports preserved. - Workspace typecheck (`pnpm -r typecheck`) and build (`pnpm build`) pass. Greptile is 5/5 on commit `89a1a3b3aa428949eba349af5c1618391f97a19c`. All CI checks pass on that commit. The full test suite runs in CI; local verification was targeted. - The first CI attempt hit a 15-second timeout in the first test of unchanged `plugin-scoped-api-routes.test.ts`. All nine tests pass locally in 6.6 seconds. That shard passed on its one retry without code changes. - Run the new tests with `pnpm exec vitest run server/src/services/heartbeat/scheduling.test.ts --maxWorkers=1`. ## Risks - Incorrect wiring could bypass suppression, duplicate a monitor wake, or lose a replacement schedule. Database tests cover these boundaries and existing integration tests still call the public heartbeat service. - This PR moves the current behavior without changing scheduling policy, schema, provider execution, or cancellation ownership. ## Model Used OpenAI Codex, based on GPT-6, with reasoning, code editing, and command execution. The session does not expose the exact model variant or context window size. ## 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 --- doc/DEVELOPING.md | 12 +- server/src/services/heartbeat.ts | 1199 +--------------- .../src/services/heartbeat/scheduling.test.ts | 284 ++++ server/src/services/heartbeat/scheduling.ts | 1248 +++++++++++++++++ 4 files changed, 1571 insertions(+), 1172 deletions(-) create mode 100644 server/src/services/heartbeat/scheduling.test.ts create mode 100644 server/src/services/heartbeat/scheduling.ts diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 8eb659935dc..624c8334674 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -1271,7 +1271,7 @@ It owns bounded run projections, database encoding checks, task session reads an writes, explicit resumes, session compaction, and usage/billing helpers. `createHeartbeatRunState(db)` binds these operations without doing database work during construction. The encoding-check cache belongs to each factory instance. -`heartbeat.ts` keeps run execution and session-goal recovery, and re-exports the +`heartbeat.ts` keeps run execution and re-exports the existing public helpers. Keep session policy changes separate from run orchestration changes. @@ -1321,6 +1321,16 @@ Forwarding callbacks preserve construction order for budget, retry, recovery, and queue services. Existing public helpers remain re-exported by `heartbeat.ts`. Keep cancellation policy changes separate from this extraction and execution. +Scheduling is in `server/src/services/heartbeat/scheduling.ts`. It owns timer +ticks, issue-monitor claims and dispatch, exhausted-monitor recovery, and +session-goal recovery. `createHeartbeatScheduling` binds explicit dependencies +without querying or starting timers. Queue admission, heartbeat policy parsing, +and atomic timer claims stay in `queue.ts`; the service supplies those operations +and the current scheduling-suppression and worktree-cutoff callbacks. Monitor +claims remain database-scoped so separate scheduler instances cannot dispatch +the same claim. The public heartbeat methods keep their existing signatures. +Keep scheduling policy changes separate from this extraction and execution. + ## Wake Context Delivery Built-in adapters deliver wake context through the run prompt, including structured diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 8e514e11ea4..39f6c0a7ce6 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,8 @@ +import { + createHeartbeatScheduling, + formatIssueIdentifierLink, + parseNativeSessionGoalControl, +} from "./heartbeat/scheduling.js"; import { agentExecutionsHaveStopped } from "./agent-execution-stop.js"; import { cancelHeartbeatNativeRun, @@ -339,7 +344,6 @@ import { buildNativeReviewRequest } from "./native-runtime/native-review-prompt. import { legacyExecutionNeedsReconciliationWithEvidence, settleInterruptedNativeBootstrap, - terminalizeLegacyExecution, } from "./legacy-execution-recovery.js"; import { adapterExecutionControls, @@ -381,12 +385,8 @@ import { desc, eq, exists, - getTableColumns, - gte, inArray, isNull, - lt, - lte, notInArray, or, sql, @@ -395,21 +395,14 @@ import type { Db } from "@paperclipai/db"; import { CHAT_PROVIDERS, ISSUE_DISPOSITION_REPAIR_RETRY_REASON, - PROVIDER_QUOTA_MONITOR_SERVICE_NAME, isEnvironmentDriverSupportedForAdapter, type ExecutionWorkspace, - type IssueExecutionMonitorClearReason, - type IssueExecutionMonitorPolicy, - type IssueExecutionMonitorRecoveryPolicy, } from "@paperclipai/shared"; import { agents, agentConfigRevisions, - agentSessionGoalActions, - agentTaskSessions, agentWakeupRequests, chatActions, - companies, completionContracts, heartbeatRunEvents, heartbeatRuns, @@ -496,7 +489,6 @@ import { describeRunnerdNativeSessionBackend, parseNativeExecutionInput, type NativeExecutionInput, - type NativeSessionGoalControl, type NativeSessionBackend, } from "../vendor/paperclip-runner/index.js"; import { createNativeSessionHandoffLoader } from "./native-runtime/native-session-handoff.js"; @@ -610,12 +602,7 @@ import { } from "./runner-goals.js"; import { resolveChatRunPresentationAuthorizationReason } from "./chat-run-publications.js"; -import { visibleIssueCondition } from "./issue-visibility.js"; - import { - buildIssueMonitorClearedPatch, - buildIssueMonitorTriggeredPatch, - normalizeIssueExecutionPolicy, parseIssueExecutionState, } from "./issue-execution-policy.js"; import { @@ -668,9 +655,7 @@ import { evaluateExecutionAllowlist, isExecutionForcedToKubernetes, } from "./execution-allowlist.js"; -import { - RECOVERY_ORIGIN_KINDS, -} from "./recovery/index.js"; + import { buildConfigurationIncompleteRecoveryNoticeSeed, buildExecutionReviewParticipantRecoveryNoticeSeed, @@ -679,7 +664,7 @@ import { SANDBOX_PROVIDER_PLUGIN_NOT_READY_REASON, type StrandedRecoveryNoticeSeed, } from "./recovery/stranded-notice.js"; -import { withRecoveryContext } from "./recovery/status-only-context.js"; + import { recoveryService } from "./recovery/service.js"; import { createRunDispatch, @@ -702,14 +687,11 @@ import { type ReleaseRecoveryBlockedNoticeKind, type RunSnapshot as WakeQueueRunSnapshot, } from "../modules/wake-queue/index.js"; -import { - REVIEW_PATH_RECOVERY_INSTRUCTION, -} from "./recovery/review-path-recovery.js"; + import { taskWatchdogService } from "./task-watchdogs.js"; import { - evaluateAgentInvokability, evaluateAgentInvokabilityFromDb, DIRECT_NON_INVOKABLE_STATUSES, } from "./agent-invokability.js"; @@ -730,9 +712,7 @@ import { readPaperclipSkillSyncPreference, selectPaperclipTaskMarkdown, } from "@paperclipai/adapter-utils/server-utils"; -import { - isUuidLike, -} from "@paperclipai/shared"; + import { environmentService } from "./environments.js"; import { parseExecutionPolicyBootstrapEnv } from "./execution-policy-bootstrap.js"; @@ -1090,36 +1070,6 @@ function readNonEmptyString(value: unknown): string | null { return typeof value === "string" && value.trim().length > 0 ? value : null; } -function parseNativeSessionGoalControl( - value: unknown, -): NativeSessionGoalControl | null { - const candidate = parseObject(value); - const requestId = readNonEmptyString(candidate.requestId); - const action = readNonEmptyString(candidate.action); - if ( - !requestId || - !action || - !["create", "edit", "replace", "pause", "resume", "clear"].includes(action) - ) - return null; - const objective = readNonEmptyString(candidate.objective); - const tokenBudget = - candidate.tokenBudget === null - ? null - : typeof candidate.tokenBudget === "number" && - Number.isSafeInteger(candidate.tokenBudget) && - candidate.tokenBudget > 0 - ? candidate.tokenBudget - : undefined; - return { - requestId, - action: action as NativeSessionGoalControl["action"], - ...(objective ? { objective } : {}), - ...(tokenBudget !== undefined ? { tokenBudget } : {}), - }; -} - - function parseIssueAssigneeAdapterOverrides( raw: unknown, ): ParsedIssueAssigneeAdapterOverrides | null { @@ -2113,871 +2063,24 @@ export function heartbeatService( return evaluateAgentInvokabilityFromDb(db, agent); } - const issueMonitorDispatchColumns = { - id: issues.id, - companyId: issues.companyId, - projectId: issues.projectId, - goalId: issues.goalId, - identifier: issues.identifier, - title: issues.title, - status: issues.status, - priority: issues.priority, - assigneeAgentId: issues.assigneeAgentId, - assigneeUserId: issues.assigneeUserId, - billingCode: issues.billingCode, - executionPolicy: issues.executionPolicy, - executionState: issues.executionState, - monitorNextCheckAt: issues.monitorNextCheckAt, - monitorWakeRequestedAt: issues.monitorWakeRequestedAt, - monitorLastTriggeredAt: issues.monitorLastTriggeredAt, - monitorAttemptCount: issues.monitorAttemptCount, - monitorNotes: issues.monitorNotes, - monitorScheduledBy: issues.monitorScheduledBy, - }; - - interface IssueMonitorDispatchRow { - id: string; - companyId: string; - projectId: string | null; - goalId: string | null; - identifier: string | null; - title: string; - status: string; - priority: string; - assigneeAgentId: string | null; - assigneeUserId: string | null; - billingCode: string | null; - executionPolicy: Record | null; - executionState: Record | null; - monitorNextCheckAt: Date | null; - monitorWakeRequestedAt: Date | null; - monitorLastTriggeredAt: Date | null; - monitorAttemptCount: number | null; - monitorNotes: string | null; - monitorScheduledBy: string | null; - } - - function parseMonitorDate(value: string | null | undefined) { - if (!value) return null; - const date = new Date(value); - return Number.isNaN(date.getTime()) ? null : date; - } - - function issueMonitorLimitClearReason(input: { - monitor: IssueExecutionMonitorPolicy | null; - nextAttemptCount: number; - now: Date; - }): IssueExecutionMonitorClearReason | null { - const timeoutAt = parseMonitorDate(input.monitor?.timeoutAt ?? null); - if (timeoutAt && input.now.getTime() >= timeoutAt.getTime()) { - return "timeout_exceeded"; - } - const maxAttempts = input.monitor?.maxAttempts ?? null; - if (maxAttempts !== null && input.nextAttemptCount > maxAttempts) { - return "max_attempts_exhausted"; - } - return null; - } - - function monitorRecoveryPolicy( - monitor: IssueExecutionMonitorPolicy | null, - ): IssueExecutionMonitorRecoveryPolicy { - return monitor?.recoveryPolicy ?? "wake_owner"; - } - - function monitorRecoveryDetails(input: { - claimed: IssueMonitorDispatchRow; - scheduledAtIso: string; - nextAttemptCount: number; - clearReason: IssueExecutionMonitorClearReason; - recoveryPolicy: IssueExecutionMonitorRecoveryPolicy; - monitor: IssueExecutionMonitorPolicy | null; - source: "manual" | "scheduled"; - }) { - return { - identifier: input.claimed.identifier, - nextCheckAt: input.scheduledAtIso, - attemptedAttemptCount: input.nextAttemptCount, - notes: input.claimed.monitorNotes ?? null, - serviceName: input.monitor?.serviceName ?? null, - timeoutAt: input.monitor?.timeoutAt ?? null, - maxAttempts: input.monitor?.maxAttempts ?? null, - clearReason: input.clearReason, - recoveryPolicy: input.recoveryPolicy, - source: input.source, - }; - } - - function formatIssueIdentifierLink( - identifier: string | null, - fallback: string, - ) { - if (!identifier) return fallback; - const prefix = identifier.split("-")[0]; - if (!prefix || !/^[A-Z][A-Z0-9]*-\d+$/.test(identifier)) return identifier; - return `[${identifier}](/${prefix}/issues/${identifier})`; - } - - function monitorRecoveryComment(input: { - issue: IssueMonitorDispatchRow; - clearReason: IssueExecutionMonitorClearReason; - recoveryPolicy: IssueExecutionMonitorRecoveryPolicy; - nextAttemptCount: number; - }) { - const label = formatIssueIdentifierLink( - input.issue.identifier, - input.issue.id, - ); - const reason = - input.clearReason === "timeout_exceeded" - ? "its timeout was reached" - : "its maximum attempt count was reached"; - return [ - `Paperclip cleared the scheduled external-service monitor for ${label} because ${reason}.`, - "", - `- Attempt count: ${input.nextAttemptCount}`, - `- Recovery policy: ${input.recoveryPolicy}`, - "", - "Next action: inspect the external service state, record the result on this issue, and restore an explicit execution or waiting path if more work remains.", - ].join("\n"); - } - - async function findOpenIssueMonitorRecoveryIssue( - claimed: IssueMonitorDispatchRow, - ) { - return db - .select() - .from(issues) - .where( - and( - eq(issues.companyId, claimed.companyId), - eq(issues.originKind, RECOVERY_ORIGIN_KINDS.strandedIssueRecovery), - eq(issues.originId, claimed.id), - visibleIssueCondition(), - notInArray(issues.status, ["done", "cancelled"]), - ), - ) - .orderBy(desc(issues.createdAt)) - .limit(1) - .then((rows) => rows[0] ?? null); - } - - async function performIssueMonitorRecovery(input: { - claimed: IssueMonitorDispatchRow; - scheduledAtIso: string; - nextAttemptCount: number; - clearReason: IssueExecutionMonitorClearReason; - recoveryPolicy: IssueExecutionMonitorRecoveryPolicy; - monitor: IssueExecutionMonitorPolicy | null; - actorType: "user" | "agent" | "system"; - actorId: string; - agentId: string | null; - runId: string | null; - activitySource: "manual" | "scheduled"; - }) { - const reviewPathLost = - input.claimed.status === "in_review" && - (await issuesSvc - .listReviewAttention(input.claimed.companyId, [input.claimed]) - .then( - (attention) => attention.get(input.claimed.id)?.state === "stalled", - )); - const reviewPathContext = reviewPathLost - ? { - reviewPathLost: true, - reviewPathConsumedRef: `monitor:${input.claimed.id}:${input.clearReason}:${input.scheduledAtIso}`, - reviewPathInstruction: REVIEW_PATH_RECOVERY_INSTRUCTION, - } - : null; - const details = monitorRecoveryDetails({ - claimed: input.claimed, - scheduledAtIso: input.scheduledAtIso, - nextAttemptCount: input.nextAttemptCount, - clearReason: input.clearReason, - recoveryPolicy: input.recoveryPolicy, - monitor: input.monitor, - source: input.activitySource, - }); - - if (input.recoveryPolicy === "create_recovery_issue") { - let recoveryIssue = await findOpenIssueMonitorRecoveryIssue( - input.claimed, - ); - if (!recoveryIssue) { - recoveryIssue = await issuesSvc.create(input.claimed.companyId, { - title: `Recover external-service monitor for ${input.claimed.identifier ?? input.claimed.title}`, - description: monitorRecoveryComment({ - issue: input.claimed, - clearReason: input.clearReason, - recoveryPolicy: input.recoveryPolicy, - nextAttemptCount: input.nextAttemptCount, - }), - status: "todo", - priority: "high", - parentId: input.claimed.id, - projectId: input.claimed.projectId, - goalId: input.claimed.goalId, - assigneeAgentId: input.claimed.assigneeAgentId, - originKind: RECOVERY_ORIGIN_KINDS.strandedIssueRecovery, - originId: input.claimed.id, - originFingerprint: `issue_monitor:${input.clearReason}`, - billingCode: input.claimed.billingCode, - }); - } - - if (recoveryIssue.assigneeAgentId) { - await enqueueWakeup(recoveryIssue.assigneeAgentId, { - source: "automation", - triggerDetail: "system", - reason: "issue_monitor_recovery_issue", - idempotencyKey: `issue-monitor-recovery-issue:${input.claimed.id}:${input.clearReason}:${input.scheduledAtIso}`, - payload: withRecoveryContext( - { issueId: recoveryIssue.id, sourceIssueId: input.claimed.id }, - "status_only", - ), - requestedByActorType: input.actorType, - requestedByActorId: input.actorId, - contextSnapshot: withRecoveryContext( - { - issueId: recoveryIssue.id, - sourceIssueId: input.claimed.id, - source: "issue.monitor.recovery_issue", - wakeReason: "issue_monitor_recovery_issue", - }, - "status_only", - ), - }); - } - - await logActivity(db, { - companyId: input.claimed.companyId, - actorType: input.actorType, - actorId: input.actorId, - agentId: input.agentId, - runId: input.runId, - action: "issue.monitor_recovery_issue_created", - entityType: "issue", - entityId: input.claimed.id, - details: { - ...details, - recoveryIssueId: recoveryIssue.id, - recoveryIdentifier: recoveryIssue.identifier, - }, - }); - return; - } - - if (input.recoveryPolicy === "escalate_to_board") { - await db.insert(issueComments).values({ - companyId: input.claimed.companyId, - issueId: input.claimed.id, - body: monitorRecoveryComment({ - issue: input.claimed, - clearReason: input.clearReason, - recoveryPolicy: input.recoveryPolicy, - nextAttemptCount: input.nextAttemptCount, - }), - }); - - await logActivity(db, { - companyId: input.claimed.companyId, - actorType: input.actorType, - actorId: input.actorId, - agentId: input.agentId, - runId: input.runId, - action: "issue.monitor_escalated_to_board", - entityType: "issue", - entityId: input.claimed.id, - details, - }); - return; - } - - await enqueueWakeup(input.claimed.assigneeAgentId!, { - source: "automation", - triggerDetail: "system", - reason: "issue_monitor_recovery", - idempotencyKey: `issue-monitor-recovery:${input.claimed.id}:${input.clearReason}:${input.scheduledAtIso}`, - payload: withRecoveryContext( - { - issueId: input.claimed.id, - monitorAttemptCount: input.nextAttemptCount, - monitorNotes: input.claimed.monitorNotes ?? null, - clearReason: input.clearReason, - serviceName: input.monitor?.serviceName ?? null, - timeoutAt: input.monitor?.timeoutAt ?? null, - maxAttempts: input.monitor?.maxAttempts ?? null, - ...(reviewPathContext ?? {}), - }, - "status_only", - ), - requestedByActorType: input.actorType, - requestedByActorId: input.actorId, - contextSnapshot: withRecoveryContext( - { - issueId: input.claimed.id, - source: "issue.monitor.recovery", - wakeReason: "issue_monitor_recovery", - monitorAttemptCount: input.nextAttemptCount, - monitorNotes: input.claimed.monitorNotes ?? null, - clearReason: input.clearReason, - serviceName: input.monitor?.serviceName ?? null, - timeoutAt: input.monitor?.timeoutAt ?? null, - maxAttempts: input.monitor?.maxAttempts ?? null, - ...(reviewPathContext ?? {}), - }, - "status_only", - ), - }); - - await logActivity(db, { - companyId: input.claimed.companyId, - actorType: input.actorType, - actorId: input.actorId, - agentId: input.agentId, - runId: input.runId, - action: "issue.monitor_recovery_wake_queued", - entityType: "issue", - entityId: input.claimed.id, - details, - }); - } - - async function clearIssueMonitorAndRecover(input: { - claimed: IssueMonitorDispatchRow; - policy: ReturnType; - scheduledAtIso: string; - nextAttemptCount: number; - clearReason: IssueExecutionMonitorClearReason; - recoveryPolicy: IssueExecutionMonitorRecoveryPolicy; - monitor: IssueExecutionMonitorPolicy | null; - now: Date; - actorType: "user" | "agent" | "system"; - actorId: string; - agentId: string | null; - runId: string | null; - activitySource: "manual" | "scheduled"; - }) { - const cleared = await db - .update(issues) - .set({ - ...monitorOnlyDispatchPatch(buildIssueMonitorClearedPatch({ - issue: input.claimed, - policy: input.policy, - clearReason: input.clearReason, - clearedAt: input.now, - })), - updatedAt: input.now, - }) - .where(issueMonitorClaimCondition(input.claimed)).returning({ id: issues.id }); - if (cleared.length === 0) return { outcome: "skipped" as const, reason: "monitor_replaced" }; - - await logActivity(db, { - companyId: input.claimed.companyId, - actorType: input.actorType, - actorId: input.actorId, - agentId: input.agentId, - runId: input.runId, - action: "issue.monitor_exhausted", - entityType: "issue", - entityId: input.claimed.id, - details: monitorRecoveryDetails({ - claimed: input.claimed, - scheduledAtIso: input.scheduledAtIso, - nextAttemptCount: input.nextAttemptCount, - clearReason: input.clearReason, - recoveryPolicy: input.recoveryPolicy, - monitor: input.monitor, - source: input.activitySource, - }), - }); - - await performIssueMonitorRecovery({ - claimed: input.claimed, - scheduledAtIso: input.scheduledAtIso, - nextAttemptCount: input.nextAttemptCount, - clearReason: input.clearReason, - recoveryPolicy: input.recoveryPolicy, - monitor: input.monitor, - actorType: input.actorType, - actorId: input.actorId, - agentId: input.agentId, - runId: input.runId, - activitySource: input.activitySource, - }); - - return { outcome: "skipped" as const, reason: input.clearReason }; - } - - function monitorOnlyDispatchPatch>(patch: T) { - // Admission and consumption are separate transactions. Preserve any review - // policy/state changes made between them; only the monitor belongs to us. - return { - ...patch, - executionPolicy: sql`nullif(${issues.executionPolicy} - 'monitor', '{}'::jsonb)`, - executionState: sql`jsonb_set(coalesce(${issues.executionState}, ${JSON.stringify(patch.executionState)}::jsonb), - '{monitor}', ${JSON.stringify(patch.executionState?.monitor ?? null)}::jsonb)`, - }; - } - - function issueMonitorClaimCondition(claimed: IssueMonitorDispatchRow) { - return and(eq(issues.id, claimed.id), eq(issues.companyId, claimed.companyId), - eq(issues.assigneeAgentId, claimed.assigneeAgentId!), isNull(issues.assigneeUserId), - eq(issues.status, claimed.status), eq(issues.monitorNextCheckAt, claimed.monitorNextCheckAt!), - eq(issues.monitorWakeRequestedAt, claimed.monitorWakeRequestedAt!)); - } - - async function dispatchClaimedIssueMonitor( - claimed: IssueMonitorDispatchRow, - input: { - now: Date; - source: "automation" | "on_demand"; - triggerDetail: "manual" | "system"; - wakeReason: string; - actorType: "user" | "agent" | "system"; - actorId: string; - agentId: string | null; - runId: string | null; - clearOnClientError: boolean; - activitySource: "manual" | "scheduled"; - }, - ) { - if (!claimed.assigneeAgentId || !claimed.monitorNextCheckAt) { - throw conflict("Issue monitor is not ready to dispatch"); - } - - const scheduledAtIso = claimed.monitorNextCheckAt.toISOString(); - const nextAttemptCount = (claimed.monitorAttemptCount ?? 0) + 1; - const policy = normalizeIssueExecutionPolicy( - claimed.executionPolicy ?? null, - ); - const monitor = policy?.monitor ?? null; - const clearReason = issueMonitorLimitClearReason({ - monitor, - nextAttemptCount, - now: input.now, - }); - const recoveryPolicy = monitorRecoveryPolicy(monitor); - const monitorMetadata = { - serviceName: monitor?.serviceName ?? null, - timeoutAt: monitor?.timeoutAt ?? null, - maxAttempts: monitor?.maxAttempts ?? null, - recoveryPolicy: monitor?.recoveryPolicy ?? null, - }; - const executionState = - claimed.status === "in_review" - ? parseIssueExecutionState(claimed.executionState) - : null; - const currentParticipant = - executionState?.status === "pending" - ? executionState.currentParticipant - : null; - const reviewParticipantAgentId = - currentParticipant?.type === "agent" ? currentParticipant.agentId : null; - const isProviderQuotaReviewMonitor = - monitor?.serviceName === PROVIDER_QUOTA_MONITOR_SERVICE_NAME && - Boolean(reviewParticipantAgentId); - const targetAgentId = isProviderQuotaReviewMonitor - ? reviewParticipantAgentId - : claimed.assigneeAgentId; - if (!targetAgentId) { - throw conflict("Issue monitor has no agent target"); - } - const wakeReason = isProviderQuotaReviewMonitor - ? EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON - : input.wakeReason; - const reviewRecoveryContext = isProviderQuotaReviewMonitor - ? { - retryReason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON, - currentStageId: executionState?.currentStageId ?? null, - currentStageType: executionState?.currentStageType ?? null, - reviewRecoveryInstruction: - "The previous reviewer run reached provider quota. Resume this execution-review stage now that the quota wait has elapsed.", - } - : {}; - - if (clearReason) { - return clearIssueMonitorAndRecover({ - claimed, - policy, - scheduledAtIso, - nextAttemptCount, - clearReason, - recoveryPolicy, - monitor, - now: input.now, - actorType: input.actorType, - actorId: input.actorId, - agentId: input.agentId, - runId: input.runId, - activitySource: input.activitySource, - }); - } - - try { - if (monitor?.serviceName === PROVIDER_QUOTA_MONITOR_SERVICE_NAME) { - // Normalized monitor projections redact externalRef. Read the claimed - // persisted policy only on this server-owned quota recovery path. - const sourceRunId = readNonEmptyString( - parseObject(parseObject(claimed.executionPolicy).monitor).externalRef, - ); - const sourceRun = - sourceRunId && isUuidLike(sourceRunId) - ? await getRun(sourceRunId, { unsafeFullResultJson: true }) - : null; - if ( - !sourceRun || - sourceRun.companyId !== claimed.companyId || - sourceRun.agentId !== targetAgentId || - sourceRun.contextSnapshot?.issueId !== claimed.id || - !["failed", "timed_out", "interrupted", "cancelled"].includes( - sourceRun.status, - ) - ) { - throw conflict( - "The quota recovery source changed; inspect the current task execution.", - ); - } - if (sourceRun.runtimeMode === "native") { - throw conflict( - "Native execution recovery owns this provider failure; a quota monitor cannot start a replacement.", - ); - } - if (await legacyExecutionNeedsReconciliationWithEvidence(db, sourceRun)) { - await terminalizeLegacyExecution({ - db, - run: sourceRun, - status: sourceRun.status, - }); - } else { - const targetAgent = await getAgent(targetAgentId); - if (!targetAgent) - throw conflict("The quota recovery agent is unavailable."); - const scheduled = await scheduleBoundedRetryForRun( - sourceRun, - targetAgent, - { - now: input.now, - ...(isProviderQuotaReviewMonitor - ? { - retryReason: - EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON, - wakeReason: - EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON, - } - : {}), - }, - ); - if (scheduled.outcome === "not_scheduled") - throw conflict(scheduled.reason); - } - } else { - const wake = await enqueueWakeup(targetAgentId, { - issueStateGuard: { statuses: [claimed.status], assigneeAgentId: claimed.assigneeAgentId, - monitorNextCheckAt: scheduledAtIso, monitorWakeRequestedAt: claimed.monitorWakeRequestedAt!.toISOString() }, - source: input.source, - triggerDetail: input.triggerDetail, - reason: wakeReason, - idempotencyKey: `issue-monitor:${claimed.id}:${scheduledAtIso}`, - payload: { - issueId: claimed.id, - nextCheckAt: scheduledAtIso, - monitorAttemptCount: nextAttemptCount, - monitorNotes: claimed.monitorNotes ?? null, - ...monitorMetadata, - ...reviewRecoveryContext, - source: input.activitySource, - }, - requestedByActorType: input.actorType, - requestedByActorId: input.actorId, - contextSnapshot: { - issueId: claimed.id, - source: isProviderQuotaReviewMonitor - ? "issue.execution_review_recovery" - : "issue.monitor", - wakeReason, - nextCheckAt: scheduledAtIso, - monitorAttemptCount: nextAttemptCount, - monitorNotes: claimed.monitorNotes ?? null, - ...monitorMetadata, - ...reviewRecoveryContext, - manualTrigger: input.activitySource === "manual", - }, - }); - if (!wake) { - await db.update(issues).set({ monitorWakeRequestedAt: null, updatedAt: input.now }) - .where(issueMonitorClaimCondition(claimed)); - return { outcome: "skipped" as const, reason: "monitor_dispatch_deferred" }; - } - } - - const consumed = await db - .update(issues) - .set({ - ...monitorOnlyDispatchPatch(buildIssueMonitorTriggeredPatch({ - issue: claimed, - policy, - triggeredAt: input.now, - })), - updatedAt: new Date(), - }) - .where(issueMonitorClaimCondition(claimed)) - .returning({ id: issues.id }); - if (consumed.length === 0) return { outcome: "skipped" as const, reason: "monitor_replaced" }; - - await logActivity(db, { - companyId: claimed.companyId, - actorType: input.actorType, - actorId: input.actorId, - agentId: input.agentId, - runId: input.runId, - action: "issue.monitor_triggered", - entityType: "issue", - entityId: claimed.id, - details: { - identifier: claimed.identifier, - nextCheckAt: scheduledAtIso, - lastTriggeredAt: input.now.toISOString(), - attemptCount: nextAttemptCount, - notes: claimed.monitorNotes ?? null, - ...monitorMetadata, - source: input.activitySource, - }, - }); - - return { outcome: "triggered" as const }; - } catch (err) { - if (err instanceof HttpError && err.status >= 400 && err.status < 500) { - if (input.clearOnClientError) { - await db - .update(issues) - .set({ - ...monitorOnlyDispatchPatch(buildIssueMonitorClearedPatch({ - issue: claimed, - policy, - clearReason: "dispatch_skipped", - clearedAt: input.now, - })), - updatedAt: new Date(), - }) - .where(issueMonitorClaimCondition(claimed)); - - await logActivity(db, { - companyId: claimed.companyId, - actorType: input.actorType, - actorId: input.actorId, - agentId: input.agentId, - runId: input.runId, - action: "issue.monitor_skipped", - entityType: "issue", - entityId: claimed.id, - details: { - identifier: claimed.identifier, - nextCheckAt: scheduledAtIso, - attemptCount: nextAttemptCount, - notes: claimed.monitorNotes ?? null, - reason: err.message, - source: input.activitySource, - }, - }); - - return { outcome: "skipped" as const, reason: err.message }; - } - - await db - .update(issues) - .set({ - monitorWakeRequestedAt: null, - updatedAt: new Date(), - }) - .where(issueMonitorClaimCondition(claimed)); - } else { - await db - .update(issues) - .set({ - monitorWakeRequestedAt: null, - updatedAt: new Date(), - }) - .where(issueMonitorClaimCondition(claimed)); - } - - throw err; - } - } - - function noActiveNativeMonitorRun() { - return sql`not exists (select 1 from ${heartbeatRuns} monitor_run - where monitor_run.company_id = ${issues.companyId} - and monitor_run.native_issue_id = ${issues.id} - and monitor_run.runtime_mode = 'native' - and monitor_run.status in ('queued', 'running', 'scheduled_retry'))`; - } - - async function triggerIssueMonitor( - issueId: string, - input?: { - now?: Date; - actorType?: "user" | "agent" | "system"; - actorId?: string | null; - agentId?: string | null; - runId?: string | null; - wakeReason?: string; - }, - ) { - const now = input?.now ?? new Date(); - const actorType = input?.actorType ?? "system"; - const actorId = - input?.actorId ?? (actorType === "system" ? "heartbeat_scheduler" : null); - if (!actorId) { - throw conflict("Issue monitor trigger requires an actor"); - } - - const issue = await db - .select(issueMonitorDispatchColumns) - .from(issues) - .where(eq(issues.id, issueId)) - .limit(1) - .then((rows) => rows[0] ?? null); - if (!issue) { - throw notFound("Issue not found"); - } - if (!issue.monitorNextCheckAt) { - throw conflict("Issue has no scheduled monitor"); - } - if (!issue.assigneeAgentId || issue.assigneeUserId) { - throw conflict("Issue monitor requires an agent assignee"); - } - if (!["in_progress", "in_review"].includes(issue.status)) { - throw conflict( - "Issue monitor can only run while the issue is in progress or in review", - ); - } - - const staleClaimThreshold = new Date(now.getTime() - 5 * 60 * 1000); - const claimed = await db.transaction(async (tx) => { - const [updated] = await tx - .update(issues) - .set({ - monitorWakeRequestedAt: now, - updatedAt: now, - }) - .where( - and( - eq(issues.id, issueId), - sql`${issues.monitorNextCheckAt} is not null`, - isNull(issues.assigneeUserId), - sql`${issues.assigneeAgentId} is not null`, - inArray(issues.status, ["in_progress", "in_review"]), - or( - isNull(issues.monitorWakeRequestedAt), - lt(issues.monitorWakeRequestedAt, staleClaimThreshold), - ), - ), - ) - .returning(); - return (updated ?? null) as IssueMonitorDispatchRow | null; - }); - - if (!claimed) { - throw conflict("Issue monitor check is already in progress"); - } - - return dispatchClaimedIssueMonitor(claimed, { - now, - source: "on_demand", - triggerDetail: "manual", - wakeReason: input?.wakeReason ?? "issue_monitor_due", - actorType, - actorId, - agentId: input?.agentId ?? null, - runId: input?.runId ?? null, - clearOnClientError: false, - activitySource: "manual", - }); - } - - async function tickDueIssueMonitors(now = new Date()) { - const staleClaimThreshold = new Date(now.getTime() - 5 * 60 * 1000); - const dueMonitors = await db - .select(issueMonitorDispatchColumns) - .from(issues) - .innerJoin(companies, eq(companies.id, issues.companyId)) - .where( - and( - eq(companies.status, "active"), - noActiveNativeMonitorRun(), - sql`${issues.monitorNextCheckAt} is not null`, - lte(issues.monitorNextCheckAt, now), - isNull(issues.assigneeUserId), - sql`${issues.assigneeAgentId} is not null`, - inArray(issues.status, ["in_progress", "in_review"]), - or( - isNull(issues.monitorWakeRequestedAt), - lt(issues.monitorWakeRequestedAt, staleClaimThreshold), - ), - ), - ) - .orderBy(asc(issues.monitorNextCheckAt), asc(issues.updatedAt)) - .limit(50); - - let triggered = 0; - let skipped = 0; - - for (const due of dueMonitors) { - const claimed = await db.transaction(async (tx) => { - const [updated] = await tx - .update(issues) - .set({ - monitorWakeRequestedAt: now, - updatedAt: now, - }) - .where( - and( - eq(issues.id, due.id), - noActiveNativeMonitorRun(), - sql`${issues.monitorNextCheckAt} is not null`, - lte(issues.monitorNextCheckAt, now), - isNull(issues.assigneeUserId), - sql`${issues.assigneeAgentId} is not null`, - inArray(issues.status, ["in_progress", "in_review"]), - or( - isNull(issues.monitorWakeRequestedAt), - lt(issues.monitorWakeRequestedAt, staleClaimThreshold), - ), - ), - ) - .returning(); - return (updated ?? null) as IssueMonitorDispatchRow | null; - }); - - if (!claimed) continue; - - try { - const result = await dispatchClaimedIssueMonitor(claimed, { - now, - source: "automation", - triggerDetail: "system", - wakeReason: "issue_monitor_due", - actorType: "system", - actorId: "heartbeat_scheduler", - agentId: null, - runId: null, - clearOnClientError: true, - activitySource: "scheduled", - }); - if (result.outcome === "triggered") triggered += 1; - if (result.outcome === "skipped") skipped += 1; - } catch (err) { - logger.error({ err, issueId: claimed.id }, "issue monitor tick failed"); - } - } - - return { - checked: dueMonitors.length, - triggered, - skipped, - }; - } + const { + triggerIssueMonitor, + tickTimers, + recoverActiveSessionGoals, + recoverPendingSessionGoalActions, + } = createHeartbeatScheduling(db, { + issuesSvc, + enqueueWakeup, + getRun, + getAgent, + scheduleBoundedRetryForRun, + getSchedulingSuppression, + getWorktreeExecutionCutoff, + toAgentOrgRow, + groupAgentOrgRowsByCompany, + parseHeartbeatPolicy, + claimDueTimerHeartbeat, + }); const { resolveReusedGitWorkspaceAnchor, @@ -3082,167 +2185,6 @@ export function heartbeatService( } } - async function recoverActiveSessionGoals() { - if ((await getSchedulingSuppression()).suppressed) { - return { scanned: 0, enqueued: 0 }; - } - const sessions = await db - .select({ - id: agentTaskSessions.id, - companyId: agentTaskSessions.companyId, - agentId: agentTaskSessions.agentId, - issueId: agentTaskSessions.taskKey, - revision: agentTaskSessions.goalRevision, - }) - .from(agentTaskSessions) - .innerJoin( - issues, - and( - sql`${issues.id}::text = ${agentTaskSessions.taskKey}`, - eq(issues.companyId, agentTaskSessions.companyId), - eq(issues.assigneeAgentId, agentTaskSessions.agentId), - ), - ) - .where( - and( - eq(agentTaskSessions.goalDesiredState, "active"), - eq(agentTaskSessions.goalStatus, "active"), - notInArray(issues.status, ["done", "cancelled"]), - ), - ); - let enqueued = 0; - for (const session of sessions) { - const run = await enqueueWakeup(session.agentId, { - source: "automation", - triggerDetail: "system", - reason: "goal_control", - payload: { issueId: session.issueId, intent: "goal_recovery" }, - idempotencyKey: `goal_recovery:${session.id}:${session.revision}`, - requestedByActorType: "system", - contextSnapshot: { - issueId: session.issueId, - taskKey: session.issueId, - resumeSessionGoalHeartbeat: true, - skipIssueComment: true, - }, - }); - if (run) enqueued += 1; - } - return { scanned: sessions.length, enqueued }; - } - - async function recoverPendingSessionGoalActions() { - if ((await getSchedulingSuppression()).suppressed) { - return { scanned: 0, enqueued: 0, alreadyQueued: 0, invalid: 0 }; - } - const pending = await db - .select({ - id: agentSessionGoalActions.id, - requestId: agentSessionGoalActions.requestId, - payload: agentSessionGoalActions.payloadJson, - companyId: agentTaskSessions.companyId, - agentId: agentTaskSessions.agentId, - adapterType: agentTaskSessions.adapterType, - issueId: agentTaskSessions.taskKey, - }) - .from(agentSessionGoalActions) - .innerJoin( - agentTaskSessions, - eq(agentTaskSessions.id, agentSessionGoalActions.sessionId), - ) - .innerJoin( - issues, - and( - sql`${issues.id}::text = ${agentTaskSessions.taskKey}`, - eq(issues.companyId, agentTaskSessions.companyId), - eq(issues.assigneeAgentId, agentTaskSessions.agentId), - ), - ) - .where( - inArray(agentSessionGoalActions.status, [ - "pending", - "delivering", - "delivered", - ]), - ) - .orderBy(asc(agentSessionGoalActions.createdAt)); - - let enqueued = 0; - let alreadyQueued = 0; - let invalid = 0; - for (const action of pending) { - const control = parseNativeSessionGoalControl(action.payload); - if (!control || control.requestId !== action.requestId) { - invalid += 1; - await failRunnerGoalAction( - db, - { - companyId: action.companyId, - issueId: action.issueId, - agentId: action.agentId, - adapterType: action.adapterType, - }, - action.requestId, - "session_goal_control_payload_invalid", - ).catch(() => undefined); - continue; - } - - const inFlight = await db - .select({ contextSnapshot: heartbeatRuns.contextSnapshot }) - .from(heartbeatRuns) - .where( - and( - eq(heartbeatRuns.companyId, action.companyId), - eq(heartbeatRuns.agentId, action.agentId), - inArray(heartbeatRuns.status, [ - "queued", - "scheduled_retry", - "running", - ]), - ), - ) - .then((runs) => - runs.some( - (run) => - readNonEmptyString( - parseObject(run.contextSnapshot).goalControlRequestId, - ) === action.requestId, - ), - ); - if (inFlight) { - alreadyQueued += 1; - continue; - } - - const run = await enqueueWakeup(action.agentId, { - source: "automation", - triggerDetail: "system", - reason: "goal_control", - payload: { - issueId: action.issueId, - requestId: action.requestId, - intent: "goal_control_recovery", - }, - idempotencyKey: `goal_control_recovery:${action.id}`, - requestedByActorType: "system", - contextSnapshot: { - issueId: action.issueId, - taskKey: action.issueId, - // Goal controls reconcile the provider session itself and remain - // valid after issue terminalization (for example, clearing a - // completed goal from its retained widget). - resumeIntent: true, - goalControlRequestId: action.requestId, - runnerGoalControl: control, - skipIssueComment: true, - }, - }); - if (run) enqueued += 1; - } - return { scanned: pending.length, enqueued, alreadyQueued, invalid }; - } - async function reconcileStrandedAssignedIssues() { return recovery.reconcileStrandedAssignedIssues({ issueCreatedAtGte: await getWorktreeExecutionCutoff(), @@ -10365,92 +9307,7 @@ export function heartbeatService( buildRunOutputSilence, - tickTimers: async (now = new Date()) => { - if ((await getSchedulingSuppression()).suppressed) { - return { - checked: 0, - enqueued: 0, - skipped: 0, - }; - } - const cutoff = await getWorktreeExecutionCutoff(); - - const allAgents = await db - .select({ ...getTableColumns(agents) }) - .from(agents) - .innerJoin(companies, eq(companies.id, agents.companyId)) - .where(eq(companies.status, "active")); - const agentsByCompany = groupAgentOrgRowsByCompany( - allAgents.map(toAgentOrgRow), - ); - let checked = 0; - let enqueued = 0; - let skipped = 0; - - for (const agent of allAgents) { - const invokability = evaluateAgentInvokability( - toAgentOrgRow(agent), - agentsByCompany.get(agent.companyId) ?? [], - ); - if (!invokability.invokable) continue; - const policy = parseHeartbeatPolicy(agent); - if (!policy.enabled || policy.intervalSec <= 0) continue; - - if (cutoff) { - const eligibleIssue = await db - .select({ id: issues.id }) - .from(issues) - .where( - and( - eq(issues.companyId, agent.companyId), - eq(issues.assigneeAgentId, agent.id), - inArray(issues.status, ["todo", "in_progress"]), - gte(issues.createdAt, cutoff), - ), - ) - .limit(1) - .then((rows) => rows[0] ?? null); - if (!eligibleIssue) continue; - } - - checked += 1; - const baseline = new Date( - agent.lastHeartbeatAt ?? agent.createdAt, - ).getTime(); - const elapsedMs = now.getTime() - baseline; - if (elapsedMs < policy.intervalSec * 1000) continue; - const timerClaim = await claimDueTimerHeartbeat( - agent, - now, - policy.intervalSec, - ); - if (!timerClaim) continue; - - const run = await enqueueWakeup(agent.id, { - source: "timer", - triggerDetail: "system", - reason: "heartbeat_timer", - requestedByActorType: "system", - requestedByActorId: "heartbeat_scheduler", - contextSnapshot: { - source: "scheduler", - reason: "interval_elapsed", - now: now.toISOString(), - timerClaimWasFirstHeartbeat: timerClaim.wasFirstHeartbeat, - }, - }); - if (run) enqueued += 1; - else skipped += 1; - } - - const issueMonitors = await tickDueIssueMonitors(now); - - return { - checked: checked + issueMonitors.checked, - enqueued: enqueued + issueMonitors.triggered, - skipped: skipped + issueMonitors.skipped, - }; - }, + tickTimers, cancelRun: (runId: string, reason?: string, options?: CancelRunOptions) => cancelRunInternal(runId, reason, options), diff --git a/server/src/services/heartbeat/scheduling.test.ts b/server/src/services/heartbeat/scheduling.test.ts new file mode 100644 index 00000000000..8a5659de786 --- /dev/null +++ b/server/src/services/heartbeat/scheduling.test.ts @@ -0,0 +1,284 @@ +import { randomUUID } from "node:crypto"; +import type { Db } from "@paperclipai/db"; +import { + agents, agentSessionGoalActions, agentTaskSessions, companies, createDb, + EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues, +} from "@paperclipai/db"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { conflict } from "../../errors.js"; +import { createHeartbeatRunPreparation } from "./run-preparation.js"; +import { createHeartbeatRunState } from "./run-state.js"; +import { createHeartbeatScheduling, type HeartbeatSchedulingDependencies } from "./scheduling.js"; + +vi.mock("../live-events.js", () => ({ publishLiveEvent: vi.fn() })); + +type Run = typeof heartbeatRuns.$inferSelect; +const now = new Date("2026-10-09T12:00:00.000Z"); +const dueAt = new Date(now.getTime() - 60_000); +const admitted = { id: "admitted-run" } as Run; +const timerPolicy = { + enabled: false, intervalSec: 60, wakeOnDemand: true, maxConcurrentRuns: 1, + skipTimerWhenNoActionableWork: false, maxDailyRuns: null, maxDailyCostCents: null, +}; + +function callbacks(db: Db) { + const state = createHeartbeatRunState(db); + const preparation = createHeartbeatRunPreparation(db); + return { + getAgent: state.getAgent, + getRun: state.getRun, + toAgentOrgRow: preparation.toAgentOrgRow, + groupAgentOrgRowsByCompany: preparation.groupAgentOrgRowsByCompany, + issuesSvc: { + listReviewAttention: vi.fn(async () => new Map()), + create: vi.fn(), + }, + enqueueWakeup: vi.fn(async () => admitted), + scheduleBoundedRetryForRun: vi.fn(), + getSchedulingSuppression: vi.fn(async () => ({ suppressed: false })), + getWorktreeExecutionCutoff: vi.fn(async () => null), + parseHeartbeatPolicy: vi.fn(() => timerPolicy), + claimDueTimerHeartbeat: vi.fn(async () => ({ wasFirstHeartbeat: true })), + } satisfies HeartbeatSchedulingDependencies; +} + +describe("heartbeat scheduling module boundary", () => { + function guardedDatabase() { + const access = vi.fn(() => { throw new Error("Unexpected scheduling database access"); }); + return { db: new Proxy({}, { get: access }) as Db, access }; + } + + it("constructs without querying, admitting work, or starting timers", () => { + const database = guardedDatabase(); + const deps = callbacks(database.db); + createHeartbeatScheduling(database.db, deps); + expect(database.access).not.toHaveBeenCalled(); + for (const callback of Object.values(deps)) if (vi.isMockFunction(callback)) expect(callback).not.toHaveBeenCalled(); + }); + + it.each(["tickTimers", "recoverActiveSessionGoals", "recoverPendingSessionGoalActions"] as const)( + "honors suppression before %s reads or admits work", async (operation) => { + const database = guardedDatabase(); + const deps = callbacks(database.db); + deps.getSchedulingSuppression.mockResolvedValue({ suppressed: true }); + const result = await createHeartbeatScheduling(database.db, deps)[operation](); + expect(Object.values(result).every(value => value === 0)).toBe(true); + expect(database.access).not.toHaveBeenCalled(); + expect(deps.enqueueWakeup).not.toHaveBeenCalled(); + expect(deps.claimDueTimerHeartbeat).not.toHaveBeenCalled(); + }, + ); +}); + +const postgres = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!postgres.supported)("heartbeat scheduling database boundary", () => { + let database: Awaited>; + let db: Db; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-scheduling-module"); + db = createDb(database.connectionString); + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + afterAll(async () => { + await db?.$client.end(); + await database?.cleanup(); + }); + beforeEach(async () => { + await db.execute(sql`truncate table companies restart identity cascade`); + }); + + async function fixture() { + const [company] = await db.insert(companies).values({ name: "Scheduling", issuePrefix: "SCH" }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Scheduled agent", adapterType: "paperclip_runner", createdAt: dueAt }).returning(); + const [issue] = await db.insert(issues).values({ + companyId: company.id, title: "Check deployment", status: "in_progress", assigneeAgentId: agent.id, + monitorNextCheckAt: dueAt, monitorAttemptCount: 0, createdAt: dueAt, + executionPolicy: { mode: "normal", commentRequired: true, stages: [], monitor: { nextCheckAt: dueAt.toISOString() } }, + }).returning(); + const deps = callbacks(db); + return { company, agent, issue, deps, scheduler: createHeartbeatScheduling(db, deps) }; + } + async function readIssue(id: string) { + return db.select().from(issues).where(eq(issues.id, id)).then(rows => rows[0]!); + } + async function session(f: Awaited>, patch: Partial = {}) { + return db.insert(agentTaskSessions).values({ + companyId: f.company.id, agentId: f.agent.id, adapterType: f.agent.adapterType, taskKey: f.issue.id, + goalDesiredState: "active", goalStatus: "active", goalRevision: 7, ...patch, + }).returning().then(rows => rows[0]!); + } + function enableTimer(deps: ReturnType) { + deps.parseHeartbeatPolicy.mockReturnValue({ ...timerPolicy, enabled: true }); + deps.parseHeartbeatPolicy.mockClear(); + } + + it("shares database monitor claims across scheduler instances", async () => { + const f = await fixture(); + let release!: () => void; + let entered!: () => void; + const started = new Promise(resolve => { entered = resolve; }); + const hold = new Promise(resolve => { release = resolve; }); + f.deps.enqueueWakeup.mockImplementationOnce(async () => { entered(); await hold; return admitted; }); + const first = f.scheduler.tickTimers(now); + try { + await started; + const second = createHeartbeatScheduling(db, f.deps); + expect(await second.tickTimers(now)).toEqual({ checked: 0, enqueued: 0, skipped: 0 }); + await expect(second.triggerIssueMonitor(f.issue.id, { now })).rejects.toThrow("already in progress"); + } finally { + release(); + await first; + } + expect(f.deps.enqueueWakeup).toHaveBeenCalledTimes(1); + 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); + f.deps.enqueueWakeup.mockImplementationOnce(async () => { + await db.update(issues).set({ monitorNextCheckAt: replacement, monitorWakeRequestedAt: null }).where(eq(issues.id, f.issue.id)); + return admitted; + }); + expect(await f.scheduler.tickTimers(now)).toEqual({ checked: 1, enqueued: 0, skipped: 1 }); + expect(await readIssue(f.issue.id)).toMatchObject({ monitorNextCheckAt: replacement, monitorAttemptCount: 0 }); + }); + + it("retains a deferred monitor and releases its claim for a later tick", async () => { + const f = await fixture(); + f.deps.enqueueWakeup.mockResolvedValueOnce(null); + expect(await f.scheduler.tickTimers(now)).toEqual({ checked: 1, enqueued: 0, skipped: 1 }); + expect(await readIssue(f.issue.id)).toMatchObject({ monitorNextCheckAt: dueAt, monitorWakeRequestedAt: null, monitorAttemptCount: 0 }); + expect(await f.scheduler.tickTimers(now)).toEqual({ checked: 1, enqueued: 1, skipped: 0 }); + const calls = f.deps.enqueueWakeup.mock.calls; + expect(calls[0]![1]!.idempotencyKey).toBe(calls[1]![1]!.idempotencyKey); + expect(calls[0]![1]!.issueStateGuard).toMatchObject({ + assigneeAgentId: f.agent.id, statuses: ["in_progress"], monitorNextCheckAt: dueAt.toISOString(), monitorWakeRequestedAt: now.toISOString(), + }); + }); + + it("keeps manual error recovery distinct from scheduled client-error handling", async () => { + const f = await fixture(); + f.deps.enqueueWakeup.mockRejectedValue(conflict("Agent is paused")); + await expect(f.scheduler.triggerIssueMonitor(f.issue.id, { now, actorType: "user", actorId: "board-user" })).rejects.toThrow("Agent is paused"); + expect(await readIssue(f.issue.id)).toMatchObject({ monitorNextCheckAt: dueAt, monitorWakeRequestedAt: null }); + expect(f.deps.enqueueWakeup).toHaveBeenLastCalledWith(f.agent.id, expect.objectContaining({ + source: "on_demand", requestedByActorType: "user", requestedByActorId: "board-user", + })); + expect(await f.scheduler.tickTimers(now)).toEqual({ checked: 1, enqueued: 0, skipped: 1 }); + expect(await readIssue(f.issue.id)).toMatchObject({ monitorNextCheckAt: null }); + }); + + it("retains monitors after transient admission failures", async () => { + const f = await fixture(); + f.deps.enqueueWakeup.mockRejectedValueOnce(new Error("Temporary admission failure")); + await f.scheduler.tickTimers(now); + expect(await readIssue(f.issue.id)).toMatchObject({ monitorNextCheckAt: dueAt, monitorWakeRequestedAt: null }); + expect(await f.scheduler.tickTimers(now)).toEqual({ checked: 1, enqueued: 1, skipped: 0 }); + }); + + it("leaves monitors alone while a native run owns the issue", async () => { + const f = await fixture(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, nativeIssueId: f.issue.id, runtimeMode: "native", status: "running", contextSnapshot: { issueId: f.issue.id } }).returning(); + expect(await f.scheduler.tickTimers(now)).toEqual({ checked: 0, enqueued: 0, skipped: 0 }); + expect(f.deps.enqueueWakeup).not.toHaveBeenCalled(); + await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, run!.id)); + expect(await f.scheduler.tickTimers(now)).toEqual({ checked: 1, enqueued: 1, skipped: 0 }); + }); + + it("skips archived companies for both timers and monitors", async () => { + const f = await fixture(); + enableTimer(f.deps); + await db.update(companies).set({ status: "archived" }).where(eq(companies.id, f.company.id)); + expect(await f.scheduler.tickTimers(now)).toEqual({ checked: 0, enqueued: 0, skipped: 0 }); + expect(f.deps.claimDueTimerHeartbeat).not.toHaveBeenCalled(); + expect(f.deps.enqueueWakeup).not.toHaveBeenCalled(); + expect(await readIssue(f.issue.id)).toMatchObject({ monitorNextCheckAt: dueAt }); + }); + + it("combines timer and monitor counts and passes the timer claim receipt to admission", async () => { + const f = await fixture(); + enableTimer(f.deps); + expect(await f.scheduler.tickTimers(now)).toEqual({ checked: 2, enqueued: 2, skipped: 0 }); + expect(f.deps.claimDueTimerHeartbeat).toHaveBeenCalledWith(expect.objectContaining({ id: f.agent.id }), now, 60); + expect(f.deps.enqueueWakeup).toHaveBeenCalledWith(f.agent.id, expect.objectContaining({ + source: "timer", reason: "heartbeat_timer", contextSnapshot: expect.objectContaining({ timerClaimWasFirstHeartbeat: true }), + })); + }); + + it("does not enqueue a timer after losing its due claim", async () => { + const f = await fixture(); + enableTimer(f.deps); + await db.update(issues).set({ monitorNextCheckAt: null }).where(eq(issues.id, f.issue.id)); + f.deps.claimDueTimerHeartbeat.mockResolvedValue(null); + expect(await f.scheduler.tickTimers(now)).toEqual({ checked: 1, enqueued: 0, skipped: 0 }); + expect(f.deps.enqueueWakeup).not.toHaveBeenCalled(); + }); + + it("enforces worktree cutoffs and agent invokability before timer claims", async () => { + const f = await fixture(); + enableTimer(f.deps); + await db.update(issues).set({ monitorNextCheckAt: null }).where(eq(issues.id, f.issue.id)); + f.deps.getWorktreeExecutionCutoff.mockResolvedValue(now); + await f.scheduler.tickTimers(now); + expect(f.deps.claimDueTimerHeartbeat).not.toHaveBeenCalled(); + await db.update(issues).set({ createdAt: now }).where(eq(issues.id, f.issue.id)); + await db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agent.id)); + await f.scheduler.tickTimers(now); + expect(f.deps.claimDueTimerHeartbeat).not.toHaveBeenCalled(); + await db.update(agents).set({ status: "idle" }).where(eq(agents.id, f.agent.id)); + expect(await f.scheduler.tickTimers(now)).toEqual({ checked: 1, enqueued: 1, skipped: 0 }); + }); + + it("recovers active goals with stable revision-based admission keys", async () => { + const f = await fixture(); + const active = await session(f); + expect(await f.scheduler.recoverActiveSessionGoals()).toEqual({ scanned: 1, enqueued: 1 }); + expect(f.deps.enqueueWakeup).toHaveBeenCalledWith(f.agent.id, expect.objectContaining({ + idempotencyKey: `goal_recovery:${active.id}:7`, contextSnapshot: { issueId: f.issue.id, taskKey: f.issue.id, resumeSessionGoalHeartbeat: true, skipIssueComment: true }, + })); + f.deps.enqueueWakeup.mockResolvedValueOnce(null); + expect(await f.scheduler.recoverActiveSessionGoals()).toEqual({ scanned: 1, enqueued: 0 }); + expect(f.deps.enqueueWakeup.mock.calls[1]![1]!.idempotencyKey).toBe(f.deps.enqueueWakeup.mock.calls[0]![1]!.idempotencyKey); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, f.issue.id)); + expect(await f.scheduler.recoverActiveSessionGoals()).toEqual({ scanned: 0, enqueued: 0 }); + }); + + it("excludes goals after reassignment or with a mismatched company binding", async () => { + const f = await fixture(); + const [other] = await db.insert(companies).values({ name: "Other company", issuePrefix: "OTH" }).returning(); + const active = await session(f, { companyId: other!.id }); + expect(await f.scheduler.recoverActiveSessionGoals()).toEqual({ scanned: 0, enqueued: 0 }); + await db.update(agentTaskSessions).set({ companyId: f.company.id }).where(eq(agentTaskSessions.id, active.id)); + await db.update(issues).set({ assigneeAgentId: null }).where(eq(issues.id, f.issue.id)); + expect(await f.scheduler.recoverActiveSessionGoals()).toEqual({ scanned: 0, enqueued: 0 }); + expect(f.deps.enqueueWakeup).not.toHaveBeenCalled(); + }); + + it.each(["queued", "scheduled_retry", "running"])("deduplicates a pending goal action against a %s run", async status => { + const f = await fixture(); + const active = await session(f); + const requestId = randomUUID(); + const [action] = await db.insert(agentSessionGoalActions).values({ companyId: f.company.id, sessionId: active.id, requestId, action: "pause", payloadJson: { requestId, action: "pause" } }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, status, contextSnapshot: { goalControlRequestId: requestId } }).returning(); + expect(await f.scheduler.recoverPendingSessionGoalActions()).toEqual({ scanned: 1, enqueued: 0, alreadyQueued: 1, invalid: 0 }); + expect(f.deps.enqueueWakeup).not.toHaveBeenCalled(); + await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, run!.id)); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, f.issue.id)); + expect(await f.scheduler.recoverPendingSessionGoalActions()).toEqual({ scanned: 1, enqueued: 1, alreadyQueued: 0, invalid: 0 }); + expect(f.deps.enqueueWakeup).toHaveBeenCalledWith(f.agent.id, expect.objectContaining({ + idempotencyKey: `goal_control_recovery:${action!.id}`, contextSnapshot: expect.objectContaining({ resumeIntent: true, goalControlRequestId: requestId, runnerGoalControl: { requestId, action: "pause" } }), + })); + }); + + it("persists invalid goal actions as failures without wake admission", async () => { + const f = await fixture(); + const active = await session(f); + const [action] = await db.insert(agentSessionGoalActions).values({ companyId: f.company.id, sessionId: active.id, requestId: "expected", action: "pause", payloadJson: { requestId: "different", action: "pause" } }).returning(); + expect(await f.scheduler.recoverPendingSessionGoalActions()).toEqual({ scanned: 1, enqueued: 0, alreadyQueued: 0, invalid: 1 }); + expect(f.deps.enqueueWakeup).not.toHaveBeenCalled(); + const failed = await db.select().from(agentSessionGoalActions).where(eq(agentSessionGoalActions.id, action!.id)); + expect(failed[0]).toMatchObject({ status: "failed", error: "session_goal_control_payload_invalid" }); + }); +}); diff --git a/server/src/services/heartbeat/scheduling.ts b/server/src/services/heartbeat/scheduling.ts new file mode 100644 index 00000000000..a0714e2d40f --- /dev/null +++ b/server/src/services/heartbeat/scheduling.ts @@ -0,0 +1,1248 @@ +import { + issues, + issueComments, + heartbeatRuns, + companies, + agentTaskSessions, + agentSessionGoalActions, + agents, +} from "@paperclipai/db"; +import { + type IssueExecutionMonitorPolicy, + type IssueExecutionMonitorClearReason, + type IssueExecutionMonitorRecoveryPolicy, + PROVIDER_QUOTA_MONITOR_SERVICE_NAME, + isUuidLike, +} from "@paperclipai/shared"; +import { + and, + eq, + notInArray, + desc, + sql, + isNull, + inArray, + or, + lt, + lte, + asc, + getTableColumns, + gte, +} from "drizzle-orm"; +import { RECOVERY_ORIGIN_KINDS } from "../recovery/index.js"; +import { visibleIssueCondition } from "../issue-visibility.js"; +import { REVIEW_PATH_RECOVERY_INSTRUCTION } from "../recovery/review-path-recovery.js"; +import { withRecoveryContext } from "../recovery/status-only-context.js"; +import { logActivity } from "../activity-log.js"; +import { + normalizeIssueExecutionPolicy, + buildIssueMonitorClearedPatch, + parseIssueExecutionState, + buildIssueMonitorTriggeredPatch, +} from "../issue-execution-policy.js"; +import { conflict, HttpError, notFound } from "../../errors.js"; +import { EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON } from "./run-state.js"; +import { parseObject } from "../../adapters/utils.js"; +import { + legacyExecutionNeedsReconciliationWithEvidence, + terminalizeLegacyExecution, +} from "../legacy-execution-recovery.js"; +import { logger } from "../../middleware/logger.js"; +import { failRunnerGoalAction } from "../runner-goals.js"; +import { evaluateAgentInvokability } from "../agent-invokability.js"; +import type { Db } from "@paperclipai/db"; +import type { issueService } from "../issues.js"; +import type { createHeartbeatQueue } from "./queue.js"; +import type { createHeartbeatRetries, HeartbeatRetryDependencies } from "./retries.js"; +import type { createHeartbeatRunPreparation } from "./run-preparation.js"; +import type { NativeSessionGoalControl } from "../../vendor/paperclip-runner/index.js"; + +type HeartbeatQueue = ReturnType; +type RunPreparation = ReturnType; + +export interface HeartbeatSchedulingDependencies + extends Pick, + Pick, + Pick { + issuesSvc: Pick, "listReviewAttention" | "create">; + scheduleBoundedRetryForRun: ReturnType["scheduleBoundedRetryForRun"]; + getSchedulingSuppression: () => Promise<{ suppressed: boolean }>; +} + +function readNonEmptyString(value: unknown): string | null { + return typeof value === "string" && value.trim().length > 0 ? value : null; +} + +export function parseNativeSessionGoalControl( + value: unknown, +): NativeSessionGoalControl | null { + const candidate = parseObject(value); + const requestId = readNonEmptyString(candidate.requestId); + const action = readNonEmptyString(candidate.action); + if ( + !requestId || + !action || + !["create", "edit", "replace", "pause", "resume", "clear"].includes(action) + ) + return null; + const objective = readNonEmptyString(candidate.objective); + const tokenBudget = + candidate.tokenBudget === null + ? null + : typeof candidate.tokenBudget === "number" && + Number.isSafeInteger(candidate.tokenBudget) && + candidate.tokenBudget > 0 + ? candidate.tokenBudget + : undefined; + return { + requestId, + action: action as NativeSessionGoalControl["action"], + ...(objective ? { objective } : {}), + ...(tokenBudget !== undefined ? { tokenBudget } : {}), + }; +} + +export function formatIssueIdentifierLink( + identifier: string | null, + fallback: string, +) { + if (!identifier) return fallback; + const prefix = identifier.split("-")[0]; + if (!prefix || !/^[A-Z][A-Z0-9]*-\d+$/.test(identifier)) return identifier; + return `[${identifier}](/${prefix}/issues/${identifier})`; +} + +const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON = + "execution_review_participant_recovery"; + +// Construction only binds dependencies. Scheduler calls retain the existing queue admission gates. +export function createHeartbeatScheduling( + db: Db, + dependencies: HeartbeatSchedulingDependencies, +) { + const { + issuesSvc, + enqueueWakeup, + getRun, + getAgent, + scheduleBoundedRetryForRun, + getSchedulingSuppression, + getWorktreeExecutionCutoff, + toAgentOrgRow, + groupAgentOrgRowsByCompany, + parseHeartbeatPolicy, + claimDueTimerHeartbeat, + } = dependencies; + + const issueMonitorDispatchColumns = { + id: issues.id, + companyId: issues.companyId, + projectId: issues.projectId, + goalId: issues.goalId, + identifier: issues.identifier, + title: issues.title, + status: issues.status, + priority: issues.priority, + assigneeAgentId: issues.assigneeAgentId, + assigneeUserId: issues.assigneeUserId, + billingCode: issues.billingCode, + executionPolicy: issues.executionPolicy, + executionState: issues.executionState, + monitorNextCheckAt: issues.monitorNextCheckAt, + monitorWakeRequestedAt: issues.monitorWakeRequestedAt, + monitorLastTriggeredAt: issues.monitorLastTriggeredAt, + monitorAttemptCount: issues.monitorAttemptCount, + monitorNotes: issues.monitorNotes, + monitorScheduledBy: issues.monitorScheduledBy, + }; + + interface IssueMonitorDispatchRow { + id: string; + companyId: string; + projectId: string | null; + goalId: string | null; + identifier: string | null; + title: string; + status: string; + priority: string; + assigneeAgentId: string | null; + assigneeUserId: string | null; + billingCode: string | null; + executionPolicy: Record | null; + executionState: Record | null; + monitorNextCheckAt: Date | null; + monitorWakeRequestedAt: Date | null; + monitorLastTriggeredAt: Date | null; + monitorAttemptCount: number | null; + monitorNotes: string | null; + monitorScheduledBy: string | null; + } + + function parseMonitorDate(value: string | null | undefined) { + if (!value) return null; + const date = new Date(value); + return Number.isNaN(date.getTime()) ? null : date; + } + + function issueMonitorLimitClearReason(input: { + monitor: IssueExecutionMonitorPolicy | null; + nextAttemptCount: number; + now: Date; + }): IssueExecutionMonitorClearReason | null { + const timeoutAt = parseMonitorDate(input.monitor?.timeoutAt ?? null); + if (timeoutAt && input.now.getTime() >= timeoutAt.getTime()) { + return "timeout_exceeded"; + } + const maxAttempts = input.monitor?.maxAttempts ?? null; + if (maxAttempts !== null && input.nextAttemptCount > maxAttempts) { + return "max_attempts_exhausted"; + } + return null; + } + + function monitorRecoveryPolicy( + monitor: IssueExecutionMonitorPolicy | null, + ): IssueExecutionMonitorRecoveryPolicy { + return monitor?.recoveryPolicy ?? "wake_owner"; + } + + function monitorRecoveryDetails(input: { + claimed: IssueMonitorDispatchRow; + scheduledAtIso: string; + nextAttemptCount: number; + clearReason: IssueExecutionMonitorClearReason; + recoveryPolicy: IssueExecutionMonitorRecoveryPolicy; + monitor: IssueExecutionMonitorPolicy | null; + source: "manual" | "scheduled"; + }) { + return { + identifier: input.claimed.identifier, + nextCheckAt: input.scheduledAtIso, + attemptedAttemptCount: input.nextAttemptCount, + notes: input.claimed.monitorNotes ?? null, + serviceName: input.monitor?.serviceName ?? null, + timeoutAt: input.monitor?.timeoutAt ?? null, + maxAttempts: input.monitor?.maxAttempts ?? null, + clearReason: input.clearReason, + recoveryPolicy: input.recoveryPolicy, + source: input.source, + }; + } + + + function monitorRecoveryComment(input: { + issue: IssueMonitorDispatchRow; + clearReason: IssueExecutionMonitorClearReason; + recoveryPolicy: IssueExecutionMonitorRecoveryPolicy; + nextAttemptCount: number; + }) { + const label = formatIssueIdentifierLink( + input.issue.identifier, + input.issue.id, + ); + const reason = + input.clearReason === "timeout_exceeded" + ? "its timeout was reached" + : "its maximum attempt count was reached"; + return [ + `Paperclip cleared the scheduled external-service monitor for ${label} because ${reason}.`, + "", + `- Attempt count: ${input.nextAttemptCount}`, + `- Recovery policy: ${input.recoveryPolicy}`, + "", + "Next action: inspect the external service state, record the result on this issue, and restore an explicit execution or waiting path if more work remains.", + ].join("\n"); + } + + async function findOpenIssueMonitorRecoveryIssue( + claimed: IssueMonitorDispatchRow, + ) { + return db + .select() + .from(issues) + .where( + and( + eq(issues.companyId, claimed.companyId), + eq(issues.originKind, RECOVERY_ORIGIN_KINDS.strandedIssueRecovery), + eq(issues.originId, claimed.id), + visibleIssueCondition(), + notInArray(issues.status, ["done", "cancelled"]), + ), + ) + .orderBy(desc(issues.createdAt)) + .limit(1) + .then((rows) => rows[0] ?? null); + } + + async function performIssueMonitorRecovery(input: { + claimed: IssueMonitorDispatchRow; + scheduledAtIso: string; + nextAttemptCount: number; + clearReason: IssueExecutionMonitorClearReason; + recoveryPolicy: IssueExecutionMonitorRecoveryPolicy; + monitor: IssueExecutionMonitorPolicy | null; + actorType: "user" | "agent" | "system"; + actorId: string; + agentId: string | null; + runId: string | null; + activitySource: "manual" | "scheduled"; + }) { + const reviewPathLost = + input.claimed.status === "in_review" && + (await issuesSvc + .listReviewAttention(input.claimed.companyId, [input.claimed]) + .then( + (attention) => attention.get(input.claimed.id)?.state === "stalled", + )); + const reviewPathContext = reviewPathLost + ? { + reviewPathLost: true, + reviewPathConsumedRef: `monitor:${input.claimed.id}:${input.clearReason}:${input.scheduledAtIso}`, + reviewPathInstruction: REVIEW_PATH_RECOVERY_INSTRUCTION, + } + : null; + const details = monitorRecoveryDetails({ + claimed: input.claimed, + scheduledAtIso: input.scheduledAtIso, + nextAttemptCount: input.nextAttemptCount, + clearReason: input.clearReason, + recoveryPolicy: input.recoveryPolicy, + monitor: input.monitor, + source: input.activitySource, + }); + + if (input.recoveryPolicy === "create_recovery_issue") { + let recoveryIssue = await findOpenIssueMonitorRecoveryIssue( + input.claimed, + ); + if (!recoveryIssue) { + recoveryIssue = await issuesSvc.create(input.claimed.companyId, { + title: `Recover external-service monitor for ${input.claimed.identifier ?? input.claimed.title}`, + description: monitorRecoveryComment({ + issue: input.claimed, + clearReason: input.clearReason, + recoveryPolicy: input.recoveryPolicy, + nextAttemptCount: input.nextAttemptCount, + }), + status: "todo", + priority: "high", + parentId: input.claimed.id, + projectId: input.claimed.projectId, + goalId: input.claimed.goalId, + assigneeAgentId: input.claimed.assigneeAgentId, + originKind: RECOVERY_ORIGIN_KINDS.strandedIssueRecovery, + originId: input.claimed.id, + originFingerprint: `issue_monitor:${input.clearReason}`, + billingCode: input.claimed.billingCode, + }); + } + + if (recoveryIssue.assigneeAgentId) { + await enqueueWakeup(recoveryIssue.assigneeAgentId, { + source: "automation", + triggerDetail: "system", + reason: "issue_monitor_recovery_issue", + idempotencyKey: `issue-monitor-recovery-issue:${input.claimed.id}:${input.clearReason}:${input.scheduledAtIso}`, + payload: withRecoveryContext( + { issueId: recoveryIssue.id, sourceIssueId: input.claimed.id }, + "status_only", + ), + requestedByActorType: input.actorType, + requestedByActorId: input.actorId, + contextSnapshot: withRecoveryContext( + { + issueId: recoveryIssue.id, + sourceIssueId: input.claimed.id, + source: "issue.monitor.recovery_issue", + wakeReason: "issue_monitor_recovery_issue", + }, + "status_only", + ), + }); + } + + await logActivity(db, { + companyId: input.claimed.companyId, + actorType: input.actorType, + actorId: input.actorId, + agentId: input.agentId, + runId: input.runId, + action: "issue.monitor_recovery_issue_created", + entityType: "issue", + entityId: input.claimed.id, + details: { + ...details, + recoveryIssueId: recoveryIssue.id, + recoveryIdentifier: recoveryIssue.identifier, + }, + }); + return; + } + + if (input.recoveryPolicy === "escalate_to_board") { + await db.insert(issueComments).values({ + companyId: input.claimed.companyId, + issueId: input.claimed.id, + body: monitorRecoveryComment({ + issue: input.claimed, + clearReason: input.clearReason, + recoveryPolicy: input.recoveryPolicy, + nextAttemptCount: input.nextAttemptCount, + }), + }); + + await logActivity(db, { + companyId: input.claimed.companyId, + actorType: input.actorType, + actorId: input.actorId, + agentId: input.agentId, + runId: input.runId, + action: "issue.monitor_escalated_to_board", + entityType: "issue", + entityId: input.claimed.id, + details, + }); + return; + } + + await enqueueWakeup(input.claimed.assigneeAgentId!, { + source: "automation", + triggerDetail: "system", + reason: "issue_monitor_recovery", + idempotencyKey: `issue-monitor-recovery:${input.claimed.id}:${input.clearReason}:${input.scheduledAtIso}`, + payload: withRecoveryContext( + { + issueId: input.claimed.id, + monitorAttemptCount: input.nextAttemptCount, + monitorNotes: input.claimed.monitorNotes ?? null, + clearReason: input.clearReason, + serviceName: input.monitor?.serviceName ?? null, + timeoutAt: input.monitor?.timeoutAt ?? null, + maxAttempts: input.monitor?.maxAttempts ?? null, + ...(reviewPathContext ?? {}), + }, + "status_only", + ), + requestedByActorType: input.actorType, + requestedByActorId: input.actorId, + contextSnapshot: withRecoveryContext( + { + issueId: input.claimed.id, + source: "issue.monitor.recovery", + wakeReason: "issue_monitor_recovery", + monitorAttemptCount: input.nextAttemptCount, + monitorNotes: input.claimed.monitorNotes ?? null, + clearReason: input.clearReason, + serviceName: input.monitor?.serviceName ?? null, + timeoutAt: input.monitor?.timeoutAt ?? null, + maxAttempts: input.monitor?.maxAttempts ?? null, + ...(reviewPathContext ?? {}), + }, + "status_only", + ), + }); + + await logActivity(db, { + companyId: input.claimed.companyId, + actorType: input.actorType, + actorId: input.actorId, + agentId: input.agentId, + runId: input.runId, + action: "issue.monitor_recovery_wake_queued", + entityType: "issue", + entityId: input.claimed.id, + details, + }); + } + + async function clearIssueMonitorAndRecover(input: { + claimed: IssueMonitorDispatchRow; + policy: ReturnType; + scheduledAtIso: string; + nextAttemptCount: number; + clearReason: IssueExecutionMonitorClearReason; + recoveryPolicy: IssueExecutionMonitorRecoveryPolicy; + monitor: IssueExecutionMonitorPolicy | null; + now: Date; + actorType: "user" | "agent" | "system"; + actorId: string; + agentId: string | null; + runId: string | null; + activitySource: "manual" | "scheduled"; + }) { + const cleared = await db + .update(issues) + .set({ + ...monitorOnlyDispatchPatch(buildIssueMonitorClearedPatch({ + issue: input.claimed, + policy: input.policy, + clearReason: input.clearReason, + clearedAt: input.now, + })), + updatedAt: input.now, + }) + .where(issueMonitorClaimCondition(input.claimed)).returning({ id: issues.id }); + if (cleared.length === 0) return { outcome: "skipped" as const, reason: "monitor_replaced" }; + + await logActivity(db, { + companyId: input.claimed.companyId, + actorType: input.actorType, + actorId: input.actorId, + agentId: input.agentId, + runId: input.runId, + action: "issue.monitor_exhausted", + entityType: "issue", + entityId: input.claimed.id, + details: monitorRecoveryDetails({ + claimed: input.claimed, + scheduledAtIso: input.scheduledAtIso, + nextAttemptCount: input.nextAttemptCount, + clearReason: input.clearReason, + recoveryPolicy: input.recoveryPolicy, + monitor: input.monitor, + source: input.activitySource, + }), + }); + + await performIssueMonitorRecovery({ + claimed: input.claimed, + scheduledAtIso: input.scheduledAtIso, + nextAttemptCount: input.nextAttemptCount, + clearReason: input.clearReason, + recoveryPolicy: input.recoveryPolicy, + monitor: input.monitor, + actorType: input.actorType, + actorId: input.actorId, + agentId: input.agentId, + runId: input.runId, + activitySource: input.activitySource, + }); + + return { outcome: "skipped" as const, reason: input.clearReason }; + } + + function monitorOnlyDispatchPatch>(patch: T) { + // Admission and consumption are separate transactions. Preserve any review + // policy/state changes made between them; only the monitor belongs to us. + return { + ...patch, + executionPolicy: sql`nullif(${issues.executionPolicy} - 'monitor', '{}'::jsonb)`, + executionState: sql`jsonb_set(coalesce(${issues.executionState}, ${JSON.stringify(patch.executionState)}::jsonb), + '{monitor}', ${JSON.stringify(patch.executionState?.monitor ?? null)}::jsonb)`, + }; + } + + function issueMonitorClaimCondition(claimed: IssueMonitorDispatchRow) { + return and(eq(issues.id, claimed.id), eq(issues.companyId, claimed.companyId), + eq(issues.assigneeAgentId, claimed.assigneeAgentId!), isNull(issues.assigneeUserId), + eq(issues.status, claimed.status), eq(issues.monitorNextCheckAt, claimed.monitorNextCheckAt!), + eq(issues.monitorWakeRequestedAt, claimed.monitorWakeRequestedAt!)); + } + + async function dispatchClaimedIssueMonitor( + claimed: IssueMonitorDispatchRow, + input: { + now: Date; + source: "automation" | "on_demand"; + triggerDetail: "manual" | "system"; + wakeReason: string; + actorType: "user" | "agent" | "system"; + actorId: string; + agentId: string | null; + runId: string | null; + clearOnClientError: boolean; + activitySource: "manual" | "scheduled"; + }, + ) { + if (!claimed.assigneeAgentId || !claimed.monitorNextCheckAt) { + throw conflict("Issue monitor is not ready to dispatch"); + } + + const scheduledAtIso = claimed.monitorNextCheckAt.toISOString(); + const nextAttemptCount = (claimed.monitorAttemptCount ?? 0) + 1; + const policy = normalizeIssueExecutionPolicy( + claimed.executionPolicy ?? null, + ); + const monitor = policy?.monitor ?? null; + const clearReason = issueMonitorLimitClearReason({ + monitor, + nextAttemptCount, + now: input.now, + }); + const recoveryPolicy = monitorRecoveryPolicy(monitor); + const monitorMetadata = { + serviceName: monitor?.serviceName ?? null, + timeoutAt: monitor?.timeoutAt ?? null, + maxAttempts: monitor?.maxAttempts ?? null, + recoveryPolicy: monitor?.recoveryPolicy ?? null, + }; + const executionState = + claimed.status === "in_review" + ? parseIssueExecutionState(claimed.executionState) + : null; + const currentParticipant = + executionState?.status === "pending" + ? executionState.currentParticipant + : null; + const reviewParticipantAgentId = + currentParticipant?.type === "agent" ? currentParticipant.agentId : null; + const isProviderQuotaReviewMonitor = + monitor?.serviceName === PROVIDER_QUOTA_MONITOR_SERVICE_NAME && + Boolean(reviewParticipantAgentId); + const targetAgentId = isProviderQuotaReviewMonitor + ? reviewParticipantAgentId + : claimed.assigneeAgentId; + if (!targetAgentId) { + throw conflict("Issue monitor has no agent target"); + } + const wakeReason = isProviderQuotaReviewMonitor + ? EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON + : input.wakeReason; + const reviewRecoveryContext = isProviderQuotaReviewMonitor + ? { + retryReason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON, + currentStageId: executionState?.currentStageId ?? null, + currentStageType: executionState?.currentStageType ?? null, + reviewRecoveryInstruction: + "The previous reviewer run reached provider quota. Resume this execution-review stage now that the quota wait has elapsed.", + } + : {}; + + if (clearReason) { + return clearIssueMonitorAndRecover({ + claimed, + policy, + scheduledAtIso, + nextAttemptCount, + clearReason, + recoveryPolicy, + monitor, + now: input.now, + actorType: input.actorType, + actorId: input.actorId, + agentId: input.agentId, + runId: input.runId, + activitySource: input.activitySource, + }); + } + + try { + if (monitor?.serviceName === PROVIDER_QUOTA_MONITOR_SERVICE_NAME) { + // Normalized monitor projections redact externalRef. Read the claimed + // persisted policy only on this server-owned quota recovery path. + const sourceRunId = readNonEmptyString( + parseObject(parseObject(claimed.executionPolicy).monitor).externalRef, + ); + const sourceRun = + sourceRunId && isUuidLike(sourceRunId) + ? await getRun(sourceRunId, { unsafeFullResultJson: true }) + : null; + if ( + !sourceRun || + sourceRun.companyId !== claimed.companyId || + sourceRun.agentId !== targetAgentId || + sourceRun.contextSnapshot?.issueId !== claimed.id || + !["failed", "timed_out", "interrupted", "cancelled"].includes( + sourceRun.status, + ) + ) { + throw conflict( + "The quota recovery source changed; inspect the current task execution.", + ); + } + if (sourceRun.runtimeMode === "native") { + throw conflict( + "Native execution recovery owns this provider failure; a quota monitor cannot start a replacement.", + ); + } + if (await legacyExecutionNeedsReconciliationWithEvidence(db, sourceRun)) { + await terminalizeLegacyExecution({ + db, + run: sourceRun, + status: sourceRun.status, + }); + } else { + const targetAgent = await getAgent(targetAgentId); + if (!targetAgent) + throw conflict("The quota recovery agent is unavailable."); + const scheduled = await scheduleBoundedRetryForRun( + sourceRun, + targetAgent, + { + now: input.now, + ...(isProviderQuotaReviewMonitor + ? { + retryReason: + EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON, + wakeReason: + EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON, + } + : {}), + }, + ); + if (scheduled.outcome === "not_scheduled") + throw conflict(scheduled.reason); + } + } else { + const wake = await enqueueWakeup(targetAgentId, { + issueStateGuard: { statuses: [claimed.status], assigneeAgentId: claimed.assigneeAgentId, + monitorNextCheckAt: scheduledAtIso, monitorWakeRequestedAt: claimed.monitorWakeRequestedAt!.toISOString() }, + source: input.source, + triggerDetail: input.triggerDetail, + reason: wakeReason, + idempotencyKey: `issue-monitor:${claimed.id}:${scheduledAtIso}`, + payload: { + issueId: claimed.id, + nextCheckAt: scheduledAtIso, + monitorAttemptCount: nextAttemptCount, + monitorNotes: claimed.monitorNotes ?? null, + ...monitorMetadata, + ...reviewRecoveryContext, + source: input.activitySource, + }, + requestedByActorType: input.actorType, + requestedByActorId: input.actorId, + contextSnapshot: { + issueId: claimed.id, + source: isProviderQuotaReviewMonitor + ? "issue.execution_review_recovery" + : "issue.monitor", + wakeReason, + nextCheckAt: scheduledAtIso, + monitorAttemptCount: nextAttemptCount, + monitorNotes: claimed.monitorNotes ?? null, + ...monitorMetadata, + ...reviewRecoveryContext, + manualTrigger: input.activitySource === "manual", + }, + }); + if (!wake) { + await db.update(issues).set({ monitorWakeRequestedAt: null, updatedAt: input.now }) + .where(issueMonitorClaimCondition(claimed)); + return { outcome: "skipped" as const, reason: "monitor_dispatch_deferred" }; + } + } + + const consumed = await db + .update(issues) + .set({ + ...monitorOnlyDispatchPatch(buildIssueMonitorTriggeredPatch({ + issue: claimed, + policy, + triggeredAt: input.now, + })), + updatedAt: new Date(), + }) + .where(issueMonitorClaimCondition(claimed)) + .returning({ id: issues.id }); + if (consumed.length === 0) return { outcome: "skipped" as const, reason: "monitor_replaced" }; + + await logActivity(db, { + companyId: claimed.companyId, + actorType: input.actorType, + actorId: input.actorId, + agentId: input.agentId, + runId: input.runId, + action: "issue.monitor_triggered", + entityType: "issue", + entityId: claimed.id, + details: { + identifier: claimed.identifier, + nextCheckAt: scheduledAtIso, + lastTriggeredAt: input.now.toISOString(), + attemptCount: nextAttemptCount, + notes: claimed.monitorNotes ?? null, + ...monitorMetadata, + source: input.activitySource, + }, + }); + + return { outcome: "triggered" as const }; + } catch (err) { + if (err instanceof HttpError && err.status >= 400 && err.status < 500) { + if (input.clearOnClientError) { + await db + .update(issues) + .set({ + ...monitorOnlyDispatchPatch(buildIssueMonitorClearedPatch({ + issue: claimed, + policy, + clearReason: "dispatch_skipped", + clearedAt: input.now, + })), + updatedAt: new Date(), + }) + .where(issueMonitorClaimCondition(claimed)); + + await logActivity(db, { + companyId: claimed.companyId, + actorType: input.actorType, + actorId: input.actorId, + agentId: input.agentId, + runId: input.runId, + action: "issue.monitor_skipped", + entityType: "issue", + entityId: claimed.id, + details: { + identifier: claimed.identifier, + nextCheckAt: scheduledAtIso, + attemptCount: nextAttemptCount, + notes: claimed.monitorNotes ?? null, + reason: err.message, + source: input.activitySource, + }, + }); + + return { outcome: "skipped" as const, reason: err.message }; + } + + await db + .update(issues) + .set({ + monitorWakeRequestedAt: null, + updatedAt: new Date(), + }) + .where(issueMonitorClaimCondition(claimed)); + } else { + await db + .update(issues) + .set({ + monitorWakeRequestedAt: null, + updatedAt: new Date(), + }) + .where(issueMonitorClaimCondition(claimed)); + } + + throw err; + } + } + + function noActiveNativeMonitorRun() { + return sql`not exists (select 1 from ${heartbeatRuns} monitor_run + where monitor_run.company_id = ${issues.companyId} + and monitor_run.native_issue_id = ${issues.id} + and monitor_run.runtime_mode = 'native' + and monitor_run.status in ('queued', 'running', 'scheduled_retry'))`; + } + + async function triggerIssueMonitor( + issueId: string, + input?: { + now?: Date; + actorType?: "user" | "agent" | "system"; + actorId?: string | null; + agentId?: string | null; + runId?: string | null; + wakeReason?: string; + }, + ) { + const now = input?.now ?? new Date(); + const actorType = input?.actorType ?? "system"; + const actorId = + input?.actorId ?? (actorType === "system" ? "heartbeat_scheduler" : null); + if (!actorId) { + throw conflict("Issue monitor trigger requires an actor"); + } + + const issue = await db + .select(issueMonitorDispatchColumns) + .from(issues) + .where(eq(issues.id, issueId)) + .limit(1) + .then((rows) => rows[0] ?? null); + if (!issue) { + throw notFound("Issue not found"); + } + if (!issue.monitorNextCheckAt) { + throw conflict("Issue has no scheduled monitor"); + } + if (!issue.assigneeAgentId || issue.assigneeUserId) { + throw conflict("Issue monitor requires an agent assignee"); + } + if (!["in_progress", "in_review"].includes(issue.status)) { + throw conflict( + "Issue monitor can only run while the issue is in progress or in review", + ); + } + + const staleClaimThreshold = new Date(now.getTime() - 5 * 60 * 1000); + const claimed = await db.transaction(async (tx) => { + const [updated] = await tx + .update(issues) + .set({ + monitorWakeRequestedAt: now, + updatedAt: now, + }) + .where( + and( + eq(issues.id, issueId), + sql`${issues.monitorNextCheckAt} is not null`, + isNull(issues.assigneeUserId), + sql`${issues.assigneeAgentId} is not null`, + inArray(issues.status, ["in_progress", "in_review"]), + or( + isNull(issues.monitorWakeRequestedAt), + lt(issues.monitorWakeRequestedAt, staleClaimThreshold), + ), + ), + ) + .returning(); + return (updated ?? null) as IssueMonitorDispatchRow | null; + }); + + if (!claimed) { + throw conflict("Issue monitor check is already in progress"); + } + + return dispatchClaimedIssueMonitor(claimed, { + now, + source: "on_demand", + triggerDetail: "manual", + wakeReason: input?.wakeReason ?? "issue_monitor_due", + actorType, + actorId, + agentId: input?.agentId ?? null, + runId: input?.runId ?? null, + clearOnClientError: false, + activitySource: "manual", + }); + } + + async function tickDueIssueMonitors(now = new Date()) { + const staleClaimThreshold = new Date(now.getTime() - 5 * 60 * 1000); + const dueMonitors = await db + .select(issueMonitorDispatchColumns) + .from(issues) + .innerJoin(companies, eq(companies.id, issues.companyId)) + .where( + and( + eq(companies.status, "active"), + noActiveNativeMonitorRun(), + sql`${issues.monitorNextCheckAt} is not null`, + lte(issues.monitorNextCheckAt, now), + isNull(issues.assigneeUserId), + sql`${issues.assigneeAgentId} is not null`, + inArray(issues.status, ["in_progress", "in_review"]), + or( + isNull(issues.monitorWakeRequestedAt), + lt(issues.monitorWakeRequestedAt, staleClaimThreshold), + ), + ), + ) + .orderBy(asc(issues.monitorNextCheckAt), asc(issues.updatedAt)) + .limit(50); + + let triggered = 0; + let skipped = 0; + + for (const due of dueMonitors) { + const claimed = await db.transaction(async (tx) => { + const [updated] = await tx + .update(issues) + .set({ + monitorWakeRequestedAt: now, + updatedAt: now, + }) + .where( + and( + eq(issues.id, due.id), + noActiveNativeMonitorRun(), + sql`${issues.monitorNextCheckAt} is not null`, + lte(issues.monitorNextCheckAt, now), + isNull(issues.assigneeUserId), + sql`${issues.assigneeAgentId} is not null`, + inArray(issues.status, ["in_progress", "in_review"]), + or( + isNull(issues.monitorWakeRequestedAt), + lt(issues.monitorWakeRequestedAt, staleClaimThreshold), + ), + ), + ) + .returning(); + return (updated ?? null) as IssueMonitorDispatchRow | null; + }); + + if (!claimed) continue; + + try { + const result = await dispatchClaimedIssueMonitor(claimed, { + now, + source: "automation", + triggerDetail: "system", + wakeReason: "issue_monitor_due", + actorType: "system", + actorId: "heartbeat_scheduler", + agentId: null, + runId: null, + clearOnClientError: true, + activitySource: "scheduled", + }); + if (result.outcome === "triggered") triggered += 1; + if (result.outcome === "skipped") skipped += 1; + } catch (err) { + logger.error({ err, issueId: claimed.id }, "issue monitor tick failed"); + } + } + + return { + checked: dueMonitors.length, + triggered, + skipped, + }; + } + + async function recoverActiveSessionGoals() { + if ((await getSchedulingSuppression()).suppressed) { + return { scanned: 0, enqueued: 0 }; + } + const sessions = await db + .select({ + id: agentTaskSessions.id, + companyId: agentTaskSessions.companyId, + agentId: agentTaskSessions.agentId, + issueId: agentTaskSessions.taskKey, + revision: agentTaskSessions.goalRevision, + }) + .from(agentTaskSessions) + .innerJoin( + issues, + and( + sql`${issues.id}::text = ${agentTaskSessions.taskKey}`, + eq(issues.companyId, agentTaskSessions.companyId), + eq(issues.assigneeAgentId, agentTaskSessions.agentId), + ), + ) + .where( + and( + eq(agentTaskSessions.goalDesiredState, "active"), + eq(agentTaskSessions.goalStatus, "active"), + notInArray(issues.status, ["done", "cancelled"]), + ), + ); + let enqueued = 0; + for (const session of sessions) { + const run = await enqueueWakeup(session.agentId, { + source: "automation", + triggerDetail: "system", + reason: "goal_control", + payload: { issueId: session.issueId, intent: "goal_recovery" }, + idempotencyKey: `goal_recovery:${session.id}:${session.revision}`, + requestedByActorType: "system", + contextSnapshot: { + issueId: session.issueId, + taskKey: session.issueId, + resumeSessionGoalHeartbeat: true, + skipIssueComment: true, + }, + }); + if (run) enqueued += 1; + } + return { scanned: sessions.length, enqueued }; + } + + async function recoverPendingSessionGoalActions() { + if ((await getSchedulingSuppression()).suppressed) { + return { scanned: 0, enqueued: 0, alreadyQueued: 0, invalid: 0 }; + } + const pending = await db + .select({ + id: agentSessionGoalActions.id, + requestId: agentSessionGoalActions.requestId, + payload: agentSessionGoalActions.payloadJson, + companyId: agentTaskSessions.companyId, + agentId: agentTaskSessions.agentId, + adapterType: agentTaskSessions.adapterType, + issueId: agentTaskSessions.taskKey, + }) + .from(agentSessionGoalActions) + .innerJoin( + agentTaskSessions, + eq(agentTaskSessions.id, agentSessionGoalActions.sessionId), + ) + .innerJoin( + issues, + and( + sql`${issues.id}::text = ${agentTaskSessions.taskKey}`, + eq(issues.companyId, agentTaskSessions.companyId), + eq(issues.assigneeAgentId, agentTaskSessions.agentId), + ), + ) + .where( + inArray(agentSessionGoalActions.status, [ + "pending", + "delivering", + "delivered", + ]), + ) + .orderBy(asc(agentSessionGoalActions.createdAt)); + + let enqueued = 0; + let alreadyQueued = 0; + let invalid = 0; + for (const action of pending) { + const control = parseNativeSessionGoalControl(action.payload); + if (!control || control.requestId !== action.requestId) { + invalid += 1; + await failRunnerGoalAction( + db, + { + companyId: action.companyId, + issueId: action.issueId, + agentId: action.agentId, + adapterType: action.adapterType, + }, + action.requestId, + "session_goal_control_payload_invalid", + ).catch(() => undefined); + continue; + } + + const inFlight = await db + .select({ contextSnapshot: heartbeatRuns.contextSnapshot }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, action.companyId), + eq(heartbeatRuns.agentId, action.agentId), + inArray(heartbeatRuns.status, [ + "queued", + "scheduled_retry", + "running", + ]), + ), + ) + .then((runs) => + runs.some( + (run) => + readNonEmptyString( + parseObject(run.contextSnapshot).goalControlRequestId, + ) === action.requestId, + ), + ); + if (inFlight) { + alreadyQueued += 1; + continue; + } + + const run = await enqueueWakeup(action.agentId, { + source: "automation", + triggerDetail: "system", + reason: "goal_control", + payload: { + issueId: action.issueId, + requestId: action.requestId, + intent: "goal_control_recovery", + }, + idempotencyKey: `goal_control_recovery:${action.id}`, + requestedByActorType: "system", + contextSnapshot: { + issueId: action.issueId, + taskKey: action.issueId, + // Goal controls reconcile the provider session itself and remain + // valid after issue terminalization (for example, clearing a + // completed goal from its retained widget). + resumeIntent: true, + goalControlRequestId: action.requestId, + runnerGoalControl: control, + skipIssueComment: true, + }, + }); + if (run) enqueued += 1; + } + return { scanned: pending.length, enqueued, alreadyQueued, invalid }; + } + + async function tickTimers(now = new Date()) { + if ((await getSchedulingSuppression()).suppressed) { + return { + checked: 0, + enqueued: 0, + skipped: 0, + }; + } + const cutoff = await getWorktreeExecutionCutoff(); + + const allAgents = await db + .select({ ...getTableColumns(agents) }) + .from(agents) + .innerJoin(companies, eq(companies.id, agents.companyId)) + .where(eq(companies.status, "active")); + const agentsByCompany = groupAgentOrgRowsByCompany( + allAgents.map(toAgentOrgRow), + ); + let checked = 0; + let enqueued = 0; + let skipped = 0; + + for (const agent of allAgents) { + const invokability = evaluateAgentInvokability( + toAgentOrgRow(agent), + agentsByCompany.get(agent.companyId) ?? [], + ); + if (!invokability.invokable) continue; + const policy = parseHeartbeatPolicy(agent); + if (!policy.enabled || policy.intervalSec <= 0) continue; + + if (cutoff) { + const eligibleIssue = await db + .select({ id: issues.id }) + .from(issues) + .where( + and( + eq(issues.companyId, agent.companyId), + eq(issues.assigneeAgentId, agent.id), + inArray(issues.status, ["todo", "in_progress"]), + gte(issues.createdAt, cutoff), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null); + if (!eligibleIssue) continue; + } + + checked += 1; + const baseline = new Date( + agent.lastHeartbeatAt ?? agent.createdAt, + ).getTime(); + const elapsedMs = now.getTime() - baseline; + if (elapsedMs < policy.intervalSec * 1000) continue; + const timerClaim = await claimDueTimerHeartbeat( + agent, + now, + policy.intervalSec, + ); + if (!timerClaim) continue; + + const run = await enqueueWakeup(agent.id, { + source: "timer", + triggerDetail: "system", + reason: "heartbeat_timer", + requestedByActorType: "system", + requestedByActorId: "heartbeat_scheduler", + contextSnapshot: { + source: "scheduler", + reason: "interval_elapsed", + now: now.toISOString(), + timerClaimWasFirstHeartbeat: timerClaim.wasFirstHeartbeat, + }, + }); + if (run) enqueued += 1; + else skipped += 1; + } + + const issueMonitors = await tickDueIssueMonitors(now); + + return { + checked: checked + issueMonitors.checked, + enqueued: enqueued + issueMonitors.triggered, + skipped: skipped + issueMonitors.skipped, + }; + } + + return { + triggerIssueMonitor, + tickTimers, + recoverActiveSessionGoals, + recoverPendingSessionGoalActions, + }; +}