improvement(mothership): run the job lane unflagged behind one indexed probe

Author: waleedlatif1

apps/sim/.env.example

@@ -29,7 +29,6 @@ NEXT_PUBLIC_APP_URL=http://localhost:3000
 # MSHIP_SYSPROMPT_OVERRIDE=             # Highest-priority instructions for Mothership; honored only when the validated API key owner is enterprise
 # NEXT_PUBLIC_CHAT_DISABLED=true        # Hides the Chat module: the workspace lands on your first workflow, and the chats list, scheduled tasks, and editor Chat panel are absent. Chat is shown when unset; `bun run setup` sets this for you if you skip the chat key
 # MOTHERSHIP_BENCHMARK_ENABLED=true      # Shows the organization Benchmark tab and enables its authenticated APIs
-# MOTHERSHIP_JOBS_ENABLED=true           # Runs scheduled agent jobs from the schedule tick (unreleased; keep off in production)
 # COPILOT_DEV_URL=http://localhost:8080  # Benchmark worker endpoint; requires the worker benchmark profile and the normal COPILOT_API_KEY
 
 # Remote Function sandboxes (Optional)

apps/sim/app/api/schedules/execute/route.test.ts

@@ -108,10 +108,9 @@ vi.mock('@/lib/execution/manual-cancellation', () => ({
 
 vi.mock('@/lib/core/async-jobs', () => asyncJobsMock)
 
-/** The job lane is off by default and has its own integration suite. */
+/** The job lane has its own integration suite. */
 vi.mock('@/lib/mothership/jobs/lane', () => ({
-  isJobLaneEnabled: () => false,
-  runJobLane: vi.fn(),
+  runJobLane: vi.fn(async () => ({ claimed: 0 })),
 }))
 
 vi.mock('drizzle-orm', () => ({

apps/sim/app/api/schedules/execute/route.ts

@@ -50,7 +50,7 @@ import {
   registerManualExecutionAborter,
   unregisterManualExecutionAborter,
 } from '@/lib/execution/manual-cancellation'
-import { isJobLaneEnabled, runJobLane } from '@/lib/mothership/jobs/lane'
+import { runJobLane } from '@/lib/mothership/jobs/lane'
 import {
   buildCarrierReconciledMetadata,
   carrierNotReconciledSql,
@@ -1789,13 +1789,11 @@ export async function runScheduleTick(requestId: string): Promise<ScheduleTickRe
     await Promise.allSettled(schedulePromises)
   }
 
-  if (isJobLaneEnabled()) {
-    try {
-      const { claimed } = await runJobLane()
-      totalSchedules += claimed
-    } catch (error) {
-      logger.error(`[${requestId}] Scheduled job lane failed`, { error: toError(error).message })
-    }
+  try {
+    const { claimed } = await runJobLane()
+    totalSchedules += claimed
+  } catch (error) {
+    logger.error(`[${requestId}] Scheduled job lane failed`, { error: toError(error).message })
   }
 
   const totalCount = totalSchedules

apps/sim/lib/core/config/env-flags.ts

@@ -111,14 +111,6 @@ export const isCopilotToolPermissionsEnabled = isTruthy(env.COPILOT_TOOL_PERMISS
 /** Deployment-only opt-in; authenticated resource permissions still apply. */
 export const isMothershipBenchmarkEnabled = isTruthy(env.MOTHERSHIP_BENCHMARK_ENABLED)
 
-/**
- * Scheduled agent jobs: the schedule tick claims due job occurrences and runs each as an
- * unattended durable turn in its job's chat. Off by default, and must stay off in production
- * until unattended runs hold production effects for human approval: until then an unattended
- * run can do anything its owner can.
- */
-export const isMothershipJobsEnabled = isTruthy(env.MOTHERSHIP_JOBS_ENABLED)
-
 /**
  * Is billing enforcement enabled.
  *

apps/sim/lib/core/config/env.ts

@@ -151,8 +151,6 @@ export const env = createEnv({
     COPILOT_TOOL_PERMISSIONS_ENABLED:      z.boolean().optional(),
     /** Enables the organization benchmark UI and its authenticated endpoints. */
     MOTHERSHIP_BENCHMARK_ENABLED:          z.boolean().optional(),
-    /** Runs scheduled agent jobs from the schedule tick. Off by default; keep off in production. */
-    MOTHERSHIP_JOBS_ENABLED:               z.boolean().optional(),
     MOTHERSHIP_BENCHMARK_URL:              z.string().url().optional(),
     SIM_AGENT_API_URL:                     z.string().url().optional(),            // URL for internal sim agent API
     MSHIP_SYSPROMPT_OVERRIDE:              z.string().min(1).optional(),           // Enterprise-only highest-priority Mothership system prompt override forwarded by Sim

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

@@ -5,7 +5,8 @@ import { jobOccurrenceId, nextJobOccurrence } from '@/lib/mothership/jobs/occurr
 /** The trigger every job run record carries; the logs surface lists runs under it. */
 export const JOB_RUN_TRIGGER = 'mothership'
 
-const dueJob = (now: Date) =>
+/** A live job whose next occurrence is due; `workflow_schedule_due_job_idx` covers it. */
+export const dueJob = (now: Date) =>
   and(
     eq(workflowSchedule.sourceType, 'job'),
     isNull(workflowSchedule.archivedAt),

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

@@ -88,13 +88,25 @@ const { redisUrl, inheritedEnv, worker } = await vi.hoisted(async () => {
   const inheritedEnv = {
     REDIS_URL: process.env.REDIS_URL,
     SIM_AGENT_API_URL: process.env.SIM_AGENT_API_URL,
+    COPILOT_API_KEY: process.env.COPILOT_API_KEY,
   }
   if (url) process.env.REDIS_URL = url
+  process.env.COPILOT_API_KEY = 'job-lane-fixture-key'
   process.env.SIM_AGENT_API_URL = `http://127.0.0.1:${port}`
   return { redisUrl: url, inheritedEnv, worker: Object.assign(worker, { server, finish }) }
 })
 
 vi.mock('@/lib/core/config/env-flags', () => envFlagsMock)
+
+/** Spied so the idle tick can prove it neither claims nor reconciles. */
+vi.mock('@/lib/mothership/jobs/claim', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('@/lib/mothership/jobs/claim')>()
+  return { ...actual, claimDueJobOccurrences: vi.fn(actual.claimDueJobOccurrences) }
+})
+vi.mock('@/lib/mothership/jobs/reconcile', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('@/lib/mothership/jobs/reconcile')>()
+  return { ...actual, reconcileJobRuns: vi.fn(actual.reconcileJobRuns) }
+})
 vi.mock('@/lib/core/config/feature-flags', () => featureFlagsMock)
 
 const { controllers } = vi.hoisted(() => ({ controllers: new Map<string, AbortController>() }))
@@ -122,15 +134,15 @@ import {
 } from '@sim/db/schema'
 import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
-import { eq, sql } from 'drizzle-orm'
+import { eq, inArray, sql } from 'drizzle-orm'
 import { closeRedisConnection, getRedisClient } from '@/lib/core/config/redis'
 import { ABANDONED_RUN_CURSOR_KEY } from '@/lib/mothership/async-runs/orphaned-runs'
 import { chatPubSub } from '@/lib/mothership/chat-status'
 import { recordJobRunSpend } from '@/lib/mothership/jobs/budget'
 import { claimDueJobOccurrences } from '@/lib/mothership/jobs/claim'
 import { dispatchJobOccurrence } from '@/lib/mothership/jobs/dispatch'
 import { finalizeJobRun } from '@/lib/mothership/jobs/finalize'
-import { runJobLane } from '@/lib/mothership/jobs/lane'
+import { jobLaneWorkQuery, runJobLane } from '@/lib/mothership/jobs/lane'
 import { jobOccurrenceId, MIN_JOB_CADENCE_MS } from '@/lib/mothership/jobs/occurrence'
 import { JOB_DISPATCH_GRACE_MS, reconcileJobRuns } from '@/lib/mothership/jobs/reconcile'
 import { takeOverAbandonedRuns } from '@/lib/mothership/request/application/abandoned-runs'
@@ -140,6 +152,18 @@ import {
   StreamControllerSupersededError,
 } from '@/lib/mothership/request/session/controller-lease'
 
+interface QueryPlan {
+  'Node Type': string
+  'Index Name'?: string
+  'Shared Hit Blocks': number
+  'Shared Read Blocks': number
+  Plans?: QueryPlan[]
+}
+
+function planNodes(plan: QueryPlan): QueryPlan[] {
+  return [plan, ...(plan.Plans ?? []).flatMap(planNodes)]
+}
+
 const userId = generateId()
 const workspaceId = generateId()
 const chatIds: string[] = []
@@ -316,6 +340,61 @@ describe.runIf(Boolean(redisUrl))('the scheduled job lane', () => {
     await closeRedisConnection()
   })
 
+  it('does nothing but one indexed probe on a tick with no live jobs', async () => {
+    /** What the archive migration leaves behind: retired jobs and their finished run logs. */
+    const archivedIds = Array.from({ length: 2_000 }, () => generateId())
+    await db.insert(workflowSchedule).values(
+      archivedIds.map((id) => ({
+        id,
+        triggerType: 'schedule',
+        sourceType: 'job',
+        status: 'disabled',
+        archivedAt: new Date(),
+        nextRunAt: new Date(Date.now() - 60_000),
+        sourceUserId: userId,
+        sourceWorkspaceId: workspaceId,
+      }))
+    )
+    await db.insert(jobExecutionLogs).values(
+      Array.from({ length: 5_000 }, () => {
+        const id = generateId()
+        return {
+          id,
+          executionId: id,
+          workspaceId,
+          level: 'info',
+          status: 'completed',
+          trigger: 'mothership',
+          startedAt: new Date(Date.now() - 60_000),
+        }
+      })
+    )
+    await db.execute(sql`ANALYZE workflow_schedule, job_execution_logs`)
+    const now = new Date()
+    vi.mocked(claimDueJobOccurrences).mockClear()
+    vi.mocked(reconcileJobRuns).mockClear()
+
+    try {
+      expect(await runJobLane(now)).toEqual({ claimed: 0 })
+      expect(claimDueJobOccurrences).not.toHaveBeenCalled()
+      expect(reconcileJobRuns).not.toHaveBeenCalled()
+
+      const plans = await db.execute<{ 'QUERY PLAN': { Plan: QueryPlan }[] }>(
+        sql`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${jobLaneWorkQuery(now)}`
+      )
+      const plan = plans[0]['QUERY PLAN'][0].Plan
+      const nodes = planNodes(plan)
+      expect(nodes.some((node) => node['Node Type'] === 'Seq Scan')).toBe(false)
+      expect(nodes.map((node) => node['Index Name']).filter(Boolean)).toEqual(
+        expect.arrayContaining(['workflow_schedule_due_job_idx', 'job_execution_logs_open_idx'])
+      )
+      expect(plan['Shared Hit Blocks'] + plan['Shared Read Blocks']).toBeLessThan(20)
+    } finally {
+      await db.delete(jobExecutionLogs).where(eq(jobExecutionLogs.workspaceId, workspaceId))
+      await db.delete(workflowSchedule).where(inArray(workflowSchedule.id, archivedIds))
+    }
+  })
+
   it('records an occurrence once when two ticks claim it at the same time, and starts it once', async () => {
     const { scheduleId, occurrenceId, dueAt } = await seedJob()
     const now = new Date()

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

@@ -0,0 +1,26 @@
+/**
+ * @vitest-environment node
+ */
+import { dbChainMockFns, resetDbChainMock } from '@sim/testing/mocks/database.mock'
+import { resetEnvMock, setEnv } from '@sim/testing/mocks/env.mock'
+import { afterEach, describe, expect, it } from 'vitest'
+import { runJobLane } from '@/lib/mothership/jobs/lane'
+
+describe('runJobLane', () => {
+  afterEach(() => {
+    resetEnvMock()
+    resetDbChainMock()
+  })
+
+  it.each([
+    { COPILOT_API_KEY: undefined, REDIS_URL: 'redis://localhost:6379' },
+    { COPILOT_API_KEY: 'key', REDIS_URL: undefined },
+  ])('touches nothing where turns cannot be durable (%o)', async (env) => {
+    setEnv(env)
+
+    expect(await runJobLane()).toEqual({ claimed: 0 })
+    expect(dbChainMockFns.execute).not.toHaveBeenCalled()
+    expect(dbChainMockFns.transaction).not.toHaveBeenCalled()
+    expect(dbChainMockFns.select).not.toHaveBeenCalled()
+  })
+})

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

@@ -1,30 +1,43 @@
+import { db, jobExecutionLogs, workflowSchedule } from '@sim/db'
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
+import { exists, sql } from 'drizzle-orm'
 import { env } from '@/lib/core/config/env'
-import { isMothershipJobsEnabled } from '@/lib/core/config/env-flags'
-import { claimDueJobOccurrences } from '@/lib/mothership/jobs/claim'
+import { claimDueJobOccurrences, dueJob } from '@/lib/mothership/jobs/claim'
 import { dispatchJobOccurrence } from '@/lib/mothership/jobs/dispatch'
-import { reconcileJobRuns } from '@/lib/mothership/jobs/reconcile'
+import { openJobRun, reconcileJobRuns } from '@/lib/mothership/jobs/reconcile'
 
 const logger = createLogger('MothershipJobLane')
 
 /** Occurrences one tick claims; the rest wait for the next tick. */
 const JOB_CLAIM_LIMIT = 50
 
 /**
- * Jobs run only where their turns are durable: the flag is on, the worker is configured, and
- * Redis holds the chat locks that let Sim take a turn over after its process dies.
+ * One row: `due`, whether a job's occurrence is due to be claimed, and `open`, whether an
+ * occurrence is still queued or running. Each leg probes its own partial index, so a deployment
+ * with no live jobs pays one index lookup per tick and nothing else.
  */
-export function isJobLaneEnabled(): boolean {
-  return isMothershipJobsEnabled && Boolean(env.COPILOT_API_KEY) && Boolean(env.REDIS_URL)
+export function jobLaneWorkQuery(now: Date) {
+  const due = db.select({ id: workflowSchedule.id }).from(workflowSchedule).where(dueJob(now))
+  const open = db
+    .select({ id: jobExecutionLogs.id })
+    .from(jobExecutionLogs)
+    .where(openJobRun(['queued', 'running']))
+  return sql`SELECT ${exists(due.limit(1))} AS due, ${exists(open.limit(1))} AS open`
 }
 
 /**
  * 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.
+ *
+ * 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
+ * nothing, not even a query.
  */
 export async function runJobLane(now: Date = new Date()): Promise<{ claimed: number }> {
-  const claimed = await claimDueJobOccurrences(now, JOB_CLAIM_LIMIT)
+  if (!env.COPILOT_API_KEY || !env.REDIS_URL) return { claimed: 0 }
+  const [work] = await db.execute<{ due: boolean; open: boolean }>(jobLaneWorkQuery(now))
+  const claimed = work?.due ? await claimDueJobOccurrences(now, JOB_CLAIM_LIMIT) : []
   for (const occurrence of claimed) {
     await dispatchJobOccurrence(occurrence.executionId).catch((error) =>
       logger.warn('Job occurrence dispatch failed; the reconciler retries it', {
@@ -33,6 +46,6 @@ export async function runJobLane(now: Date = new Date()): Promise<{ claimed: num
       })
     )
   }
-  await reconcileJobRuns(now)
+  if (work?.open || claimed.length > 0) await reconcileJobRuns(now)
   return { claimed: claimed.length }
 }

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

@@ -1,7 +1,7 @@
 import { copilotRuns, db, jobExecutionLogs, workflowSchedule } from '@sim/db'
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
-import { and, asc, eq, isNotNull, sql } from 'drizzle-orm'
+import { and, asc, eq, inArray, isNotNull, sql } from 'drizzle-orm'
 import { requestRunStop } from '@/lib/mothership/async-runs/repository'
 import { JOB_RUN_TRIGGER } from '@/lib/mothership/jobs/claim'
 import { dispatchJobOccurrence } from '@/lib/mothership/jobs/dispatch'
@@ -22,7 +22,17 @@ const JOB_RUN_TIME_LIMIT_MS = 6 * 60 * 60_000
 
 const RECONCILE_BATCH_SIZE = 100
 
-function openJobRuns(status: 'queued' | 'running') {
+type OpenJobRunStatus = 'queued' | 'running'
+
+/** An occurrence's run record that has not ended; `job_execution_logs_open_idx` covers it. */
+export const openJobRun = (statuses: OpenJobRunStatus[]) =>
+  and(
+    inArray(jobExecutionLogs.status, statuses),
+    eq(jobExecutionLogs.trigger, JOB_RUN_TRIGGER),
+    isNotNull(jobExecutionLogs.occurrenceAt)
+  )
+
+function openJobRuns(status: OpenJobRunStatus) {
   return db
     .select({
       executionId: jobExecutionLogs.executionId,
@@ -39,13 +49,7 @@ function openJobRuns(status: 'queued' | 'running') {
     .from(jobExecutionLogs)
     .innerJoin(workflowSchedule, eq(workflowSchedule.id, jobExecutionLogs.scheduleId))
     .leftJoin(copilotRuns, eq(copilotRuns.streamId, jobExecutionLogs.executionId))
-    .where(
-      and(
-        eq(jobExecutionLogs.status, status),
-        eq(jobExecutionLogs.trigger, JOB_RUN_TRIGGER),
-        isNotNull(jobExecutionLogs.occurrenceAt)
-      )
-    )
+    .where(openJobRun([status]))
     .orderBy(asc(jobExecutionLogs.startedAt))
     .limit(RECONCILE_BATCH_SIZE)
 }