feat(server): threads resume on their own after a usage limit resets

apps/server/src/mcp/toolkits/threads/handlers.ts

@@ -21,6 +21,7 @@ import * as Crypto from "effect/Crypto";
 import * as DateTime from "effect/DateTime";
 import * as Effect from "effect/Effect";
 import * as Option from "effect/Option";
+import * as Scope from "effect/Scope";
 import * as Stream from "effect/Stream";
 
 import * as OrchestrationEngine from "../../../orchestration/Services/OrchestrationEngine.ts";
@@ -43,6 +44,7 @@ import {
   unrecordedChildReportState,
 } from "./childReportState.ts";
 import { isSubagentThreadId, makeSubagentThreadId, parentThreadIdOf } from "./subagentThreadId.ts";
+import { RESUME_TEXT, resumeAfterUsageLimitReset } from "./usageLimitResume.ts";
 import {
   type SubagentStatus,
   type ThreadScope,
@@ -444,17 +446,63 @@ const make = Effect.gen(function* () {
     );
   });
 
+  const scope = yield* Scope.Scope;
+  /** Top-level thread id -> the failed turn a usage-limit resume is pending for. */
+  const pendingResumes = new Map<string, string>();
+
+  /**
+   * Schedules one "continue" after the usage limit that failed a top-level
+   * thread's turn resets (see usageLimitResume.ts). Starting another turn
+   * first, or archiving the thread, cancels it.
+   */
+  const resumeAfterUsageLimit = Effect.fn("ThreadsToolkit.resumeAfterUsageLimit")(function* (
+    event: OrchestrationEvent,
+  ) {
+    if (event.type !== "thread.session-set" || event.payload.session.status !== "error") return;
+    const threadId = event.payload.threadId;
+    if (isSubagentThreadId(threadId)) return;
+    const failed = yield* threadShell(threadId);
+    const turnId = failed?.latestTurn?.turnId;
+    if (!failed || !turnId || pendingResumes.get(threadId) === turnId) return;
+    const instanceId = event.payload.session.providerInstanceId ?? failed.modelSelection.instanceId;
+    pendingResumes.set(threadId, turnId);
+    yield* resumeAfterUsageLimitReset(
+      {
+        thread: threadShell,
+        providers: registry.getProviders,
+        resume: (thread) => startTurn(thread, RESUME_TEXT),
+      },
+      { threadId, turnId, instanceId },
+    ).pipe(
+      Effect.catchCause((cause) =>
+        Effect.logWarning("usage-limit resume failed", { threadId, cause: Cause.pretty(cause) }),
+      ),
+      Effect.ensuring(
+        Effect.sync(() => {
+          if (pendingResumes.get(threadId) === turnId) pendingResumes.delete(threadId);
+        }),
+      ),
+      Effect.forkIn(scope),
+    );
+  });
+
+  const skipOnFailure = (name: string, event: OrchestrationEvent) =>
+    Effect.catchCause((cause: Cause.Cause<unknown>) =>
+      Effect.logWarning(`threads toolkit ${name} skipped an event`, {
+        eventType: event.type,
+        cause: Cause.pretty(cause),
+      }),
+    );
+
   // Consume the hot stream like the upstream reactors do. This layer builds
   // with the HTTP routes, so it must not acquire an engine subscription at
   // build time (`subscribeDomainEvents`).
   yield* Effect.forkScoped(
     Stream.runForEach(engine.streamDomainEvents, (event) =>
       bridge(event).pipe(
-        Effect.catchCause((cause) =>
-          Effect.logWarning("threads toolkit bridge skipped an event", {
-            eventType: event.type,
-            cause: Cause.pretty(cause),
-          }),
+        skipOnFailure("bridge", event),
+        Effect.andThen(
+          resumeAfterUsageLimit(event).pipe(skipOnFailure("usage-limit resume", event)),
         ),
       ),
     ),

apps/server/src/mcp/toolkits/threads/mobileShell.ts

@@ -12,11 +12,9 @@ import * as SqlClient from "effect/unstable/sql/SqlClient";
 
 import { isSubagentThreadId } from "./subagentThreadId.ts";
 
-export const hidesChildThreads = (surface: string | null | undefined) => surface === "mobile";
+const hidesChildThreads = (surface: string | null | undefined) => surface === "mobile";
 
-export const withoutChildThreads = (
-  snapshot: OrchestrationShellSnapshot,
-): OrchestrationShellSnapshot => ({
+const withoutChildThreads = (snapshot: OrchestrationShellSnapshot): OrchestrationShellSnapshot => ({
   ...snapshot,
   threads: snapshot.threads.filter((thread) => !isSubagentThreadId(thread.id)),
 });
@@ -25,7 +23,7 @@ const isChildThreadEvent = (item: OrchestrationShellStreamItem) =>
   (item.kind === "thread-upserted" && isSubagentThreadId(item.thread.id)) ||
   (item.kind === "thread-removed" && isSubagentThreadId(item.threadId));
 
-export const withoutChildThreadItems = <A extends OrchestrationShellStreamItem, E, R>(
+const withoutChildThreadItems = <A extends OrchestrationShellStreamItem, E, R>(
   stream: Stream.Stream<A, E, R>,
 ): Stream.Stream<A, E, R> =>
   stream.pipe(

apps/server/src/mcp/toolkits/threads/usageLimitResume.test.ts

@@ -0,0 +1,141 @@
+import {
+  ProjectId,
+  ProviderInstanceId,
+  ThreadId,
+  TurnId,
+  type OrchestrationThreadShell,
+  type ServerProvider,
+  type ServerProviderUsageWindow,
+} from "@t3tools/contracts";
+import { assert, describe, it } from "@effect/vitest";
+import * as Clock from "effect/Clock";
+import * as DateTime from "effect/DateTime";
+import * as Effect from "effect/Effect";
+import * as Fiber from "effect/Fiber";
+import * as TestClock from "effect/testing/TestClock";
+
+import {
+  RESUME_DELAY_MS,
+  resumeAfterUsageLimitReset,
+  usageLimitResetAt,
+} from "./usageLimitResume.ts";
+
+const MINUTE = 60_000;
+const iso = (ms: number) => DateTime.formatIso(DateTime.makeUnsafe(ms));
+
+const window = (usedPercent: number, resetsAt?: number): ServerProviderUsageWindow => ({
+  id: `window-${usedPercent}-${resetsAt}`,
+  kind: "session",
+  label: "Session",
+  usedPercent,
+  ...(resetsAt === undefined ? {} : { resetsAt: iso(resetsAt) }),
+});
+
+const provider = (windows: ReadonlyArray<ServerProviderUsageWindow>) =>
+  ({
+    instanceId: ProviderInstanceId.make("claudeAgent"),
+    usageLimits: { checkedAt: iso(0), windows },
+  }) as unknown as ServerProvider;
+
+const thread = (
+  turnId: string,
+  overrides: Partial<OrchestrationThreadShell> = {},
+): OrchestrationThreadShell => ({
+  id: ThreadId.make("thread-user"),
+  projectId: ProjectId.make("project-1"),
+  title: "Thread",
+  modelSelection: { instanceId: ProviderInstanceId.make("claudeAgent"), model: "claude-opus" },
+  runtimeMode: "full-access",
+  interactionMode: "default",
+  branch: null,
+  worktreePath: null,
+  pullRequests: [],
+  latestTurn: {
+    turnId: TurnId.make(turnId),
+    state: "error",
+    requestedAt: iso(0),
+    startedAt: iso(0),
+    completedAt: iso(0),
+    assistantMessageId: null,
+  },
+  createdAt: iso(0),
+  updatedAt: iso(0),
+  archivedAt: null,
+  settledOverride: null,
+  settledAt: null,
+  session: null,
+  latestUserMessageAt: null,
+  hasPendingApprovals: false,
+  hasPendingUserInput: false,
+  hasActionableProposedPlan: false,
+  ...overrides,
+});
+
+describe("usageLimitResetAt", () => {
+  it("returns the latest reset among exhausted windows only", () => {
+    const now = 0;
+    const resetAt = usageLimitResetAt(
+      provider([window(100, 10 * MINUTE), window(100, 30 * MINUTE), window(80, 90 * MINUTE)]),
+      now,
+    );
+    assert.equal(resetAt, 30 * MINUTE);
+  });
+
+  it("returns null without an exhausted window with a future reset", () => {
+    assert.isNull(usageLimitResetAt(provider([window(99, 10 * MINUTE)]), 0));
+    assert.isNull(usageLimitResetAt(provider([window(100)]), 0));
+    assert.isNull(usageLimitResetAt(provider([window(100, 10 * MINUTE)]), 10 * MINUTE));
+    assert.isNull(usageLimitResetAt(undefined, 0));
+  });
+});
+
+/** Runs a resume for a limit resetting at 20 minutes, against `atReset` as the thread then. */
+const runResume = (usedPercent: number, atReset: OrchestrationThreadShell) =>
+  Effect.gen(function* () {
+    const windows = [window(usedPercent, (yield* Clock.currentTimeMillis) + 20 * MINUTE)];
+    const resumed: string[] = [];
+    const fiber = yield* resumeAfterUsageLimitReset(
+      {
+        thread: () => Effect.succeed(atReset),
+        providers: Effect.succeed([provider(windows)]),
+        resume: (resumedThread) => Effect.sync(() => void resumed.push(resumedThread.id)),
+      },
+      { threadId: "thread-user", turnId: "turn-failed", instanceId: "claudeAgent" },
+    ).pipe(Effect.forkChild);
+    // At the reset itself nothing is sent yet; the margin has not passed.
+    yield* TestClock.adjust(20 * MINUTE);
+    const beforeReset = [...resumed];
+    yield* TestClock.adjust(RESUME_DELAY_MS);
+    return { outcome: yield* Fiber.join(fiber), beforeReset, resumed };
+  });
+
+describe("resumeAfterUsageLimitReset", () => {
+  it.effect("resumes once, a minute after the limit resets", () =>
+    Effect.gen(function* () {
+      const result = yield* runResume(100, thread("turn-failed"));
+      assert.equal(result.outcome, "resumed");
+      assert.deepEqual(result.beforeReset, []);
+      assert.deepEqual(result.resumed, ["thread-user"]);
+    }),
+  );
+
+  it.effect("does nothing when the provider is not out of usage", () =>
+    Effect.gen(function* () {
+      const result = yield* runResume(40, thread("turn-failed"));
+      assert.equal(result.outcome, "not-limited");
+      assert.deepEqual(result.resumed, []);
+    }),
+  );
+
+  it.effect("stands down when the user started another turn or archived the thread", () =>
+    Effect.gen(function* () {
+      const newerTurn = yield* runResume(100, thread("turn-newer"));
+      assert.equal(newerTurn.outcome, "superseded");
+      assert.deepEqual(newerTurn.resumed, []);
+
+      const archived = yield* runResume(100, thread("turn-failed", { archivedAt: iso(0) }));
+      assert.equal(archived.outcome, "superseded");
+      assert.deepEqual(archived.resumed, []);
+    }),
+  );
+});

apps/server/src/mcp/toolkits/threads/usageLimitResume.ts

@@ -0,0 +1,71 @@
+/**
+ * Resume after a usage limit (toolboxmd/chromeria#61): a thread the user
+ * started whose turn fails while its provider is out of usage gets one
+ * "continue" shortly after the limit resets. Child threads are left alone:
+ * Model Router reroutes or blocks Prism jobs itself, and a late "continue" to
+ * an old worker would duplicate work. Upstream Orchestrator V2 has its own
+ * "Resume at reset", so this goes away with the V2 port.
+ */
+import type { OrchestrationThreadShell, ServerProvider } from "@t3tools/contracts";
+import * as Clock from "effect/Clock";
+import * as Effect from "effect/Effect";
+
+export const RESUME_TEXT = "Continue. (Sent automatically after the usage limit reset.)";
+
+/** Lets the provider's usage windows catch up with the rate-limit event that ended the turn. */
+const SETTLE_DELAY_MS = 5_000;
+
+/** Margin after the reset so the first resumed request is not refused again. */
+export const RESUME_DELAY_MS = 60_000;
+
+/**
+ * The latest reset among the provider's exhausted usage windows, in epoch
+ * milliseconds, or null when nothing is exhausted or no reset time is known.
+ */
+export function usageLimitResetAt(
+  provider: Pick<ServerProvider, "usageLimits"> | undefined,
+  nowMs: number,
+): number | null {
+  let latest: number | null = null;
+  for (const window of provider?.usageLimits?.windows ?? []) {
+    if (window.usedPercent < 100 || window.resetsAt === undefined) continue;
+    const resetMs = Date.parse(window.resetsAt);
+    if (!Number.isFinite(resetMs) || resetMs <= nowMs) continue;
+    latest = latest === null ? resetMs : Math.max(latest, resetMs);
+  }
+  return latest;
+}
+
+export interface UsageLimitResumeInput {
+  readonly threadId: string;
+  /** The turn that failed; any later turn cancels the resume. */
+  readonly turnId: string;
+  readonly instanceId: string;
+}
+
+/**
+ * Waits for the usage limit behind a failed turn to reset, then resumes the
+ * thread once, unless the user started another turn or archived it first.
+ */
+export const resumeAfterUsageLimitReset = <E, R>(
+  deps: {
+    readonly thread: (threadId: string) => Effect.Effect<OrchestrationThreadShell | undefined>;
+    readonly providers: Effect.Effect<ReadonlyArray<ServerProvider>>;
+    readonly resume: (thread: OrchestrationThreadShell) => Effect.Effect<void, E, R>;
+  },
+  input: UsageLimitResumeInput,
+) =>
+  Effect.gen(function* () {
+    yield* Effect.sleep(SETTLE_DELAY_MS);
+    const providers = yield* deps.providers;
+    const provider = providers.find((candidate) => candidate.instanceId === input.instanceId);
+    const resetAt = usageLimitResetAt(provider, yield* Clock.currentTimeMillis);
+    if (resetAt === null) return "not-limited" as const;
+    yield* Effect.sleep(resetAt + RESUME_DELAY_MS - (yield* Clock.currentTimeMillis));
+    const thread = yield* deps.thread(input.threadId);
+    if (!thread || thread.archivedAt !== null || thread.latestTurn?.turnId !== input.turnId) {
+      return "superseded" as const;
+    }
+    yield* deps.resume(thread);
+    return "resumed" as const;
+  });

docs/fork-features.md

@@ -115,11 +115,13 @@ for per-match decisions and the absorption PR record.
   "issues": [
     "https://github.com/toolboxmd/t3code/issues/8",
     "https://github.com/toolboxmd/t3code/issues/48",
-    "https://github.com/toolboxmd/t3code/issues/59"
+    "https://github.com/toolboxmd/t3code/issues/59",
+    "https://github.com/toolboxmd/t3code/issues/61"
   ],
   "prs": [
     "https://github.com/toolboxmd/t3code/pull/10",
-    "https://github.com/toolboxmd/t3code/pull/60"
+    "https://github.com/toolboxmd/t3code/pull/60",
+    "https://github.com/toolboxmd/t3code/pull/63"
   ],
   "newFiles": [
     "apps/server/src/mcp/toolkits/threads/childReportState.test.ts",
@@ -131,6 +133,8 @@ for per-match decisions and the absorption PR record.
     "apps/server/src/mcp/toolkits/threads/subagentThreadId.test.ts",
     "apps/server/src/mcp/toolkits/threads/subagentThreadId.ts",
     "apps/server/src/mcp/toolkits/threads/tools.ts",
+    "apps/server/src/mcp/toolkits/threads/usageLimitResume.test.ts",
+    "apps/server/src/mcp/toolkits/threads/usageLimitResume.ts",
     "apps/web/src/components/AgentThreadLink.tsx",
     "apps/web/src/components/subagentThreads.test.ts",
     "apps/web/src/components/subagentThreads.ts"