Skip to content

Commit bec4639

Browse files
authored
feat(mothership): run scheduled jobs as durable unattended chat turns (#9331)
* feat(mothership): run unattended turns durably on the leased chat path A scheduled job's occurrence must outlive the Sim process driving it. The headless lifecycle the wake door uses takes no controller lease, so a pod death leaves its run unrecoverable. An unattended turn now admits like a typed send (user message, run and recovery intent in one commit under the chat lock), runs detached from any HTTP response, and records a new unattended_stream recovery kind. Takeover restores it non-interactive with the secret authority it was admitted with, instead of the interactive default, and the abandoned-run sweep, stream liveness and billing-admission recording treat both durable kinds alike. Durable start now also forwards a caller's billing-admission hook, so a job can key its per-run budget on the run's billing request. The integration test kills the turn's controller three times mid-run; it fails on a headless implementation, on a sweep blind to the new kind, and on a takeover that restores the interactive default. * feat(db): add scheduled job occurrence columns and archive legacy jobs Scheduled jobs return as Sim-owned schedules whose occurrences run as durable chat turns. A job's run record (job_execution_logs) gains the occurrence it runs (unique per schedule), its outcome and summary, and the billing request its charges settle under; a schedule gains a per-run spend cap. Indexes are built concurrently. Every job row written for the retired Go runner is archived, never deleted, so none can start once the tick claims jobs again. The archive is a script migration committing per keyed batch, so it finishes on any row count and resumes after a crash. * feat(mothership): run scheduled jobs from the schedule tick Behind MOTHERSHIP_JOBS_ENABLED (off by default, and to stay off in production until unattended runs hold production effects for approval), the schedule tick gains a job lane: - Claim: each due job's occurrence is recorded (queued) and the schedule advances past it in one transaction, skipping locked rows, so two ticks record it once and a crash never leaves a stale claim. Missed occurrences collapse into one; cadence is never under 15 minutes; a one-shot job (no cron) runs once and completes; max_runs completes. - Dispatch: the occurrence starts as an unattended durable turn in the job's chat under its occurrence id (UUIDv5 of schedule and instant), so every retry finds the turn it admitted. A busy chat skips the occurrence; a payer over its plan limit skips it; an owner who lost access turns the job off. Only secrets a person named for the job mount. - Reconcile: a queued occurrence whose dispatch was lost is dispatched again after two minutes; a running one whose turn ended is finalized; one past six hours is stopped. - Finalize: exactly once under a row lock. A reported outcome is kept; a turn that ended well without one is not_met, summarized by its last reply. Three failed or interrupted runs in a row turn the job off and email its people; a met until_complete job completes. - Budget: each billing callback records the run's cumulative settled cost on its record (keyed by the run's billing request), and a run at its cap pauses at its next step with a job_run_budget card. The cost is cumulative per billing request, which recovered legs reuse, so the cap survives worker hand-offs and Sim takeovers. * improvement(mothership): run the job lane unflagged behind one indexed probe * improvement(mothership): simplify the job lane and stop jobs through the shared Stop delivery * chore(db): format the scheduled job migration metadata * improvement(mothership): keep the durable-run predicates out of the chat protocol graph * fix(mothership): settle job runs before claiming, keep spend-cap links and failure counts durable * fix(mothership): stop orphaned job runs at their limit and skip any number of deleted occurrences * fix(mothership): end a running job record whose run was removed
1 parent ff98c86 commit bec4639

49 files changed

Lines changed: 34815 additions & 119 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎apps/sim/app/api/billing/update-cost/route.ts‎

Lines changed: 28 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@ import { BILLING_CALLBACK_OUTCOME } from '@/lib/mothership/generated/billing-pro
4949
import { BillingRouteOutcome } from '@/lib/mothership/generated/trace-attribute-values-v1'
5050
import { TraceAttr } from '@/lib/mothership/generated/trace-attributes-v1'
5151
import { TraceSpan } from '@/lib/mothership/generated/trace-spans-v1'
52+
import { recordJobRunSpend } from '@/lib/mothership/jobs/budget'
5253
import { checkInternalApiKey } from '@/lib/mothership/request/http'
5354
import { withIncomingGoSpan } from '@/lib/mothership/request/otel'
5455

@@ -122,6 +123,28 @@ async function readUsageStanding(
122123
}
123124
}
124125

126+
/**
127+
* Adds a scheduled job run's own spend cap to the payer's standing. A payer over its plan limit
128+
* keeps the plan card; otherwise a job run at its cap pauses with the job's card. A cap that
129+
* cannot be read answers as the payer's standing alone, like a standing read that fails.
130+
*/
131+
async function withJobRunBudget(
132+
verdict: BillingUsageVerdict,
133+
billingRequestId: string | undefined,
134+
cumulativeCostUsd: number
135+
): Promise<BillingUsageVerdict> {
136+
if (verdict.usageExceeded || !billingRequestId) return verdict
137+
try {
138+
const card = await recordJobRunSpend(billingRequestId, cumulativeCostUsd)
139+
return card ? { usageExceeded: true, usageUpgrade: card } : verdict
140+
} catch (error) {
141+
logger.warn('Job run spend could not be recorded; answering the payer standing alone', {
142+
error: toError(error).message,
143+
})
144+
return verdict
145+
}
146+
}
147+
125148
function getBillingResolution(
126149
isMarkerlessLegacy: boolean,
127150
billingAttribution: BillingAttributionSnapshot | undefined
@@ -442,7 +465,11 @@ async function updateCostInner(req: NextRequest, span: Span): Promise<NextRespon
442465
await checkAndBillOverageThreshold(userId, undefined, { onError: 'throw' })
443466
}
444467

445-
const usageVerdict = await readUsageStanding(userId, billingAttribution, accountDecision)
468+
const usageVerdict = await withJobRunBudget(
469+
await readUsageStanding(userId, billingAttribution, accountDecision),
470+
billingRequestId,
471+
cost
472+
)
446473
const duration = Date.now() - startTime
447474

448475
// Same-or-lower cumulative than already recorded: nothing new to bill.

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,11 @@ vi.mock('@/lib/execution/manual-cancellation', () => ({
108108

109109
vi.mock('@/lib/core/async-jobs', () => asyncJobsMock)
110110

111+
/** The job lane has its own integration suite. */
112+
vi.mock('@/lib/mothership/jobs/lane', () => ({
113+
runJobLane: vi.fn(async () => ({ claimed: 0 })),
114+
}))
115+
111116
vi.mock('drizzle-orm', () => ({
112117
and: vi.fn((...conditions: unknown[]) => ({ type: 'and', conditions })),
113118
eq: vi.fn((field: unknown, value: unknown) => ({ field, value, type: 'eq' })),

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

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@ import {
5050
registerManualExecutionAborter,
5151
unregisterManualExecutionAborter,
5252
} from '@/lib/execution/manual-cancellation'
53+
import { runJobLane } from '@/lib/mothership/jobs/lane'
5354
import {
5455
buildCarrierReconciledMetadata,
5556
carrierNotReconciledSql,
@@ -1788,6 +1789,13 @@ export async function runScheduleTick(requestId: string): Promise<ScheduleTickRe
17881789
await Promise.allSettled(schedulePromises)
17891790
}
17901791

1792+
try {
1793+
const { claimed } = await runJobLane()
1794+
totalSchedules += claimed
1795+
} catch (error) {
1796+
logger.error(`[${requestId}] Scheduled job lane failed`, { error: toError(error).message })
1797+
}
1798+
17911799
const totalCount = totalSchedules
17921800
const durationMs = Date.now() - tickStart
17931801
logger.info(

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

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -523,3 +523,28 @@ describe('CredentialDisplay link tag', () => {
523523
act(() => root.unmount())
524524
})
525525
})
526+
527+
describe('UsageUpgradeDisplay', () => {
528+
it("shows a job run's own spend cap without an account billing action", () => {
529+
;(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true
530+
const container = document.createElement('div')
531+
const root = createRoot(container)
532+
act(() => {
533+
root.render(
534+
<SpecialTags
535+
segment={{
536+
type: 'usage_upgrade',
537+
data: {
538+
reason: 'job_run_budget',
539+
action: 'increase_limit',
540+
message: 'This scheduled job run reached its $5.00 spend limit.',
541+
},
542+
}}
543+
/>
544+
)
545+
})
546+
547+
expect(container.querySelector('a')).toBeNull()
548+
act(() => root.unmount())
549+
})
550+
})

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

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3443,6 +3443,43 @@ function MothershipErrorDisplay({ data }: { data: MothershipErrorTagData }) {
34433443
}
34443444

34453445
function UsageUpgradeDisplay({ data }: { data: UsageUpgradeTagData }) {
3446+
return data.reason === 'job_run_budget' ? (
3447+
<JobRunBudgetDisplay message={data.message} />
3448+
) : (
3449+
<PlanUsageLimitDisplay data={data} />
3450+
)
3451+
}
3452+
3453+
/**
3454+
* A scheduled job's run reached its own spend cap. The cap belongs to the job, not to the
3455+
* account, so the card offers no billing action: the next run starts with a fresh cap.
3456+
*/
3457+
interface JobRunBudgetDisplayProps {
3458+
message: string
3459+
}
3460+
3461+
function JobRunBudgetDisplay({ message }: JobRunBudgetDisplayProps) {
3462+
return (
3463+
<InteractionCard
3464+
title={
3465+
<span className='inline-flex items-center gap-2'>
3466+
<TriangleAlert
3467+
aria-hidden
3468+
className='size-[14px] shrink-0 text-[var(--badge-amber-text)]'
3469+
/>
3470+
Run Spend Limit Reached
3471+
</span>
3472+
}
3473+
>
3474+
<div className='px-2 pb-2'>
3475+
<p className='text-[var(--text-body)] text-small leading-5'>{message}</p>
3476+
</div>
3477+
</InteractionCard>
3478+
)
3479+
}
3480+
3481+
/** The payer's plan limit: the card links to where the plan or the limit can change. */
3482+
function PlanUsageLimitDisplay({ data }: { data: UsageUpgradeTagData }) {
34463483
const { data: session } = useSession()
34473484
const hostContext = useOptionalWorkspaceHostContext()
34483485
const organizationContext = useOptionalOrganizationContext()

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

Lines changed: 30 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@ import { useQueryClient } from '@tanstack/react-query'
2424
import { defaultRangeExtractor, type Range, useVirtualizer } from '@tanstack/react-virtual'
2525
import { ComputerUseActivity } from '@/components/computer-use/activity'
2626
import { SMOOTH_CHASE_RATE } from '@/lib/core/utils/smooth-bottom-chase'
27+
import type { TurnOrigin } from '@/lib/mothership/generated/protocol'
2728
import type { WorkspaceFileRecord } from '@/lib/uploads/contexts/workspace'
2829
import { inter } from '@/app/_styles/fonts/inter/inter'
2930
import { FindBar } from '@/app/workspace/[workspaceId]/components/find-bar'
@@ -187,24 +188,36 @@ interface UserMessageRowProps {
187188
}
188189

189190
/**
190-
* A background-task notification that opened a turn (mothership 21-background-tasks.md
191-
* §6.4): a muted system chip, never a user bubble — the user did not type it. Shows the
192-
* outcome line; the provenance header and the agent's own note stay out of the way.
191+
* A turn automation opened, never a user bubble — the user did not type it: a muted system chip
192+
* with one line. A background-task notification (mothership 21-background-tasks.md §6.4) shows
193+
* its outcome line; a scheduled job's run shows the brief's first line (the job and its run).
193194
*/
194-
const TaskNotificationRow = memo(function TaskNotificationRow({
195-
content,
196-
rowClassName,
197-
}: {
195+
interface AutomatedTurnRowProps {
196+
origin: TurnOrigin
198197
content: string
199198
rowClassName: string
200-
}) {
199+
}
200+
201+
const AutomatedTurnRow = memo(function AutomatedTurnRow({
202+
origin,
203+
content,
204+
rowClassName,
205+
}: AutomatedTurnRowProps) {
201206
const lines = content.split('\n').filter((line) => line.length > 0)
202-
const outcome = lines.find((line) => line.startsWith('Task ')) ?? lines[lines.length - 1] ?? ''
207+
const summary =
208+
origin === 'task'
209+
? (lines.find((line) => line.startsWith('Task ')) ?? lines[lines.length - 1] ?? '').replace(
210+
/^Task [0-9a-f-]+ /,
211+
''
212+
)
213+
: (lines[0] ?? '').replace(/^Job: /, '')
203214
return (
204215
<div className={cn('flex w-full justify-center', rowClassName)}>
205216
<div className='max-w-[85%] rounded-full border border-[var(--border)] bg-[var(--bg)] px-3 py-1 text-[12px] text-[var(--text-secondary)]'>
206-
<span className='mr-1.5 font-medium text-[var(--text-primary)]'>Background task</span>
207-
{outcome.replace(/^Task [0-9a-f-]+ /, '')}
217+
<span className='mr-1.5 font-medium text-[var(--text-primary)]'>
218+
{origin === 'task' ? 'Background task' : 'Scheduled job'}
219+
</span>
220+
{summary}
208221
</div>
209222
</div>
210223
)
@@ -930,8 +943,12 @@ export function MothershipChat({
930943
style={{ top: virtualItem.start }}
931944
>
932945
{msg.role === 'user' ? (
933-
interactionPairing.hiddenUserByIndex[index] ? null : msg.origin === 'task' ? (
934-
<TaskNotificationRow content={msg.content} rowClassName={styles.rowGap} />
946+
interactionPairing.hiddenUserByIndex[index] ? null : msg.origin ? (
947+
<AutomatedTurnRow
948+
origin={msg.origin}
949+
content={msg.content}
950+
rowClassName={styles.rowGap}
951+
/>
935952
) : (
936953
<UserMessageRow
937954
content={msg.content}

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

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,18 @@ describe('useChatFind', () => {
112112
expect(find.current).toMatchObject({ isOpen: false, query: '', count: 0 })
113113
})
114114

115+
it('leaves turns automation opened out of the search, whatever opened them', () => {
116+
const find = renderFind({
117+
messages: [
118+
{ id: 't', role: 'user', origin: 'task', content: 'Task abc finished: triage done.' },
119+
{ id: 'j', role: 'user', origin: 'job', content: 'Job: Nightly triage (Run 1)' },
120+
assistant('a', 'The triage is done.'),
121+
],
122+
})
123+
find.search('triage')
124+
expect(find.current.count).toBe(1)
125+
})
126+
115127
it('counts the same footnoted matches in the index as it highlights in the rendered message', async () => {
116128
const find = renderFind({
117129
chatId: 'chat',

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -163,7 +163,7 @@ export function useChatFind({
163163
if (term) {
164164
for (let messageIndex = messages.length - 1; messageIndex >= 0; messageIndex--) {
165165
const message = messages[messageIndex]
166-
if (!message || hiddenUserByIndex[messageIndex] || message.origin === 'task') continue
166+
if (!message || hiddenUserByIndex[messageIndex] || message.origin) continue
167167
const text = getMessageSearchText(message, textCache)
168168
const occurrences: ChatFindMatch[] = []
169169
let found = 0

‎apps/sim/app/workspace/[workspaceId]/home/types.ts‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ import type { WorkspaceSearchFilters } from '@/lib/api/contracts/knowledge/searc
22
import type { MothershipChat } from '@/lib/api/contracts/mothership-chats'
33
import type { ManagedMcpConnectorId } from '@/lib/credential-groups/managed-mcp-connectors'
44
import type { AssistantSearchLevel } from '@/lib/mothership/generated/assistant'
5-
import type { ToolActivity } from '@/lib/mothership/generated/protocol'
5+
import type { ToolActivity, TurnOrigin } from '@/lib/mothership/generated/protocol'
66
import type { ChatContext } from '@/stores/panel'
77
import type { BrowserTextSelection, TerminalTextSelection } from '@/stores/panel/types'
88

@@ -197,8 +197,8 @@ export interface ChatMessage {
197197
role: 'user' | 'assistant'
198198
requestMode?: 'agent' | 'assistant' | 'plan'
199199
content: string
200-
/** "task": a background-task notification opened this turn, not the user (a system chip, not a bubble). */
201-
origin?: 'task'
200+
/** Automation opened this turn, not the user: a system chip, not a bubble. */
201+
origin?: TurnOrigin
202202
contentBlocks?: ContentBlock[]
203203
attachments?: ChatMessageAttachment[]
204204
fileAttachments?: FileAttachmentForApi[]

‎apps/sim/lib/api/contracts/subscription.ts‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,11 @@ import {
55
usageLimitReadSchema,
66
usageLimitUpdateSchema,
77
} from '@/lib/billing/application/usage-limit-validation'
8-
import { BillingCallbackBody, BillingCallbackHeaders } from '@/lib/mothership/generated/billing'
8+
import {
9+
BillingCallbackBody,
10+
BillingCallbackHeaders,
11+
UsageUpgrade,
12+
} from '@/lib/mothership/generated/billing'
913

1014
const booleanQueryParamSchema = z
1115
.preprocess((value) => {
@@ -315,8 +319,9 @@ export const billingSwitchPlanResponseSchema = z.object({
315319
* limit writes the tag with this payload verbatim into its durable log.
316320
*/
317321
export const usageUpgradePayloadSchema = z.object({
318-
reason: z.literal('usage_limit'),
319-
action: z.enum(['upgrade_plan', 'increase_limit']),
322+
/** `job_run_budget`: a scheduled job's run reached its own spend cap, not the payer's plan. */
323+
reason: UsageUpgrade.shape.reason,
324+
action: UsageUpgrade.shape.action,
320325
message: z.string(),
321326
})
322327
export type UsageUpgradePayload = z.infer<typeof usageUpgradePayloadSchema>

0 commit comments

Comments
 (0)