PaperClipAI/server/src/services/execution-control-reconciliation.ts

import { notifyNativeEnvironmentSettlement } from "./native-environment-settlement-notifications.js";
import { notifyExternalAgentSettlement } from "../modules/external-agents/index.js";
import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
import { notifyNativeFinalizationWork } from "./native-finalization-notifications.js";
import { notifyUnsafeWorkspaceExportRunChange } from "./unsafe-workspace-export-notifications.js";
import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js";
import { notifyInstructionCollection } from "./instruction-collection-notifications.js";
import { notifyExecutionRunWork } from "./execution-control-notifications.js";
import { randomUUID } from "node:crypto";
import { logger } from "../middleware/logger.js";
import { and, asc, eq, isNotNull, lte, sql } from "drizzle-orm";
import {
  agents,
  heartbeatRuns,
  issues,
  nativeRunFinalizations,
  type Db,
} from "@paperclipai/db";
import { parseIssueExecutionState } from "./issue-execution-policy.js";
import { issueRecoveryActionService } from "./issue-recovery-actions.js";
import { hasLiveLegacyController } from "./legacy-controller-lease.js";
import { reportRunFailure } from "./run-failure-report.js";
import { getNativeReviewAssignment, readNativeReviewAssignmentContext } from "./native-runtime/native-review-participant.js";
import { preserveNativeFailedWorkspace } from "./native-runtime/native-failed-workspace-recovery.js";
import { EXECUTION_FINALIZATION_DEADLINE_EXCEEDED, EXECUTION_FINALIZATION_DEADLINE_MESSAGE,
  EXECUTION_RECONCILIATION_INTERVAL_MS } from "./execution-control-deadline.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
import { NATIVE_STARTUP_CLEANUP_PENDING, readNativeStartupContext, promoteNativeStartupCleanup } from "./native-runtime/native-startup-cleanup.js";
import { releaseAbandonedNativeOwners } from "./native-runtime/native-failed-run-owner-release.js";

