diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 600764704e57f..8eb3362af0c4e 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -5418,6 +5418,20 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const failure = latestRootProviderFailure(run, projection.turnItems); const recovery = projection.thread.limitRecovery; const now = yield* DateTime.now; + const parentId = + projection.thread.lineage.relationshipToParent === "subagent" + ? projection.thread.lineage.parentThreadId + : null; + // A subagent of a settled parent is abandoned work; resuming it would reopen the parent. + const parentSettled = + parentId !== null && + (yield* projectionStore + .getThreadShell(parentId) + .pipe( + Effect.mapError( + (cause) => new OrchestratorProjectionError({ threadId: parentId, cause }), + ), + ))?.settledOverride === "settled"; if ( run?.id !== command.usageLimitContinuationOfRunId || failure?.class !== "usage_limit" || @@ -5430,11 +5444,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio projection.thread.archivedAt !== null || projection.thread.deletedAt !== null || projection.thread.settledOverride === "settled" || + parentSettled || projection.thread.providerInstanceId !== run.providerInstanceId || projection.runtimeRequests.some((request) => request.status === "pending") || (projection.thread.snoozedUntil != null && DateTime.toEpochMillis(projection.thread.snoozedUntil) > DateTime.toEpochMillis(now)) ) { + const recoveryMatches = + recovery?.autoResume && + recovery.requestId === command.usageLimitRecoveryRequestId && + recovery.runId === command.usageLimitContinuationOfRunId; yield* emit( events, command, @@ -5444,13 +5463,15 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio occurredAt: now, payload: { ...projection.thread, - ...(recovery?.autoResume && - recovery.requestId === command.usageLimitRecoveryRequestId && - recovery.runId === command.usageLimitContinuationOfRunId && - projection.thread.snoozedUntil != null && - DateTime.toEpochMillis(projection.thread.snoozedUntil) > DateTime.toEpochMillis(now) - ? { limitRecovery: { ...recovery, requestId: command.commandId } } - : {}), + // A settled parent retires the resume; leaving it armed would replay this receipt. + ...(recoveryMatches && parentSettled + ? { limitRecovery: { ...recovery, autoResume: false, requestId: command.commandId } } + : recoveryMatches && + projection.thread.snoozedUntil != null && + DateTime.toEpochMillis(projection.thread.snoozedUntil) > + DateTime.toEpochMillis(now) + ? { limitRecovery: { ...recovery, requestId: command.commandId } } + : {}), }, }); return; diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index b38fb737c91db..2dfec54539479 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -8524,6 +8524,166 @@ it.layer(layerTest)("usage-limit recovery", (it) => { } }), ); + + it.effect("retires a usage-limited subagent resume once its parent settles", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const events = yield* EventSink.EventSinkV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const parentId = ThreadId.make("recovery:settled-parent:parent"); + const childId = ThreadId.make("recovery:settled-parent:child"); + const projectId = ProjectId.make("recovery:settled-parent:project"); + yield* seedProject({ + projectId, + title: "Recovery project", + workspaceRoot: process.cwd(), + defaultModelSelection: modelSelection, + createdAt: DateTime.formatIso(yield* DateTime.now), + }); + for (const threadId of [parentId, childId]) + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(`${threadId}:create`), + threadId, + projectId, + title: "Limited thread", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("recovery:settled-parent:message"), + threadId: childId, + messageId: MessageId.make("recovery:settled-parent:message"), + text: "Work on this.", + attachments: [], + dispatchMode: { type: "defer_start" }, + createdBy: "user", + creationSource: "web", + }); + const child = yield* orchestrator.getThreadProjection(childId); + const run = child.runs[0]!; + const now = yield* DateTime.now; + const resetAt = DateTime.formatIso(DateTime.add(now, { minutes: 1 })); + yield* events.write({ + events: [ + { + id: EventId.make("recovery:settled-parent:lineage"), + type: "thread.metadata-updated", + threadId: childId, + occurredAt: now, + payload: { + ...child.thread, + lineage: { + parentThreadId: parentId, + rootThreadId: parentId, + relationshipToParent: "subagent", + }, + }, + }, + { + id: EventId.make("recovery:settled-parent:run"), + type: "run.updated", + threadId: childId, + occurredAt: now, + payload: { ...run, status: "failed", completedAt: now }, + }, + { + id: EventId.make("recovery:settled-parent:error"), + type: "turn-item.updated", + threadId: childId, + occurredAt: now, + payload: { + id: TurnItemId.make("recovery:settled-parent:error"), + type: "error", + threadId: childId, + runId: run.id, + nodeId: run.rootNodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 2, + status: "failed", + title: "Usage limit reached", + startedAt: now, + completedAt: now, + updatedAt: now, + failure: { + class: "usage_limit", + message: "Plan limit reached.", + code: "usageLimitExceeded", + retryable: null, + resetAt, + }, + }, + }, + ], + }); + const sweep = Effect.fnUntraced(function* (nowMs: number) { + const candidate = (yield* projections.getLimitRecoveryCandidates({ + now: DateTime.makeUnsafe(nowMs), + autoResume: true, + snooze: false, + })).find((row) => row.id === childId); + return candidate === undefined ? null : limitRecoveryCommand(candidate, true, nowMs); + }); + const arm = yield* sweep(DateTime.toEpochMillis(now)); + assert(arm?.type === "thread.metadata.update" && arm.limitRecovery != null); + yield* orchestrator.autoArmLimitRecovery({ + commandId: arm.commandId, + threadId: childId, + limitRecovery: arm.limitRecovery, + }); + const resume = yield* sweep(Date.parse(resetAt)); + assert.equal(resume?.type, "message.dispatch"); + yield* orchestrator.dispatch({ + type: "thread.settle", + commandId: CommandId.make("recovery:settled-parent:settle"), + threadId: parentId, + }); + yield* TestClock.adjust("1 minute"); + yield* orchestrator.dispatch(resume!); + let current = yield* orchestrator.getThreadProjection(childId); + assert.lengthOf(current.runs, 1); + assert.isFalse(current.thread.limitRecovery!.autoResume); + assert.equal( + (yield* orchestrator.getThreadProjection(parentId)).thread.settledOverride, + "settled", + ); + yield* orchestrator.dispatch({ + type: "thread.unsettle", + commandId: CommandId.make("recovery:settled-parent:unsettle"), + threadId: parentId, + reason: "user", + }); + assert.isNull(yield* sweep(DateTime.toEpochMillis(yield* DateTime.now))); + yield* orchestrator.dispatch(resume!); + current = yield* orchestrator.getThreadProjection(childId); + assert.lengthOf(current.runs, 1); + assert.isFalse(current.thread.limitRecovery!.autoResume); + yield* orchestrator.dispatch({ + type: "thread.metadata.update", + commandId: CommandId.make("recovery:settled-parent:rearm"), + threadId: childId, + limitRecovery: { runId: run.id, resetAt, autoResume: true }, + }); + const rearmed = (yield* orchestrator.getThreadProjection(childId)).thread.limitRecovery; + assert.isTrue(rearmed!.autoResume); + yield* orchestrator.dispatch({ + ...resume!, + commandId: CommandId.make("recovery:settled-parent:stale-resume"), + }); + current = yield* orchestrator.getThreadProjection(childId); + assert.lengthOf(current.runs, 1); + assert.deepEqual(current.thread.limitRecovery, rearmed); + }), + ); }); const queueResumeDisabledInstanceProjectId = ProjectId.make( diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index bed4ed73a40c5..3fc960625552c 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -249,7 +249,9 @@ continuation. You can cancel it from the thread. Enable **Resume after usage limits** in **Settings → Threads** to schedule limit stops by default. The environment must be running when the reset arrives; it resumes overdue continuations after a restart. Stopping the thread, sending a new message, -archiving, or settling it prevents a pending continuation from starting. +archiving, or settling it prevents a pending continuation from starting. A +task the thread delegated does not continue at its reset while the thread is +settled; its continuation is cancelled instead. Choose **Snooze until reset** to hide the thread until its allowance returns. Snooze and auto-resume are independent: snooze alone wakes the thread without