fix(server): auto-resume OpenCode Go threads when usage limits reset

apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts

@@ -24,6 +24,7 @@ import {
   type OrchestrationV2ProviderThread,
   type OrchestrationV2ProviderTurn,
   type ProviderReplayEntry,
+  type ServerProviderUsageLimits,
 } from "@t3tools/contracts";
 import * as Cause from "effect/Cause";
 import * as DateTime from "effect/DateTime";
@@ -326,6 +327,7 @@ const resumed = (
     readonly resolveExtractedTextPath?: (path: string) => Effect.Effect<string | undefined>;
     readonly fileSystem?: FileSystem.FileSystem;
     readonly onPrompt?: (input: unknown) => void;
+    readonly readUsageLimits?: Effect.Effect<ServerProviderUsageLimits>;
     readonly borrowers?: { current: number };
   },
 ) =>
@@ -364,6 +366,9 @@ const resumed = (
         ...(options?.resolveExtractedTextPath === undefined
           ? {}
           : { resolveExtractedTextPath: options.resolveExtractedTextPath }),
+        ...(options?.readUsageLimits === undefined
+          ? {}
+          : { readUsageLimits: options.readUsageLimits }),
       },
     );
     const thread = yield* runtime.resumeThread({
@@ -528,6 +533,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {
   it.effect.each([
     ["provider.quota", "usage_limit"],
     ["provider.rate-limit", "usage_limit"],
+    ["provider.go-limit", "usage_limit"],
     ["provider", "provider_error"],
   ] as const)("classifies a %s execution failure as %s", ([code, failureClass]) =>
     Effect.gen(function* () {
@@ -579,6 +585,15 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {
         },
       },
     },
+    {
+      name: "an out-of-funds provider.go-limit (402)",
+      failureClass: "provider_error",
+      error: {
+        type: "provider.go-limit",
+        message: "Upstream request failed: Insufficient account funds",
+        status: 402,
+      },
+    },
     {
       name: "a provider.quota account budget (429)",
       failureClass: "usage_limit",
@@ -662,6 +677,265 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {
       }).pipe(Effect.scoped),
   );
 
+  const goLimitMessage =
+    "OpenCode Go 5 hour usage limit reached. Switch to a free model, or turn on Extra Usage on the Go page in OpenCode Console to keep going.";
+  const goUsage = (rollingPercent: number): ServerProviderUsageLimits => ({
+    checkedAt: "2026-11-06T01:49:10.000Z",
+    windows: [
+      {
+        id: "go_rolling",
+        kind: "session",
+        label: "Go · Session",
+        usedPercent: rollingPercent,
+        resetsAt: "2026-11-06T04:12:00.000Z",
+      },
+      {
+        id: "go_weekly",
+        kind: "weekly",
+        label: "Go · Weekly",
+        usedPercent: 41,
+        resetsAt: "2026-11-09T00:00:00.000Z",
+      },
+    ],
+  });
+  const failWith = (error: Record<string, unknown>) =>
+    event("session.execution.failed", { sessionID: SESSION, error });
+  const endedBy = (
+    error: Record<string, unknown>,
+    readUsageLimits: Effect.Effect<ServerProviderUsageLimits>,
+  ) =>
+    Effect.gen(function* () {
+      const { runtime, thread } = yield* resumed(
+        [
+          out("session.prompt", { sessionID: SESSION, text: "<any>" }),
+          promptAccepted,
+          failWith(error),
+        ],
+        { readUsageLimits },
+      );
+      const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped);
+      yield* runtime.startTurn(turnInput(thread));
+      return yield* Fiber.join(terminal);
+    });
+
+  // regression: OpenCode Go threads stayed failed after the Go window reset instead of resuming
+  it.effect("reports a spent Go window as a usage limit that resets with that window", () =>
+    Effect.gen(function* () {
+      // OpenCode's usage endpoint can lag the gateway, so the named window wins even below 100%.
+      const ended = yield* endedBy(
+        { type: "provider.go-limit", message: goLimitMessage },
+        Effect.succeed(goUsage(97)),
+      );
+      assert.equal(ended?.status, "failed");
+      assert.deepEqual(ended?.failure, {
+        class: "usage_limit",
+        code: "provider.go-limit",
+        message: goLimitMessage,
+        retryable: null,
+        resetAt: "2026-11-06T04:12:00.000Z",
+      });
+    }).pipe(Effect.scoped),
+  );
+
+  it.effect("reads the reset for an older provider.quota Go limit from the spent window", () =>
+    Effect.gen(function* () {
+      const ended = yield* endedBy(
+        {
+          type: "provider.quota",
+          message: "Subscription quota exceeded.",
+          status: 429,
+          response: {
+            body: JSON.stringify({
+              type: "error",
+              error: { type: "GoUsageLimitError", message: "Subscription quota exceeded." },
+            }),
+          },
+        },
+        Effect.succeed(goUsage(100)),
+      );
+      assert.equal(ended?.failure?.class, "usage_limit");
+      assert.equal(ended?.failure?.resetAt, "2026-11-06T04:12:00.000Z");
+    }).pipe(Effect.scoped),
+  );
+
+  it.effect("keeps a Go limit without a reset when Go usage cannot be read", () =>
+    Effect.gen(function* () {
+      const ended = yield* endedBy(
+        { type: "provider.go-limit", message: goLimitMessage },
+        Effect.succeed({
+          checkedAt: "2026-11-06T01:49:10.000Z",
+          windows: [],
+          unavailable: { reason: "probeFailed" },
+        }),
+      );
+      assert.deepEqual(ended?.failure, {
+        class: "usage_limit",
+        code: "provider.go-limit",
+        message: goLimitMessage,
+        retryable: null,
+      });
+    }).pipe(Effect.scoped),
+  );
+
+  it.effect("reads Go usage off the event lock, so Stop still ends the limited turn", () =>
+    Effect.gen(function* () {
+      const reading = yield* Deferred.make<void>();
+      const { runtime, thread } = yield* resumed(
+        [
+          out("session.prompt", { sessionID: SESSION, text: "<any>" }),
+          promptAccepted,
+          failWith({ type: "provider.go-limit", message: goLimitMessage, status: 429 }),
+          out("session.interrupt", { sessionID: SESSION }),
+          reply("session.interrupt", { interrupted: false }),
+        ],
+        {
+          readUsageLimits: Deferred.succeed(reading, undefined).pipe(Effect.andThen(Effect.never)),
+        },
+      );
+      const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped);
+      yield* runtime.startTurn(turnInput(thread));
+      yield* Deferred.await(reading);
+      yield* runtime.interruptTurn({
+        providerThread: thread,
+        providerTurnId: yield* providerTurnId,
+      });
+      assert.equal((yield* Fiber.join(terminal))?.status, "interrupted");
+    }).pipe(Effect.scoped),
+  );
+
+  it.effect("does not read Go usage for limits from other OpenCode providers", () =>
+    Effect.gen(function* () {
+      const ended = yield* endedBy(
+        {
+          type: "provider.rate-limit",
+          message: "Rate limit exceeded: free-models-per-min.",
+          status: 429,
+        },
+        Effect.die("Go usage must not be read for a non-Go limit"),
+      );
+      assert.deepEqual(ended?.failure, {
+        class: "usage_limit",
+        code: "provider.rate-limit",
+        message: "Rate limit exceeded: free-models-per-min.",
+        retryable: null,
+      });
+    }).pipe(Effect.scoped),
+  );
+
+  // An event for a session T3 Code does not know; it only holds the replay at its label.
+  const marker = (label: string) =>
+    event(
+      "session.model.selected",
+      {
+        sessionID: "ses_f148ca2deffeUnknownSession0",
+        model: { providerID: "opencode", id: "big-pickle", variant: "default" },
+      },
+      label,
+    );
+
+  // regression: a Stop sent during the Go usage read ended the turn failed, so it resumed at the reset
+  it.effect("keeps a Stop sent while Go usage is read, so the thread does not resume", () =>
+    Effect.gen(function* () {
+      const gate = makeProviderReplayGate(["interrupting"]);
+      const reading = yield* Deferred.make<void>();
+      const usage = yield* Deferred.make<ServerProviderUsageLimits>();
+      const { runtime, thread } = yield* resumed(
+        [
+          out("session.prompt", { sessionID: SESSION, text: "<any>" }),
+          promptAccepted,
+          failWith({ type: "provider.go-limit", message: goLimitMessage, status: 429 }),
+          out("session.interrupt", { sessionID: SESSION }),
+          // Holds the interrupt reply until the usage read has finished.
+          marker("interrupting"),
+          reply("session.interrupt", { interrupted: false }),
+        ],
+        {
+          replayGate: gate,
+          readUsageLimits: Deferred.succeed(reading, undefined).pipe(
+            Effect.andThen(Deferred.await(usage)),
+          ),
+        },
+      );
+      const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped);
+      yield* runtime.startTurn(turnInput(thread));
+      yield* Deferred.await(reading);
+      const stop = yield* runtime
+        .interruptTurn({ providerThread: thread, providerTurnId: yield* providerTurnId })
+        .pipe(Effect.forkScoped);
+      assert.isTrue(yield* Effect.promise(() => gate.waitForReached("interrupting")));
+      yield* Deferred.succeed(usage, goUsage(100));
+      const ended = yield* Fiber.join(terminal);
+      gate.release("interrupting");
+      yield* Fiber.join(stop);
+      assert.equal(ended?.status, "interrupted");
+      assert.notExists(ended?.failure);
+    }).pipe(Effect.scoped),
+  );
+
+  // regression: an undecodable start left the Go usage read able to end the restarted execution
+  it.effect("keeps a Go-limited turn that restarts with a start this build cannot decode", () =>
+    Effect.gen(function* () {
+      const gate = makeProviderReplayGate(["end"]);
+      const restarted = yield* Deferred.make<void>();
+      const { runtime, thread } = yield* resumed(
+        [
+          out("session.prompt", { sessionID: SESSION, text: "<any>" }),
+          promptAccepted,
+          failWith({ type: "provider.go-limit", message: goLimitMessage, status: 429 }),
+          {
+            type: "emit_inbound",
+            frame: {
+              type: "sdk.event",
+              event: {
+                id: "evt_executionstartedx",
+                created: 1,
+                type: "session.execution.started",
+                data: { sessionID: SESSION },
+                durable: "not-an-envelope",
+              },
+            },
+          },
+          event("session.text.started", {
+            sessionID: SESSION,
+            assistantMessageID: "msg_0eb735d5b001oAFVeY5jz3WD4Z",
+            ordinal: 0,
+          }),
+          event("session.text.ended", {
+            sessionID: SESSION,
+            assistantMessageID: "msg_0eb735d5b001oAFVeY5jz3WD4Z",
+            ordinal: 0,
+            text: "DONE",
+          }),
+          event("session.execution.succeeded", { sessionID: SESSION }, "end"),
+        ],
+        {
+          replayGate: gate,
+          // The restarted execution ends only after the read, whose end is queued first.
+          readUsageLimits: Deferred.await(restarted).pipe(
+            Effect.as(goUsage(100)),
+            Effect.ensuring(Effect.sync(() => gate.release("end"))),
+          ),
+        },
+      );
+      const terminal = yield* runtime.events.pipe(
+        Stream.tap((event) =>
+          event.type === "turn_item.updated" &&
+          event.turnItem.type === "assistant_message" &&
+          event.turnItem.text === "DONE"
+            ? Deferred.succeed(restarted, undefined)
+            : Effect.void,
+        ),
+        Stream.filter((event) => event.type === "turn.terminal"),
+        Stream.runHead,
+        Effect.map(Option.getOrUndefined),
+        Effect.forkScoped,
+      );
+      yield* runtime.startTurn(turnInput(thread));
+      const ended = yield* Fiber.join(terminal);
+      assert.deepInclude(ended, { type: "turn.terminal", status: "completed" });
+    }).pipe(Effect.scoped),
+  );
+
   it.effect("ends the turn when its terminal event is one this build cannot decode", () =>
     Effect.gen(function* () {
       const { runtime, thread } = yield* resumed([
@@ -2696,6 +2970,39 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {
       }).pipe(Effect.scoped),
   );
 
+  // OpenCode already delivered this execution's end; a listing taken at reconnect does not undo it.
+  it.effect("ends a Go limit after a reconnect even while OpenCode still lists the session", () =>
+    Effect.gen(function* () {
+      const gate = makeProviderReplayGate(["reconnected"]);
+      const { runtime, thread } = yield* resumed(
+        [
+          out("session.prompt", { sessionID: SESSION, text: "<any>" }),
+          promptAccepted,
+          failWith({ type: "provider.go-limit", message: goLimitMessage, status: 429 }),
+          { type: "runtime_exit", status: "success" } as const,
+          out("event.subscribe"),
+          out("session.active"),
+          replyData("session.active", { [SESSION]: { type: "running" } }),
+          ...openRequests([]),
+          marker("reconnected"),
+        ],
+        {
+          replayGate: gate,
+          // The read ends after the reconnect has listed the session as running.
+          readUsageLimits: Effect.promise(() => gate.waitForReached("reconnected")).pipe(
+            Effect.tap(() => Effect.sync(() => gate.release("reconnected"))),
+            Effect.as(goUsage(100)),
+          ),
+        },
+      );
+      const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped);
+      yield* runtime.startTurn(turnInput(thread));
+      const ended = yield* Fiber.join(terminal);
+      assert.equal(ended?.failure?.class, "usage_limit");
+      assert.equal(ended?.failure?.resetAt, "2026-11-06T04:12:00.000Z");
+    }).pipe(Effect.scoped),
+  );
+
   it.effect("keeps a turn still running after a reconnect open for its next events", () =>
     Effect.gen(function* () {
       const { runtime, thread } = yield* resumed([

apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts

@@ -6,6 +6,7 @@ import {
   ProviderSessionId,
   ThreadId,
   type ProviderReplayTranscript,
+  type ServerProviderUsageLimits,
 } from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
 import * as FileSystem from "effect/FileSystem";
@@ -257,13 +258,17 @@ const makeReplayAdapter = (
     readonly onPrompt?: (input: unknown) => void;
     readonly replayGate?: ProviderReplayGate;
     readonly borrowers?: { current: number };
+    readonly readUsageLimits?: Effect.Effect<ServerProviderUsageLimits>;
   },
 ) =>
   Effect.gen(function* () {
     const server = yield* replayServer(transcript, options);
-    return yield* OpenCode2Adapter.make(ProviderInstanceId.make("opencode")).pipe(
-      Effect.provideService(OpenCode2Server.OpenCode2Server, server),
-    );
+    return yield* OpenCode2Adapter.make(
+      ProviderInstanceId.make("opencode"),
+      options?.readUsageLimits === undefined
+        ? undefined
+        : { readUsageLimits: options.readUsageLimits },
+    ).pipe(Effect.provideService(OpenCode2Server.OpenCode2Server, server));
   });
 
 const layerReplayHost = TestProviderHost.layer().pipe(
@@ -314,6 +319,7 @@ export const openCode2ReplayRuntime = (
     readonly attachmentsDir?: string;
     readonly resolveExtractedTextPath?: (path: string) => Effect.Effect<string | undefined>;
     readonly fileSystem?: FileSystem.FileSystem;
+    readonly readUsageLimits?: Effect.Effect<ServerProviderUsageLimits>;
   },
 ) =>
   Effect.gen(function* () {