Skip to content

Commit 1049e99

Browse files
committed
improvement(mothership): simplify the job lane and stop jobs through the shared Stop delivery
1 parent 7ed96b5 commit 1049e99

23 files changed

Lines changed: 259 additions & 217 deletions

File tree

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

Lines changed: 3 additions & 6 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,11 +197,8 @@ export interface ChatMessage {
197197
role: 'user' | 'assistant'
198198
requestMode?: 'agent' | 'assistant' | 'plan'
199199
content: string
200-
/**
201-
* "task": a background-task notification opened this turn, not the user (a system chip, not a
202-
* bubble). "job": a scheduled job's occurrence opened it with the job brief.
203-
*/
204-
origin?: 'task' | 'job'
200+
/** Automation opened this turn, not the user: a system chip, not a bubble. */
201+
origin?: TurnOrigin
205202
contentBlocks?: ContentBlock[]
206203
attachments?: ChatMessageAttachment[]
207204
fileAttachments?: FileAttachmentForApi[]

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

Lines changed: 7 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) => {
@@ -316,8 +320,8 @@ export const billingSwitchPlanResponseSchema = z.object({
316320
*/
317321
export const usageUpgradePayloadSchema = z.object({
318322
/** `job_run_budget`: a scheduled job's run reached its own spend cap, not the payer's plan. */
319-
reason: z.enum(['usage_limit', 'job_run_budget']),
320-
action: z.enum(['upgrade_plan', 'increase_limit']),
323+
reason: UsageUpgrade.shape.reason,
324+
action: UsageUpgrade.shape.action,
321325
message: z.string(),
322326
})
323327
export type UsageUpgradePayload = z.infer<typeof usageUpgradePayloadSchema>

‎apps/sim/lib/mothership/async-runs/repository.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -84,7 +84,7 @@ import { chatSandboxSessionKey } from '@/lib/mothership/tools/sandbox-session-ke
8484

8585
const logger = createLogger('CopilotAsyncRunsRepo')
8686
const WORKFLOW_EXECUTION_CLAIM_PREFIX = 'workflow:'
87-
const TERMINAL_RUN_STATUSES: CopilotRunStatus[] = ['complete', 'error', 'cancelled']
87+
export const TERMINAL_RUN_STATUSES: CopilotRunStatus[] = ['complete', 'error', 'cancelled']
8888
// Resolve the tracer lazily per-call to avoid capturing the NoOp tracer
8989
// before NodeSDK installs the global TracerProvider (Next.js 16/Turbopack
9090
// can evaluate modules before instrumentation-node.ts finishes).

‎apps/sim/lib/mothership/chat/application/admit-turn.ts‎

Lines changed: 46 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import type { z } from 'zod'
77
import { defineWorkspaceOperation } from '@/lib/core/application'
88
import { defineOrganizationOperation } from '@/lib/core/application/organization-operation'
99
import { OrchestrationError } from '@/lib/core/orchestration/types'
10+
import type { DbTransaction } from '@/lib/db/types'
1011
import { resolveTurnDesktopDevice } from '@/lib/desktop/application/executor'
1112
import { requireOrganizationSearchAvailable } from '@/lib/knowledge/access/availability'
1213
import { insertRunSegment, withRunAdmissionLock } from '@/lib/mothership/async-runs/repository'
@@ -58,6 +59,50 @@ export async function turnDesktopDevice(
5859
return resolveTurnDesktopDevice(principal, offeredDeviceId)
5960
}
6061

62+
interface TurnChat {
63+
userId: string
64+
workspaceId?: string | null
65+
organizationId?: string | null
66+
chatId: string
67+
}
68+
69+
/**
70+
* Makes an admitted turn its chat's current conversation, inside the admission's transaction:
71+
* the chat's conversation mode follows the turn's, and a send's effort pick stays as the chat's
72+
* choice. A chat deleted or moved since its context was resolved is `not_found`.
73+
*/
74+
export async function moveChatToTurn(
75+
tx: DbTransaction,
76+
chat: TurnChat,
77+
turn: { messageId: string; mode?: string },
78+
effortChoice?: MothershipEffort
79+
): Promise<{ model: string | null }> {
80+
const turnConfig = sql`COALESCE(${copilotChats.config}, '{}'::jsonb) || jsonb_build_object('conversationMode', ${turn.mode ?? 'agent'}::text)`
81+
const [updated] = await tx
82+
.update(copilotChats)
83+
.set({
84+
conversationId: turn.messageId,
85+
updatedAt: new Date(),
86+
config: effortChoice ? withChatEffortChoice(turnConfig, effortChoice) : turnConfig,
87+
})
88+
.where(
89+
and(
90+
eq(copilotChats.id, chat.chatId),
91+
eq(copilotChats.userId, chat.userId),
92+
chat.workspaceId
93+
? eq(copilotChats.workspaceId, chat.workspaceId)
94+
: isNull(copilotChats.workspaceId),
95+
chat.organizationId
96+
? eq(copilotChats.organizationId, chat.organizationId)
97+
: isNull(copilotChats.organizationId),
98+
isNull(copilotChats.deletedAt)
99+
)
100+
)
101+
.returning({ model: copilotChats.model })
102+
if (!updated) throw new OrchestrationError('not_found', 'Chat not found')
103+
return updated
104+
}
105+
61106
/** The accepted message, its start intent and retry destination commit together. */
62107
export const admitChatTurn = defineAuthorizedChatUseCase({
63108
operation: defineWorkspaceOperation({
@@ -102,32 +147,8 @@ export const admitChatTurn = defineAuthorizedChatUseCase({
102147
}
103148
await assertChatStreamLease(input.lease)
104149
const desktopDeviceId = await turnDesktopDevice(principal, workspaceId, input.desktopDeviceId)
105-
const turnConfig = sql`COALESCE(${copilotChats.config}, '{}'::jsonb) || jsonb_build_object('conversationMode', ${request.mode ?? 'agent'}::text)`
106150
return withRunAdmissionLock(userId, request.messageId, async (tx) => {
107-
const [chat] = await tx
108-
.update(copilotChats)
109-
.set({
110-
conversationId: request.messageId,
111-
updatedAt: new Date(),
112-
config: input.effortChoice
113-
? withChatEffortChoice(turnConfig, input.effortChoice)
114-
: turnConfig,
115-
})
116-
.where(
117-
and(
118-
eq(copilotChats.id, chatId),
119-
eq(copilotChats.userId, userId),
120-
workspaceId
121-
? eq(copilotChats.workspaceId, workspaceId)
122-
: isNull(copilotChats.workspaceId),
123-
organizationId
124-
? eq(copilotChats.organizationId, organizationId)
125-
: isNull(copilotChats.organizationId),
126-
isNull(copilotChats.deletedAt)
127-
)
128-
)
129-
.returning({ model: copilotChats.model })
130-
if (!chat) throw new OrchestrationError('not_found', 'Chat not found')
151+
const chat = await moveChatToTurn(tx, context, request, input.effortChoice)
131152
await appendCopilotChatMessages(
132153
chatId,
133154
[buildPersistedUserMessage(input.message)],

‎apps/sim/lib/mothership/chat/application/admit-unattended-turn.ts‎

Lines changed: 20 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,12 @@
11
import type { DelegatedPrincipal, OrganizationDelegatedPrincipal } from '@sim/auth/principal'
2-
import { copilotChats, copilotRuns } from '@sim/db/schema'
3-
import { and, eq, isNull } from 'drizzle-orm'
2+
import { copilotRuns } from '@sim/db/schema'
3+
import { eq } from 'drizzle-orm'
44
import type { z } from 'zod'
55
import { defineWorkspaceOperation } from '@/lib/core/application'
66
import { defineOrganizationOperation } from '@/lib/core/application/organization-operation'
77
import { OrchestrationError } from '@/lib/core/orchestration/types'
88
import { insertRunSegment, withRunAdmissionLock } from '@/lib/mothership/async-runs/repository'
9+
import { moveChatToTurn } from '@/lib/mothership/chat/application/admit-turn'
910
import { defineAuthorizedChatUseCase } from '@/lib/mothership/chat/application/authorized-chat-use-case'
1011
import { resolveOwnedWorkspaceChatContext } from '@/lib/mothership/chat/application/context'
1112
import { appendCopilotChatMessages } from '@/lib/mothership/chat/messages-store'
@@ -21,19 +22,13 @@ import type { TurnSecretAuthority } from '@/lib/mothership/secret-mount-policy'
2122
export const UNATTENDED_TURN_AUDIENCE = 'sim:copilot-jobs'
2223

2324
interface AdmitUnattendedTurnInput {
24-
chatId: string
25-
/** The occurrence's id: the run, its stream and its user message all carry it. */
26-
messageId: string
27-
executionId: string
28-
requestId: string
29-
message: string
30-
origin: 'job'
25+
/** The turn's request is its identity: its message id names the run, its stream and its user message. */
3126
recovery: z.input<typeof StreamRecoveryConfigSchema>
3227
lease: ChatStreamLease
3328
secretAuthority: TurnSecretAuthority
3429
}
3530

36-
export type UnattendedAdmission =
31+
type UnattendedAdmission =
3732
| { kind: 'admitted'; run: Awaited<ReturnType<typeof insertRunSegment>> }
3833
| { kind: 'duplicate' }
3934

@@ -69,7 +64,7 @@ export const admitUnattendedTurn = defineAuthorizedChatUseCase({
6964
}: {
7065
principal: DelegatedPrincipal | OrganizationDelegatedPrincipal
7166
input: AdmitUnattendedTurnInput
72-
}) => resolveOwnedWorkspaceChatContext(principal, input.chatId),
67+
}) => resolveOwnedWorkspaceChatContext(principal, input.recovery.request.chatId),
7368
authorizationOptions: {
7469
delegation: {
7570
audience: UNATTENDED_TURN_AUDIENCE,
@@ -85,61 +80,46 @@ export const admitUnattendedTurn = defineAuthorizedChatUseCase({
8580
const request = recovery.request
8681
if (
8782
recovery.kind !== 'unattended_stream' ||
83+
!request.origin ||
8884
request.userId !== userId ||
8985
request.workspaceId !== workspaceId ||
9086
request.organizationId !== undefined ||
91-
request.chatId !== chatId ||
92-
request.messageId !== input.messageId ||
93-
request.message !== input.message ||
94-
request.origin !== input.origin
87+
request.chatId !== chatId
9588
) {
9689
throw new OrchestrationError('validation', 'Turn identity does not match its chat')
9790
}
91+
const { messageId, message, origin } = request
9892
await assertChatStreamLease(input.lease)
99-
return withRunAdmissionLock(userId, input.messageId, async (tx) => {
93+
return withRunAdmissionLock(userId, messageId, async (tx) => {
10094
const [existing] = await tx
10195
.select({ id: copilotRuns.id })
10296
.from(copilotRuns)
103-
.where(eq(copilotRuns.streamId, input.messageId))
97+
.where(eq(copilotRuns.streamId, messageId))
10498
.limit(1)
10599
if (existing) return { kind: 'duplicate' } as const
106-
const [chat] = await tx
107-
.update(copilotChats)
108-
.set({ conversationId: input.messageId, updatedAt: new Date() })
109-
.where(
110-
and(
111-
eq(copilotChats.id, chatId),
112-
eq(copilotChats.userId, userId),
113-
workspaceId
114-
? eq(copilotChats.workspaceId, workspaceId)
115-
: isNull(copilotChats.workspaceId),
116-
isNull(copilotChats.deletedAt)
117-
)
118-
)
119-
.returning({ model: copilotChats.model })
120-
if (!chat) throw new OrchestrationError('not_found', 'Chat not found')
100+
const chat = await moveChatToTurn(tx, context, request)
121101
await appendCopilotChatMessages(
122102
chatId,
123103
[
124104
buildPersistedUserMessage({
125-
id: input.messageId,
126-
content: input.message,
127-
origin: input.origin,
105+
id: messageId,
106+
content: message,
107+
origin,
128108
requestMode: 'agent',
129109
}),
130110
],
131-
{ streamId: input.messageId, chatModel: chat.model },
111+
{ streamId: messageId, chatModel: chat.model },
132112
tx
133113
)
134114
const run = await insertRunSegment(tx, {
135-
id: input.messageId,
136-
executionId: input.executionId,
115+
id: messageId,
116+
executionId: messageId,
137117
chatId,
138118
userId,
139119
workspaceId,
140-
streamId: input.messageId,
120+
streamId: messageId,
141121
requestContext: {
142-
requestId: input.requestId,
122+
requestId: messageId,
143123
controllerToken: input.lease.value,
144124
recovery,
145125
secretAuthority: input.secretAuthority,

‎apps/sim/lib/mothership/chat/persisted-message.ts‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ import {
2424
MothershipStreamV1ToolOutcome,
2525
MothershipStreamV1ToolPhase,
2626
} from '@/lib/mothership/generated/mothership-stream-v1'
27-
import { ToolActivity } from '@/lib/mothership/generated/protocol'
27+
import { ToolActivity, TurnOrigin } from '@/lib/mothership/generated/protocol'
2828
import type {
2929
ContentBlock,
3030
LocalToolCallStatus,
@@ -137,11 +137,8 @@ export interface PersistedMessage {
137137
requestMode?: 'agent' | 'assistant' | 'plan'
138138
content: string
139139
timestamp: string
140-
/**
141-
* "task": a background-task notification opened this turn, not the user (rendered as a system
142-
* chip). "job": a scheduled job's occurrence opened it with the job brief.
143-
*/
144-
origin?: 'task' | 'job'
140+
/** Automation opened this turn, not the user (rendered as a system chip). */
141+
origin?: TurnOrigin
145142
requestId?: string
146143
contentBlocks?: PersistedContentBlock[]
147144
fileAttachments?: PersistedFileAttachment[]
@@ -574,7 +571,7 @@ export interface UserMessageParams {
574571
content: string
575572
fileAttachments?: PersistedFileAttachment[]
576573
contexts?: PersistedMessageContext[]
577-
origin?: 'task' | 'job'
574+
origin?: TurnOrigin
578575
}
579576

580577
export function buildPersistedUserMessage(params: UserMessageParams): PersistedMessage {
@@ -899,7 +896,8 @@ export function normalizeMessage(raw: Record<string, unknown>): PersistedMessage
899896
if (raw.requestId && typeof raw.requestId === 'string') {
900897
msg.requestId = raw.requestId
901898
}
902-
if (raw.origin === 'task' || raw.origin === 'job') msg.origin = raw.origin
899+
const origin = TurnOrigin.safeParse(raw.origin)
900+
if (origin.success) msg.origin = origin.data
903901

904902
const rawBlocks = raw.contentBlocks as RawBlock[] | undefined
905903
const rawToolCalls = raw.toolCalls as LegacyToolCall[] | undefined

‎apps/sim/lib/mothership/generated/protocol.ts‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -146,6 +146,12 @@ export const ENTITLEMENTS = { dashboards: "dashboards", tests: "tests", changelo
146146
export type Entitlement = (typeof ENTITLEMENTS)[keyof typeof ENTITLEMENTS];
147147
export const Entitlements = z.array(z.string().min(1).max(64)).max(32);
148148

149+
/** What opened a turn when no person typed it: "task", Sim's wake for a background-task
150+
* notification (21-background-tasks.md); "job", Sim's schedule for one occurrence of a scheduled
151+
* job (D92), its message the job brief. Absent on a person's turn. */
152+
export const TurnOrigin = z.enum(["task", "job"]);
153+
export type TurnOrigin = z.infer<typeof TurnOrigin>;
154+
149155
export const ChatPayloadSchema = z
150156
.strictObject({
151157
desktop: DesktopContextSchema.optional(),
@@ -192,10 +198,8 @@ export const ChatPayloadSchema = z
192198
/** Explicit client-executor declaration (see contracts ChatRequest.clientCapabilities);
193199
* the worker accepts and ignores it — dispatch semantics live on the sim side. */
194200
clientCapabilities: z.array(z.string()).default([]),
195-
/** "task": sim opened this turn for a background-task notification, not for a typed
196-
* message (21-background-tasks.md); "job": sim's schedule opened it for one occurrence of a
197-
* scheduled job (D92), its message the job brief. Recorded on the turn's user_message event. */
198-
origin: z.enum(["task", "job"]).optional(),
201+
/** Recorded on the turn's user_message event. */
202+
origin: TurnOrigin.optional(),
199203
/** Per-turn effort dial (user-selected in the composer); absent = deployment default. */
200204
effort: z.enum(["none", "low", "medium", "high", "xhigh", "max"]).optional(),
201205
modelSelection: ModelSelectionSchema.optional(),

‎apps/sim/lib/mothership/jobs/brief.ts‎

Lines changed: 6 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,20 +1,16 @@
11
import { db, jobExecutionLogs } from '@sim/db'
22
import { truncate } from '@sim/utils/string'
33
import { and, desc, eq, isNotNull, lt } from 'drizzle-orm'
4+
import type { JobRun } from '@/lib/mothership/jobs/runs'
45

56
/** How many earlier runs' outcomes a brief carries. */
67
const BRIEF_RECENT_RUNS = 5
78
const BRIEF_SUMMARY_CHARS = 400
89

9-
export interface JobBriefSchedule {
10-
id: string
11-
jobTitle: string | null
12-
prompt: string | null
13-
successCondition: string | null
14-
lifecycle: string
15-
maxRuns: number | null
16-
runCount: number
17-
}
10+
type JobBriefSchedule = Pick<
11+
JobRun,
12+
'scheduleId' | 'jobTitle' | 'prompt' | 'successCondition' | 'lifecycle' | 'maxRuns' | 'runCount'
13+
>
1814

1915
/**
2016
* The user message of one occurrence's turn: what the job is, how far along it is, and how its
@@ -35,7 +31,7 @@ export async function buildJobBrief(
3531
.from(jobExecutionLogs)
3632
.where(
3733
and(
38-
eq(jobExecutionLogs.scheduleId, schedule.id),
34+
eq(jobExecutionLogs.scheduleId, schedule.scheduleId),
3935
isNotNull(jobExecutionLogs.outcome),
4036
lt(jobExecutionLogs.occurrenceAt, occurrenceAt)
4137
)

0 commit comments

Comments
 (0)