From 7f7227a179d0f9feab08621615b8a4e5b9250a3d Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 10 Sep 2026 21:42:58 +0000 Subject: [PATCH 1/4] perf(chat): use lightweight stop markers Co-Authored-By: David Cramer --- .../junior/src/chat/task-execution/state.ts | 23 ++++++++++++ .../junior/src/chat/task-execution/store.ts | 1 + .../junior/src/chat/task-execution/worker.ts | 10 ++--- .../task-execution/conversation-work.test.ts | 37 ++++++++++++++++++- 4 files changed, 64 insertions(+), 7 deletions(-) diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index 03ec653cf0..082b96028c 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/packages/junior/src/chat/task-execution/state.ts @@ -268,6 +268,10 @@ function conversationKey(conversationId: string): string { return `${CONVERSATION_PREFIX}:${conversationId}`; } +function conversationStopKey(conversationId: string, runId: string): string { + return `${CONVERSATION_PREFIX}:stop:${conversationId}:${runId}`; +} + function indexLockKey(indexKey: string): string { return `${indexKey}:lock`; } @@ -1740,6 +1744,19 @@ function isHumanFacingMessage(message: InboundMessage): boolean { return message.source === "web" || message.source === "slack"; } +/** Return whether a durable stop request exists for one Conversation run. */ +export async function hasConversationStopRequest(args: { + conversationId: string; + runId: string; + state?: StateAdapter; +}): Promise { + const state = await getConnectedState(args.state); + return ( + (await state.get(conversationStopKey(args.conversationId, args.runId))) !== + null + ); +} + /** Persist a stop request for the current run without process affinity. */ export async function stopConversationWork(args: { conversationId: string; @@ -1770,6 +1787,11 @@ export async function stopConversationWork(args: { nowMs, ), ); + await state.set( + conversationStopKey(args.conversationId, runId), + true, + JUNIOR_THREAD_STATE_TTL_MS, + ); return { runId, status: "requested" }; }); } @@ -1823,6 +1845,7 @@ export async function completeConversationStop(args: { nowMs, ), ); + await state.delete(conversationStopKey(args.conversationId, args.runId)); return { status: "cleared", removedInboundMessageIds }; }); } diff --git a/packages/junior/src/chat/task-execution/store.ts b/packages/junior/src/chat/task-execution/store.ts index b2dc648e11..2b0d26b90d 100644 --- a/packages/junior/src/chat/task-execution/store.ts +++ b/packages/junior/src/chat/task-execution/store.ts @@ -14,6 +14,7 @@ export { CONVERSATION_WORK_MAX_DELIVERY_ATTEMPTS, CONVERSATION_WORK_MAX_RETRIES, CONVERSATION_WORK_STALE_ENQUEUE_MS, + hasConversationStopRequest, isFinalAttempt, isInvalidConversationRecordError, type AgentInput, diff --git a/packages/junior/src/chat/task-execution/worker.ts b/packages/junior/src/chat/task-execution/worker.ts index 03937b019f..672caf089c 100644 --- a/packages/junior/src/chat/task-execution/worker.ts +++ b/packages/junior/src/chat/task-execution/worker.ts @@ -24,6 +24,7 @@ import { drainConversationMailbox, ensureConversationWake, getConversationWorkState, + hasConversationStopRequest, isFinalAttempt, isInvalidConversationRecordError, recordAttemptFailure, @@ -285,7 +286,7 @@ function startLeaseCheckIn(args: { return timer; } -/** Poll shared state only when a worker adapter asks to observe remote stops. */ +/** Poll the run-scoped stop marker only when an adapter observes remote stops. */ function createConversationStopSignal(args: { conversationId: string; initialStopRunId?: string; @@ -311,13 +312,12 @@ function createConversationStopSignal(args: { if (checking || controller.signal.aborted) return; checking = true; try { - const current = await getConversationWorkState({ + const stopped = await hasConversationStopRequest({ conversationId: args.conversationId, + runId: args.runId, state: args.options.state, }); - if (current?.execution.stop?.runId === args.runId) { - requestStop(); - } + if (stopped) requestStop(); } catch (error) { if (!failureCaptured) { failureCaptured = true; diff --git a/packages/junior/tests/component/task-execution/conversation-work.test.ts b/packages/junior/tests/component/task-execution/conversation-work.test.ts index 534c7494a3..d13d20c5f3 100644 --- a/packages/junior/tests/component/task-execution/conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/conversation-work.test.ts @@ -1695,7 +1695,25 @@ describe("conversation work execution", () => { vi.useFakeTimers({ now: 1_000 }); let currentNowMs = 1_000; const queue = createConversationWorkQueueTestAdapter(); - await appendInboundMessage({ message: inboundMessage("m1"), nowMs: 1_000 }); + const state = getStateAdapter(); + const workerGetKeys: string[] = []; + const workerState = new Proxy(state, { + get(target, prop) { + if (prop === "get") { + return async (key: string) => { + workerGetKeys.push(key); + return target.get(key); + }; + } + const value = readProxyProperty(target, prop); + return typeof value === "function" ? value.bind(target) : value; + }, + }) as StateAdapter; + await appendInboundMessage({ + message: inboundMessage("m1"), + nowMs: 1_000, + state, + }); const entered = deferred(); const stopObserved = deferred(); const finishRun = deferred(); @@ -1704,6 +1722,7 @@ describe("conversation work execution", () => { nowMs: () => currentNowMs, queue, softYieldAfterMs: 1_000, + state: workerState, run: async (context) => { await context.attempt.ack(); const signal = context.stopSignal?.(); @@ -1723,22 +1742,35 @@ describe("conversation work execution", () => { }); await entered.promise; + const conversationReadsBeforeStop = workerGetKeys.filter( + (key) => key === CONVERSATION_WORK_STATE_KEY, + ).length; await appendInboundMessage({ message: inboundMessage("m2", { createdAtMs: 1_500, receivedAtMs: 1_500, }), nowMs: 1_500, + state, }); await expect( stopConversationWork({ conversationId: CONVERSATION_ID, nowMs: 2_000, + state, }), ).resolves.toMatchObject({ status: "requested" }); await vi.advanceTimersByTimeAsync(500); await stopObserved.promise; + expect( + workerGetKeys.filter((key) => key === CONVERSATION_WORK_STATE_KEY), + ).toHaveLength(conversationReadsBeforeStop); + expect( + workerGetKeys.some((key) => + key.startsWith(`junior:conversation:v2:stop:${CONVERSATION_ID}:`), + ), + ).toBe(true); await appendInboundMessage({ message: inboundMessage("m3", { @@ -1748,13 +1780,14 @@ describe("conversation work execution", () => { receivedAtMs: 2_000, }), nowMs: 2_000, + state, }); currentNowMs = 2_000; finishRun.resolve(); await expect(running).resolves.toEqual({ status: "pending_requeued" }); await expect( - getConversationWorkState({ conversationId: CONVERSATION_ID }), + getConversationWorkState({ conversationId: CONVERSATION_ID, state }), ).resolves.toMatchObject({ execution: { stop: undefined }, messages: [expect.objectContaining({ inboundMessageId: "m3" })], From f90c9f96ec505b27557fe2f71473054e4cc22154 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 10 Sep 2026 21:59:03 +0000 Subject: [PATCH 2/4] refactor(chat): make stop marker canonical Co-Authored-By: David Cramer --- .../junior/src/chat/task-execution/state.ts | 87 +++++++++---------- .../junior/src/chat/task-execution/store.ts | 2 +- .../junior/src/chat/task-execution/worker.ts | 16 ++-- .../task-execution/conversation-work.test.ts | 44 +++++----- 4 files changed, 75 insertions(+), 74 deletions(-) diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index 082b96028c..7cb0127293 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/packages/junior/src/chat/task-execution/state.ts @@ -135,7 +135,7 @@ export interface Lease { token: string; } -/** Durable request to stop the current Conversation run. */ +/** Durable request to stop one Conversation run. */ export interface ConversationStop { inboundMessageIds: string[]; runId: string; @@ -152,7 +152,6 @@ export interface ConversationExecution { pendingMessages: InboundMessage[]; runId?: string; status: ExecutionStatus; - stop?: ConversationStop; updatedAtMs?: number; } @@ -268,8 +267,8 @@ function conversationKey(conversationId: string): string { return `${CONVERSATION_PREFIX}:${conversationId}`; } -function conversationStopKey(conversationId: string, runId: string): string { - return `${CONVERSATION_PREFIX}:stop:${conversationId}:${runId}`; +function conversationStopKey(conversationId: string): string { + return `${CONVERSATION_PREFIX}:stop:${conversationId}`; } function indexLockKey(indexKey: string): string { @@ -473,8 +472,6 @@ function normalizeExecution( const lease = normalizeLease(value.lease); const runId = toOptionalString(value.runId); - const normalizedStop = normalizeStop(value.stop); - const stop = normalizedStop?.runId === runId ? normalizedStop : undefined; const normalizedStatus = status === "idle" && lease ? "running" @@ -496,7 +493,6 @@ function normalizeExecution( lastProgressAtMs: toOptionalNumber(value.lastProgressAtMs), retryCount: toOptionalNumber(value.retryCount), runId, - stop, updatedAtMs: toOptionalNumber(value.updatedAtMs), }; } @@ -1015,13 +1011,7 @@ async function writeConversation( ...conversation, execution, }; - const fenced = await state.extendLock( - lock, - CONVERSATION_MUTATION_LOCK_TTL_MS, - ); - if (!fenced) { - throw new ConversationMutationFencedError(next.conversationId); - } + await fenceConversationMutation(state, lock, next.conversationId); // TODO(dcramer): Remove the stored publishExternally field after no deployed // mailbox reader requires it. Destination is used only to write this old // Redis shape. Current workers ignore the field. @@ -1556,6 +1546,9 @@ export async function startConversationWork(args: { expiresAtMs: nowMs + CONVERSATION_WORK_LEASE_TTL_MS, }; const startsNewRun = current.execution.runId === undefined; + const stop = startsNewRun + ? await readConversationStop(state, args.conversationId) + : undefined; await writeConversation( state, lock, @@ -1565,7 +1558,7 @@ export async function startConversationWork(args: { ...current.execution, lease, status: current.execution.status === "paused" ? "paused" : "running", - runId: current.execution.runId ?? randomUUID(), + runId: current.execution.runId ?? stop?.runId ?? randomUUID(), lastEnqueuedAtMs: undefined, retryCount: startsNewRun ? 0 : current.execution.retryCount, }, @@ -1744,16 +1737,33 @@ function isHumanFacingMessage(message: InboundMessage): boolean { return message.source === "web" || message.source === "slack"; } -/** Return whether a durable stop request exists for one Conversation run. */ -export async function hasConversationStopRequest(args: { +async function readConversationStop( + state: StateAdapter, + conversationId: string, +): Promise { + return normalizeStop(await state.get(conversationStopKey(conversationId))); +} + +async function fenceConversationMutation( + state: StateAdapter, + lock: Lock, + conversationId: string, +): Promise { + if (!(await state.extendLock(lock, CONVERSATION_MUTATION_LOCK_TTL_MS))) { + throw new ConversationMutationFencedError(conversationId); + } +} + +/** Return whether one Conversation run has a durable stop request. */ +export async function hasConversationStop(args: { conversationId: string; runId: string; state?: StateAdapter; }): Promise { const state = await getConnectedState(args.state); return ( - (await state.get(conversationStopKey(args.conversationId, args.runId))) !== - null + (await readConversationStop(state, args.conversationId))?.runId === + args.runId ); } @@ -1763,7 +1773,6 @@ export async function stopConversationWork(args: { nowMs?: number; state?: StateAdapter; }): Promise { - const nowMs = args.nowMs ?? now(); return await withConversationMutation(args, async (state, lock) => { const current = await readConversation(state, args.conversationId); if (!current || !hasRunnableWork(current)) { @@ -1774,22 +1783,10 @@ export async function stopConversationWork(args: { const inboundMessageIds = current.execution.pendingMessages .filter(isHumanFacingMessage) .map((message) => message.inboundMessageId); - await writeConversation( - state, - lock, - withExecutionUpdate( - current, - { - ...current.execution, - runId, - stop: { inboundMessageIds, runId }, - }, - nowMs, - ), - ); + await fenceConversationMutation(state, lock, args.conversationId); await state.set( - conversationStopKey(args.conversationId, runId), - true, + conversationStopKey(args.conversationId), + { inboundMessageIds, runId } satisfies ConversationStop, JUNIOR_THREAD_STATE_TTL_MS, ); return { runId, status: "requested" }; @@ -1810,7 +1807,7 @@ export async function completeConversationStop(args: { if (!current || current.execution.lease?.token !== args.leaseToken) { return { status: "lost_lease", removedInboundMessageIds: [] }; } - const stop = current.execution.stop; + const stop = await readConversationStop(state, args.conversationId); if (!stop || stop.runId !== args.runId) { return { status: "none", removedInboundMessageIds: [] }; } @@ -1840,12 +1837,12 @@ export async function completeConversationStop(args: { ? undefined : current.execution.lastEnqueuedAtMs, pendingMessages, - stop: undefined, }, nowMs, ), ); - await state.delete(conversationStopKey(args.conversationId, args.runId)); + await fenceConversationMutation(state, lock, args.conversationId); + await state.delete(conversationStopKey(args.conversationId)); return { status: "cleared", removedInboundMessageIds }; }); } @@ -1907,7 +1904,6 @@ export async function cancelHumanFacingPendingMessages(args: { retryCount: becomesIdle ? 0 : current.execution.retryCount, runId: becomesIdle ? undefined : current.execution.runId, status: becomesIdle ? "idle" : current.execution.status, - stop: becomesIdle ? undefined : current.execution.stop, }, nowMs, ), @@ -2046,7 +2042,6 @@ export async function recordConversationRetry(args: { pendingMessages: stopped ? [] : current.execution.pendingMessages, runId: stopped ? undefined : current.execution.runId, status: stopped ? "failed" : "paused", - stop: stopped ? undefined : current.execution.stop, }, nowMs, ), @@ -2072,9 +2067,11 @@ export async function completeConversationWork(args: { return "lost_lease"; } const hasPending = pendingMessages(current).length > 0; - const needsRun = - current.execution.status === "paused" || - (args.resumeIfStopped === true && current.execution.stop !== undefined); + const stopped = + args.resumeIfStopped === true && + (await readConversationStop(state, args.conversationId))?.runId === + current.execution.runId; + const needsRun = current.execution.status === "paused" || stopped; const runnable = needsRun || hasPending; await writeConversation( state, @@ -2094,7 +2091,6 @@ export async function completeConversationWork(args: { ? current.execution.retryCount : 0, runId: runnable ? current.execution.runId : undefined, - stop: runnable ? current.execution.stop : undefined, }, nowMs, ), @@ -2229,7 +2225,6 @@ export async function deadLetterAttempt(args: { lease: undefined, status: runnable ? "pending" : "failed", runId: runnable ? current.execution.runId : undefined, - stop: runnable ? current.execution.stop : undefined, }, nowMs, ), @@ -2268,7 +2263,6 @@ export async function clearExpiredConversationLease(args: { pendingMessages: stopped ? [] : current.execution.pendingMessages, runId: stopped ? undefined : current.execution.runId, status: stopped ? "failed" : "paused", - stop: stopped ? undefined : current.execution.stop, }, nowMs, ), @@ -2284,6 +2278,7 @@ export async function deleteConversationState(args: { }): Promise { await withConversationMutation(args, async (state) => { await state.delete(conversationKey(args.conversationId)); + await state.delete(conversationStopKey(args.conversationId)); await removeIndexEntry({ state, indexKey: CONVERSATION_ACTIVE_INDEX_KEY, diff --git a/packages/junior/src/chat/task-execution/store.ts b/packages/junior/src/chat/task-execution/store.ts index 2b0d26b90d..ae4f8de389 100644 --- a/packages/junior/src/chat/task-execution/store.ts +++ b/packages/junior/src/chat/task-execution/store.ts @@ -14,7 +14,7 @@ export { CONVERSATION_WORK_MAX_DELIVERY_ATTEMPTS, CONVERSATION_WORK_MAX_RETRIES, CONVERSATION_WORK_STALE_ENQUEUE_MS, - hasConversationStopRequest, + hasConversationStop, isFinalAttempt, isInvalidConversationRecordError, type AgentInput, diff --git a/packages/junior/src/chat/task-execution/worker.ts b/packages/junior/src/chat/task-execution/worker.ts index 672caf089c..95683c8473 100644 --- a/packages/junior/src/chat/task-execution/worker.ts +++ b/packages/junior/src/chat/task-execution/worker.ts @@ -24,7 +24,7 @@ import { drainConversationMailbox, ensureConversationWake, getConversationWorkState, - hasConversationStopRequest, + hasConversationStop, isFinalAttempt, isInvalidConversationRecordError, recordAttemptFailure, @@ -289,7 +289,7 @@ function startLeaseCheckIn(args: { /** Poll the run-scoped stop marker only when an adapter observes remote stops. */ function createConversationStopSignal(args: { conversationId: string; - initialStopRunId?: string; + initiallyStopped: boolean; options: ProcessConversationWorkOptions; runId: string; }) { @@ -304,15 +304,13 @@ function createConversationStopSignal(args: { controller.abort(new Error("Conversation work stopped")); } }; - if (args.initialStopRunId === args.runId) { - requestStop(); - } + if (args.initiallyStopped) requestStop(); const check = async (): Promise => { if (checking || controller.signal.aborted) return; checking = true; try { - const stopped = await hasConversationStopRequest({ + const stopped = await hasConversationStop({ conversationId: args.conversationId, runId: args.runId, state: args.options.state, @@ -633,7 +631,11 @@ async function processConversationWorkInContext( }; const stop = createConversationStopSignal({ conversationId, - initialStopRunId: leasedWork.execution.stop?.runId, + initiallyStopped: await hasConversationStop({ + conversationId, + runId, + state: options.state, + }), options, runId, }); diff --git a/packages/junior/tests/component/task-execution/conversation-work.test.ts b/packages/junior/tests/component/task-execution/conversation-work.test.ts index d13d20c5f3..ec40af74f8 100644 --- a/packages/junior/tests/component/task-execution/conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/conversation-work.test.ts @@ -17,6 +17,7 @@ import { countPendingConversationMessages, drainConversationMailbox, getConversationWorkState, + hasConversationStop, listActiveConversationIds, listConversationsByActivity, ackMessages, @@ -1767,9 +1768,7 @@ describe("conversation work execution", () => { workerGetKeys.filter((key) => key === CONVERSATION_WORK_STATE_KEY), ).toHaveLength(conversationReadsBeforeStop); expect( - workerGetKeys.some((key) => - key.startsWith(`junior:conversation:v2:stop:${CONVERSATION_ID}:`), - ), + workerGetKeys.includes(`junior:conversation:v2:stop:${CONVERSATION_ID}`), ).toBe(true); await appendInboundMessage({ @@ -1789,7 +1788,6 @@ describe("conversation work execution", () => { await expect( getConversationWorkState({ conversationId: CONVERSATION_ID, state }), ).resolves.toMatchObject({ - execution: { stop: undefined }, messages: [expect.objectContaining({ inboundMessageId: "m3" })], }); }); @@ -1827,14 +1825,16 @@ describe("conversation work execution", () => { finish.resolve(); await expect(running).resolves.toEqual({ status: "pending_requeued" }); - await expect( - getConversationWorkState({ conversationId: CONVERSATION_ID }), - ).resolves.toMatchObject({ - execution: { - status: "paused", - stop: expect.objectContaining({ inboundMessageIds: [] }), - }, + const paused = await getConversationWorkState({ + conversationId: CONVERSATION_ID, }); + expect(paused?.execution.status).toBe("paused"); + await expect( + hasConversationStop({ + conversationId: CONVERSATION_ID, + runId: paused!.execution.runId!, + }), + ).resolves.toBe(true); await expect( processConversationWork(queue.takeMessage(), { @@ -1851,7 +1851,7 @@ describe("conversation work execution", () => { await expect( getConversationWorkState({ conversationId: CONVERSATION_ID }), ).resolves.toMatchObject({ - execution: { status: "idle", stop: undefined }, + execution: { status: "idle" }, }); }); @@ -1908,14 +1908,18 @@ describe("conversation work execution", () => { releaseCompletion.resolve(); await expect(running).resolves.toEqual({ status: "pending_requeued" }); - await expect( - getConversationWorkState({ conversationId: CONVERSATION_ID, state }), - ).resolves.toMatchObject({ - execution: { - status: "paused", - stop: expect.objectContaining({ runId: expect.any(String) }), - }, + const paused = await getConversationWorkState({ + conversationId: CONVERSATION_ID, + state, }); + expect(paused?.execution.status).toBe("paused"); + await expect( + hasConversationStop({ + conversationId: CONVERSATION_ID, + runId: paused!.execution.runId!, + state, + }), + ).resolves.toBe(true); await expect( processConversationWork(queue.takeMessage(), { @@ -1932,7 +1936,7 @@ describe("conversation work execution", () => { await expect( getConversationWorkState({ conversationId: CONVERSATION_ID, state }), ).resolves.toMatchObject({ - execution: { status: "idle", stop: undefined }, + execution: { status: "idle" }, }); }); From 1543f2fd18412fd8b469a1f4a0ebab71f9357ce0 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 10 Sep 2026 22:29:42 +0000 Subject: [PATCH 3/4] fix(chat): ignore stale stop markers for new runs --- .../junior/src/chat/task-execution/state.ts | 5 +- .../task-execution/conversation-work.test.ts | 46 +++++++++++++++++++ 2 files changed, 47 insertions(+), 4 deletions(-) diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index 7cb0127293..0d4adcafc0 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/packages/junior/src/chat/task-execution/state.ts @@ -1546,9 +1546,6 @@ export async function startConversationWork(args: { expiresAtMs: nowMs + CONVERSATION_WORK_LEASE_TTL_MS, }; const startsNewRun = current.execution.runId === undefined; - const stop = startsNewRun - ? await readConversationStop(state, args.conversationId) - : undefined; await writeConversation( state, lock, @@ -1558,7 +1555,7 @@ export async function startConversationWork(args: { ...current.execution, lease, status: current.execution.status === "paused" ? "paused" : "running", - runId: current.execution.runId ?? stop?.runId ?? randomUUID(), + runId: current.execution.runId ?? randomUUID(), lastEnqueuedAtMs: undefined, retryCount: startsNewRun ? 0 : current.execution.retryCount, }, diff --git a/packages/junior/tests/component/task-execution/conversation-work.test.ts b/packages/junior/tests/component/task-execution/conversation-work.test.ts index ec40af74f8..ca1518db6e 100644 --- a/packages/junior/tests/component/task-execution/conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/conversation-work.test.ts @@ -1792,6 +1792,52 @@ describe("conversation work execution", () => { }); }); + it("does not apply a stale stop marker to a new run", async () => { + const queue = createConversationWorkQueueTestAdapter(); + const entered = deferred(); + const finish = deferred(); + await appendInboundMessage({ message: inboundMessage("m1"), nowMs: 1_000 }); + + const first = processConversationWork(conversationQueueMessage(), { + queue, + run: async (context) => { + await context.attempt.ack(); + entered.resolve(); + await finish.promise; + return { status: "completed" }; + }, + }); + await entered.promise; + await expect( + stopConversationWork({ + conversationId: CONVERSATION_ID, + nowMs: 2_000, + }), + ).resolves.toMatchObject({ status: "requested" }); + finish.resolve(); + await expect(first).resolves.toEqual({ status: "completed" }); + + await appendInboundMessage({ + message: inboundMessage("m2", { + createdAtMs: 3_000, + receivedAtMs: 3_000, + }), + nowMs: 3_000, + }); + await expect( + processConversationWork(conversationQueueMessage(), { + queue, + run: async (context) => { + const signal = context.stopSignal?.(); + if (!signal) throw new Error("Expected a Conversation stop signal"); + expect(signal.aborted).toBe(false); + await context.attempt.ack(); + return { status: "completed" }; + }, + }), + ).resolves.toEqual({ status: "completed" }); + }); + it("resumes a paused Turn when its stop missed the live poll", async () => { vi.useFakeTimers({ now: 1_000 }); let currentNowMs = 1_000; From c3dd17d0e0fc5fae8f56476c98953a85a3c4b5a0 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 10 Sep 2026 22:37:02 +0000 Subject: [PATCH 4/4] fix(chat): bind queued stops to their run --- packages/junior/src/chat/task-execution/state.ts | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index 0d4adcafc0..2fda06e081 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/packages/junior/src/chat/task-execution/state.ts @@ -1770,6 +1770,7 @@ export async function stopConversationWork(args: { nowMs?: number; state?: StateAdapter; }): Promise { + const nowMs = args.nowMs ?? now(); return await withConversationMutation(args, async (state, lock) => { const current = await readConversation(state, args.conversationId); if (!current || !hasRunnableWork(current)) { @@ -1780,6 +1781,17 @@ export async function stopConversationWork(args: { const inboundMessageIds = current.execution.pendingMessages .filter(isHumanFacingMessage) .map((message) => message.inboundMessageId); + if (current.execution.runId === undefined) { + await writeConversation( + state, + lock, + withExecutionUpdate( + current, + { ...current.execution, runId }, + nowMs, + ), + ); + } await fenceConversationMutation(state, lock, args.conversationId); await state.set( conversationStopKey(args.conversationId),