diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index a235ab12885..5bffe606284 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -434,6 +434,14 @@ An external wait counts as a live or waiting path only when the next move surviv A one-shot issue monitor consumes its persisted `nextCheckAt` when it dispatches the assignee wake. If that monitor-consuming run is lost before it records a new disposition or future monitor, Paperclip restores exactly one bounded continuation using the existing process-loss retry limit; if that continuation is also lost, the normal recovery-action escalation owns the next step instead of creating another monitor loop. +A due monitor remains scheduled when queue admission defers it. Execution recovery +holds still prevent provider work until the existing reconciliation path clears +those holds. `issue.monitor_deferred` activity records that decision and a safe +blocker cause when available. Unexpected admission failures retain the schedule +and record `issue.monitor_dispatch_failed`, with no raw provider error. Repeated +identical diagnostics for one schedule are limited to one per five minutes. +Diagnostics never authorize execution or consume a monitor attempt. + Native `set_task_monitor` arguments put scheduling fields inside `monitor`: `{ "idempotencyKey": "check-ci", "monitor": { "nextCheckAt": "", "notes": "Check pending CI." } }`. Use `monitor: null` to clear a schedule. A runner schema rejection executes no diff --git a/maps/agent-access-map/tasks.md b/maps/agent-access-map/tasks.md index 3f15a1df402..ce8ec6f181c 100644 --- a/maps/agent-access-map/tasks.md +++ b/maps/agent-access-map/tasks.md @@ -127,6 +127,14 @@ one `issue_monitor_due` wake. [Scheduler integration coverage](../../server/src/ uses real PostgreSQL, the native tool authority, and the heartbeat scheduler with a no-op process agent. It does not prove live model recovery. +Monitor deferral diagnostics: an unresolved execution-recovery hold keeps a due +schedule intact and prevents execution. Read `issue.monitor_deferred` activity +for the safe blocker cause; unexpected dispatch failures record +`issue.monitor_dispatch_failed`. Repeated identical diagnostics are throttled for +five minutes. The scheduler regression proves that a held source does not replay +and that clearing the hold permits one later wake. This is local PostgreSQL +evidence, not live provider recovery. + Gaps: Dedicated tools differ in scope and fields. Generic `call_api` forwards status, assignment and dependency fields to existing route authorization. diff --git a/maps/feature-map/tasks.md b/maps/feature-map/tasks.md index 7ee1c856bca..fa9f530047a 100644 --- a/maps/feature-map/tasks.md +++ b/maps/feature-map/tasks.md @@ -75,6 +75,9 @@ task. Reload and confirm the persisted monitor. After the agent yields, verify one later wake and a consumed schedule. For malformed-call recovery, use the [agent access recipe](../agent-access-map/tasks.md#properties) and verify that schema feedback supports a corrected call before a wait is claimed. +For a due monitor held by execution recovery, confirm that the schedule remains +and task activity explains the deferral. Repeated checks must not replay the held +work. Resolve the hold through its existing recovery action, then verify one wake. Manual: Add an ordinary Board comment while a task retry is queued or running. Confirm that the same execution keeps working and receives the saved message, diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 758910670f2..5c89c2c2538 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -3123,6 +3123,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende const deferBlockedExecution = async ( executionBlocker: NonNullable>>, ) => { + opts.issueStateGuard?.onExecutionDeferred?.(executionBlocker.cause); const condition = { recoveryActionId: executionBlocker.recoveryActionId, ...continuationWait }; if (executionWaitRequestId) { await tx.update(agentWakeupRequests).set({ diff --git a/server/src/services/heartbeat/run-preparation.ts b/server/src/services/heartbeat/run-preparation.ts index 374412e1276..31149baed5a 100644 --- a/server/src/services/heartbeat/run-preparation.ts +++ b/server/src/services/heartbeat/run-preparation.ts @@ -860,6 +860,8 @@ export interface WakeupOptions { statusVersion?: number; monitorNextCheckAt?: string; monitorWakeRequestedAt?: string; + /** Internal diagnostic only; the callback cannot grant admission. */ + onExecutionDeferred?: (blockerCause: string) => void; }; /** Keep causally distinct external chat continuations out of an existing run. */ allowRunCoalescing?: boolean; diff --git a/server/src/services/heartbeat/scheduling.test.ts b/server/src/services/heartbeat/scheduling.test.ts index 74d5a62bb40..27fb76a2249 100644 --- a/server/src/services/heartbeat/scheduling.test.ts +++ b/server/src/services/heartbeat/scheduling.test.ts @@ -213,6 +213,47 @@ describe.skipIf(!postgres.supported)("heartbeat scheduling database boundary", ( expect(JSON.stringify(events)).not.toContain("PRIVATE_PROVIDER_FAILURE"); }); + it("repeats a persistent diagnostic after its bounded quiet window", async () => { + const f = await fixture(); + f.deps.enqueueWakeup.mockResolvedValue(null); + await f.scheduler.tickTimers(now); + await f.scheduler.tickTimers(new Date(now.getTime() + 30_000)); + expect(await db.select().from(activityLog).where(eq(activityLog.entityId, f.issue.id))).toHaveLength(1); + await db.update(activityLog).set({ createdAt: new Date(Date.now() - 6 * 60_000) }).where(eq(activityLog.entityId, f.issue.id)); + await f.scheduler.tickTimers(new Date(now.getTime() + 60_000)); + expect(await db.select().from(activityLog).where(eq(activityLog.entityId, f.issue.id))).toHaveLength(2); + expect(await readIssue(f.issue.id)).toMatchObject({ monitorNextCheckAt: dueAt, monitorAttemptCount: 0 }); + }); + + it("does not copy an unrecognized blocker cause into monitor activity", async () => { + const f = await fixture(); + f.deps.enqueueWakeup.mockImplementationOnce(async (_agentId, options) => { + options?.issueStateGuard?.onExecutionDeferred?.("PRIVATE_BLOCKER_CAUSE"); + return null; + }); + await f.scheduler.tickTimers(now); + const events = await db.select().from(activityLog).where(eq(activityLog.entityId, f.issue.id)); + expect(events).toMatchObject([{ action: "issue.monitor_deferred", details: { reason: "execution_reconciliation_required" } }]); + expect(JSON.stringify(events)).not.toContain("PRIVATE_BLOCKER_CAUSE"); + }); + + it("releases a monitor claim when diagnostic storage fails", async () => { + const f = await fixture(); + await db.execute(sql`create function fail_monitor_diagnostic_test() returns trigger language plpgsql as $$ + begin if new.action = 'issue.monitor_deferred' then raise exception 'Synthetic diagnostic storage failure'; end if; return new; end; $$`); + await db.execute(sql`create trigger fail_monitor_diagnostic_test before insert on activity_log + for each row execute function fail_monitor_diagnostic_test()`); + try { + 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 }); + } finally { + await db.execute(sql`drop trigger fail_monitor_diagnostic_test on activity_log`); + await db.execute(sql`drop function fail_monitor_diagnostic_test()`); + } + }); + it("does not attach old dispatch diagnostics to a replacement monitor", async () => { const f = await fixture(); const replacement = new Date(now.getTime() + 60_000); diff --git a/server/src/services/heartbeat/scheduling.ts b/server/src/services/heartbeat/scheduling.ts index a0714e2d40f..38835c6fdf1 100644 --- a/server/src/services/heartbeat/scheduling.ts +++ b/server/src/services/heartbeat/scheduling.ts @@ -6,12 +6,14 @@ import { agentTaskSessions, agentSessionGoalActions, agents, + activityLog, } from "@paperclipai/db"; import { type IssueExecutionMonitorPolicy, type IssueExecutionMonitorClearReason, type IssueExecutionMonitorRecoveryPolicy, PROVIDER_QUOTA_MONITOR_SERVICE_NAME, + EXECUTION_RECONCILIATION_CAUSES, isUuidLike, } from "@paperclipai/shared"; import { @@ -33,7 +35,7 @@ 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 { logActivity, publishActivity, type ActivityPublication } from "../activity-log.js"; import { normalizeIssueExecutionPolicy, buildIssueMonitorClearedPatch, @@ -539,6 +541,52 @@ export function createHeartbeatScheduling( eq(issues.monitorWakeRequestedAt, claimed.monitorWakeRequestedAt!)); } + async function releaseIssueMonitorWithDiagnostic( + claimed: IssueMonitorDispatchRow, + input: Parameters[1], + diagnostic: { action: "issue.monitor_deferred" | "issue.monitor_dispatch_failed"; + reason: string; blockerCause?: string; httpStatus?: number }, + ) { + const publications: ActivityPublication[] = []; + try { + await db.transaction(async (tx) => { + const released = await tx.update(issues) + .set({ monitorWakeRequestedAt: null, updatedAt: input.now }) + .where(issueMonitorClaimCondition(claimed)).returning({ id: issues.id }); + if (released.length === 0) return; + // The issue lock serializes diagnostics from competing scheduler instances. + // Bound recurring reads to this issue's recent indexed activity history. + const [recent] = await tx.select({ id: activityLog.id }).from(activityLog).where(and( + eq(activityLog.companyId, claimed.companyId), eq(activityLog.entityType, "issue"), + eq(activityLog.entityId, claimed.id), eq(activityLog.action, diagnostic.action), + gte(activityLog.createdAt, new Date(Date.now() - 5 * 60_000)), + sql`${activityLog.details}->>'nextCheckAt' = ${claimed.monitorNextCheckAt!.toISOString()}`, + sql`${activityLog.details}->>'reason' = ${diagnostic.reason}`, + sql`coalesce(${activityLog.details}->>'blockerCause', '') = ${diagnostic.blockerCause ?? ""}`, + )).limit(1); + if (recent) return; + await logActivity(tx as unknown as Db, { + companyId: claimed.companyId, actorType: input.actorType, actorId: input.actorId, + agentId: input.agentId, runId: input.runId, action: diagnostic.action, + entityType: "issue", entityId: claimed.id, + details: { nextCheckAt: claimed.monitorNextCheckAt!.toISOString(), + checkedAt: input.now.toISOString(), attemptCount: claimed.monitorAttemptCount ?? 0, + reason: diagnostic.reason, source: input.activitySource, + ...(diagnostic.blockerCause ? { blockerCause: diagnostic.blockerCause } : {}), + ...(diagnostic.httpStatus ? { httpStatus: diagnostic.httpStatus } : {}), + }, + }, publications); + }); + for (const publication of publications) publishActivity(publication); + } catch (err) { + // Diagnostics are optional. Release the original claim even if audit storage + // failed, and never clobber a monitor that was replaced in the meantime. + await db.update(issues).set({ monitorWakeRequestedAt: null, updatedAt: input.now }) + .where(issueMonitorClaimCondition(claimed)); + logger.warn({ err, issueId: claimed.id }, "issue monitor diagnostic could not be recorded"); + } + } + async function dispatchClaimedIssueMonitor( claimed: IssueMonitorDispatchRow, input: { @@ -684,9 +732,11 @@ export function createHeartbeatScheduling( throw conflict(scheduled.reason); } } else { + let executionDeferredCause: string | null = null; const wake = await enqueueWakeup(targetAgentId, { issueStateGuard: { statuses: [claimed.status], assigneeAgentId: claimed.assigneeAgentId, - monitorNextCheckAt: scheduledAtIso, monitorWakeRequestedAt: claimed.monitorWakeRequestedAt!.toISOString() }, + monitorNextCheckAt: scheduledAtIso, monitorWakeRequestedAt: claimed.monitorWakeRequestedAt!.toISOString(), + onExecutionDeferred: cause => { executionDeferredCause = cause; } }, source: input.source, triggerDetail: input.triggerDetail, reason: wakeReason, @@ -717,9 +767,14 @@ export function createHeartbeatScheduling( }, }); 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 blockerCause = executionDeferredCause && + ([...EXECUTION_RECONCILIATION_CAUSES, "execution_owner_active"] as readonly string[]).includes(executionDeferredCause) + ? executionDeferredCause : undefined; + const reason = executionDeferredCause ? "execution_reconciliation_required" : "monitor_dispatch_deferred"; + await releaseIssueMonitorWithDiagnostic(claimed, input, { + action: "issue.monitor_deferred", reason, ...(blockerCause ? { blockerCause } : {}), + }); + return { outcome: "skipped" as const, reason }; } } @@ -795,24 +850,11 @@ export function createHeartbeatScheduling( 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)); } - + await releaseIssueMonitorWithDiagnostic(claimed, input, { + action: "issue.monitor_dispatch_failed", reason: "monitor_dispatch_failed", + ...(err instanceof HttpError ? { httpStatus: err.status } : {}), + }); throw err; } }