fix(heartbeat): explain deferred monitor dispatch

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/services/heartbeat/queue.ts

@@ -3123,6 +3123,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
           const deferBlockedExecution = async (
             executionBlocker: NonNullable<Awaited<ReturnType<typeof getExecutionBlocker>>>,
           ) => {
+            opts.issueStateGuard?.onExecutionDeferred?.(executionBlocker.cause);
             const condition = { recoveryActionId: executionBlocker.recoveryActionId, ...continuationWait };
             if (executionWaitRequestId) {
               await tx.update(agentWakeupRequests).set({

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;

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);

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<typeof dispatchClaimedIssueMonitor>[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;
     }
   }