Skip to content

Commit 50e66ae

Browse files
committed
fix(mothership): settle a too-late call as never run only under open admission
1 parent 5457cfa commit 50e66ae

5 files changed

Lines changed: 101 additions & 28 deletions

File tree

‎apps/sim/lib/mothership/async-runs/orphaned-runs.integration.ts‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -262,6 +262,56 @@ describe.runIf(Boolean(redisUrl))('Chat runs no controller owns', () => {
262262
return { ...run, marker: chat?.conversationId ?? null }
263263
}
264264

265+
describe('a Sim claim past its start window', () => {
266+
async function lateTool(
267+
runId: string,
268+
fields: Partial<{ status: 'running'; claimedBy: string }> = {}
269+
) {
270+
const tool = await dispatchedTool(runId)
271+
await db
272+
.update(copilotAsyncToolCalls)
273+
.set({ executionDeadlineAt: sql`clock_timestamp() + interval '30 seconds'`, ...fields })
274+
.where(eq(copilotAsyncToolCalls.toolCallId, tool.toolCallId))
275+
return tool
276+
}
277+
278+
async function storedTool(toolCallId: string) {
279+
const [row] = await db
280+
.select()
281+
.from(copilotAsyncToolCalls)
282+
.where(eq(copilotAsyncToolCalls.toolCallId, toolCallId))
283+
return row
284+
}
285+
286+
it('leaves an admitted pending call unclaimed for its caller to settle as never run', async () => {
287+
const run = await admittedRun()
288+
const tool = await lateTool(run.runId)
289+
290+
expect(await claimToolExecution(tool)).toEqual({ outcome: 'too_late' })
291+
expect(await storedTool(tool.toolCallId)).toMatchObject({
292+
status: 'pending',
293+
executionStartedAt: null,
294+
executionOwnerToken: null,
295+
})
296+
})
297+
298+
it('reports a stopped run as closed before the start window', async () => {
299+
const run = await admittedRun({ stopped: true })
300+
const tool = await lateTool(run.runId)
301+
302+
expect(await claimToolExecution(tool)).toEqual({ outcome: 'closed' })
303+
})
304+
305+
it('still claims a call a client already claimed, and one without a deadline', async () => {
306+
const run = await admittedRun()
307+
const clientClaimed = await lateTool(run.runId, { status: 'running', claimedBy: 'client' })
308+
const undated = await dispatchedTool(run.runId)
309+
310+
expect(await claimToolExecution(clientClaimed)).toEqual({ outcome: 'claimed' })
311+
expect(await claimToolExecution(undated)).toEqual({ outcome: 'claimed' })
312+
})
313+
})
314+
265315
it('settles a run whose controller died once its recovery window has passed', async () => {
266316
const orphan = await admittedRun({ idleMinutes: 90, status: 'paused_waiting_for_tool' })
267317

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

Lines changed: 23 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -868,6 +868,12 @@ export type ToolExecutionClaim =
868868
/** Already claimed or settled. */
869869
| { outcome: 'existing' }
870870

871+
/** A Sim handler's claim, which also reports an unclaimed call too late to start. */
872+
export type SimToolExecutionClaim =
873+
| ToolExecutionClaim
874+
/** Admitted but unclaimed, with too little of its deadline left to start: nothing ran it. */
875+
| { outcome: 'too_late' }
876+
871877
/**
872878
* Every condition at once, typed as a condition: drizzle's `and` may return undefined, which a
873879
* negation would render as `NOT (NULL)` and match nothing.
@@ -904,6 +910,13 @@ const tooLateToStart = sql`(${copilotAsyncToolCalls.executionDeadlineAt} IS NOT
904910
*/
905911
const pastResumeGiveUp = sql`(${copilotAsyncToolCalls.executionDeadlineAt} IS NOT NULL AND ${copilotAsyncToolCalls.executionDeadlineAt} + ${TOOL_WATCHDOG_RESUME_GRACE_MS} * interval '1 millisecond' <= clock_timestamp())`
906912

913+
/** A call nothing claimed or started whose deadline leaves too little to start it. */
914+
const pendingTooLateToStart = allOf(
915+
eq(copilotAsyncToolCalls.status, ASYNC_TOOL_STATUS.pending),
916+
isNull(copilotAsyncToolCalls.executionStartedAt),
917+
tooLateToStart
918+
)
919+
907920
/**
908921
* A Sim execution whose owner died before recording a result: its lease lapsed, which a live
909922
* owner's heartbeat never lets happen. A takeover runs it again until it has started
@@ -932,11 +945,12 @@ const rerunnableSimExecution = allOf(
932945
* Sim claims a call it is about to execute under an execution lease; a terminal result never
933946
* releases it. The claim takes a call's first execution, or re-runs one whose owner died
934947
* mid-execution: the new owner token fences the dead owner out, since its renewals and its
935-
* result both require the token and an unexpired lease.
948+
* result both require the token and an unexpired lease. A pending call too late to start is left
949+
* unclaimed, under the same admission check, for its caller to settle as never run.
936950
*/
937951
export async function claimToolExecution(
938952
owner: SimToolExecutionOwner
939-
): Promise<ToolExecutionClaim> {
953+
): Promise<SimToolExecutionClaim> {
940954
return await claimUnderRunAdmission(owner, 'sim-stream', true, async (tx, thisCall) => {
941955
const startedAt = new Date()
942956
const [claimed] = await traceMothershipQuery('UPDATE', 'copilot_async_tool_calls', () =>
@@ -961,7 +975,8 @@ export async function claimToolExecution(
961975
inArray(copilotAsyncToolCalls.status, [
962976
ASYNC_TOOL_STATUS.pending,
963977
ASYNC_TOOL_STATUS.running,
964-
])
978+
]),
979+
not(pendingTooLateToStart)
965980
),
966981
interruptedSimExecution
967982
)
@@ -971,10 +986,13 @@ export async function claimToolExecution(
971986
)
972987
if (claimed) return { outcome: 'claimed' } as const
973988
const [record] = await traceMothershipQuery('SELECT', 'copilot_async_tool_calls', () =>
974-
tx.select({ id: copilotAsyncToolCalls.id }).from(copilotAsyncToolCalls).where(thisCall)
989+
tx
990+
.select({ tooLate: sql<boolean>`${pendingTooLateToStart}` })
991+
.from(copilotAsyncToolCalls)
992+
.where(thisCall)
975993
)
976994
if (!record) throw new Error('Tool execution record is unavailable')
977-
return { outcome: 'existing' } as const
995+
return record.tooLate ? ({ outcome: 'too_late' } as const) : ({ outcome: 'existing' } as const)
978996
})
979997
}
980998

‎apps/sim/lib/mothership/request/tools/execution-lifetime.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,9 +15,9 @@ import {
1515
prepareWorkbenchAccess,
1616
recordSimSandboxProcess,
1717
renewSimToolExecutionLease,
18+
type SimToolExecutionClaim,
1819
settleSimSandboxProcess,
1920
settleSimToolExecution,
20-
type ToolExecutionClaim,
2121
} from '@/lib/mothership/async-runs/repository'
2222
import { TraceAttr } from '@/lib/mothership/generated/trace-attributes-v1'
2323
import { TraceSpan } from '@/lib/mothership/generated/trace-spans-v1'
@@ -30,7 +30,7 @@ export interface ToolExecutionLifetime {
3030
readonly signal: AbortSignal
3131
readonly owner?: SimToolExecutionOwner
3232
complete(input: CompleteAsyncToolCallInput): Promise<void>
33-
claim(runId: string, userId: string): Promise<ToolExecutionClaim>
33+
claim(runId: string, userId: string): Promise<SimToolExecutionClaim>
3434
hold<T>(work: Promise<T>): Promise<T>
3535
}
3636

‎apps/sim/lib/mothership/request/tools/executor.test.ts‎

Lines changed: 13 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -369,8 +369,10 @@ describe('executeToolAndReport provenance isolation', () => {
369369
beforeEach(() => {
370370
executeTool.mockReset()
371371
completeAsyncToolCall.mockResolvedValue(null)
372+
completePendingAsyncToolCall.mockReset()
372373
markAsyncToolRunning.mockResolvedValue(null)
373374
upsertAsyncToolCall.mockResolvedValue(null)
375+
claimToolExecution.mockReset()
374376
claimToolExecution.mockResolvedValue({ outcome: 'claimed' })
375377
settleSimToolExecution.mockResolvedValue(undefined)
376378
waitForToolConfirmation.mockReset()
@@ -698,7 +700,8 @@ describe('executeToolAndReport provenance isolation', () => {
698700
expect(tool.error).toBe(TOOL_DEADLINE_BEFORE_START_MESSAGE)
699701
})
700702

701-
it('settles an unclaimed call past its start window as never run, without claiming it', async () => {
703+
it('settles a call its claim found too late to start as never run, without claiming it', async () => {
704+
claimToolExecution.mockResolvedValueOnce({ outcome: 'too_late' })
702705
completePendingAsyncToolCall.mockImplementationOnce(async (input) => ({ ...input }))
703706
const tool: ToolCallState = {
704707
...buildPendingToolCall(),
@@ -711,7 +714,6 @@ describe('executeToolAndReport provenance isolation', () => {
711714
resolvedSecretTraceRegistry: new ResolvedSecretTraceRegistry(),
712715
})
713716

714-
expect(claimToolExecution).not.toHaveBeenCalled()
715717
expect(executeTool).not.toHaveBeenCalled()
716718
expect(completePendingAsyncToolCall).toHaveBeenCalledExactlyOnceWith(
717719
expect.objectContaining({
@@ -727,22 +729,25 @@ describe('executeToolAndReport provenance isolation', () => {
727729
})
728730
})
729731

730-
it('leaves a call past its start window that is already claimed to its claim', async () => {
731-
completePendingAsyncToolCall.mockResolvedValueOnce(null)
732+
it('settles the call of a stopped run as stopped even past its start window', async () => {
733+
claimToolExecution.mockResolvedValueOnce({ outcome: 'closed' })
734+
completePendingAsyncToolCall.mockImplementation(async (input) => ({ ...input }))
732735
const tool: ToolCallState = {
733736
...buildPendingToolCall(),
734737
name: 'generate_audio',
735738
deadlineAt: Date.now() + 30_000,
736739
}
737-
await executeToolAndReport(tool.id, buildStreamingContext(tool), {
740+
const completion = await executeToolAndReport(tool.id, buildStreamingContext(tool), {
738741
userId: 'user-1',
739742
workflowId: 'workflow-1',
740743
resolvedSecretTraceRegistry: new ResolvedSecretTraceRegistry(),
741744
})
742745

743-
expect(claimToolExecution).toHaveBeenCalledOnce()
744-
expect(executeTool).not.toHaveBeenCalled()
745-
expect(tool.error).toBe(TOOL_DEADLINE_BEFORE_START_MESSAGE)
746+
expect(completePendingAsyncToolCall).not.toHaveBeenCalled()
747+
expect(completion).toMatchObject({
748+
status: 'cancelled',
749+
message: 'Run stopped before tool admission',
750+
})
746751
})
747752

748753
it('claims a call without a deadline without trying to settle it first', async () => {

‎apps/sim/lib/mothership/request/tools/executor.ts‎

Lines changed: 13 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -754,22 +754,22 @@ async function executeToolAndReportInner(
754754
})
755755
/** A re-run after a takeover gets what remains of the call's first deadline, never a fresh one. */
756756
if (record?.executionDeadlineAt) toolCall.deadlineAt = record.executionDeadlineAt.getTime()
757-
// Too late to start: an unclaimed call is settled as never run rather than claimed first, so a
758-
// crash before its handler refuses to start cannot leave it recovered as an unknown outcome.
759-
if (
760-
toolStartWindowClosed(toolCall.deadlineAt, Date.now()) &&
761-
(await failUnstartedToolCall(
762-
toolCall.id,
763-
context,
764-
execContext,
765-
TOOL_DEADLINE_BEFORE_START_MESSAGE
766-
))
767-
)
768-
return terminalCompletionFromToolCall(toolCall)
769757
if (context.runId) {
770758
const claim = await lifetime.claim(context.runId, execContext.userId)
771759
if (claim.outcome === 'closed') return settleCancelled('Run stopped before tool admission')
772-
if (claim.outcome === 'existing') {
760+
// Settled as never run instead of claimed, so a crash before the handler refuses to start
761+
// cannot leave it recovered as an unknown outcome. Losing that settlement leaves another writer's.
762+
if (
763+
claim.outcome === 'too_late' &&
764+
(await failUnstartedToolCall(
765+
toolCall.id,
766+
context,
767+
execContext,
768+
TOOL_DEADLINE_BEFORE_START_MESSAGE
769+
))
770+
)
771+
return terminalCompletionFromToolCall(toolCall)
772+
if (claim.outcome !== 'claimed') {
773773
/** The winning controller owns execution; this promise observes its durable result. */
774774
const completion = await waitForToolConfirmation(
775775
toolCall.id,

0 commit comments

Comments
 (0)