diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 5053e8d568767..d57a7927329d5 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -25,6 +25,7 @@ import * as Deferred from "effect/Deferred"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import * as CheckpointStore from "../checkpointing/CheckpointStore.ts"; import * as ServerConfig from "../config.ts"; @@ -42,6 +43,7 @@ import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as EventSink from "./EventSink.ts"; import * as Orchestrator from "./Orchestrator.ts"; import { continueRestartedRun } from "./RestartContinuation.ts"; +import { limitRecoveryCommand } from "./UsageLimitRecoveryWorker.ts"; import * as RuntimeLayer from "./runtimeLayer.ts"; import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; import * as ProviderTurnStartServiceTestkit from "./ProviderTurnStartService.testkit.ts"; @@ -1749,25 +1751,34 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( "restarted", "stopped", "overlapping", + "quota-completed", + "quota-stopped", ] as const; it.effect.each(outcomes)( "delivers a %s follow-up independently of the original task", - (outcome) => + (scenario) => Effect.gen(function* () { + const quotaRecovery = scenario === "quota-completed" || scenario === "quota-stopped"; + const outcome = + scenario === "quota-completed" + ? "completed" + : scenario === "quota-stopped" + ? "stopped" + : scenario; const orchestrator = yield* Orchestrator.OrchestratorV2; const eventSink = yield* EventSink.EventSinkV2; const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; const now = yield* DateTime.now; - const parentThreadId = ThreadId.make(`thread:followup-${outcome}`); - const projectId = ProjectId.make(`project:followup-${outcome}`); - const parentRunId = RunId.make(`run:followup-${outcome}`); - const rootNodeId = NodeId.make(`node:followup-${outcome}`); + const parentThreadId = ThreadId.make(`thread:followup-${scenario}`); + const projectId = ProjectId.make(`project:followup-${scenario}`); + const parentRunId = RunId.make(`run:followup-${scenario}`); + const rootNodeId = NodeId.make(`node:followup-${scenario}`); yield* seedParentWithTerminalTask({ threadId: parentThreadId, projectId, runId: parentRunId, rootNodeId, - taskId: NodeId.make(`node:seed-${outcome}`), + taskId: NodeId.make(`node:seed-${scenario}`), deliveryState: "acknowledged", now, }); @@ -1776,7 +1787,7 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( projectId, parentRunId, rootNodeId, - name: `followup-child-${outcome}`, + name: `followup-child-${scenario}`, completionWake: "always", continuationPending: false, runStatus: "completed", @@ -1788,7 +1799,7 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( assert.equal(firstWake.delegatedCompletion?.parentRunId, parentRunId); yield* orchestrator.dispatch({ type: "delegated_task.completion-delivery.acknowledge", - commandId: CommandId.make(`command:followup-ack-${outcome}`), + commandId: CommandId.make(`command:followup-ack-${scenario}`), parentThreadId, taskId: child.taskId, observedByRunId: parentRunId, @@ -1800,10 +1811,10 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( (run) => run.id === parentRunId, )!; yield* eventSink.write({ - commandId: reconcileCommandId(`followup-parent-finished-${outcome}`), + commandId: reconcileCommandId(`followup-parent-finished-${scenario}`), events: [ { - id: EventId.make(`event:followup-parent-finished-${outcome}`), + id: EventId.make(`event:followup-parent-finished-${scenario}`), type: "run.updated", threadId: parentThreadId, runId: parentRunId, @@ -1812,19 +1823,86 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( }, ], }); - const messageId = MessageId.make(`message:followup-${outcome}`); - yield* orchestrator.dispatch({ - type: "message.dispatch", - createdBy: outcome === "failed" ? "user" : "agent", - creationSource: outcome === "failed" ? "web" : "mcp", - commandId: CommandId.make(`command:followup-send-${outcome}`), - threadId: child.childThreadId, - senderThreadId: parentThreadId, - messageId, - text: "Check the same work again.", - attachments: [], - dispatchMode: { type: "start_immediately" }, - }); + if (quotaRecovery) { + const projection = yield* orchestrator.getThreadProjection(child.childThreadId); + const source = projection.runs.find((run) => run.id === child.childRunId)!; + const resetAt = DateTime.formatIso(DateTime.add(now, { minutes: 1 })); + yield* eventSink.write({ + events: [ + { + ...runEvent({ + threadId: child.childThreadId, + runId: source.id, + ordinal: source.ordinal, + status: "failed", + now, + }), + payload: { ...source, status: "failed", completedAt: now }, + }, + { + id: EventId.make(`event:quota-recovery:${scenario}`), + type: "thread.metadata-updated", + threadId: child.childThreadId, + occurredAt: now, + payload: { + ...projection.thread, + limitRecovery: { runId: source.id, resetAt, autoResume: true, snooze: false }, + }, + }, + { + id: EventId.make(`event:quota-failure:${scenario}`), + type: "turn-item.updated", + threadId: child.childThreadId, + occurredAt: now, + payload: { + id: TurnItemId.make(`item:quota-failure:${scenario}`), + type: "error", + threadId: child.childThreadId, + runId: source.id, + nodeId: source.rootNodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + 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, + }, + }, + }, + ], + }); + yield* TestClock.adjust("1 minute"); + } + const shell = (yield* orchestrator.getShellSnapshot()).threads.find( + (thread) => thread.id === child.childThreadId, + )!; + const command = quotaRecovery + ? limitRecoveryCommand(shell, true, DateTime.toEpochMillis(yield* DateTime.now)) + : ({ + type: "message.dispatch", + createdBy: outcome === "failed" ? "user" : "agent", + creationSource: outcome === "failed" ? "web" : "mcp", + commandId: CommandId.make(`command:followup-send-${scenario}`), + threadId: child.childThreadId, + senderThreadId: parentThreadId, + messageId: MessageId.make(`message:followup-${scenario}`), + text: "Check the same work again.", + attachments: [], + dispatchMode: { type: "start_immediately" }, + } as const); + assert(command?.type === "message.dispatch"); + const messageId = command.messageId; + yield* orchestrator.dispatch(command); const childProjection = yield* orchestrator.getThreadProjection(child.childThreadId); let run = childProjection.runs.find((row) => row.userMessageId === messageId)!; assert.isDefined(run.delegatedTaskId); @@ -1899,14 +1977,14 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( yield* orchestrator.dispatch({ type: "thread.stop", threadId: parentThreadId, - commandId: CommandId.make(`command:followup-stop-${outcome}`), + commandId: CommandId.make(`command:followup-stop-${scenario}`), }); } yield* eventSink.write({ - commandId: reconcileCommandId(`followup-finish-${outcome}`), + commandId: reconcileCommandId(`followup-finish-${scenario}`), events: [ { - id: EventId.make(`event:followup-finish-${outcome}`), + id: EventId.make(`event:followup-finish-${scenario}`), type: "run.updated", threadId: child.childThreadId, runId: run.id, @@ -1979,7 +2057,7 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( yield* orchestrator.recoverDelegatedTasks; const recovered = yield* orchestrator.getThreadProjection(parentThreadId); assert.equal(recovered.contextTransfers.length, finished.contextTransfers.length); - if (outcome === "completed") { + if (scenario === "completed") { const automaticMessageId = MessageId.make("message:followup-automatic-wake"); yield* orchestrator.dispatch({ type: "message.dispatch", diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index b0a52315923e8..37d24a206a54b 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -634,7 +634,11 @@ function mayStartSubagentFollowup( if (command.scheduledTaskId !== undefined || command.restartContinuationOfRunId !== undefined) return false; if (command.notification !== undefined) return command.notification.source.kind !== "monitor"; - return command.createdBy === "user" || command.creationSource === "mcp"; + return ( + command.createdBy === "user" || + command.creationSource === "mcp" || + command.usageLimitContinuationOfRunId !== undefined + ); } function isBlockingRun(run: OrchestrationV2Run): boolean { @@ -7628,6 +7632,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio child.lineage.relationshipToParent !== "subagent" || parentThreadId === null || (command.notification === undefined && + command.usageLimitContinuationOfRunId === undefined && command.createdBy !== "user" && command.senderThreadId !== parentThreadId) ) diff --git a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts index 2e077314b8212..56c73c062ed45 100644 --- a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts +++ b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts @@ -1,5 +1,5 @@ import { assert, it } from "@effect/vitest"; -import type { OrchestrationV2DomainEvent } from "@t3tools/contracts"; +import { type OrchestrationV2DomainEvent, RunId, ThreadId } from "@t3tools/contracts"; import * as Clock from "effect/Clock"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -7,6 +7,7 @@ import * as Layer from "effect/Layer"; import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; +import { describe, expect } from "vite-plus/test"; import * as Scheduler from "../scheduling/Scheduler.ts"; import * as ServerSettings from "../serverSettings.ts"; @@ -68,3 +69,44 @@ it.effect("idle sweeps skip SQL until the next wake time or a relevant event", ( }).pipe(Effect.provide(layer)); }), ); + +const stoppedAt = "2026-10-18T10:00:00.000Z"; +const resetAt = "2026-10-18T11:00:00.000Z"; +const candidate: ProjectionStore.ProjectionLimitRecoveryCandidate = { + id: ThreadId.make("quota-test"), + status: "failed", + lastErrorClass: "usage_limit", + latestRunId: RunId.make("stopped-run"), + usageLimitResetAt: resetAt, + archivedAt: null, + settledOverride: "active", + pendingRuntimeRequest: null, + latestRunCompletedAt: DateTime.makeUnsafe(stoppedAt), + updatedAt: DateTime.makeUnsafe(stoppedAt), + limitRecovery: null, + snoozedUntil: null, +}; + +describe("quota reset continuation provenance", () => { + it("attributes reset continuations to T3 while preserving the original recovery identity", () => { + const armed = { + ...candidate, + limitRecovery: { runId: candidate.latestRunId!, resetAt, autoResume: true, snooze: true }, + }; + expect( + UsageLimitRecoveryWorker.limitRecoveryCommand(armed, true, Date.parse(resetAt) - 1), + ).toBeNull(); + const resume = UsageLimitRecoveryWorker.limitRecoveryCommand(armed, true, Date.parse(resetAt)); + expect(resume).toMatchObject({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "server", + usageLimitContinuationOfRunId: candidate.latestRunId, + text: "Continue where you left off.", + }); + expect(resume?.commandId).toEqual( + UsageLimitRecoveryWorker.limitRecoveryCommand(armed, true, Date.parse(resetAt) + 1) + ?.commandId, + ); + }); +}); diff --git a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts index 224eecbbe5f69..401b116233ca0 100644 --- a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts +++ b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts @@ -70,7 +70,7 @@ export function limitRecoveryCommand( text: "Continue where you left off.", attachments: [], dispatchMode: { type: "start_immediately" }, - createdBy: "user", + createdBy: "agent", creationSource: "server", }; }