fix(server): settle routine status from committed work

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

server/src/services/delivery-work-notifications.ts

@@ -114,6 +114,7 @@ export const DELIVERY_QUEUES = {
   pluginLifecycle: "plugin-resource-lifecycle",
   toolAction: "tool-action-receipts",
   terminalWorkspaces: "terminal-workspaces",
+  routineExecutionStatus: "routine-execution-status",
   routineSchedules: "routine-schedules",
   issueMonitors: "issue-monitors",
   agentTimers: "agent-timers",

server/src/services/execution-control-notifications.ts

@@ -1,7 +1,9 @@
+import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js";
 import type { Db } from "@paperclipai/db";
 import { DELIVERY_QUEUES, notifyDeliveryWork, type DeliveryQueue } from "./delivery-work-notifications.js";
 
-export async function notifyExecutionRunWork(transaction: object) {
+export async function notifyExecutionRunWork(transaction: object, issueId?: string) {
+  await notifyRoutineExecutionStatus(transaction, issueId);
   for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization,
     DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors]) {
     await notifyDeliveryWork(transaction, queue);
@@ -14,6 +16,7 @@ export function writeExecutionControl<T>(
   write: (tx: Db) => PromiseLike<T>,
 ): Promise<T> {
   return db.transaction(async transaction => {
+    await notifyRoutineExecutionStatus(transaction);
     await notifyDeliveryWork(transaction, DELIVERY_QUEUES.issueMonitors);
     for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic);
     return write(transaction as unknown as Db);

server/src/services/heartbeat/run-lifecycle.ts

@@ -1044,7 +1044,8 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl
       if (previous && isHeartbeatRunTerminalStatus(status))
         await notifyRetainedNativeSessionCleanup(tx, { companyId: previous.companyId, runId: previous.id });
       if (isHeartbeatRunTerminalStatus(status)) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, companyId);
-      await notifyExecutionRunWork(tx);
+      await notifyExecutionRunWork(tx, previous?.nativeIssueId ?? previous?.issueId
+        ?? readNonEmptyString(parseObject(previous?.contextSnapshot).issueId) ?? undefined);
       if (previous) await notifyChatRunMilestones(tx, { companyId: previous.companyId, runId: previous.id,
         issueId: previous.nativeIssueId ?? previous.issueId ?? readNonEmptyString(parseObject(previous.contextSnapshot).issueId) ?? undefined });
       return write(tx);

server/src/services/issue-review-state-notifications.ts

@@ -1,9 +1,11 @@
+import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
 import { notifyNativeReviewResponse } from "./native-review-response-notifications.js";
 import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
 
 export async function notifyIssueReviewState(tx: object, scope: { companyId: string; issueId?: string; runId?: string; agentId?: string }) {
+  await notifyRoutineExecutionStatus(tx, scope.issueId);
   await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
   await notifyExternalConversationSettlement(tx, scope);
   await notifyNativeReviewResponse(tx, scope);

server/src/services/issues.ts

@@ -1,3 +1,4 @@
+import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js";
 import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
 import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js";
 import { notifyNativeRunProjection } from "./native-run-projection-notifications.js";
@@ -10606,6 +10607,7 @@ export function issueService(db: Db) {
         }
         if (values.monitorNextCheckAt) await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
+        if (values.originKind === "routine_execution") await notifyRoutineExecutionStatus(tx, values.id);
         const [issue] = await tx.insert(issues).values(values).returning();
         await attachOwnedDraftImages(tx, issue, issue.description, { userId: issueData.createdByUserId, agentId: issueData.createdByAgentId });
         if (issue.visibility === "private" && !inheritedPrivateSubtree && issueData.createdByAgentId) {
@@ -11494,6 +11496,7 @@ export function issueService(db: Db) {
         if (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined
           || patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined) {
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
+          await notifyRoutineExecutionStatus(tx, id);
         }
         if (patch.monitorNextCheckAt !== undefined || (receiptExisting.monitorNextCheckAt &&
           (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined))) {
@@ -11542,6 +11545,9 @@ export function issueService(db: Db) {
           await notifyNativeWorkspaceFinalization(tx, { companyId: receiptExisting.companyId, issueId: id });
           await notifyNativeStatusReassessment(tx, { companyId: receiptExisting.companyId, issueId: id });
         }
+        if (patch.originKind !== undefined || patch.originRunId !== undefined || patch.originId !== undefined) {
+          await notifyRoutineExecutionStatus(tx, id);
+        }
         const updated = await tx
           .update(issues)
           .set(patch)
@@ -11958,6 +11964,7 @@ export function issueService(db: Db) {
           await notifyDecisionTargetIssueRemoval(tx, id);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
+          await notifyRoutineExecutionStatus(tx, id);
           removedIssue = await tx
             .delete(issues)
             .where(eq(issues.id, id))
@@ -12086,6 +12093,7 @@ export function issueService(db: Db) {
       const updated = await db.transaction(async tx => {
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
+        await notifyRoutineExecutionStatus(tx, id);
         return tx
           .update(issues)
           .set({
@@ -12215,6 +12223,7 @@ export function issueService(db: Db) {
           const adopted = await db.transaction(async tx => {
             await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
             await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
+            await notifyRoutineExecutionStatus(tx, id);
             return tx
               .update(issues)
               .set(adoptionSet)
@@ -12470,6 +12479,7 @@ export function issueService(db: Db) {
         await notifyNativeStatusReassessment(tx, { companyId: existing.companyId, issueId: id });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
+        await notifyRoutineExecutionStatus(tx, id);
         const updated = await tx
           .update(issues)
           .set({
@@ -12530,6 +12540,7 @@ export function issueService(db: Db) {
         }
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
+        await notifyRoutineExecutionStatus(tx, id);
         const updated = await tx
           .update(issues)
           .set(patch)

server/src/services/native-runtime/native-routine-status.test.ts

@@ -1,3 +1,4 @@
+import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
 import { randomUUID } from "node:crypto";
 import { eq } from "drizzle-orm";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
@@ -8,13 +9,14 @@ import { routineService } from "../routines.js";
 import { commitNativeStatusDecision } from "./status-decision-committer.js";
 import { NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeStatusDecision } from "./status-arbiter.js";
 
-const support = await getEmbeddedPostgresTestSupport();
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_ROUTINE_STATUS_TEST_DATABASE_URL");
+const support = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport();
 if (!support.supported) console.warn(`Skipping native routine status tests: ${support.reason}`);
 (support.supported ? describe : describe.skip)("native routine history status projection", () => {
   let db: ReturnType<typeof createDb>, database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
   const companyId = randomUUID(), agentId = randomUUID();
   beforeAll(async () => {
-    database = await startEmbeddedPostgresTestDatabase("native-routine-status-"); db = createDb(database.connectionString);
+    if (!external) database = await startEmbeddedPostgresTestDatabase("native-routine-status-"); db = createDb(external ?? database.connectionString, { maxConnections: 6 });
     await db.insert(companies).values({ id: companyId, name: "Routine status fixture", issuePrefix: "NRS" });
     await db.insert(agents).values({ id: agentId, companyId, name: "Routine worker", adapterType: "paperclip_runner", status: "running" });
   }, 60_000);

server/src/services/routine-execution-status-notifications.ts

@@ -0,0 +1,5 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+
+export function notifyRoutineExecutionStatus(tx: object, issueId?: string) {
+  return notifyDeliveryWork(tx, DELIVERY_QUEUES.routineExecutionStatus, issueId ?? "*");
+}

server/src/services/routine-execution-status-work.ts

@@ -0,0 +1,18 @@
+import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
+import { registerScopedDeadlineWork } from "./scoped-deadline-work.js";
+import type { routineExecutionStatusService } from "./routine-run-status.js";
+
+export function registerRoutineExecutionStatusWork(
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  service: ReturnType<typeof routineExecutionStatusService>,
+  canRun: () => boolean,
+) {
+  return registerScopedDeadlineWork(coordinator, {
+    queue: DELIVERY_QUEUES.routineExecutionStatus,
+    decodeScope: hint => hint === "*" ? undefined : hint,
+    matches: (entry, issueId?: string) => issueId === undefined || entry.issueId === issueId,
+    load: async (issueId?: string) => (await service.load(issueId)).map(row => ({ ...row, dueAt: Date.now() })),
+    check: async entry => { await service.settle(entry); return null; },
+  }, canRun);
+}

server/src/services/routine-run-status.ts

@@ -1,6 +1,7 @@
-import { and, eq } from "drizzle-orm";
+import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js";
+import { and, eq, inArray, or, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
-import { issues, routineRuns, routines } from "@paperclipai/db";
+import { issues, pluginManagedResources, routineRuns, routines } from "@paperclipai/db";
 
 const EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE = "execution_issue_status";
 const EXECUTION_ISSUE_TRANSIENT_FAILURE_STATUSES = ["blocked", "cancelled"] as const;
@@ -37,6 +38,7 @@ function legacyExecutionIssueTransientFailureStatus(
 }
 
 export async function finalizeRoutineRun(executor: Db, runId: string, patch: Partial<typeof routineRuns.$inferInsert>) {
+  if (patch.linkedIssueId) await notifyRoutineExecutionStatus(executor, patch.linkedIssueId);
   return executor
     .update(routineRuns)
     .set({
@@ -67,14 +69,16 @@ export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string
       status: routineRuns.status,
       failureReason: routineRuns.failureReason,
       triggerPayload: routineRuns.triggerPayload,
+      completedAt: routineRuns.completedAt,
     })
     .from(routineRuns)
     .innerJoin(routines, and(eq(routines.id, routineRuns.routineId), eq(routines.companyId, issue.companyId)))
     .where(and(eq(routineRuns.id, issue.originRunId), eq(routineRuns.companyId, issue.companyId),
       eq(routineRuns.linkedIssueId, issue.id)))
     .then((rows) => rows[0] ?? null);
-  if (!run) return null;
+  if (!run || ["received", "skipped", "coalesced"].includes(run.status)) return null;
   if (issue.status === "done") {
+    if (run.status === "completed" && run.failureReason === null && run.completedAt) return null;
     const transientFailureStatus = executionIssueTransientFailureStatusFromPayload(run.triggerPayload)
       ?? legacyExecutionIssueTransientFailureStatus(run.failureReason);
     const transientFailureClearedAt = executionIssueTransientFailureClearedAtFromPayload(run.triggerPayload);
@@ -99,6 +103,7 @@ export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string
   }
   if (issue.status === "blocked" || issue.status === "cancelled") {
     const failureReason = executionIssueTransientFailureReason(issue.status);
+    if (run.status === "failed" && run.failureReason === failureReason && run.completedAt) return null;
     return finalizeRoutineRun(executor, issue.originRunId, {
       status: "failed",
       failureReason,
@@ -135,3 +140,60 @@ export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string
   return null;
 }
 
+
+export function routineExecutionOriginCondition() {
+  return or(sql`${issues.originId} = ${routineRuns.routineId}::text`, sql`exists (
+    select 1 from ${pluginManagedResources} resource
+    where resource.company_id = ${routineRuns.companyId}
+      and resource.resource_kind = 'routine' and resource.resource_id = ${routineRuns.routineId}
+      and resource.defaults_json -> 'issueTemplate' ->> 'originId' = ${issues.originId}
+  )`);
+}
+
+export function routineExecutionStatusDebtCondition() {
+  const transientFailure = sql`(${routineRuns.triggerPayload} -> 'transientFailure' ->> 'code' = ${EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE}
+    or ${routineRuns.failureReason} in ('Execution issue moved to blocked', 'Execution issue moved to cancelled'))`;
+  return or(
+    and(eq(routineRuns.status, "issue_created"), inArray(issues.status, ["done", "blocked", "cancelled"])),
+    and(eq(routineRuns.status, "completed"), or(inArray(issues.status, ["blocked", "cancelled"]),
+      and(eq(issues.status, "done"), sql`(${routineRuns.completedAt} is null or ${routineRuns.failureReason} is not null)`))),
+    and(eq(routineRuns.status, "failed"), transientFailure,
+      sql`(${issues.status} not in ('blocked', 'cancelled')
+        or ${routineRuns.completedAt} is null
+        or ${routineRuns.failureReason} is distinct from 'Execution issue moved to ' || ${issues.status})`),
+  );
+}
+
+export function routineExecutionStatusService(db: Db) {
+  async function load(issueId?: string) {
+    return db.select({ id: routineRuns.id, companyId: routineRuns.companyId,
+      issueId: issues.id, routineId: routineRuns.routineId })
+      .from(routineRuns)
+      .innerJoin(routines, and(eq(routines.id, routineRuns.routineId), eq(routines.companyId, routineRuns.companyId)))
+      .innerJoin(issues, and(eq(issues.id, routineRuns.linkedIssueId), eq(issues.companyId, routineRuns.companyId),
+        eq(issues.originKind, "routine_execution"), sql`${issues.originRunId} = ${routineRuns.id}::text`, routineExecutionOriginCondition()))
+      .where(and(routineExecutionStatusDebtCondition(), ...(issueId ? [eq(issues.id, issueId)] : [])))
+      .orderBy(routineRuns.id);
+  }
+  async function settle(candidate: Awaited<ReturnType<typeof load>>[number]) {
+    return db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '500ms'`);
+      await tx.execute(sql`SET LOCAL statement_timeout = '5s'`);
+      const [routine] = await tx.select({ id: routines.id }).from(routines).where(and(
+        eq(routines.id, candidate.routineId), eq(routines.companyId, candidate.companyId))).for("share");
+      if (!routine) return false;
+      const [issue] = await tx.select().from(issues).where(and(eq(issues.id, candidate.issueId),
+        eq(issues.companyId, candidate.companyId), eq(issues.originKind, "routine_execution"),
+        eq(issues.originRunId, candidate.id))).for("update");
+      if (!issue) return false;
+      const [run] = await tx.select({ id: routineRuns.id }).from(routineRuns)
+        .innerJoin(issues, eq(issues.id, routineRuns.linkedIssueId))
+        .where(and(eq(routineRuns.id, candidate.id), eq(routineRuns.companyId, candidate.companyId),
+          eq(routineRuns.routineId, candidate.routineId), eq(routineRuns.linkedIssueId, issue.id),
+          routineExecutionOriginCondition(), routineExecutionStatusDebtCondition())).for("update", { of: routineRuns });
+      if (!run) return false;
+      return (await syncRoutineRunStatusForIssue(tx as unknown as Db, issue.id)) !== null;
+    });
+  }
+  return { load, settle };
+}

server/src/services/routines.ts

@@ -73,7 +73,7 @@ import { secretService } from "./secrets.js";
 import { getSecretProvider } from "../secrets/provider-registry.js";
 import { parseCron, validateCron } from "./cron.js";
 import { heartbeatService } from "./heartbeat.js";
-import { finalizeRoutineRun, syncRoutineRunStatusForIssue } from "./routine-run-status.js";
+import { finalizeRoutineRun, routineExecutionStatusService, syncRoutineRunStatusForIssue } from "./routine-run-status.js";
 import {
   instanceSettingsService,
   isTruthyRuntimeEnvValue,
@@ -2001,7 +2001,6 @@ export function routineService(
     return run;
   }
 
-  let statusReconcileCursor: string | null = null;
 
   async function syncRunStatusForIssue(issueId: string, executor: Db = db) {
     return syncRoutineRunStatusForIssue(executor, issueId);
@@ -3245,41 +3244,13 @@ export function routineService(
       return { triggered };
     },
 
-    // Repair a missed issue-status hook through the same status transition as
-    // normal issue writes. Never discard queued input or infer run completion
-    // from age. Lock both owners and re-read the issue before applying repair.
     reconcileExecutionRunStatuses: async (): Promise<{ scanned: number; settled: number; deferred: number }> => {
-      const candidates = await db.select({ id: routineRuns.id, companyId: routineRuns.companyId,
-        issueId: issues.id, routineId: routineRuns.routineId })
-        .from(routineRuns).innerJoin(routines, and(eq(routines.id, routineRuns.routineId), eq(routines.companyId, routineRuns.companyId)))
-        .innerJoin(issues, and(eq(issues.id, routineRuns.linkedIssueId),
-          eq(issues.companyId, routineRuns.companyId), eq(issues.originKind, "routine_execution"),
-          sql`${issues.originRunId} = ${routineRuns.id}::text`, sql`${issues.originId} = ${routineRuns.routineId}::text`))
-        .where(and(eq(routineRuns.status, "issue_created"), inArray(issues.status, ["done", "cancelled", "blocked"]),
-          ...(statusReconcileCursor ? [sql`${routineRuns.id}::text > ${statusReconcileCursor}`] : [])))
-        .orderBy(asc(routineRuns.id)).limit(50);
-      statusReconcileCursor = candidates.length === 50 ? candidates.at(-1)!.id : null;
-      let settled = 0;
-      let deferred = 0;
+      const service = routineExecutionStatusService(db);
+      const candidates = await service.load();
+      let settled = 0, deferred = 0;
       for (const candidate of candidates) {
         try {
-          const repaired = await db.transaction(async tx => {
-            await tx.execute(sql`SET LOCAL lock_timeout = '500ms'`);
-            await tx.execute(sql`SET LOCAL statement_timeout = '5s'`);
-            const [routine] = await tx.select({ id: routines.id }).from(routines).where(and(
-              eq(routines.id, candidate.routineId), eq(routines.companyId, candidate.companyId))).for("share");
-            if (!routine) return false;
-            const [issue] = await tx.select().from(issues).where(and(eq(issues.id, candidate.issueId),
-              eq(issues.companyId, candidate.companyId), eq(issues.originKind, "routine_execution"),
-              eq(issues.originRunId, candidate.id), eq(issues.originId, candidate.routineId))).for("update");
-            if (!issue || !["done", "cancelled", "blocked"].includes(issue.status)) return false;
-            const [run] = await tx.select().from(routineRuns).where(and(eq(routineRuns.id, candidate.id),
-              eq(routineRuns.companyId, candidate.companyId), eq(routineRuns.routineId, candidate.routineId),
-              eq(routineRuns.linkedIssueId, issue.id), eq(routineRuns.status, "issue_created"))).for("update");
-            if (!run) return false;
-            return (await syncRunStatusForIssue(issue.id, tx as unknown as Db)) !== null;
-          });
-          if (repaired) settled += 1;
+          if (await service.settle(candidate)) settled++;
         } catch (error) {
           let current: unknown = error;
           let retryable = false;
@@ -3289,7 +3260,7 @@ export function routineService(
             current = cause.cause;
           }
           if (!retryable) throw error;
-          deferred += 1;
+          deferred++;
         }
       }
       return { scanned: candidates.length, settled, deferred };