From 6aa0e0cfd3620c9a3c2bede3cfdc9240bffbab90 Mon Sep 17 00:00:00 2001 From: sunlover7 <328539788+sunlover7@users.noreply.github.com> Date: Sun, 18 Oct 2026 17:41:39 -0400 Subject: [PATCH 1/2] fix(server): attribute quota continuations to the agent --- .../UsageLimitRecoveryWorker.test.ts | 43 +++++++++++++++++++ .../UsageLimitRecoveryWorker.ts | 2 +- 2 files changed, 44 insertions(+), 1 deletion(-) create mode 100644 apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts diff --git a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts new file mode 100644 index 000000000000..89cd43fe55eb --- /dev/null +++ b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts @@ -0,0 +1,43 @@ +import { RunId, ThreadId } from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import { describe, expect, it } from "vite-plus/test"; +import type { ProjectionLimitRecoveryCandidate } from "./ProjectionStore.ts"; +import { limitRecoveryCommand } from "./UsageLimitRecoveryWorker.ts"; + +const stoppedAt = "2026-10-18T10:00:00.000Z"; +const resetAt = "2026-10-18T11:00:00.000Z"; +const candidate: 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(limitRecoveryCommand(armed, true, Date.parse(resetAt) - 1)).toBeNull(); + const resume = 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( + 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 352c71ef62fe..ff7e003c8c33 100644 --- a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts +++ b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts @@ -65,7 +65,7 @@ export function limitRecoveryCommand( text: "Continue where you left off.", attachments: [], dispatchMode: { type: "start_immediately" }, - createdBy: "user", + createdBy: "agent", creationSource: "server", }; } From 6407aa370cdd98d0562cf5368d92f9a7fc89e4e5 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 30 Oct 2026 17:50:26 +0000 Subject: [PATCH 2/2] fix(server): retain parent tracking for quota recovery --- .../DelegatedCompletionDelivery.test.ts | 132 ++++++++++++++---- .../src/orchestration-v2/Orchestrator.ts | 7 +- 2 files changed, 111 insertions(+), 28 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index ad2bc395d083..e2744dbdc3fe 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"; @@ -1690,25 +1692,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, }); @@ -1717,7 +1728,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", @@ -1729,7 +1740,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, @@ -1741,10 +1752,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, @@ -1753,19 +1764,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); @@ -1840,14 +1918,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, @@ -1920,7 +1998,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 50f9a2acc0f3..66aa9a4670d3 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 { @@ -7583,6 +7587,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) )