Skip to content
3 changes: 2 additions & 1 deletion apps/server/src/orchestration-v2/Orchestrator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4540,7 +4540,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
projection.thread.providerInstanceId !== run.providerInstanceId ||
projection.runtimeRequests.some((request) => request.status === "pending") ||
(projection.thread.snoozedUntil != null &&
DateTime.toEpochMillis(projection.thread.snoozedUntil) > DateTime.toEpochMillis(now))
DateTime.toEpochMillis(projection.thread.snoozedUntil) > DateTime.toEpochMillis(now)) ||
(yield* stopReachedRun(command, command.threadId, run.id))
) {
yield* emit(
events,
Expand Down
31 changes: 31 additions & 0 deletions apps/server/src/orchestration-v2/ProjectionStore.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3596,6 +3596,37 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
now: reset,
})).find((row) => row.id === threadId)!;
assert.deepEqual(due.limitRecovery, recovery);
const interruptRequestId = TurnItemId.make("limit-shell:interrupt-request");
yield* store.apply({
id: EventId.make("event:limit-shell:interrupt-request"),
type: "turn-item.updated",
threadId,
occurredAt: now,
payload: {
id: interruptRequestId,
threadId,
runId: original.id,
nodeId: original.rootNodeId,
providerThreadId: null,
providerTurnId: null,
nativeItemRef: null,
parentItemId: null,
ordinal: 3,
status: "completed",
title: "Interrupt requested",
startedAt: now,
completedAt: now,
updatedAt: now,
type: "run_interrupt_request",
message: "Interrupt requested",
},
});
assert.isUndefined(
(yield* store.getLimitRecoveryCandidates({ ...recoveryOptions, now: reset })).find(
(row) => row.id === threadId,
),
);
yield* sql`DELETE FROM orchestration_v2_projection_turn_items WHERE turn_item_id = ${interruptRequestId}`;
yield* sql`INSERT INTO orchestration_v2_projection_runtime_requests
(runtime_request_id, thread_id, node_id, kind, status, created_at, payload_json)
VALUES ('limit-shell:pending-request', ${threadId}, ${original.rootNodeId}, 'approval', 'pending', ${DateTime.formatIso(now)}, '{}')`;
Expand Down
6 changes: 6 additions & 0 deletions apps/server/src/orchestration-v2/ProjectionStore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3611,6 +3611,12 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
SELECT 1 FROM orchestration_v2_projection_runtime_requests request
WHERE request.thread_id = t.thread_id AND request.status = 'pending'
)
-- Stop reached this run: only a fresh user action may resume it.
AND NOT EXISTS (
SELECT 1 FROM orchestration_v2_projection_turn_items stop
WHERE stop.thread_id = t.thread_id AND stop.run_id = r.run_id
AND stop.type = 'run_interrupt_request'
)
ORDER BY t.thread_id
`;
const candidates: Array<ProjectionLimitRecoveryCandidate> = [];
Expand Down
27 changes: 27 additions & 0 deletions apps/server/src/orchestration-v2/runtimeLayer.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5249,6 +5249,7 @@ it.layer(layerTest)("usage-limit recovery", (it) => {
"independent-patches",
"expired-snooze",
"wake",
"stop",
] as const)("guards a scheduled usage-limit continuation against %s", (scenario) =>
Effect.gen(function* () {
const orchestrator = yield* Orchestrator.OrchestratorV2;
Expand Down Expand Up @@ -5619,6 +5620,12 @@ it.layer(layerTest)("usage-limit recovery", (it) => {
commandId: CommandId.make(`recovery:settle:${scenario}`),
threadId,
});
if (scenario === "stop")
yield* orchestrator.dispatch({
type: "thread.stop",
commandId: CommandId.make(`recovery:stop:${scenario}`),
threadId,
});
if (scenario === "replacement") {
const current = yield* orchestrator.getThreadProjection(threadId);
const error = current.turnItems.find((item) => item.type === "error")!;
Expand Down Expand Up @@ -5697,6 +5704,26 @@ it.layer(layerTest)("usage-limit recovery", (it) => {
? 1
: 0),
);
if (scenario === "stop") {
assert.isTrue(
after.turnItems.some(
(item) => item.type === "run_interrupt_request" && item.runId === run.id,
),
);
yield* orchestrator.dispatch({
type: "message.dispatch",
commandId: CommandId.make("recovery:manual-resume:stop"),
messageId: MessageId.make("recovery:manual-resume:stop"),
threadId,
manualContinuationOfRunId: run.id,
text: "Continue where you left off.",
attachments: [],
dispatchMode: { type: "start_immediately" },
createdBy: "user",
creationSource: "web",
});
assert.lengthOf((yield* orchestrator.getThreadProjection(threadId)).runs, 2);
}
if (scenario === "queued-resume") {
assert.equal(after.runs[1]?.status, "queued");
assert.isTrue(after.runs[1]?.queueHeld);
Expand Down
4 changes: 2 additions & 2 deletions docs/user/thread-sidebar.md
Original file line number Diff line number Diff line change
Expand Up @@ -205,8 +205,8 @@ continuation. You can cancel it from the thread. Enable **Auto-resume limited
threads** in **Settings → General** on web and desktop, or **Settings → Thread
behavior** on mobile, to schedule limit stops by default.
The environment must be running when the reset arrives; it resumes overdue
continuations after a restart. Sending a new message, archiving, or settling the
thread prevents a pending continuation from starting.
continuations after a restart. Stopping the thread, sending a new message,
archiving, or settling it prevents a pending continuation from starting.

Choose **Snooze until reset** to hide the thread until its allowance returns.
Snooze and auto-resume are independent: snooze alone wakes the thread without
Expand Down
Loading