/** Only newly recorded control deadlines are eligible. Upgrades never replay ambiguous historical runs. */
export async function reconcileAbandonedExecutionControl(
  db: Db,
  now = new Date(),
  options: { startupCleanup?: (scope: { companyId: string; runId: string }) => Promise<unknown> } = {},
) {
  const nativeDue = await db
    .select({
      runId: nativeRunFinalizations.runId,
      issueId: nativeRunFinalizations.issueId,
      companyId: nativeRunFinalizations.companyId,
      failureCode: nativeRunFinalizations.failureCode,
      leaseOwner: nativeRunFinalizations.leaseOwner,
      controllerGeneration: nativeRunFinalizations.controllerGeneration,
      attempt: nativeRunFinalizations.attempt,
    })
    .from(nativeRunFinalizations)
    .where(
      and(
        isNotNull(nativeRunFinalizations.controlDeadlineAt),
        lte(nativeRunFinalizations.controlDeadlineAt, now),
      ),
    )
    .orderBy(asc(nativeRunFinalizations.controlDeadlineAt), asc(nativeRunFinalizations.runId))
    .limit(50);
  const controlDue = await db
    .select({
      runId: heartbeatRuns.id,
      companyId: heartbeatRuns.companyId,
      context: heartbeatRuns.contextSnapshot,
      nativeIssueId: nativeRunFinalizations.issueId,
      failureCode: nativeRunFinalizations.failureCode,
      leaseOwner: nativeRunFinalizations.leaseOwner,
      controllerGeneration: nativeRunFinalizations.controllerGeneration,
      attempt: nativeRunFinalizations.attempt,
    })
    .from(heartbeatRuns)
    .leftJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id))
    .where(
      and(
        isNotNull(heartbeatRuns.executionControlDeadlineAt),
        lte(heartbeatRuns.executionControlDeadlineAt, now),
      ),
    )
    .orderBy(asc(heartbeatRuns.executionControlDeadlineAt), asc(heartbeatRuns.id))
    .limit(50);
  const due = [
    ...new Map(
      [
        ...nativeDue,
        ...controlDue.map((row) => ({
          ...row,
          issueId: row.nativeIssueId ?? (
            typeof row.context?.issueId === "string"
              ? row.context.issueId
              : null),
        })),
      ].map((row) => [row.runId, row]),
    ).values(),
  ];
  let surfaced = 0;
  // Bound contention latency across independent tasks: a locked task must not
  // consume the entire reconciliation window for every task behind it.
  let nextCandidate = 0;
  await Promise.all(Array.from({ length: Math.min(5, due.length) }, async () => {
    while (nextCandidate < due.length) {
      const candidate = due[nextCandidate++]!;
      try {
        if (candidate.failureCode === NATIVE_STARTUP_CLEANUP_PENDING) {
          if (!options.startupCleanup) continue;
          try {
            await options.startupCleanup({ companyId: candidate.companyId, runId: candidate.runId });
          } finally {
            await db.transaction(async tx => {
              if (candidate.issueId) await tx.select({ id: issues.id }).from(issues).where(eq(issues.id, candidate.issueId)).for("update");
              const [owner] = await tx.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, candidate.runId)).for("update");
              if (owner?.failureCode !== NATIVE_STARTUP_CLEANUP_PENDING || !owner.controlDeadlineAt || owner.controlDeadlineAt > now ||
                  owner.leaseOwner !== candidate.leaseOwner || owner.controllerGeneration !== candidate.controllerGeneration || owner.attempt !== candidate.attempt) return;
              const cleanupRetryAt = new Date(Math.max(now.getTime(), Date.now()) + EXECUTION_RECONCILIATION_INTERVAL_MS);
              await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionFinalization);
              await tx.update(nativeRunFinalizations).set({ controlDeadlineAt: cleanupRetryAt }).where(eq(nativeRunFinalizations.runId, candidate.runId));
              await tx.update(heartbeatRuns).set({ executionControlDeadlineAt: cleanupRetryAt }).where(and(eq(heartbeatRuns.id, candidate.runId),
                isNotNull(heartbeatRuns.executionControlDeadlineAt), lte(heartbeatRuns.executionControlDeadlineAt, now)));
            });
          }
          continue;
        }
        // Set inside the transaction only when the write below genuinely
        // transitions the run into "failed". Read after the transaction
        // commits, so a rolled-back write never reports a false failure.
        let terminalRunToReport: typeof heartbeatRuns.$inferSelect | null = null;
        const repaired = await db.transaction(async (tx) => {
          await tx.execute(
            sql`select set_config('statement_timeout', '15000', true), set_config('lock_timeout', '1000', true)`,
          );
          const task = candidate.issueId
            ? (
                await tx
                  .select()
                  .from(issues)
                  .where(
                    and(
                      eq(issues.id, candidate.issueId),
                      eq(issues.companyId, candidate.companyId),
                    ),
                  )
                  .for("update")
              )[0]
            : null;
          const [coordinator] = await tx
            .select()
            .from(nativeRunFinalizations)
            .where(
              and(
                eq(nativeRunFinalizations.runId, candidate.runId),
                eq(nativeRunFinalizations.companyId, candidate.companyId),
              ),
            )
            .for("update");
          const [run] = await tx
            .select()
            .from(heartbeatRuns)
            .where(
              and(
                eq(heartbeatRuns.id, candidate.runId),
                eq(heartbeatRuns.companyId, candidate.companyId),
              ),
            )
            .for("update");
          if (
            !run ||
            ![
              coordinator?.controlDeadlineAt,
              run.executionControlDeadlineAt,
            ].some((deadline) => deadline && deadline <= now)
          )
            return false;
          if (coordinator?.failureCode === NATIVE_STARTUP_CLEANUP_PENDING) return false;
          const startupContext = coordinator && readNativeStartupContext(run, coordinator);
          if (startupContext) {
            if (!await promoteNativeStartupCleanup(tx as unknown as Db, run, coordinator!, now)) {
              const retryAt = new Date(Math.max(now.getTime() + EXECUTION_RECONCILIATION_INTERVAL_MS,
                startupContext.deadlineAtMs, coordinator!.leaseExpiresAt?.getTime() ?? 0));
              await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionFinalization);
              await tx.update(nativeRunFinalizations).set({ controlDeadlineAt: retryAt }).where(eq(nativeRunFinalizations.runId, run.id));
              await tx.update(heartbeatRuns).set({ executionControlDeadlineAt: retryAt }).where(eq(heartbeatRuns.id, run.id));
            }
            return false;
          }
          if (coordinator && !coordinator.resultId && run.status === "running") {
            const nativeOwnerLeaseAt = coordinator.leaseOwner && coordinator.leaseExpiresAt && coordinator.leaseExpiresAt > now
              ? coordinator.leaseExpiresAt : null;
            const sameRunResumeScheduled = coordinator.phase === "retryable_failure" && !coordinator.leaseOwner;
            const nativeSettlementDue = Boolean(coordinator.controlDeadlineAt && coordinator.controlDeadlineAt <= now);
            if (!nativeSettlementDue && (nativeOwnerLeaseAt || sameRunResumeScheduled)) {
              await tx.update(heartbeatRuns).set({ executionControlDeadlineAt: null }).where(eq(heartbeatRuns.id, run.id));
              return false;
            }
            if (nativeOwnerLeaseAt) {
              await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionFinalization);
              await tx.update(nativeRunFinalizations).set({ controlDeadlineAt: nativeOwnerLeaseAt })
                .where(eq(nativeRunFinalizations.runId, run.id));
              await tx.update(heartbeatRuns).set({ executionControlDeadlineAt: run.executionControlDeadlineAt ? nativeOwnerLeaseAt : null })
                .where(eq(heartbeatRuns.id, run.id));
              return false;
            }
          }
          await tx
            .update(heartbeatRuns)
            .set({ executionControlDeadlineAt: null })
            .where(eq(heartbeatRuns.id, run.id));
          if (
            [
              "succeeded",
              "failed",
              "cancelled",
              "timed_out",
              "interrupted",
            ].includes(run.status)
          ) {
            if (coordinator?.controlDeadlineAt)
              await tx
                .update(nativeRunFinalizations)
                .set({ controlDeadlineAt: null })
                .where(eq(nativeRunFinalizations.runId, run.id));
            return true;
          }
          // A persisted result belongs to the existing finalizer, not a replacement provider.
          if (coordinator?.resultId) {
            await notifyNativeFinalizationWork(tx, { companyId: run.companyId, runId: run.id });
            await tx
              .update(nativeRunFinalizations)
              .set({
                controlDeadlineAt: null,
                leaseOwner: null,
                leaseExpiresAt: null,
                updatedAt: now,
              })
              .where(eq(nativeRunFinalizations.runId, run.id));
            return true;
          }
          if (await hasLiveLegacyController(tx as unknown as Db, run)) {
            await tx
              .update(heartbeatRuns)
              .set({ executionControlDeadlineAt: run.controllerLeaseExpiresAt })
              .where(eq(heartbeatRuns.id, run.id));
            return false;
          }
          const cause = EXECUTION_FINALIZATION_DEADLINE_EXCEEDED;
          const nextAction = EXECUTION_FINALIZATION_DEADLINE_MESSAGE;
          // The deadline replaces the run's error code. Keep the earlier
          // diagnosis as its root cause, as the native finalizer does.
          const priorFailureCode = coordinator?.failureCode ?? run.errorCode;
          const recordsOriginalFailure = Boolean(priorFailureCode) && priorFailureCode !== cause &&
            typeof run.resultJson?.originalFailureCode !== "string";
          if (coordinator)
            await tx
              .update(nativeRunFinalizations)
              .set({
                phase: "terminal_failure",
                controlDeadlineAt: null,
                leaseOwner: null,
                leaseExpiresAt: null,
                recoveryState: "blocked",
                nextAttemptAt: null,
                failureCode: cause,
                failureDetail: {
                  ...coordinator.failureDetail,
                  nextAction,
                  recoveryOwner: { kind: "board" },
                },
                updatedAt: now,
              })
              .where(eq(nativeRunFinalizations.runId, run.id));
          await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id });
          await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id });
          await notifyNativeEnvironmentSettlement(tx, { companyId: run.companyId, runId: run.id });
          await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id });
          await notifyUnsafeWorkspaceExportRunChange(tx, run);
          await notifyExecutionRunWork(tx, { companyId: run.companyId });
          const [updatedRun] = await tx
            .update(heartbeatRuns)
            .set({
              status: "failed",
              executionStatusDeliveryId: randomUUID(),
              finishedAt: now,
              ...(coordinator
                ? { nativePhase: "terminal_failure", nativePhaseUpdatedAt: now }
                : {}),
              errorCode: cause,
              error: nextAction,
              nextAction,
              ...(recordsOriginalFailure
                ? { resultJson: { ...run.resultJson, originalFailureCode: priorFailureCode } }
                : {}),
              updatedAt: now,
            })
            .where(eq(heartbeatRuns.id, run.id))
            .returning();
          if (updatedRun && updatedRun.status !== run.status) {
            terminalRunToReport = updatedRun;
          }
          await tx
            .update(agents)
            .set({ status: "idle", updatedAt: now })
            .where(
              and(
                eq(agents.id, run.agentId),
                eq(agents.companyId, run.companyId),
                eq(agents.status, "running"),
                eq(agents.lifecycleState, "ready"),
                sql`not exists (select 1 from ${heartbeatRuns} where ${heartbeatRuns.agentId} = ${run.agentId} and ${heartbeatRuns.status} = 'running')`,
              ),
            );
          const nativeReviewContext = readNativeReviewAssignmentContext(run.contextSnapshot);
          const nativeReviewAssignment = nativeReviewContext && task
            ? await getNativeReviewAssignment(tx as unknown as Db, {
                companyId: run.companyId,
                issueId: task.id,
                agentId: run.agentId,
                contextSnapshot: nativeReviewContext,
              })
            : null;
          const review = task?.status === "in_review" ? parseIssueExecutionState(task.executionState) : null;
          const isCurrentReviewer = (review?.status === "pending" &&
            review.currentParticipant?.type === "agent" && review.currentParticipant.agentId === run.agentId)
            || nativeReviewAssignment?.interaction.status === "pending";
          if (
            !task ||
            (task.assigneeAgentId !== run.agentId && !isCurrentReviewer) ||
            ["done", "cancelled"].includes(task.status) ||
            (task.executionRunId && task.executionRunId !== run.id) ||
            (task.checkoutRunId && task.checkoutRunId !== run.id)
          ) {
            if (updatedRun) await preserveNativeFailedWorkspace(tx as unknown as Db, updatedRun);
            return true;
          }
          await notifyNativeStatusReassessment(tx, { companyId: task.companyId, issueId: task.id });
          await tx
            .update(issues)
            .set({ executionRunId: null, checkoutRunId: null, updatedAt: now })
            .where(eq(issues.id, task.id));
          await issueRecoveryActionService(
            tx as unknown as Db,
          ).upsertSourceScoped({
            companyId: run.companyId,
            sourceIssueId: task.id,
            kind: "active_run_watchdog",
            ownerType: "board",
            ownerAgentId: null,
            returnOwnerAgentId: task.assigneeAgentId,
            cause,
            fingerprint: `execution-control:${run.id}`,
            evidence: {
              runId: run.id,
              ...(isCurrentReviewer ? { reviewParticipantAgentId: run.agentId } : {}),
              originalFailureCode: coordinator?.failureCode ?? run.errorCode,
              providerOwnership: "unverified",
            },
            nextAction,
            wakePolicy: null,
            maxAttempts: 3,
            supersedeOnIdentityChange: true,
          });
          if (updatedRun) await preserveNativeFailedWorkspace(tx as unknown as Db, updatedRun);
          return true;
        });
        if (repaired) surfaced += 1;
        if (terminalRunToReport) void reportRunFailure(db, terminalRunToReport);
      } catch {
        logger.warn(
          { runId: candidate.runId },
          "Execution finalization reconciliation remains pending; continuing with other runs",
        );
      }
    }
  }));
  // Owners left on runs that already ended outside the native executor have no
  // control deadline; their expired lease is the due time.
  const releasedOwners = await releaseAbandonedNativeOwners(db, now).catch((error: unknown) => {
    logger.warn({ err: error }, "Abandoned native owner release remains pending");
    return 0;
  });
  return { scanned: due.length, surfaced, releasedOwners };
}