Fix scheduled trigger invocations being skipped (#2777)

* Fix scheduled trigger invocations being skipped when trigger is edited without changing the next execution time

* claude comments

Author: shagun-singh-inkeep

.changeset/full-shirts-shop.md

@@ -0,0 +1,5 @@
+---
+"@inkeep/agents-api": patch
+---
+
+Fix scheduled trigger invocations being skipped when trigger is edited without changing the next execution time

agents-api/src/domains/run/workflow/functions/scheduledTriggerRunner.ts

@@ -36,6 +36,7 @@ import {
   markCompletedStep,
   markFailedStep,
   markRunningStep,
+  resetInvocationToPendingStep,
 } from '../steps/scheduledTriggerSteps';
 
 const logger = getLogger('workflow-scheduled-trigger-runner');
@@ -215,14 +216,86 @@ async function _scheduledTriggerRunnerWorkflow(payload: ScheduledTriggerRunnerPa
     });
     invocation = result.invocation;
 
-    if (isOneTime && result.alreadyExists && invocation.status !== 'pending') {
+    if (
+      isOneTime &&
+      result.alreadyExists &&
+      invocation.status !== 'pending' &&
+      invocation.status !== 'cancelled'
+    ) {
       await logStep('One-time trigger already executed', {
         scheduledTriggerId,
         invocationId: invocation.id,
         status: invocation.status,
       });
       return { status: 'already_executed', invocationId: invocation.id };
     }
+
+    if (result.alreadyExists && invocation.status === 'cancelled') {
+      const scheduledTime = new Date(scheduledFor).getTime();
+      const now = Date.now();
+
+      if (scheduledTime > now) {
+        await logStep('Cancelled idempotent invocation still in future, resetting to pending', {
+          scheduledTriggerId,
+          invocationId: invocation.id,
+          scheduledFor,
+        });
+        let resetResult: Awaited<ReturnType<typeof resetInvocationToPendingStep>> | undefined;
+        try {
+          resetResult = await resetInvocationToPendingStep({
+            tenantId,
+            projectId,
+            agentId,
+            scheduledTriggerId,
+            invocationId: invocation.id,
+          });
+        } catch (err) {
+          await logStep('Failed to reset cancelled invocation to pending', {
+            scheduledTriggerId,
+            invocationId: invocation.id,
+            error: err instanceof Error ? err.message : String(err),
+          });
+          throw err;
+        }
+
+        if (!resetResult) {
+          await logStep('Reset skipped — invocation status changed concurrently, exiting', {
+            scheduledTriggerId,
+            invocationId: invocation.id,
+          });
+          return { status: 'skipped' as const, reason: 'concurrent_status_change' };
+        }
+        invocation = { ...invocation, status: 'pending' };
+      } else {
+        await logStep('Cancelled idempotent invocation in the past, skipping to next iteration', {
+          scheduledTriggerId,
+          invocationId: invocation.id,
+          scheduledFor,
+        });
+
+        if (!isOneTime) {
+          try {
+            await startNextIterationStep({
+              tenantId,
+              projectId,
+              agentId,
+              scheduledTriggerId,
+              lastScheduledFor: scheduledFor,
+              currentRunId: runnerId,
+            });
+          } catch (err) {
+            await logStep('Failed to chain after cancelled idempotent invocation', {
+              scheduledTriggerId,
+              invocationId: invocation.id,
+              error: err instanceof Error ? err.message : String(err),
+            });
+            throw err;
+          }
+        }
+
+        return { status: 'skipped_cancelled', invocationId: invocation.id };
+      }
+    }
   }
 
   await logStep('Got next pending invocation', {

agents-api/src/domains/run/workflow/steps/scheduledTriggerSteps.ts

@@ -21,6 +21,7 @@ import {
   markScheduledTriggerInvocationFailed,
   markScheduledTriggerInvocationRunning,
   type Part,
+  resetCancelledInvocationToPending,
   resolveRef,
   type ScheduledTriggerInvocation,
   updateScheduledTriggerInvocationStatus,
@@ -489,6 +490,45 @@ export async function incrementAttemptStep(params: {
   });
 }
 
+/**
+ * Step: Reset a cancelled invocation back to pending.
+ * Used when a restarted workflow finds a cancelled invocation via idempotency
+ * that is still scheduled for a future time.
+ */
+export async function resetInvocationToPendingStep(params: {
+  tenantId: string;
+  projectId: string;
+  agentId: string;
+  scheduledTriggerId: string;
+  invocationId: string;
+}) {
+  'use step';
+
+  const updated = await resetCancelledInvocationToPending(runDbClient)({
+    scopes: {
+      tenantId: params.tenantId,
+      projectId: params.projectId,
+      agentId: params.agentId,
+    },
+    scheduledTriggerId: params.scheduledTriggerId,
+    invocationId: params.invocationId,
+  });
+
+  if (updated) {
+    logger.info(
+      { scheduledTriggerId: params.scheduledTriggerId, invocationId: params.invocationId },
+      'Reset cancelled invocation to pending'
+    );
+  } else {
+    logger.warn(
+      { scheduledTriggerId: params.scheduledTriggerId, invocationId: params.invocationId },
+      'Skipped reset — invocation status changed concurrently (no longer cancelled)'
+    );
+  }
+
+  return updated;
+}
+
 /**
  * Step: Execute the scheduled trigger using executeAgentAsync.
  *

packages/agents-core/src/data-access/runtime/scheduledTriggerInvocations.ts

@@ -278,6 +278,38 @@ export const markScheduledTriggerInvocationFailed =
     return result[0] as ScheduledTriggerInvocation | undefined;
   };
 
+/**
+ * Reset a cancelled invocation back to pending.
+ * Only updates if the current status is 'cancelled' to prevent race conditions.
+ */
+export const resetCancelledInvocationToPending =
+  (db: AgentsRunDatabaseClient) =>
+  async (params: {
+    scopes: AgentScopeConfig;
+    scheduledTriggerId: string;
+    invocationId: string;
+  }): Promise<ScheduledTriggerInvocation | undefined> => {
+    const result = await db
+      .update(scheduledTriggerInvocations)
+      .set({
+        status: 'pending',
+        attemptNumber: 1,
+        startedAt: null,
+        completedAt: null,
+      })
+      .where(
+        and(
+          agentScopedWhere(scheduledTriggerInvocations, params.scopes),
+          eq(scheduledTriggerInvocations.scheduledTriggerId, params.scheduledTriggerId),
+          eq(scheduledTriggerInvocations.id, params.invocationId),
+          eq(scheduledTriggerInvocations.status, 'cancelled')
+        )
+      )
+      .returning();
+
+    return result[0] as ScheduledTriggerInvocation | undefined;
+  };
+
 /**
  * Add a conversation ID to the invocation's conversationIds array
  * Used to track all conversations created during retries