refactor(server): simplify the wake-queue module (#13139)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The server uses a deferred wake queue to release work at the correct time > - The wake queue had repeated decisions, helpers, queries, and recovery data > - This repetition made the module harder to read and left a drain invariant implicit > - This pull request moves pure decisions into the policy layer and simplifies the queue flow > - The benefit is a smaller, clearer module with the same behavior and direct test coverage ## Linked Issues or Issue Description Refs #13136 **Problem:** The deferred wake queue carried repeated logic across the application and database layers. **Expected behavior:** The queue keeps the same wake, release, recovery, and escalation behavior after the refactor. **Solution:** Move pure decisions into the policy layer, share repeated data and predicates, and state the drain invariant in the application layer. ## What Changed - Remove the unused database port method and carry the blocked release notice kind as a typed field. - Move four pre-drain decisions into pure policy functions with table-driven tests. - Resolve the responsible user once in the application layer. - Use the canonical helper for agent invokability checks. - Share string helpers and run predicates across the module. - Split the deferred-wake decision flow and share recovery facts with a discriminator. - Bound the drain loop and throw when it processes a wake identifier twice. - Rename queue ports to describe their behavior. - Share row-loading code between stranded-issue escalation adapters. - Restore the interaction-continuation integration test case. ## Verification - Run the three wake-queue module test files. They pass 61 cases locally. - Run `pnpm check:module-boundaries`. It passes locally. - Run the `server/` type-check and confirm that no error names the changed wake-queue module or `heartbeat.ts`. - Run continuous integration and confirm that `promotes an interaction continuation after removing a coalesced self-authored comment` passes. ## Risks The change refactors queue control flow and database adapter boundaries. The main risk is a behavior change in deferred wake release or recovery. The new policy tests and the restored integration case cover these paths. The local environment cannot run the integration test because the same dependency failure occurs on the base branch. ## Model Used OpenAI Codex, GPT-5, context window not exposed by the runtime, with reasoning, tool use, and code execution. ## 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-24 18:19:05 +02:00 · 2026-09-10 00:03:22 -07:00
1 parent 6dd48cad43
commit 0bff1d5cb6
@@ -11,7 +11,7 @@ import {
  nativeRunFinalizations,
 } from "@paperclipai/db";
 import { legacyExecutionNeedsReconciliation } from "../../../services/legacy-execution-recovery.js";
-import { evaluateAgentInvokability } from "../../../services/agent-invokability.js";
+import { evaluateAgentInvokabilityFromDb } from "../../../services/agent-invokability.js";
 import { issueTreeControlService, isVerifiedIssueTreeControlInteractionWake } from "../../../services/issue-tree-control.js";
 import { isAutomaticRecoverySuppressedByPauseHold } from "../../../services/recovery/pause-hold-guard.js";
 import { issueService } from "../../../services/issues.js";
@@ -19,18 +19,20 @@ import { issueRecoveryActionService } from "../../../services/issue-recovery-act
 import { readContinuationAttempt } from "../../../services/recovery/run-liveness-continuations.js";
 import { withRecoveryContext } from "../../../services/recovery/status-only-context.js";
 import { parseIssueExecutionState } from "../../../services/issue-execution-policy.js";
-import {
-  buildConfigurationIncompleteRecoveryNoticeSeed,
-  buildExecutionReviewParticipantRecoveryNoticeSeed,
-  buildImmediateExecutionPathRecoveryNoticeSeed,
-  buildWorkspaceValidationRecoveryNoticeSeed,
-} from "../../../services/recovery/stranded-notice.js";
 import {
  queuedCommentIdsFromWakePayload,
  withQueuedCommentIdsInWakePayload,
 } from "../../../services/issue-queued-comment-queue.js";
 import { extractWakeCommentIds } from "../../run-dispatch/index.js";
 import { hasInteractionContinuationWakeContext } from "../domain/context.js";
+import { decidePreDrain, type PreDrainFacts } from "../domain/policy.js";
+import {
+  EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON,
+  isConfigurationIncompleteFailedRun,
+  isWorkspaceValidationFailedRun,
+  parseObject,
+  readNonEmptyString,
+} from "../domain/values.js";
 import type {
  DeferredWakeCandidate,
  InvokableAgentSnapshot,
@@ -39,48 +41,28 @@ import type {
  LockedIssueExecution,
  ReleaseTransactionResult,
  RunSnapshot,
-  WakeQueueReader,
-  WakeQueueWriter,
+  WakeQueueHost,
+  WakeQueueTransaction,
 } from "../application/ports.js";
 import type { RunSummary } from "../application/types.js";
-import { WakeQueueApplicationError } from "../application/types.js";

 const DEFERRED_WAKE_STATUS = "deferred_issue_execution";
 const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext";
-const WORKSPACE_VALIDATION_FAILURE_CODE = "workspace_validation_failed";
-const CONFIGURATION_INCOMPLETE_FAILURE_CODE = "configuration_incomplete";
-const WORKSPACE_VALIDATION_RECOVERY_CAUSE = "workspace_validation_failed";
-const CONFIGURATION_INCOMPLETE_RECOVERY_CAUSE = "configuration_incomplete";
-const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_CAUSE = "execution_review_participant_recovery";
-const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON = "execution_review_participant_recovery";
-const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON = "execution_review_participant_recovery";
 const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const;

 type HeartbeatRunRow = typeof heartbeatRuns.$inferSelect;
 type IssueRow = typeof issues.$inferSelect;

-function parseObject(value: unknown): Record<string, unknown> {
-  return typeof value === "object" && value !== null && !Array.isArray(value)
-    ? (value as Record<string, unknown>)
-    : {};
-}
-
-function readNonEmptyString(value: unknown): string | null {
-  return typeof value === "string" && value.trim().length > 0 ? value : null;
-}
-
 function normalizeAgentNameKey(value: string | null | undefined): string | null {
  if (typeof value !== "string") return null;
  const normalized = value.trim().toLowerCase();
  return normalized.length > 0 ? normalized : null;
 }

-function isWorkspaceValidationFailedRun(run: Pick<HeartbeatRunRow, "errorCode">): boolean {
-  return run.errorCode === WORKSPACE_VALIDATION_FAILURE_CODE;
-}
-
-function isConfigurationIncompleteFailedRun(run: Pick<HeartbeatRunRow, "errorCode">): boolean {
-  return run.errorCode === CONFIGURATION_INCOMPLETE_FAILURE_CODE || run.errorCode === "model_not_found";
+function toRequestedByActorType(value: string | null): "user" | "agent" | "system" | null {
+  // The database column is free text; map any value outside the union to
+  // null instead of widening the type back to string.
+  return value === "user" || value === "agent" || value === "system" ? value : null;
 }

 function toRunSnapshot(row: HeartbeatRunRow): RunSnapshot {
@@ -152,7 +134,7 @@ function toDeferredWakeCandidate(row: typeof agentWakeupRequests.$inferSelect):
    reason: row.reason,
    source: row.source,
    triggerDetail: row.triggerDetail,
-    requestedByActorType: row.requestedByActorType,
+    requestedByActorType: toRequestedByActorType(row.requestedByActorType),
    requestedByActorId: row.requestedByActorId,
    payload,
    queuedCommentIds,
@@ -164,12 +146,23 @@ function toDeferredWakeCandidate(row: typeof agentWakeupRequests.$inferSelect):
 }

 export type WakeQueuePostgresAdapterDeps = {
-  resolveResponsibleUserId: WakeQueueReader["resolveResponsibleUserId"];
-  getRoutineEnv: WakeQueueReader["getRoutineEnv"];
-  resolveSessionBeforeForWakeup: WakeQueueReader["resolveSessionBeforeForWakeup"];
+  resolveResponsibleUserId: WakeQueueHost["resolveResponsibleUserId"];
+  getRoutineEnv: WakeQueueHost["getRoutineEnv"];
+  resolveSessionBeforeForWakeup: WakeQueueHost["resolveSessionBeforeForWakeup"];
 };

-function buildReader(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueReader {
+function buildHost(_tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueHost {
+  return {
+    resolveResponsibleUserId: deps.resolveResponsibleUserId,
+    getRoutineEnv: deps.getRoutineEnv,
+    resolveSessionBeforeForWakeup: deps.resolveSessionBeforeForWakeup,
+  };
+}
+
+function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueTransaction {
+  const treeControlSvc = issueTreeControlService(tx);
+  const issuesSvc = issueService(tx);
+
  return {
    async findInvokableAgent({ companyId, agentId }): Promise<InvokableAgentSnapshot | null> {
      const agent = await tx
@@ -178,25 +171,11 @@ function buildReader(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueReade
        .where(and(eq(agents.id, agentId), eq(agents.companyId, companyId)))
        .then((rows) => rows[0] ?? null);
      if (!agent) return null;
-      const companyAgents = await tx
-        .select({ id: agents.id, companyId: agents.companyId, name: agents.name, reportsTo: agents.reportsTo, status: agents.status })
-        .from(agents)
-        .where(eq(agents.companyId, companyId));
-      const invokability = evaluateAgentInvokability(agent, companyAgents);
+      const invokability = await evaluateAgentInvokabilityFromDb(tx, agent);
      return { id: agent.id, companyId: agent.companyId, name: agent.name, invokable: invokability.invokable };
    },
-    resolveResponsibleUserId: deps.resolveResponsibleUserId,
-    getRoutineEnv: deps.getRoutineEnv,
-    resolveSessionBeforeForWakeup: deps.resolveSessionBeforeForWakeup,
-  };
-}

-function buildWriter(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueWriter {
-  const treeControlSvc = issueTreeControlService(tx);
-  const issuesSvc = issueService(tx);
-
-  return {
-    async claimNextDeferredWake({ companyId, issueId }) {
+    async findNextDeferredWake({ companyId, issueId }) {
      const row = await tx
        .select()
        .from(agentWakeupRequests)
@@ -446,25 +425,6 @@ function buildWriter(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueWrite
      return isAutomaticRecoverySuppressedByPauseHold(tx, companyId, issueId, treeControlSvc);
    },

-    async buildBlockedRecoveryNotice({ noticeKind, issueStatus, finishingRun }) {
-      if (noticeKind === "workspace_validation") {
-        return { notice: buildWorkspaceValidationRecoveryNoticeSeed(), recoveryCause: WORKSPACE_VALIDATION_RECOVERY_CAUSE };
-      }
-      if (noticeKind === "configuration_incomplete") {
-        return {
-          notice: buildConfigurationIncompleteRecoveryNoticeSeed(finishingRun.configurationIncompletePayload),
-          recoveryCause: CONFIGURATION_INCOMPLETE_RECOVERY_CAUSE,
-        };
-      }
-      if (noticeKind === "execution_review_participant") {
-        return {
-          notice: buildExecutionReviewParticipantRecoveryNoticeSeed(),
-          recoveryCause: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_CAUSE,
-        };
-      }
-      return { notice: buildImmediateExecutionPathRecoveryNoticeSeed({ status: issueStatus }), recoveryCause: null };
-    },
-
    async queueReviewParticipantRecoveryRun({ companyId, issue, finishingRun, recoveryAgent, sessionBefore, now }) {
      const executionState = parseIssueExecutionState(issue.executionState);
      const wakeupRequest = await tx
@@ -474,7 +434,7 @@ function buildWriter(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueWrite
          agentId: recoveryAgent.id,
          source: "automation",
          triggerDetail: "system",
-          reason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON,
+          reason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON,
          payload: withRecoveryContext(
            {
              issueId: issue.id,
@@ -506,7 +466,7 @@ function buildWriter(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueWrite
            {
              issueId: issue.id,
              taskId: issue.id,
-              wakeReason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON,
+              wakeReason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON,
              retryReason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON,
              source: "issue.execution_review_recovery",
              retryOfRunId: finishingRun.id,
@@ -542,48 +502,17 @@ function buildWriter(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueWrite
      return toRunSummary(queuedRun);
    },

-    async queueImmediateRecoveryRun({ companyId, issue, finishingRun, recoveryAgent, sessionBefore, now }) {
-      const retryReason = issue.status === "todo" ? "assignment_recovery" : "issue_continuation_needed";
-      const recoveryReason = issue.status === "todo" ? "issue_assignment_recovery" : "issue_continuation_needed";
-      const recoverySource = issue.status === "todo" ? "issue.assignment_recovery" : "issue.continuation_recovery";
-      const recoveryContextSnapshot = withRecoveryContext(
-        {
-          issueId: issue.id,
-          taskId: issue.id,
-          wakeReason: recoveryReason,
-          retryReason,
-          source: recoverySource,
-          retryOfRunId: finishingRun.id,
-        },
-        "normal_model",
-      );
-
-      const routineEnvContext = await deps.getRoutineEnv({ companyId, issue });
-      const responsibleUserId = await deps.resolveResponsibleUserId({
-        companyId,
-        contextSnapshot: recoveryContextSnapshot,
-        issue,
-        routineEnvContext,
-        requestedByActorType: "system",
-        requestedByActorId: null,
-        source: "automation",
-        triggerDetail: "system",
-        existingRunResponsibleUserId: finishingRun.responsibleUserId,
-      });
-      if (!responsibleUserId) {
-        throw new WakeQueueApplicationError(
-          "responsible_user_unresolved",
-          "Unable to resolve responsible user for recovery heartbeat run",
-          {
-            runId: finishingRun.id,
-            agentId: recoveryAgent.id,
-            companyId,
-            issueId: issue.id,
-            wakeReason: recoveryReason,
-          },
-        );
-      }
-
+    async queueImmediateRecoveryRun({
+      companyId,
+      issue,
+      finishingRun,
+      recoveryAgent,
+      reason,
+      contextSnapshot,
+      responsibleUserId,
+      sessionBefore,
+      now,
+    }) {
      const wakeupRequest = await tx
        .insert(agentWakeupRequests)
        .values({
@@ -591,7 +520,7 @@ function buildWriter(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueWrite
          agentId: recoveryAgent.id,
          source: "automation",
          triggerDetail: "system",
-          reason: recoveryReason,
+          reason,
          payload: withRecoveryContext({ issueId: issue.id, retryOfRunId: finishingRun.id }, "normal_model"),
          status: "queued",
          requestedByActorType: "system",
@@ -610,7 +539,7 @@ function buildWriter(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueWrite
          triggerDetail: "system",
          status: "queued",
          wakeupRequestId: wakeupRequest.id,
-          contextSnapshot: recoveryContextSnapshot,
+          contextSnapshot,
          responsibleUserId,
          sessionIdBefore: sessionBefore,
          retryOfRunId: finishingRun.id,
@@ -767,49 +696,51 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd
        const issueRow =
          (contextIssueId ? candidateIssues.find((candidate) => candidate.id === contextIssueId) : candidateIssues[0]) ?? null;

-        if (!issueRow || (issueRow.executionRunId && issueRow.executionRunId !== run.id)) {
+        const preDrainFacts: PreDrainFacts = {
+          issueRowPresent: issueRow !== null,
+          executionRunIdMatchesRun: !issueRow || !issueRow.executionRunId || issueRow.executionRunId === run.id,
+          isWorkspaceValidationFailedRun: isWorkspaceValidationFailedRun(run),
+          isConfigurationIncompleteFailedRun: isConfigurationIncompleteFailedRun(run),
+          issueStatus: issueRow?.status ?? "",
+          hasAssigneeUser: Boolean(issueRow?.assigneeUserId),
+          assigneeAgentMatchesRunAgent: issueRow?.assigneeAgentId === run.agentId,
+          legacyExecutionNeedsReconciliation: legacyExecutionNeedsReconciliation(run),
+          // An operator stop never promotes old queued work by itself. The
+          // next explicit wake adopts those messages atomically when it
+          // queues a run.
+          executionCancellationAcknowledged:
+            run.status === "cancelled" && parseObject(run.resultJson?.executionCancellation).state === "acknowledged",
+        };
+        const preDrain = decidePreDrain(preDrainFacts);
+
+        if (preDrain.kind === "released") {
          return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot };
        }

-        if (
-          (isWorkspaceValidationFailedRun(run) || isConfigurationIncompleteFailedRun(run)) &&
-          (issueRow.status === "todo" || issueRow.status === "in_progress") &&
-          !issueRow.assigneeUserId &&
-          issueRow.assigneeAgentId === run.agentId
-        ) {
-          const configurationIncomplete = isConfigurationIncompleteFailedRun(run);
-          const notice = configurationIncomplete
-            ? buildConfigurationIncompleteRecoveryNoticeSeed(runSnapshot.configurationIncompletePayload)
-            : buildWorkspaceValidationRecoveryNoticeSeed();
+        // decidePreDrain only returns "blocked" or "proceed" when the issue row is present.
+        if (!issueRow) {
+          throw new Error(`wake-queue: pre-drain decision ${preDrain.kind} reached without an issue row`);
+        }
+
+        if (preDrain.kind === "blocked") {
          return {
            outcome: {
              kind: "blocked",
              issue: toIssueSnapshot(issueRow),
              previousStatus: issueRow.status as "todo" | "in_progress",
-              notice,
-              recoveryCause: configurationIncomplete ? CONFIGURATION_INCOMPLETE_RECOVERY_CAUSE : WORKSPACE_VALIDATION_RECOVERY_CAUSE,
+              noticeKind: preDrain.noticeKind,
            },
            postCommitEffects: [],
            run: runSnapshot,
          };
        }

-        if (legacyExecutionNeedsReconciliation(run)) {
-          return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot };
-        }
-
-        // An operator stop never promotes old queued work by itself. The next
-        // explicit wake adopts those messages atomically when it queues a run.
-        if (run.status === "cancelled" && parseObject(run.resultJson?.executionCancellation).state === "acknowledged") {
-          return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot };
-        }
-
        if (await recordNativeTerminalRecoveryIfNeeded(tx, run, issueRow, input.now)) {
          return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot };
        }

        const locked: LockedIssueExecution = { primaryIssue: toIssueSnapshot(issueRow), run: runSnapshot };
-        const result = await fn(locked, { reader: buildReader(tx, deps), writer: buildWriter(tx, deps) });
+        const result = await fn(locked, { host: buildHost(tx, deps), transaction: buildTransaction(tx, deps) });
        return { ...result, run: runSnapshot };
      });
    },