fix(mothership): end a running job record whose run was removed

Author: waleedlatif1

apps/sim/lib/mothership/jobs/finalize.ts

@@ -79,7 +79,8 @@ async function lastAssistantText(tx: DbOrTx, chatId: string, streamId: string):
  * under a row lock, so the turn's own end and a reconciler pass settle it exactly once: one
  * outcome, one change to the job's failure streak. An outcome the agent reported is kept; a
  * turn that ended well without one is `not_met`, summarized by its last reply, and is not a
- * failure. Returns false while the turn is still live.
+ * failure. A run whose row is gone (its chat was deleted) ends `interrupted`. Returns false while
+ * the turn is still live.
  */
 export async function finalizeJobRun(executionId: string): Promise<boolean> {
   let disabledScheduleId: string | null = null
@@ -108,13 +109,17 @@ export async function finalizeJobRun(executionId: string): Promise<boolean> {
       .from(copilotRuns)
       .where(eq(copilotRuns.streamId, executionId))
       .limit(1)
-    if (!run || !TERMINAL_RUN_STATUSES.includes(run.status)) return false
+    if (run && !TERMINAL_RUN_STATUSES.includes(run.status)) return false
 
     const reported = REPORTED_JOB_OUTCOMES.find((outcome) => outcome === record.outcome)
     const deadlineStopped = toRecord(record.executionData).deadlineStopRequestedAt !== undefined
     let outcome: JobRunOutcome
     let summary: string
-    if (run.status === 'complete' && reported) {
+    if (!run) {
+      /** A started run's row is gone (its chat was deleted): nothing will ever end it otherwise. */
+      outcome = 'interrupted'
+      summary = 'The run was removed before it finished.'
+    } else if (run.status === 'complete' && reported) {
       outcome = reported
       summary = record.summary ?? ''
     } else if (run.status === 'complete') {
@@ -127,7 +132,7 @@ export async function finalizeJobRun(executionId: string): Promise<boolean> {
       outcome = 'failed'
       summary = run.error ?? 'The run failed.'
     }
-    const endedAt = run.completedAt ?? new Date()
+    const endedAt = run?.completedAt ?? new Date()
     await tx
       .update(jobExecutionLogs)
       .set({

apps/sim/lib/mothership/jobs/lane.integration.ts

@@ -648,6 +648,24 @@ describe.runIf(Boolean(redisUrl))('the scheduled job lane', () => {
     await reconcileUntilSettled(occurrenceId)
   })
 
+  it('ends a running occurrence whose run was removed, instead of re-reading it forever', async () => {
+    const { chatId, scheduleId, occurrenceId } = await seedJob()
+    worker.legs.set(occurrenceId, [{ text: 'Working.', mode: 'hang' }])
+    await runJobLane(new Date())
+    await waitFor(async () => worker.open.has(occurrenceId), 'the leg')
+    controllers.get(occurrenceId)?.abort(new StreamControllerSupersededError())
+    worker.open.get(occurrenceId)?.response.socket?.destroy()
+    worker.open.delete(occurrenceId)
+    await waitFor(async () => !(await redis().get(chatStreamLockKey(chatId))), 'the lock to lapse')
+    await db.delete(copilotRuns).where(eq(copilotRuns.streamId, occurrenceId))
+
+    await reconcileJobRuns(new Date(Date.now() + 6 * 60 * 60_000 + 60_000))
+
+    expect(await runsOf(scheduleId)).toEqual([
+      expect.objectContaining({ status: 'complete', outcome: 'interrupted' }),
+    ])
+  }, 60_000)
+
   it('ends a queued occurrence whose job was deleted, so the lane goes idle', async () => {
     const { scheduleId, occurrenceId } = await seedJob()
     await claimDueJobOccurrences(new Date(), 10)

apps/sim/lib/mothership/jobs/reconcile.ts

@@ -46,15 +46,16 @@ const admittedBillingRequestId = sql<
 
 /**
  * Open records that need this pass, oldest first. Every queued record does; a running one only
- * once its turn ended, its time limit passed without a Stop, or its spend-cap link is missing,
- * so a backlog of healthy long runs never hides the ones that finished.
+ * once its turn ended or its run row is gone, its time limit passed without a Stop, or its
+ * spend-cap link is missing, so a backlog of healthy long runs never hides the ones that finished.
  */
 function openJobRuns(status: OpenJobRunStatus, now: Date) {
   const needsAction =
     status === 'queued'
       ? undefined
       : or(
           inArray(copilotRuns.status, TERMINAL_RUN_STATUSES),
+          isNull(copilotRuns.id),
           and(
             lte(jobExecutionLogs.startedAt, new Date(now.getTime() - JOB_RUN_TIME_LIMIT_MS)),
             sql`${jobExecutionLogs.executionData}->>'deadlineStopRequestedAt' IS NULL`
@@ -101,7 +102,10 @@ export async function reconcileJobRuns(now: Date): Promise<void> {
   for (const row of [...queued, ...running]) {
     try {
       if (!row.runStatus) {
-        if (row.status !== 'queued') continue
+        if (row.status === 'running') {
+          await finalizeJobRun(row.executionId)
+          continue
+        }
         if (!row.scheduleId) {
           await endUnstartedJobRun(
             row.executionId,