fix(mothership): settle job runs before claiming, keep spend-cap links and failure counts durable

Author: waleedlatif1

apps/sim/app/workspace/[workspaceId]/home/components/message-content/components/special-tags/special-tags.test.tsx

@@ -523,3 +523,30 @@ describe('CredentialDisplay link tag', () => {
     act(() => root.unmount())
   })
 })
+
+describe('UsageUpgradeDisplay', () => {
+  it("shows a job run's own spend cap without an account billing action", () => {
+    ;(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true
+    const container = document.createElement('div')
+    const root = createRoot(container)
+    act(() => {
+      root.render(
+        <SpecialTags
+          segment={{
+            type: 'usage_upgrade',
+            data: {
+              reason: 'job_run_budget',
+              action: 'increase_limit',
+              message: 'This scheduled job run reached its $5.00 spend limit.',
+            },
+          }}
+        />
+      )
+    })
+
+    expect(container.textContent).toContain('Run Spend Limit Reached')
+    expect(container.textContent).toContain('This scheduled job run reached its $5.00 spend limit.')
+    expect(container.querySelector('a')).toBeNull()
+    act(() => root.unmount())
+  })
+})

apps/sim/app/workspace/[workspaceId]/home/components/message-content/components/special-tags/special-tags.tsx

@@ -3443,6 +3443,39 @@ function MothershipErrorDisplay({ data }: { data: MothershipErrorTagData }) {
 }
 
 function UsageUpgradeDisplay({ data }: { data: UsageUpgradeTagData }) {
+  return data.reason === 'job_run_budget' ? (
+    <JobRunBudgetDisplay message={data.message} />
+  ) : (
+    <PlanUsageLimitDisplay data={data} />
+  )
+}
+
+/**
+ * A scheduled job's run reached its own spend cap. The cap belongs to the job, not to the
+ * account, so the card offers no billing action: the next run starts with a fresh cap.
+ */
+function JobRunBudgetDisplay({ message }: { message: string }) {
+  return (
+    <InteractionCard
+      title={
+        <span className='inline-flex items-center gap-2'>
+          <TriangleAlert
+            aria-hidden
+            className='size-[14px] shrink-0 text-[var(--badge-amber-text)]'
+          />
+          Run Spend Limit Reached
+        </span>
+      }
+    >
+      <div className='px-2 pb-2'>
+        <p className='text-[var(--text-body)] text-small leading-5'>{message}</p>
+      </div>
+    </InteractionCard>
+  )
+}
+
+/** The payer's plan limit: the card links to where the plan or the limit can change. */
+function PlanUsageLimitDisplay({ data }: { data: UsageUpgradeTagData }) {
   const { data: session } = useSession()
   const hostContext = useOptionalWorkspaceHostContext()
   const organizationContext = useOptionalOrganizationContext()

apps/sim/app/workspace/[workspaceId]/home/components/mothership-chat/mothership-chat.tsx

@@ -24,6 +24,7 @@ import { useQueryClient } from '@tanstack/react-query'
 import { defaultRangeExtractor, type Range, useVirtualizer } from '@tanstack/react-virtual'
 import { ComputerUseActivity } from '@/components/computer-use/activity'
 import { SMOOTH_CHASE_RATE } from '@/lib/core/utils/smooth-bottom-chase'
+import type { TurnOrigin } from '@/lib/mothership/generated/protocol'
 import type { WorkspaceFileRecord } from '@/lib/uploads/contexts/workspace'
 import { inter } from '@/app/_styles/fonts/inter/inter'
 import { FindBar } from '@/app/workspace/[workspaceId]/components/find-bar'
@@ -187,24 +188,34 @@ interface UserMessageRowProps {
 }
 
 /**
- * A background-task notification that opened a turn (mothership 21-background-tasks.md
- * §6.4): a muted system chip, never a user bubble — the user did not type it. Shows the
- * outcome line; the provenance header and the agent's own note stay out of the way.
+ * A turn automation opened, never a user bubble — the user did not type it: a muted system chip
+ * with one line. A background-task notification (mothership 21-background-tasks.md §6.4) shows
+ * its outcome line; a scheduled job's run shows the brief's first line (the job and its run).
  */
-const TaskNotificationRow = memo(function TaskNotificationRow({
+const AutomatedTurnRow = memo(function AutomatedTurnRow({
+  origin,
   content,
   rowClassName,
 }: {
+  origin: TurnOrigin
   content: string
   rowClassName: string
 }) {
   const lines = content.split('\n').filter((line) => line.length > 0)
-  const outcome = lines.find((line) => line.startsWith('Task ')) ?? lines[lines.length - 1] ?? ''
+  const summary =
+    origin === 'task'
+      ? (lines.find((line) => line.startsWith('Task ')) ?? lines[lines.length - 1] ?? '').replace(
+          /^Task [0-9a-f-]+ /,
+          ''
+        )
+      : (lines[0] ?? '').replace(/^Job: /, '')
   return (
     <div className={cn('flex w-full justify-center', rowClassName)}>
       <div className='max-w-[85%] rounded-full border border-[var(--border)] bg-[var(--bg)] px-3 py-1 text-[12px] text-[var(--text-secondary)]'>
-        <span className='mr-1.5 font-medium text-[var(--text-primary)]'>Background task</span>
-        {outcome.replace(/^Task [0-9a-f-]+ /, '')}
+        <span className='mr-1.5 font-medium text-[var(--text-primary)]'>
+          {origin === 'task' ? 'Background task' : 'Scheduled job'}
+        </span>
+        {summary}
       </div>
     </div>
   )
@@ -930,8 +941,12 @@ export function MothershipChat({
                     style={{ top: virtualItem.start }}
                   >
                     {msg.role === 'user' ? (
-                      interactionPairing.hiddenUserByIndex[index] ? null : msg.origin === 'task' ? (
-                        <TaskNotificationRow content={msg.content} rowClassName={styles.rowGap} />
+                      interactionPairing.hiddenUserByIndex[index] ? null : msg.origin ? (
+                        <AutomatedTurnRow
+                          origin={msg.origin}
+                          content={msg.content}
+                          rowClassName={styles.rowGap}
+                        />
                       ) : (
                         <UserMessageRow
                           content={msg.content}

apps/sim/app/workspace/[workspaceId]/home/components/mothership-chat/use-chat-find.test.tsx

@@ -112,6 +112,18 @@ describe('useChatFind', () => {
     expect(find.current).toMatchObject({ isOpen: false, query: '', count: 0 })
   })
 
+  it('leaves turns automation opened out of the search, whatever opened them', () => {
+    const find = renderFind({
+      messages: [
+        { id: 't', role: 'user', origin: 'task', content: 'Task abc finished: triage done.' },
+        { id: 'j', role: 'user', origin: 'job', content: 'Job: Nightly triage (Run 1)' },
+        assistant('a', 'The triage is done.'),
+      ],
+    })
+    find.search('triage')
+    expect(find.current.count).toBe(1)
+  })
+
   it('counts the same footnoted matches in the index as it highlights in the rendered message', async () => {
     const find = renderFind({
       chatId: 'chat',

apps/sim/app/workspace/[workspaceId]/home/components/mothership-chat/use-chat-find.ts

@@ -163,7 +163,7 @@ export function useChatFind({
     if (term) {
       for (let messageIndex = messages.length - 1; messageIndex >= 0; messageIndex--) {
         const message = messages[messageIndex]
-        if (!message || hiddenUserByIndex[messageIndex] || message.origin === 'task') continue
+        if (!message || hiddenUserByIndex[messageIndex] || message.origin) continue
         const text = getMessageSearchText(message, textCache)
         const occurrences: ChatFindMatch[] = []
         let found = 0

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

@@ -1,4 +1,3 @@
-import { db } from '@sim/db'
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
 import {
@@ -9,7 +8,7 @@ import { isHosted } from '@/lib/core/config/env-flags'
 import { asOrchestrationError } from '@/lib/core/orchestration/types'
 import { buildJobBrief } from '@/lib/mothership/jobs/brief'
 import { jobRunBudgetUsd } from '@/lib/mothership/jobs/budget'
-import { settleJobSchedule } from '@/lib/mothership/jobs/finalize'
+import { failUnstartedJobRun } from '@/lib/mothership/jobs/finalize'
 import {
   countJobDispatchFailure,
   disableJob,
@@ -119,14 +118,7 @@ export async function dispatchJobOccurrence(executionId: string): Promise<JobDis
       error: getErrorMessage(error),
     })
     if (failures < MAX_JOB_DISPATCH_ATTEMPTS) return 'retry'
-    if (await endUnstartedJobRun(executionId, 'failed', 'The run could not be started.')) {
-      if (await settleJobSchedule(db, run.scheduleId, 'failed')) {
-        await notifyScheduleAutoDisabled({
-          scheduleId: run.scheduleId,
-          reason: 'consecutive_failures',
-        })
-      }
-    }
+    await failUnstartedJobRun(executionId, run.scheduleId, 'The run could not be started.')
     return 'failed'
   }
 }

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

@@ -5,6 +5,8 @@ import { and, desc, eq, sql } from 'drizzle-orm'
 import type { DbOrTx } from '@/lib/db/types'
 import { TERMINAL_RUN_STATUSES } from '@/lib/mothership/async-runs/repository'
 import {
+  endedRunStatus,
+  endUnstartedJobRun,
   type JobRunOutcome,
   MAX_CONSECUTIVE_JOB_FAILURES,
   REPORTED_JOB_OUTCOMES,
@@ -19,7 +21,7 @@ const SUMMARY_CHARS = 2_000
  * turns the job off. A met `until_complete` job completes. Returns whether the job was turned
  * off, so the caller can tell its people.
  */
-export async function settleJobSchedule(
+async function settleJobSchedule(
   tx: DbOrTx,
   scheduleId: string,
   outcome: JobRunOutcome
@@ -95,7 +97,7 @@ export async function finalizeJobRun(executionId: string): Promise<boolean> {
         and(eq(jobExecutionLogs.executionId, executionId), eq(jobExecutionLogs.status, 'running'))
       )
       .for('update')
-    if (!record?.scheduleId) return false
+    if (!record) return false
     const [run] = await tx
       .select({
         status: copilotRuns.status,
@@ -129,15 +131,15 @@ export async function finalizeJobRun(executionId: string): Promise<boolean> {
     await tx
       .update(jobExecutionLogs)
       .set({
-        status: outcome === 'failed' ? 'error' : 'complete',
-        level: outcome === 'failed' ? 'error' : 'info',
+        ...endedRunStatus(outcome),
         outcome,
         summary,
         endedAt,
         totalDurationMs: Math.max(0, endedAt.getTime() - record.startedAt.getTime()),
       })
       .where(eq(jobExecutionLogs.executionId, executionId))
-    if (await settleJobSchedule(tx, record.scheduleId, outcome)) {
+    /** A record whose job was deleted mid-run still ends; there is no streak left to count. */
+    if (record.scheduleId && (await settleJobSchedule(tx, record.scheduleId, outcome))) {
       disabledScheduleId = record.scheduleId
     }
     return true
@@ -150,3 +152,19 @@ export async function finalizeJobRun(executionId: string): Promise<boolean> {
   }
   return settled
 }
+
+/**
+ * Ends an occurrence that could never be started as failed, counting it toward the job's failure
+ * streak in the same transaction, so a crash cannot close the record without the count.
+ */
+export async function failUnstartedJobRun(
+  executionId: string,
+  scheduleId: string,
+  summary: string
+): Promise<void> {
+  const disabled = await db.transaction(async (tx) => {
+    if (!(await endUnstartedJobRun(executionId, 'failed', summary, tx))) return false
+    return settleJobSchedule(tx, scheduleId, 'failed')
+  })
+  if (disabled) await notifyScheduleAutoDisabled({ scheduleId, reason: 'consecutive_failures' })
+}

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

@@ -602,14 +602,67 @@ describe.runIf(Boolean(redisUrl))('the scheduled job lane', () => {
       .set({ outcome: 'met', summary: 'Queue empty.' })
       .where(eq(jobExecutionLogs.executionId, occurrenceId))
     worker.finish(occurrenceId, chatId)
-    await reconcileUntilSettled(occurrenceId)
+    await waitFor(async () => {
+      const [run] = await db
+        .select({ status: copilotRuns.status })
+        .from(copilotRuns)
+        .where(eq(copilotRuns.streamId, occurrenceId))
+      return run?.status === 'complete'
+    }, 'the turn to complete')
+    /** The next occurrence comes due before any pass settled the met run. */
+    await dueAgain(scheduleId)
+    await runJobLane(new Date())
 
     expect(await runsOf(scheduleId)).toEqual([
       expect.objectContaining({ status: 'complete', outcome: 'met', summary: 'Queue empty.' }),
     ])
     expect(await schedule(scheduleId)).toMatchObject({ status: 'completed', nextRunAt: null })
   })
 
+  it('gives a recovered run back the spend-cap link its dead process never wrote', 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')
+    const billingRequestId = generateId()
+    /** Its admission was recorded on the run, then the process died before linking the record. */
+    await db
+      .update(copilotRuns)
+      .set({
+        requestContext: sql`jsonb_set(${copilotRuns.requestContext}, '{recovery,billingAdmission}', ${JSON.stringify({ billingRequestId, serializedAttribution: 'fixture' })}::jsonb)`,
+      })
+      .where(eq(copilotRuns.streamId, occurrenceId))
+    await db
+      .update(jobExecutionLogs)
+      .set({ billingRequestId: null })
+      .where(eq(jobExecutionLogs.executionId, occurrenceId))
+
+    await reconcileJobRuns(new Date())
+
+    const [record] = await db
+      .select({ billingRequestId: jobExecutionLogs.billingRequestId })
+      .from(jobExecutionLogs)
+      .where(eq(jobExecutionLogs.scheduleId, scheduleId))
+    expect(record?.billingRequestId).toBe(billingRequestId)
+    worker.finish(occurrenceId, chatId)
+    await reconcileUntilSettled(occurrenceId)
+  })
+
+  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)
+    await db.delete(workflowSchedule).where(eq(workflowSchedule.id, scheduleId))
+
+    await reconcileJobRuns(new Date())
+
+    const [record] = await db
+      .select({ status: jobExecutionLogs.status, outcome: jobExecutionLogs.outcome })
+      .from(jobExecutionLogs)
+      .where(eq(jobExecutionLogs.executionId, occurrenceId))
+    expect(record).toEqual({ status: 'complete', outcome: 'blocked' })
+    await db.delete(jobExecutionLogs).where(eq(jobExecutionLogs.executionId, occurrenceId))
+  })
+
   it('turns a job off after three failed runs in a row, and a verdict clears the streak', async () => {
     const { scheduleId, occurrenceId } = await seedJob()
     const fail = (id: string) =>

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

@@ -27,8 +27,10 @@ export function jobLaneWorkQuery(now: Date) {
 }
 
 /**
- * The schedule tick's job lane: claims due occurrences, starts each as a durable turn, then
- * drives every open occurrence (including ones a crashed tick left) towards its end.
+ * The schedule tick's job lane: drives every open occurrence (including ones a crashed tick
+ * left) towards its end, then claims due occurrences and starts each as a durable turn. Settling
+ * first means a run that met its condition, or a third failure that turned its job off, stops
+ * the job before its next occurrence is claimed.
  *
  * Jobs run only where their turns are durable: the worker is configured, and Redis holds the
  * chat locks that let Sim take a turn over after its process dies. Elsewhere the lane does
@@ -37,6 +39,7 @@ export function jobLaneWorkQuery(now: Date) {
 export async function runJobLane(now: Date = new Date()): Promise<{ claimed: number }> {
   if (!env.COPILOT_API_KEY || !env.REDIS_URL) return { claimed: 0 }
   const [work] = await db.execute<{ due: boolean; open: boolean }>(jobLaneWorkQuery(now))
+  if (work?.open) await reconcileJobRuns(now)
   const claimed = work?.due ? await claimDueJobOccurrences(now, JOB_CLAIM_LIMIT) : []
   for (const occurrence of claimed) {
     await dispatchJobOccurrence(occurrence.executionId).catch((error) =>
@@ -46,6 +49,5 @@ export async function runJobLane(now: Date = new Date()): Promise<{ claimed: num
       })
     )
   }
-  if (work?.open || claimed.length > 0) await reconcileJobRuns(now)
   return { claimed: claimed.length }
 }

apps/sim/lib/mothership/jobs/occurrence.test.ts

@@ -0,0 +1,36 @@
+/**
+ * @vitest-environment node
+ */
+import { describe, expect, it } from 'vitest'
+import { nextJobOccurrence } from '@/lib/mothership/jobs/occurrence'
+
+const quarterHourly = {
+  cronExpression: '*/15 * * * *',
+  timezone: 'UTC',
+  excludedDates: null,
+  endsAt: null,
+}
+
+describe('nextJobOccurrence', () => {
+  it('keeps the next slot when the occurrence is claimed on time', () => {
+    const occurrenceAt = new Date('2026-10-22T10:00:00Z')
+    expect(
+      nextJobOccurrence(quarterHourly, occurrenceAt, new Date('2026-10-22T10:00:30Z'))
+    ).toEqual(new Date('2026-10-22T10:15:00Z'))
+  })
+
+  it('never runs again within the minimum cadence of a late claim', () => {
+    const occurrenceAt = new Date('2026-10-22T10:00:00Z')
+    expect(
+      nextJobOccurrence(quarterHourly, occurrenceAt, new Date('2026-10-22T10:14:00Z'))
+    ).toEqual(new Date('2026-10-22T10:30:00Z'))
+  })
+
+  it('runs a cron tighter than the minimum cadence at most every fifteen minutes', () => {
+    const everyMinute = { ...quarterHourly, cronExpression: '* * * * *' }
+    const occurrenceAt = new Date('2026-10-22T10:00:00Z')
+    expect(nextJobOccurrence(everyMinute, occurrenceAt, occurrenceAt)).toEqual(
+      new Date('2026-10-22T10:15:00Z')
+    )
+  })
+})