fix(server): attribute quota continuations to the agent

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",

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)
     )

apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts

@@ -1,12 +1,13 @@
 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";
 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,
+    );
+  });
+});

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",
   };
 }