fix(runner): restore task monitors and durable timed waits (#15446)

## Thinking Path > - Paperclip manages AI agents and their task execution. > - Agents need a durable way to return to work after a delayed check. > - The issue monitor scheduler already provides a one-shot wake for an assignee. > - Native runners reject generic execution-policy writes and had no bound monitor tool. > - Scheduling alone is insufficient because native completion also needs to accept a timed wait. > - This pull request adds an authorized monitor tool and connects it to completion and the existing scheduler. > - An agent can now schedule its next check, end the run, and resume on the same task. ## Linked Issues or Issue Description **What happened?** A native runner could not set its own task monitor. `call_api` correctly rejected execution-policy writes, while `schedule_wake` had no production binding. `paperclip_finish` also rejected monitor waits. **Expected behavior** A standard native run can set a one-shot monitor on its current task or another accessible task assigned to the same agent. After a confirmed schedule on the current task, it can yield. The scheduler later delivers `issue_monitor_due`. **Steps to reproduce** 1. Start a standard native task. 2. Ask the agent to check the task again later and end its current run. 3. Inspect available tools and try the generic issue execution-policy update. 4. Observe the missing native tool and the lifecycle-write denial. Related PRs: #14680 concerns monitor notes in the shared wake prompt. #11919 changes attempt-limit scope. This PR adds native scheduling and completion authority and retains the existing cumulative attempt bounds. It does not depend on either PR. ## What Changed - Add provider-neutral `set_task_monitor` with a default current-task target, future timestamp, required notes, existing bounds, and explicit clearing. - Check company, task visibility, ownership, runtime permissions, work mode, and active-run authority. Preserve review-only restrictions. Reject the reserved server-owned quota-recovery name before saving or accepting a native wait. - Commit the monitor, audit event, and retry receipt together. Retry receipts survive a successor run without re-arming cleared or consumed timers. - Permit `paperclip_finish` to yield to a persisted monitor. Recheck ownership and the schedule when committing final disposition. Release execution without an immediate continuation. - Preserve due monitors during native execution. Fence wake admission and consumption against replacement, clearing, reassignment, and completion. Preserve unrelated review policy. - Expose scheduled and consumed monitor instructions in task context. Update provider schemas, Rust validation, generated contracts, and execution documentation. - Add an opt-in live Codex smoke script with isolated data and explicit run/session/runner/process evidence. ## Verification - Repository `pnpm -r typecheck` and `pnpm build` passed after rebase. Server typecheck passed again after review fixes. All CI test shards pass on `3def77b1b`, including runner TypeScript/Rust, server, serialized server, workspace, and browser tests. All CI gates are green, including the canary dry run. Greptile is 5/5 on the same commit with zero unresolved threads. - The local monolithic `pnpm test:run`, started before the rebase, was interrupted after current-head CI test coverage passed. It is not counted as a standalone full-suite pass; the focused local regression suites passed. - Targeted server tests cover scheduling, replacement, clearing, policy preservation, cumulative bounds, cross-run retries, permissions, provider-neutral discovery, review restrictions, completion authority, and scheduler/finalizer races. - Runner contract/catalog/semantic tests and Rust terminal-tool tests cover the new operation and monitor completion. - Live Codex test passed twice (latest live run on `e0bcd63e6`) in a temporary database and workspace, with a 300,000 ms warm window. First run `67bd7709-c089-4d4a-9d2b-0d6b618a34b0` yielded at `2026-10-07T13:15:03.274Z`. Second run `97c323f9-595a-4cc5-a007-db5a2fbb937c` started at `13:15:30.952Z`, received `issue_monitor_due`, and completed the same task. Exactly one monitor wake was recorded. - Both live runs used native session `7b1dd753-1c9b-4e7a-b22f-a125dbc3748c`, runner `e90d9a1b-3502-4ee3-b15e-edc024c555d4`, provider session `01a11680-6d01-70c0-9a55-db7246ed66c3`, and PID `64218` with the same process start time. This proves warm reuse for that local Codex test, not only successful scheduling. - Reproduce the paid live test with `node --import ./server/node_modules/tsx/dist/loader.mjs server/scripts/smoke-native-task-monitor.ts --run`, with the installed Codex binary on `PATH` and a valid local login. ## Risks - The scheduler now defers monitor dispatch while the task has an active native run. A stuck run still depends on the existing recovery lifecycle. - Idempotency uses the existing run ledger; no table or migration is added. - Other providers share the tested tool and completion contracts. Only Codex received a live model test. - Existing `call_api` lifecycle restrictions remain enforced. Monitor waits do not bypass task blockers, reviews, or approvals. ## Model Used OpenAI Codex, GPT-6 family, with tool use, code execution, and TypeScript/Rust editing. The session does not expose the exact deployed model ID 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 <noreply@paperclip.ing>
2026-10-23 00:53:38 +02:00 · 2026-10-07 08:41:24 -05:00
1 parent b315580640
commit 8cfedd7df8
@@ -3786,6 +3786,8 @@ interface WakeupOptions {
    statuses: string[];
    assigneeAgentId: string;
    statusVersion?: number;
+    monitorNextCheckAt?: string;
+    monitorWakeRequestedAt?: string;
  };
  /** Keep causally distinct external chat continuations out of an existing run. */
  allowRunCoalescing?: boolean;
@@ -11661,18 +11663,19 @@ export function heartbeatService(
    runId: string | null;
    activitySource: "manual" | "scheduled";
  }) {
-    await db
+    const cleared = await db
      .update(issues)
      .set({
-        ...buildIssueMonitorClearedPatch({
+        ...monitorOnlyDispatchPatch(buildIssueMonitorClearedPatch({
          issue: input.claimed,
          policy: input.policy,
          clearReason: input.clearReason,
          clearedAt: input.now,
-        }),
+        })),
        updatedAt: input.now,
      })
-      .where(eq(issues.id, input.claimed.id));
+      .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,
@@ -11711,6 +11714,24 @@ export function heartbeatService(
    return { outcome: "skipped" as const, reason: input.clearReason };
  }

+  function monitorOnlyDispatchPatch<T extends ReturnType<typeof buildIssueMonitorClearedPatch>>(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: {
@@ -11855,8 +11876,10 @@ export function heartbeatService(
          if (scheduled.outcome === "not_scheduled")
            throw conflict(scheduled.reason);
        }
-      } else
-        await enqueueWakeup(targetAgentId, {
+      } 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,
@@ -11886,18 +11909,26 @@ export function heartbeatService(
            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" };
+        }
+      }

-      await db
+      const consumed = await db
        .update(issues)
        .set({
-          ...buildIssueMonitorTriggeredPatch({
+          ...monitorOnlyDispatchPatch(buildIssueMonitorTriggeredPatch({
            issue: claimed,
            policy,
            triggeredAt: input.now,
-          }),
+          })),
          updatedAt: new Date(),
        })
-        .where(eq(issues.id, claimed.id));
+        .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,
@@ -11926,15 +11957,15 @@ export function heartbeatService(
          await db
            .update(issues)
            .set({
-              ...buildIssueMonitorClearedPatch({
+              ...monitorOnlyDispatchPatch(buildIssueMonitorClearedPatch({
                issue: claimed,
                policy,
                clearReason: "dispatch_skipped",
                clearedAt: input.now,
-              }),
+              })),
              updatedAt: new Date(),
            })
-            .where(eq(issues.id, claimed.id));
+            .where(issueMonitorClaimCondition(claimed));

          await logActivity(db, {
            companyId: claimed.companyId,
@@ -11964,7 +11995,7 @@ export function heartbeatService(
            monitorWakeRequestedAt: null,
            updatedAt: new Date(),
          })
-          .where(eq(issues.id, claimed.id));
+          .where(issueMonitorClaimCondition(claimed));
      } else {
        await db
          .update(issues)
@@ -11972,13 +12003,21 @@ export function heartbeatService(
            monitorWakeRequestedAt: null,
            updatedAt: new Date(),
          })
-          .where(eq(issues.id, claimed.id));
+          .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?: {
@@ -12071,6 +12110,7 @@ export function heartbeatService(
      .where(
        and(
          eq(companies.status, "active"),
+          noActiveNativeMonitorRun(),
          sql`${issues.monitorNextCheckAt} is not null`,
          lte(issues.monitorNextCheckAt, now),
          isNull(issues.assigneeUserId),
@@ -12099,6 +12139,7 @@ export function heartbeatService(
          .where(
            and(
              eq(issues.id, due.id),
+              noActiveNativeMonitorRun(),
              sql`${issues.monitorNextCheckAt} is not null`,
              lte(issues.monitorNextCheckAt, now),
              isNull(issues.assigneeUserId),
@@ -27839,6 +27880,9 @@ export function heartbeatService(
              executionWorkspacePreference: issues.executionWorkspacePreference,
              executionWorkspaceSettings: issues.executionWorkspaceSettings,
              assigneeAgentId: issues.assigneeAgentId,
+              assigneeUserId: issues.assigneeUserId,
+              monitorNextCheckAt: issues.monitorNextCheckAt,
+              monitorWakeRequestedAt: issues.monitorWakeRequestedAt,
              executionRunId: issues.executionRunId,
              executionAgentNameKey: issues.executionAgentNameKey,
              createdAt: issues.createdAt,
@@ -27870,6 +27914,54 @@ export function heartbeatService(
            return { kind: "skipped" as const };
          }

+          const issueStateGuard = opts.issueStateGuard;
+          const activeMonitorRun = issueStateGuard?.monitorNextCheckAt === undefined ? null
+            : await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
+              eq(heartbeatRuns.companyId, issue.companyId), eq(heartbeatRuns.nativeIssueId, issue.id),
+              eq(heartbeatRuns.runtimeMode, "native"), inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]),
+            )).limit(1).then(rows => rows[0] ?? null);
+          if (
+            issueStateGuard &&
+            (!issueStateGuard.statuses.includes(issue.status) ||
+              issue.assigneeAgentId !== issueStateGuard.assigneeAgentId ||
+              (issueStateGuard.statusVersion !== undefined && issue.statusVersion !== issueStateGuard.statusVersion) ||
+              (issueStateGuard.monitorNextCheckAt !== undefined && (
+                activeMonitorRun !== null || issue.assigneeUserId !== null ||
+                issue.monitorNextCheckAt?.toISOString() !== issueStateGuard.monitorNextCheckAt ||
+                issue.monitorWakeRequestedAt?.toISOString() !== issueStateGuard.monitorWakeRequestedAt
+              )))
+          ) {
+            // A deferred monitor retains its schedule; do not create a receipt
+            // that could suppress its next admission attempt.
+            if (issueStateGuard.monitorNextCheckAt !== undefined) return { kind: "skipped" as const };
+            await tx.insert(agentWakeupRequests).values({
+              ...durableReceiptFields,
+              companyId: agent.companyId,
+              agentId,
+              source,
+              triggerDetail,
+              reason: "issue_state_guard_mismatch",
+              payload: {
+                ...(payload ?? {}),
+                heartbeatSkip: {
+                  reason:
+                    "Issue status or assignee changed before the wake could be queued.",
+                  issueId: issue.id,
+                  expectedStatuses: issueStateGuard.statuses,
+                  actualStatus: issue.status,
+                  expectedAssigneeAgentId: issueStateGuard.assigneeAgentId,
+                  actualAssigneeAgentId: issue.assigneeAgentId,
+                },
+              },
+              status: "skipped",
+              requestedByActorType: opts.requestedByActorType ?? null,
+              requestedByActorId: opts.requestedByActorId ?? null,
+              idempotencyKey: opts.idempotencyKey ?? null,
+              finishedAt: new Date(),
+            });
+            return { kind: "skipped" as const };
+          }
+
          if (opts.failedRunId) {
            // The issue lock makes double-clicks and network retries adopt the
            // same successor, including after it has already finished.
@@ -28068,40 +28160,6 @@ export function heartbeatService(
            onBlocked: (reason, message) => { continuationWait = { reason, message }; },
          }))) return deferBlockedExecution(executionBlocker);

-          const issueStateGuard = opts.issueStateGuard;
-          if (
-            issueStateGuard &&
-            (!issueStateGuard.statuses.includes(issue.status) ||
-              issue.assigneeAgentId !== issueStateGuard.assigneeAgentId ||
-              (issueStateGuard.statusVersion !== undefined && issue.statusVersion !== issueStateGuard.statusVersion))
-          ) {
-            await tx.insert(agentWakeupRequests).values({
-              ...durableReceiptFields,
-              companyId: agent.companyId,
-              agentId,
-              source,
-              triggerDetail,
-              reason: "issue_state_guard_mismatch",
-              payload: {
-                ...(payload ?? {}),
-                heartbeatSkip: {
-                  reason:
-                    "Issue status or assignee changed before the wake could be queued.",
-                  issueId: issue.id,
-                  expectedStatuses: issueStateGuard.statuses,
-                  actualStatus: issue.status,
-                  expectedAssigneeAgentId: issueStateGuard.assigneeAgentId,
-                  actualAssigneeAgentId: issue.assigneeAgentId,
-                },
-              },
-              status: "skipped",
-              requestedByActorType: opts.requestedByActorType ?? null,
-              requestedByActorId: opts.requestedByActorId ?? null,
-              idempotencyKey: opts.idempotencyKey ?? null,
-              finishedAt: new Date(),
-            });
-            return { kind: "skipped" as const };
-          }

          if (
            worktreeExecutionCutoff &&
@@ -0,0 +1,166 @@
+import { issueExecutionMonitorPolicySchema, PROVIDER_QUOTA_MONITOR_SERVICE_NAME } from "@paperclipai/shared";
+import type { issues } from "@paperclipai/db";
+import { z } from "zod";
+import { forbidden, unprocessable } from "../errors.js";
+import type { accessService } from "./access.js";
+import { authorizationDeniedDetails, type AuthorizationActor } from "./authorization.js";
+import { applyIssueMonitorPolicyTransition, normalizeIssueExecutionPolicy, parseIssueExecutionState, redactIssueMonitorExternalRef, setIssueExecutionPolicyMonitorScheduledBy } from "./issue-execution-policy.js";
+
+type NormalizedExecutionPolicy = ReturnType<typeof normalizeIssueExecutionPolicy>;
+
+export function monitorPoliciesEqual(
+  left: NormalizedExecutionPolicy | null,
+  right: NormalizedExecutionPolicy | null,
+) {
+  return (
+    JSON.stringify(left?.monitor ?? null) ===
+    JSON.stringify(right?.monitor ?? null)
+  );
+}
+
+export function applyActorMonitorScheduledBy(
+  policy: NormalizedExecutionPolicy | null,
+  actorType: "agent" | "user",
+) {
+  return setIssueExecutionPolicyMonitorScheduledBy(
+    policy,
+    actorType === "user" ? "board" : "assignee",
+  );
+}
+
+export async function assertCanManageIssueMonitor(
+  accessSvc: Pick<ReturnType<typeof accessService>, "decide">,
+  req: { actor: AuthorizationActor },
+  companyId: string,
+  assigneeAgentId: string | null,
+  monitorChanged: boolean,
+) {
+  if (!monitorChanged) return;
+  if (req.actor.type === "board") return;
+  const runtimeDecision = await accessSvc.decide({
+    actor: req.actor,
+    action: "runtime:manage",
+    resource: { type: "company", companyId },
+  });
+  if (!runtimeDecision.allowed) {
+    throw forbidden(
+      runtimeDecision.explanation,
+      authorizationDeniedDetails(runtimeDecision),
+    );
+  }
+  if (
+    req.actor.type === "agent" &&
+    req.actor.agentId &&
+    req.actor.agentId === assigneeAgentId
+  )
+    return;
+  throw forbidden(
+    "Only the assignee agent or a board user can manage issue monitors",
+  );
+}
+
+export function summarizeIssueMonitor(
+  issue: {
+    monitorNextCheckAt?: Date | null;
+    monitorLastTriggeredAt?: Date | null;
+    monitorAttemptCount?: number | null;
+    monitorNotes?: string | null;
+    monitorScheduledBy?: string | null;
+    executionState?: unknown;
+  },
+  policy: NormalizedExecutionPolicy | null,
+) {
+  const state = parseIssueExecutionState(issue.executionState);
+  return {
+    nextCheckAt:
+      issue.monitorNextCheckAt?.toISOString() ??
+      policy?.monitor?.nextCheckAt ??
+      null,
+    lastTriggeredAt:
+      issue.monitorLastTriggeredAt?.toISOString() ??
+      state?.monitor?.lastTriggeredAt ??
+      null,
+    attemptCount:
+      issue.monitorAttemptCount ?? state?.monitor?.attemptCount ?? 0,
+    notes:
+      policy?.monitor?.notes ??
+      issue.monitorNotes ??
+      state?.monitor?.notes ??
+      null,
+    scheduledBy:
+      issue.monitorScheduledBy ??
+      policy?.monitor?.scheduledBy ??
+      state?.monitor?.scheduledBy ??
+      null,
+    kind: policy?.monitor?.kind ?? state?.monitor?.kind ?? null,
+    serviceName:
+      policy?.monitor?.serviceName ?? state?.monitor?.serviceName ?? null,
+    externalRef: redactIssueMonitorExternalRef(
+      policy?.monitor?.externalRef ?? state?.monitor?.externalRef ?? null,
+    ),
+    timeoutAt: policy?.monitor?.timeoutAt ?? state?.monitor?.timeoutAt ?? null,
+    maxAttempts:
+      policy?.monitor?.maxAttempts ?? state?.monitor?.maxAttempts ?? null,
+    recoveryPolicy:
+      policy?.monitor?.recoveryPolicy ?? state?.monitor?.recoveryPolicy ?? null,
+    status: state?.monitor?.status ?? (policy?.monitor ? "scheduled" : null),
+    clearReason: state?.monitor?.clearReason ?? null,
+  };
+}
+
+export const setTaskMonitorSchema = z.object({
+  taskId: z.string().uuid().optional(),
+  idempotencyKey: z.string().trim().min(1).max(240),
+  monitor: issueExecutionMonitorPolicySchema.omit({ scheduledBy: true }).extend({
+    notes: z.string().trim().min(1).max(500),
+  }).strict().refine(monitor => monitor.serviceName !== PROVIDER_QUOTA_MONITOR_SERVICE_NAME, {
+    message: "This serviceName is reserved for server-owned quota recovery; use a different service name for an ordinary task check",
+    path: ["serviceName"],
+  }).nullable(),
+}).strict();
+
+/** Caller holds the issue lock; merge only monitor state, never review policy. */
+export function prepareIssueMonitorUpdate(
+  issue: typeof issues.$inferSelect,
+  monitor: z.infer<typeof setTaskMonitorSchema>["monitor"],
+  actor: AuthorizationActor,
+  now = new Date(),
+) {
+  if (monitor && Date.parse(monitor.nextCheckAt) <= now.getTime()) {
+    throw unprocessable("Monitor nextCheckAt must be in the future");
+  }
+  if (monitor?.timeoutAt && Date.parse(monitor.timeoutAt) <= Date.parse(monitor.nextCheckAt)) {
+    throw unprocessable("Monitor timeoutAt must be later than nextCheckAt");
+  }
+  const previousPolicy = normalizeIssueExecutionPolicy(issue.executionPolicy);
+  const policy = applyActorMonitorScheduledBy(normalizeIssueExecutionPolicy({
+    ...previousPolicy, monitor,
+  }), actor.type === "board" ? "user" : "agent");
+  const transition = applyIssueMonitorPolicyTransition({
+    issue, policy, previousPolicy, requestedAssigneePatch: {},
+    actor: { agentId: actor.agentId ?? null, userId: actor.userId ?? null },
+    monitorExplicitlyUpdated: true,
+  });
+  // Normalization supplies monitor defaults, but must not rewrite unrelated
+  // review/authorization settings (including forward-compatible policy fields).
+  const storedPolicy = { ...(issue.executionPolicy ?? {}) };
+  if (policy?.monitor) storedPolicy.monitor = policy.monitor;
+  else delete storedPolicy.monitor;
+  return { ...transition.patch, executionPolicy: Object.keys(storedPolicy).length ? storedPolicy : null } as Partial<typeof issues.$inferInsert>;
+}
+
+/** Timestamp presence alone is not authority to park a native task. */
+export function eligibleIssueMonitorWait(
+  issue: typeof issues.$inferSelect,
+  agentId: string,
+  now = new Date(),
+): string | null {
+  if (issue.assigneeAgentId !== agentId || issue.assigneeUserId ||
+      !["in_progress", "in_review"].includes(issue.status) || !issue.monitorNextCheckAt) return null;
+  const monitor = normalizeIssueExecutionPolicy(issue.executionPolicy)?.monitor;
+  if (!monitor || monitor.serviceName === PROVIDER_QUOTA_MONITOR_SERVICE_NAME ||
+      Date.parse(monitor.nextCheckAt) !== issue.monitorNextCheckAt.getTime() ||
+      (monitor.timeoutAt && Date.parse(monitor.timeoutAt) <= now.getTime()) ||
+      (monitor.maxAttempts != null && issue.monitorAttemptCount >= monitor.maxAttempts)) return null;
+  return issue.monitorNextCheckAt.toISOString();
+}
@@ -1,3 +1,4 @@
+import { eligibleIssueMonitorWait } from "../issue-monitors.js";
 import { activeIssueInteractionCondition, ordinaryQuestionCondition } from "../issue-question-context.js";
 import { publishedTaskDocuments, validateNativeDeliverableEvidence } from "./native-deliverable-feedback.js";
 import { findAutomaticCompletionReviews } from "./automatic-completion-reviews.js";
@@ -64,6 +65,7 @@ export async function nativeCompletionFeedback(
  if (!issue) throw new Error("Completion task no longer exists.");
  const reviewContext = readNativeReviewAssignmentContext(run.contextSnapshot);
  if (reviewContext) {
+    if (result.continuation?.kind === "monitor") throw new Error("Review-only runs cannot yield to a task monitor; resolve the assigned review or report its blocker.");
    const review = await getNativeReviewAssignment(db, {
      companyId: run.companyId, issueId: issue.id, agentId: run.agentId,
      contextSnapshot: reviewContext, allowResolvedByRunId: run.id,
@@ -96,6 +98,12 @@ export async function nativeCompletionFeedback(
    }

  }
+  if (result.reportedWorkDisposition === "yielded" && result.continuation?.kind === "monitor") {
+    if (issue.workMode !== "standard" || issue.executionRunId !== run.id ||
+        !eligibleIssueMonitorWait(issue, run.agentId)) {
+      throw new Error("A monitor wait requires a persisted, eligible monitor on this task. Call set_task_monitor for the current task, confirm its schedule, then report yielded with continuation.kind monitor.");
+    }
+  }
  const signals = normalizePrpResultSignals(result);
  if (
    result.reportedWorkDisposition === "done" &&
@@ -215,6 +223,9 @@ export async function nativeCompletionFeedback(
  if (readiness.unresolvedBlockerCount > 0) {
    return `Completion report accepted; this task still has unresolved dependencies. Explain the blockers on [this task](/issues/${issue.identifier ?? issue.id}); do not say the task is done.`;
  }
+  if (result.reportedWorkDisposition === "yielded" && result.continuation?.kind === "monitor") {
+    return `Monitor wait accepted. The task remains active and Paperclip will wake its assignee at or after ${issue.monitorNextCheckAt!.toISOString()} with issue_monitor_due. End this turn; do not poll or mark the task done.`;
+  }
  if (
    !isConversation(issue) &&
    result.reportedWorkDisposition === "yielded" &&
@@ -1,3 +1,4 @@
+import { eligibleIssueMonitorWait } from "../issue-monitors.js";
 import { isNativePlanWaitResult, readNativePlanWait } from "./native-plan-wait.js";
 import { activeIssueInteractionCondition } from "../issue-question-context.js";
 import { hasPendingNativeChildCompletion } from "./native-child-completion-delivery.js";
@@ -1292,7 +1293,9 @@ export async function finalizeNativeRun(input: {
    };
    const hasPendingChildCompletion = !reviewContext &&
      await hasPendingNativeChildCompletion(input.db, childCompletionRecipient);
+    const monitorWaitAt = eligibleIssueMonitorWait(authoritativeIssue, run.agentId);
    const proposedDecision = resolveNativeFinalizerStatus({
+      monitorWaitAuthorized: authoritativeIssue.workMode === "standard" && authoritativeIssue.executionRunId === run.id && monitorWaitAt !== null,
      planWaitAuthorized: planWait !== null,
      hasPendingChildCompletion,
      providerModelRejected: providerFailure?.errorCode === "native_provider_model_rejected" && ownsProviderFailureDecision,
@@ -1381,6 +1384,8 @@ export async function finalizeNativeRun(input: {
        priorStatusVersion: Number(authoritativeIssue.statusVersion),
        priorDecisionId: authoritativeIssue.lastStatusDecisionId,
        decision,
+        requireMonitorWait: decision.reasonCode === "scheduled_monitor_waiting" && monitorWaitAt
+          ? { agentId: run.agentId, nextCheckAt: monitorWaitAt } : undefined,
        requirePlanWaitSource:
          decision.reasonCode === "native_plan_accepted_waiting_for_continuation"
            ? planWait?.source : undefined,
@@ -102,7 +102,7 @@ describe("PaperclipRunnerToolAuthority", () => {
      issueId,
      runId,
    });
-    expect(authority.definitions()).toHaveLength(38);
+    expect(authority.definitions()).toHaveLength(39);
    const questions = authority.definitions().find(tool => tool.name === "request_human_input")!;
    expect(questions.description).toContain("ask only the next unanswered question");
    expect(questions.description).toContain("Never fabricate answers");
@@ -128,6 +128,7 @@ describe("PaperclipRunnerToolAuthority", () => {
        "request_human_input",
        "create_task",
        "set_dependencies",
+        "set_task_monitor",
        "list_documents",
        "read_document",
        "list_document_revisions",
@@ -1,3 +1,4 @@
+import { assertCanManageIssueMonitor, prepareIssueMonitorUpdate, setTaskMonitorSchema, summarizeIssueMonitor } from "../issue-monitors.js";
 import { readTaskQuestionContext } from "../issue-question-context.js";
 import { isConversation } from "../agent-conversations.js";
 import { setIssueTitle } from "../issue-title.js";
@@ -14,7 +15,7 @@ import type { createAssignedMcpTools } from "./assigned-mcp-tools.js";
 import { assertAssignableAgent } from "../agent-assignability.js";
 import { approvalReadSqlCondition, canActorReadApproval, authorizationService, issueReadSqlCondition, type AuthorizationActor } from "../authorization.js";
 import { resolveCoreTrustPreset } from "../trust-preset-resolver.js";
-import { normalizeIssueExecutionPolicy } from "../issue-execution-policy.js";
+import { normalizeIssueExecutionPolicy, redactIssueMonitorExternalRef } from "../issue-execution-policy.js";
 import { buildLowTrustSourceTrust } from "../source-trust.js";
 import { handoffPlanContext } from "./handoff-plan-context.js";
 import { callCreateSkillTool, callUpdateSkillTool } from "../skill-tools.js";
@@ -101,7 +102,7 @@ const IMPLEMENTED_OPERATIONS = new Set([
  "submit_complaint", "submit_suggestion",
  "read_agent_instructions", "update_agent_instructions", "get_agent_instruction_history", "restore_agent_instructions",
  "search_api", "call_api", "hire_agent",
-  "get_task_context", "get_task_history", "search_tasks", "report_progress", "set_task_title",
+  "get_task_context", "get_task_history", "search_tasks", "report_progress", "set_task_title", "set_task_monitor",
  "request_human_input",
  "create_skill", "update_skill", "create_task", "reassign_task", "set_dependencies", "create_project", "list_project_repositories", "list_projects", "register_deliverable",
  "list_documents", "read_document", "list_document_revisions", "write_document",
@@ -582,6 +583,7 @@ export class PaperclipRunnerToolAuthority {
        (await captureRunIdentity(this.db, this.binding)).context?.id ?? null);
      case "reassign_task": return this.#reassignTask(input);
      case "set_task_title": return this.#setTaskTitle(input);
+      case "set_task_monitor": return this.#setTaskMonitor(input);
      case "set_dependencies": return this.#setDependencies(input);
      case "register_deliverable": return this.#registerDeliverable(input);
      default: throw new Error("paperclip_runner_tool_not_bound");
@@ -1349,6 +1351,84 @@ export class PaperclipRunnerToolAuthority {
    return result;
  }

+  async #setTaskMonitor(value: Record<string, unknown>): Promise<unknown> {
+    const input = setTaskMonitorSchema.parse(value);
+    if (input.monitor?.externalRef) input.monitor.externalRef = redactIssueMonitorExternalRef(input.monitor.externalRef);
+    const targetId = input.taskId ?? this.binding.issueId;
+    let publication: Parameters<typeof publishActivity>[0] | null = null;
+    let replayed = false;
+    let replayStatus: string | null = null;
+    let replayMonitor: ReturnType<typeof summarizeIssueMonitor> | null = null;
+    const authorizeTarget = async (tx: Db, context: { run: typeof heartbeatRuns.$inferSelect }) => {
+      // A second owned task may itself be writing another task: fail on lock
+      // contention instead of taking opposite issue/run locks and deadlocking.
+      const [target] = await tx.select().from(issues).where(and(
+        eq(issues.id, targetId), eq(issues.companyId, this.binding.companyId),
+      )).for("update", { noWait: true });
+      if (!target) throw notFound("Task not found");
+      const actor = this.#privacyActor(context.run);
+      const access = authorizationService(tx);
+      for (const action of ["issue:read", "issue:mutate"] as const) {
+        const decision = await access.decide({ actor, action, resource: {
+          type: "issue", companyId: this.binding.companyId, issueId: target.id,
+          status: target.status, assigneeAgentId: target.assigneeAgentId, assigneeUserId: target.assigneeUserId,
+        } });
+        if (!decision.allowed) throw forbidden(decision.explanation);
+      }
+      await assertCanManageIssueMonitor(access, { actor }, target.companyId, target.assigneeAgentId, true);
+      if (target.assigneeAgentId !== this.binding.agentId || target.assigneeUserId ||
+          !["in_progress", "in_review"].includes(target.status)) {
+        throw forbidden("Monitors require an in-progress or in-review task assigned to this agent");
+      }
+      return { target, actor };
+    };
+    const result = await this.#withMutationReceipt("set_task_monitor", input.idempotencyKey, input, async (tx, context) => {
+      const { target, actor } = await authorizeTarget(tx, context);
+      // The task lock also serializes retries after a new run resumes the task.
+      // Keep the durable receipt in the existing run ledger, without a new table.
+      const [earlier] = await tx.select({
+        receipt: sql<ToolReceipt>`${heartbeatRuns.resultJson}->'semanticToolReceipts'->${input.idempotencyKey}`,
+      }).from(heartbeatRuns).where(and(
+        eq(heartbeatRuns.companyId, this.binding.companyId),
+        eq(heartbeatRuns.agentId, this.binding.agentId),
+        eq(heartbeatRuns.nativeIssueId, this.binding.issueId),
+        sql`${heartbeatRuns.resultJson}->'semanticToolReceipts' ? ${input.idempotencyKey}`,
+      )).orderBy(desc(heartbeatRuns.createdAt)).limit(1);
+      if (earlier?.receipt) {
+        if (earlier.receipt.operationId !== "set_task_monitor" || canonicalJson(earlier.receipt.input) !== canonicalJson(input)) {
+          throw new Error("paperclip_runner_tool_idempotency_conflict");
+        }
+        replayed = true;
+        replayStatus = target.status;
+        replayMonitor = summarizeIssueMonitor(target, normalizeIssueExecutionPolicy(target.executionPolicy));
+        return { taskId: target.id, status: target.status, monitor: replayMonitor };
+      }
+      const patch = prepareIssueMonitorUpdate(target, input.monitor, actor);
+      const [updated] = await tx.update(issues).set({ ...patch, updatedAt: new Date() })
+        .where(and(eq(issues.id, target.id), eq(issues.companyId, this.binding.companyId))).returning();
+      if (!updated) throw notFound("Task not found");
+      const monitor = summarizeIssueMonitor(updated, normalizeIssueExecutionPolicy(updated.executionPolicy));
+      const activity = await persistActivity(tx, {
+        companyId: this.binding.companyId, actorType: "agent", actorId: this.binding.agentId,
+        agentId: this.binding.agentId, runId: this.binding.runId,
+        action: input.monitor ? "issue.monitor_scheduled" : "issue.monitor_cleared",
+        entityType: "issue", entityId: target.id,
+        details: { identifier: target.identifier, nextCheckAt: monitor.nextCheckAt,
+          notes: monitor.notes, scheduledBy: monitor.scheduledBy, serviceName: monitor.serviceName,
+          timeoutAt: monitor.timeoutAt, maxAttempts: monitor.maxAttempts, recoveryPolicy: monitor.recoveryPolicy },
+      });
+      publication = activity.publication;
+      return { taskId: updated.id, status: updated.status, monitor };
+    }, { beforeReceiptReplay: async (tx, context) => {
+      const { target } = await authorizeTarget(tx, context);
+      replayed = true;
+      replayStatus = target.status;
+      replayMonitor = summarizeIssueMonitor(target, normalizeIssueExecutionPolicy(target.executionPolicy));
+    } });
+    if (publication) publishActivity(publication);
+    return { ...record(result), ...(replayed ? { status: replayStatus, monitor: replayMonitor } : {}), replayed };
+  }
+
  async #setDependencies(input: Record<string, unknown>): Promise<unknown> {
    const idempotencyKey = requiredString(input.idempotencyKey);
    if (!Array.isArray(input.blockedByTaskIds)) {
@@ -2037,6 +2117,9 @@ function redactedTask(task: typeof issues.$inferSelect) {
    workMode: task.workMode,
    assigneeAgentId: task.assigneeAgentId,
    executionRunId: task.executionRunId,
+    ...(task.monitorNextCheckAt || task.monitorLastTriggeredAt ? {
+      monitor: summarizeIssueMonitor(task, normalizeIssueExecutionPolicy(task.executionPolicy)),
+    } : {}),
    parentId: task.parentId,
    projectId: task.projectId,
    goalId: task.goalId,
@@ -44,6 +44,20 @@ function arbitrate(
 }

 describe("native status authority", () => {
+  it("waits on persisted monitors without an immediate continuation, even with unfinished work", () => {
+    const pending = assessment({ reportedDisposition: "yielded", hasBlockingRemainingWork: true,
+      continuation: { kind: "monitor", summary: "Check CI", idempotencyKey: "ci" } });
+    for (const priorIssueStatus of ["in_progress", "in_review"] as const) {
+      expect(arbitrate({ assessment: pending, priorIssueStatus, monitorWaitAuthorized: true }))
+        .toMatchObject({ statusAction: "preserve", toStatus: priorIssueStatus,
+          reasonCode: "scheduled_monitor_waiting", effects: [{ kind: "release_checkout" }] });
+    }
+    expect(arbitrate({ assessment: pending })).toMatchObject({ reasonCode: "monitor_wait_authority_lost",
+      effects: [{ kind: "record_finalization_error" }] });
+    expect(arbitrate({ assessment: pending, monitorWaitAuthorized: true, hasUnresolvedIssueBlockers: true }).toStatus).toBe("blocked");
+    expect(arbitrate({ assessment: pending, monitorWaitAuthorized: true, governanceGate: { kind: "approval", id: "pending" } }).toStatus).toBe("in_review");
+  });
+
  it("keeps a pending child result non-terminal without adding another continuation", () => {
    expect(arbitrate({ hasPendingChildCompletion: true })).toMatchObject({
      statusAction: "in_progress", toStatus: "in_progress", reasonCode: "native_child_completion_pending",
@@ -150,7 +164,7 @@ describe("native status authority", () => {
      toStatus: "in_review",
      effects: [expect.objectContaining({ kind: "create_interaction" })],
    });
-    for (const kind of ["same_agent", "retry", "monitor"] as const) {
+    for (const kind of ["same_agent", "retry"] as const) {
      expect(
        arbitrate({
          assessment: {
@@ -562,7 +576,7 @@ describe("native status authority", () => {
      expect.objectContaining({
        statusAction: "blocked",
        toStatus: "blocked",
-        policyVersion: "phase6-v10",
+        policyVersion: "phase6-v11",
        reasonCode: "current_track_blocker_waiting",
        unblockDescriptor: {
          owner: "board",
@@ -1,7 +1,7 @@
 import type { NativeEvidenceAssessment } from "./evidence-classifier.js";
 import { NATIVE_MODEL_REJECTION_MESSAGE, NATIVE_PROVIDER_CAPACITY_MAX_RETRIES, NATIVE_PROVIDER_OVERLOADED_CODE, NATIVE_PROVIDER_OVERLOADED_MESSAGE } from "./native-provider-failure.js";

-export const NATIVE_STATUS_ARBITER_POLICY_VERSION = "phase6-v10";
+export const NATIVE_STATUS_ARBITER_POLICY_VERSION = "phase6-v11";

 export type NativeAuthoritativeIssueStatus =
  | "backlog"
@@ -113,6 +113,7 @@ export function arbitrateNativeStatus(input: {
    "authorized" | "revoked" | "not_applicable";
  /** Server-verified accepted plan request and normal provider terminal. */
  planWaitAuthorized?: boolean;
+  monitorWaitAuthorized?: boolean;
  boardResponseWaitAuthorized?: boolean;
  boardResponseWaitOrigin?: boolean;
  isConversation?: boolean;
@@ -325,7 +326,7 @@ export function arbitrateNativeStatus(input: {
    input.assessment.hasBlockingRemainingWork;
  if (
    input.hasUnresolvedIssueBlockers === true &&
-    (["done", "blocked"].includes(input.assessment.reportedDisposition) || unfinishedResponseWait)
+    (["done", "blocked"].includes(input.assessment.reportedDisposition) || unfinishedResponseWait || input.assessment.continuation?.kind === "monitor")
  ) {
    const owner = input.assessment.blocker?.boardOwned
      ? ("board" as const)
@@ -461,6 +462,25 @@ export function arbitrateNativeStatus(input: {
      ],
    };
  }
+  if (input.assessment.reportedDisposition === "yielded" && input.assessment.continuation?.kind === "monitor") {
+    if (input.monitorWaitAuthorized) {
+      return {
+        policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION,
+        statusAction: "preserve", toStatus: input.priorIssueStatus,
+        reasonCode: "scheduled_monitor_waiting", unblockDescriptor: null,
+        // The persisted timer owns delivery, never enqueue an immediate wake.
+        effects: [{ kind: "release_checkout" }],
+      };
+    }
+    return {
+      policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION,
+      statusAction: "preserve", toStatus: input.priorIssueStatus,
+      reasonCode: "monitor_wait_authority_lost", unblockDescriptor: null,
+      effects: [{ kind: "record_finalization_error", cause: "monitor_wait_authority_lost",
+        nextAction: "The scheduled monitor was cleared or became ineligible. Re-establish a valid task action path.",
+        agentId: input.agentId }],
+    };
+  }
  if (
    input.assessment.reportedDisposition === "yielded" &&
    input.assessment.continuation?.kind === "response_wake" &&
@@ -1,3 +1,4 @@
+import { eligibleIssueMonitorWait } from "../issue-monitors.js";
 import { readNativePlanWait, type NativePlanWaitSource } from "./native-plan-wait.js";
 import { activeIssueInteractionCondition } from "../issue-question-context.js";
 import { isConversation } from "../agent-conversations.js";
@@ -732,15 +733,13 @@ async function materializeDecisionEffect(input: {
          },
        };
    }
+    if (effect.continuationKind === "monitor") throw new Error("Monitor delivery belongs to the issue monitor scheduler");
    const wakeId = await enqueueWake({
      tx: input.tx,
      companyId: input.companyId,
      issueId: input.issue.id,
      agentId: effect.agentId,
-      reason:
-        effect.continuationKind === "monitor"
-          ? "monitor_due"
-          : "issue_status_changed",
+      reason: "issue_status_changed",
      idempotencyKey: `native-status:${input.decisionId}:continuation`,
      payload: {
        nativeDecisionId: input.decisionId,
@@ -1556,6 +1555,7 @@ export async function commitNativeStatusDecision(input: {
  supersedesCommittedDecisionId?: string;
  requireExternalChatResponseWaitAuthorization?: { agentId: string };
  requirePlanWaitSource?: NativePlanWaitSource;
+  requireMonitorWait?: { agentId: string; nextCheckAt: string };
  requireNoPendingChildCompletion?: NativeChildCompletionRecipient;
  requireModelRejectionOwner?: { agentId: string; reviewContext: NativeReviewAssignmentContext | null };
  requireProviderFailureOwner?: { agentId: string; reviewContext: NativeReviewAssignmentContext | null };
@@ -1646,6 +1646,15 @@ export async function commitNativeStatusDecision(input: {
    ) {
      throw new NativeStatusRaceError();
    }
+    if (reasonCode === "scheduled_monitor_waiting") {
+      const expected = input.requireMonitorWait;
+      if (!expected || issue.workMode !== "standard" || issue.executionRunId !== input.runId ||
+          input.decision.statusAction !== "preserve" || input.decision.effects.length !== 1 ||
+          input.decision.effects[0]?.kind !== "release_checkout" ||
+          eligibleIssueMonitorWait(issue, expected.agentId) !== expected.nextCheckAt) {
+        throw new NativeStatusRaceError();
+      }
+    }
    // Recheck under the parent status lock: a child can finish between the
    // finalizer's read and this commit. Re-arbitrate instead of discarding its wake.
    if (input.requireNoPendingChildCompletion && await hasPendingNativeChildCompletion(
@@ -0,0 +1,204 @@
+import { randomUUID } from "node:crypto";
+import { and, eq } from "drizzle-orm";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { activityLog, agents, authUsers, companyMemberships, companies, createDb, heartbeatRuns, issues } from "@paperclipai/db";
+import { PROVIDER_QUOTA_MONITOR_SERVICE_NAME } from "@paperclipai/shared";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
+import { PaperclipRunnerToolAuthority } from "./paperclip-runner-tool-authority.js";
+import { buildIssueMonitorTriggeredPatch, normalizeIssueExecutionPolicy } from "../issue-execution-policy.js";
+import { nativeCompletionFeedback } from "./native-completion-feedback.js";
+import type { PrpStructuredRunResult } from "../../vendor/paperclip-runner/index.js";
+
+const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
+const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
+if (!embeddedPostgresSupport.supported) {
+  console.warn(`Skipping embedded Postgres native task monitor tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`);
+}
+
+describeEmbeddedPostgres("native task monitors", () => {
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let db: ReturnType<typeof createDb>;
+  beforeAll(async () => {
+    temporary = await startEmbeddedPostgresTestDatabase("native-task-monitor-");
+    db = createDb(temporary.connectionString);
+  });
+  afterAll(async () => { await temporary?.cleanup(); });
+
+  async function fixture() {
+    const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Monitors", issuePrefix: `M${companyId.slice(0, 7)}` });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Worker", adapterType: "paperclip_runner", status: "active" });
+    await db.insert(issues).values({ id: issueId, companyId, title: "Deferred check", status: "in_progress", assigneeAgentId: agentId });
+    await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, runtimeMode: "native", status: "running", contextSnapshot: { issueId } });
+    await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId));
+    const binding = { companyId, agentId, issueId, runId };
+    const authority = new PaperclipRunnerToolAuthority(db, binding);
+    const monitor = { nextCheckAt: new Date(Date.now() + 60_000).toISOString(), notes: "Compare this task's run and session identities." };
+    const call = (args: Record<string, unknown>) => authority.execute({ tool: "set_task_monitor", callId: randomUUID(), arguments: args });
+    const read = async () => (await db.select().from(issues).where(eq(issues.id, issueId)))[0]!;
+    return { ...binding, binding, authority, monitor, call, read };
+  }
+
+  it("advertises a real binding only in standard execution", async () => {
+    const value = await fixture();
+    expect(value.authority.definitions().some(tool => tool.name === "set_task_monitor")).toBe(true);
+    for (const workMode of ["planning", "ask"] as const) {
+      const authority = new PaperclipRunnerToolAuthority(db, { ...value.binding, workMode });
+      expect(authority.definitions().some(tool => tool.name === "set_task_monitor")).toBe(false);
+      await db.update(issues).set({ workMode }).where(eq(issues.id, value.issueId));
+      await expect(authority.execute({ tool: "set_task_monitor", callId: randomUUID(), arguments: { idempotencyKey: "forbidden", monitor: value.monitor } })).rejects.toThrow();
+    }
+  });
+
+  it("keeps provider-neutral discovery and review-only restrictions", async () => {
+    const value = await fixture();
+    for (const provider of ["codex", "opencode", "claude_managed", "aws_agentcore", "acpx"]) {
+      await db.update(agents).set({ adapterConfig: { provider } }).where(eq(agents.id, value.agentId));
+      expect(value.authority.definitions().map(tool => tool.name)).toContain("set_task_monitor");
+    }
+    const authority = new PaperclipRunnerToolAuthority(db, { ...value.binding,
+      nativeReview: { nativeReviewInteractionId: randomUUID(), nativeReviewDecisionId: randomUUID() } });
+    expect(authority.definitions().map(tool => tool.name)).not.toContain("set_task_monitor");
+    await expect(authority.execute({ tool: "set_task_monitor", callId: randomUUID(), arguments: { idempotencyKey: "review", monitor: value.monitor } })).rejects.toThrow("review run");
+  });
+
+  it("rechecks the responsible user's permissions before mutation and replay", async () => {
+    const value = await fixture(), userId = randomUUID();
+    await db.insert(authUsers).values({ id: userId, name: "Owner", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() });
+    await db.insert(companyMemberships).values({ companyId: value.companyId, principalType: "user", principalId: userId, membershipRole: "owner" });
+    await db.update(heartbeatRuns).set({ responsibleUserId: userId }).where(eq(heartbeatRuns.id, value.runId));
+    const original = { idempotencyKey: "authorized", monitor: value.monitor };
+    await value.call(original);
+    await db.update(companyMemberships).set({ membershipRole: "viewer" }).where(eq(companyMemberships.companyId, value.companyId));
+    await expect(value.call(original)).rejects.toThrow();
+    await expect(value.call({ idempotencyKey: "clear", monitor: null })).rejects.toThrow();
+    expect((await value.read()).monitorNextCheckAt?.toISOString()).toBe(value.monitor.nextCheckAt);
+  });
+
+  it("persists a monitor and its audit atomically while preserving review policy", async () => {
+    const value = await fixture();
+    const policy = { mode: "normal", commentRequired: false, stages: [{ id: randomUUID(), type: "review", approvalsNeeded: 1, participants: [{ type: "user", userId: "reviewer" }] }] };
+    await db.update(issues).set({ executionPolicy: policy }).where(eq(issues.id, value.issueId));
+    const result = await value.call({ idempotencyKey: "schedule", monitor: value.monitor });
+    expect(result).toMatchObject({ taskId: value.issueId, monitor: { ...value.monitor, scheduledBy: "assignee", status: "scheduled" }, replayed: false });
+    const saved = await value.read();
+    expect(saved.executionPolicy).toMatchObject(policy);
+    expect(saved.monitorNextCheckAt?.toISOString()).toBe(value.monitor.nextCheckAt);
+    expect(saved.status).toBe("in_progress");
+    const audits = await db.select().from(activityLog).where(and(eq(activityLog.entityId, value.issueId), eq(activityLog.action, "issue.monitor_scheduled")));
+    expect(audits).toHaveLength(1);
+    expect(audits[0]).toMatchObject({ runId: value.runId, agentId: value.agentId });
+  });
+
+  it("replaces and clears schedules without re-arming a receipt on replay", async () => {
+    const value = await fixture();
+    const original = { idempotencyKey: "first", monitor: value.monitor };
+    await value.call(original);
+    const replacement = { ...value.monitor, nextCheckAt: new Date(Date.now() + 120_000).toISOString() };
+    await value.call({ idempotencyKey: "replace", monitor: replacement });
+    expect(await value.call(original)).toMatchObject({ monitor: { nextCheckAt: replacement.nextCheckAt }, replayed: true });
+    await value.call({ idempotencyKey: "clear", monitor: null });
+    expect(await value.call(original)).toMatchObject({ monitor: { nextCheckAt: null, status: "cleared" }, replayed: true });
+    expect((await value.read()).monitorNextCheckAt).toBeNull();
+    await expect(value.call({ ...original, monitor: replacement })).rejects.toThrow("idempotency_conflict");
+    const audits = await db.select().from(activityLog).where(eq(activityLog.entityId, value.issueId));
+    expect(audits.filter(row => row.action.startsWith("issue.monitor_"))).toHaveLength(3);
+  });
+
+  it("exposes consumed monitor instructions to a resumed task and does not re-arm them on retry", async () => {
+    const value = await fixture();
+    const input = { idempotencyKey: "consumed", monitor: value.monitor };
+    await value.call(input);
+    const scheduled = await value.read();
+    await db.update(issues).set(buildIssueMonitorTriggeredPatch({ issue: scheduled,
+      policy: normalizeIssueExecutionPolicy(scheduled.executionPolicy), triggeredAt: new Date() })).where(eq(issues.id, value.issueId));
+    expect(await value.authority.execute({ tool: "get_task_context", callId: randomUUID(), arguments: {} }))
+      .toMatchObject({ activeTask: { monitor: { status: "triggered", nextCheckAt: null, notes: value.monitor.notes, attemptCount: 1 } } });
+    expect(await value.call(input)).toMatchObject({ replayed: true, monitor: { nextCheckAt: null, status: "triggered" } });
+    expect((await value.read()).monitorNextCheckAt).toBeNull();
+  });
+
+  it("keeps a cleared schedule cleared when a successor run retries the original key", async () => {
+    const value = await fixture();
+    const original = { idempotencyKey: "durable-schedule", monitor: value.monitor };
+    await value.call(original);
+    await value.call({ idempotencyKey: "durable-clear", monitor: null });
+    await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, value.runId));
+    const nextRunId = randomUUID();
+    await db.insert(heartbeatRuns).values({ id: nextRunId, companyId: value.companyId, agentId: value.agentId,
+      nativeIssueId: value.issueId, runtimeMode: "native", status: "running", contextSnapshot: { issueId: value.issueId } });
+    await db.update(issues).set({ executionRunId: nextRunId }).where(eq(issues.id, value.issueId));
+    const authority = new PaperclipRunnerToolAuthority(db, { ...value.binding, runId: nextRunId });
+    expect(await authority.execute({ tool: "set_task_monitor", callId: randomUUID(), arguments: original }))
+      .toMatchObject({ replayed: true, monitor: { nextCheckAt: null, status: "cleared" } });
+    expect((await value.read()).monitorNextCheckAt).toBeNull();
+    await expect(value.call(original)).rejects.toThrow();
+  });
+
+  it("allows another owned task but rejects foreign ownership and company scope", async () => {
+    const value = await fixture(), other = await fixture(), targetId = randomUUID();
+    await db.insert(issues).values({ id: targetId, companyId: value.companyId, title: "Another owned check", status: "in_review", assigneeAgentId: value.agentId });
+    expect(await value.call({ taskId: targetId, idempotencyKey: "other", monitor: value.monitor })).toMatchObject({ taskId: targetId });
+    await expect(value.call({ taskId: other.issueId, idempotencyKey: "foreign", monitor: value.monitor })).rejects.toThrow("Task not found");
+    const otherAgent = randomUUID();
+    await db.insert(agents).values({ id: otherAgent, companyId: value.companyId, name: "Other", adapterType: "paperclip_runner" });
+    await db.update(issues).set({ assigneeAgentId: otherAgent }).where(eq(issues.id, targetId));
+    await expect(value.call({ taskId: targetId, idempotencyKey: "other", monitor: value.monitor })).rejects.toThrow();
+  });
+
+  it("rejects invalid schedules, exhausted bounds and ineligible tasks without writes", async () => {
+    const value = await fixture();
+    for (const monitor of [
+      { ...value.monitor, notes: "" },
+      { ...value.monitor, nextCheckAt: new Date(0).toISOString() },
+      { ...value.monitor, scheduledBy: "board" },
+      { ...value.monitor, timeoutAt: value.monitor.nextCheckAt },
+      { ...value.monitor, maxAttempts: 101 },
+    ]) await expect(value.call({ idempotencyKey: randomUUID(), monitor })).rejects.toThrow();
+    await db.update(issues).set({ monitorAttemptCount: 1 }).where(eq(issues.id, value.issueId));
+    await expect(value.call({ idempotencyKey: "exhausted", monitor: { ...value.monitor, maxAttempts: 1 } })).rejects.toThrow();
+    for (const status of ["blocked", "backlog", "done", "cancelled"] as const) {
+      await db.update(issues).set({ status }).where(eq(issues.id, value.issueId));
+      await expect(value.call({ idempotencyKey: randomUUID(), monitor: value.monitor })).rejects.toThrow();
+    }
+    expect((await value.read()).monitorNextCheckAt).toBeNull();
+  });
+
+  it("accepts unfinished monitor waits only after scheduling the current task", async () => {
+    const value = await fixture();
+    const result: PrpStructuredRunResult = {
+      schema: "paperclip.run_result.v1", reportedWorkDisposition: "yielded", summary: "I will compare the next run.",
+      completionClaim: { contractRevision: "test", objectiveSatisfied: false, criteria: [], remainingWork: [{ description: "Compare next run", blocksCompletion: true }] },
+      continuation: { kind: "monitor", summary: "Wait for the scheduled check", idempotencyKey: "wait" },
+      evidence: [], verification: [], attentionRequests: [], artifacts: [],
+    };
+    await expect(nativeCompletionFeedback(db, value.runId, result)).rejects.toThrow("persisted, eligible monitor");
+    const otherId = randomUUID();
+    await db.insert(issues).values({ id: otherId, companyId: value.companyId, title: "Other", status: "in_progress", assigneeAgentId: value.agentId });
+    await value.call({ taskId: otherId, idempotencyKey: "other", monitor: value.monitor });
+    await expect(nativeCompletionFeedback(db, value.runId, result)).rejects.toThrow("persisted, eligible monitor");
+    await value.call({ idempotencyKey: "self", monitor: value.monitor });
+    await expect(nativeCompletionFeedback(db, value.runId, result)).resolves.toContain("issue_monitor_due");
+    // Legacy APIs may persist a server-owned quota monitor. It cannot justify
+    // a native wait because its dispatcher explicitly excludes native runs.
+    await db.update(issues).set({ executionPolicy: { monitor: { ...value.monitor, serviceName: PROVIDER_QUOTA_MONITOR_SERVICE_NAME } } }).where(eq(issues.id, value.issueId));
+    await expect(nativeCompletionFeedback(db, value.runId, result)).rejects.toThrow("persisted, eligible monitor");
+    await value.call({ idempotencyKey: "clear", monitor: null });
+    await expect(nativeCompletionFeedback(db, value.runId, result)).rejects.toThrow("persisted, eligible monitor");
+  });
+
+  it("rejects the server-owned quota recovery name without promising a wake", async () => {
+    const value = await fixture();
+    await expect(value.call({ idempotencyKey: "quota", monitor: {
+      ...value.monitor, serviceName: PROVIDER_QUOTA_MONITOR_SERVICE_NAME, externalRef: value.runId,
+    } })).rejects.toThrow("reserved for server-owned quota recovery");
+    expect((await value.read()).monitorNextCheckAt).toBeNull();
+    const audits = await db.select().from(activityLog).where(eq(activityLog.entityId, value.issueId));
+    expect(audits).toHaveLength(0);
+    const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, value.runId));
+    expect(run.resultJson?.semanticToolReceipts).toBeUndefined();
+    await expect(value.call({ idempotencyKey: "quota", monitor: {
+      ...value.monitor, serviceName: "Provider usage dashboard",
+    } })).resolves.toMatchObject({ monitor: { status: "scheduled" }, replayed: false });
+  });
+});