fix(server): a usage-limited thread no longer auto-settles

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

@@ -9,6 +9,7 @@ import {
   ProjectId,
   ProviderDriverKind,
   ProviderInstanceId,
+  ProviderSessionId,
   ProviderThreadId,
   RunId,
   RuntimeRequestId,
@@ -20,6 +21,7 @@ import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 
 import * as SqlitePersistence from "../persistence/Sqlite.ts";
+import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
 import * as ProjectionStore from "./ProjectionStore.ts";
 import {
   isAutoSettlementCandidate,
@@ -77,6 +79,7 @@ const createRun = Effect.fn(function* (
   threadId: ThreadId,
   status: OrchestrationV2Run["status"] = "completed",
   ordinal = 1,
+  completedAt = old,
 ) {
   const store = yield* ProjectionStore.ProjectionStoreV2;
   const runId = RunId.make(`run:${threadId}:${ordinal}`);
@@ -98,7 +101,7 @@ const createRun = Effect.fn(function* (
       status,
       requestedAt: old,
       startedAt: status === "queued" ? null : old,
-      completedAt: ["completed", "rolled_back", "failed"].includes(status) ? old : null,
+      completedAt: ["completed", "rolled_back", "failed"].includes(status) ? completedAt : null,
       checkpointId: null,
       contextHandoffId: null,
     },
@@ -141,6 +144,69 @@ const createItem = Effect.fn(function* (
   });
 });
 
+const createFailure = Effect.fn(function* (
+  threadId: ThreadId,
+  runId: RunId,
+  failureClass: "usage_limit" | "provider_error",
+) {
+  const store = yield* ProjectionStore.ProjectionStoreV2;
+  yield* store.apply({
+    id: EventId.make(`event:failure:${runId}`),
+    type: "turn-item.updated",
+    threadId,
+    occurredAt: old,
+    payload: {
+      id: TurnItemId.make(`failure:${runId}`),
+      threadId,
+      runId,
+      nodeId: null,
+      providerThreadId: null,
+      providerTurnId: null,
+      nativeItemRef: null,
+      parentItemId: null,
+      ordinal: 1,
+      type: "error",
+      status: "failed",
+      title: failureClass,
+      startedAt: old,
+      completedAt: old,
+      updatedAt: old,
+      failure: {
+        class: failureClass,
+        message: failureClass,
+        code: null,
+        retryable: true,
+        resetAt: DateTime.formatIso(DateTime.add(now, { hours: 1 })),
+      },
+    },
+  });
+});
+
+const attachSessionError = Effect.fn(function* (threadId: ThreadId, lastError: string) {
+  const store = yield* ProjectionStore.ProjectionStoreV2;
+  const driver = ProviderDriverKind.make("codex");
+  yield* store.apply({
+    id: EventId.make(`event:session:${threadId}`),
+    type: "provider-session.attached",
+    threadId,
+    driver,
+    providerInstanceId,
+    occurredAt: old,
+    payload: {
+      id: ProviderSessionId.make(`provider-session:${threadId}`),
+      driver,
+      providerInstanceId,
+      status: "error",
+      cwd: "/workspace",
+      model: modelSelection.model,
+      capabilities: CodexProviderCapabilitiesV2,
+      createdAt: old,
+      updatedAt: old,
+      lastError,
+    },
+  });
+});
+
 it.effect.each([["sql", layerSql]] as const)(
   "%s: pending approvals wake a snoozed question even with older creation times",
   ([, testLayer]) =>
@@ -284,6 +350,49 @@ it.effect(
         snoozedAt: DateTime.subtract(old, { days: 1 }),
       });
       yield* createRun(woke, "failed");
+      const limited = yield* createThread("usage-limited");
+      yield* createFailure(limited, yield* createRun(limited, "failed"), "usage_limit");
+      // A message queued behind the limit is the newest run, but the limit still blocks.
+      const queuedBehindLimit = yield* createThread("queued-behind-limit");
+      yield* createFailure(
+        queuedBehindLimit,
+        yield* createRun(queuedBehindLimit, "failed"),
+        "usage_limit",
+      );
+      yield* createRun(queuedBehindLimit, "queued", 2);
+      // Held queues run out of order; the run that ended last decides, not the ordinal.
+      const limitEndedLast = yield* createThread("limit-ended-last");
+      yield* createFailure(
+        limitEndedLast,
+        yield* createRun(limitEndedLast, "failed", 1, DateTime.add(old, { hours: 1 })),
+        "usage_limit",
+      );
+      yield* createFailure(
+        limitEndedLast,
+        yield* createRun(limitEndedLast, "failed", 2),
+        "provider_error",
+      );
+      // threadErrorSummary: a matching session error keeps the limit, a differing one replaces it.
+      const sessionErrorMatchingLimit = yield* createThread("session-error-matching-limit");
+      yield* createFailure(
+        sessionErrorMatchingLimit,
+        yield* createRun(sessionErrorMatchingLimit, "failed"),
+        "usage_limit",
+      );
+      yield* attachSessionError(sessionErrorMatchingLimit, "usage_limit");
+      const sessionErrorAfterLimit = yield* createThread("session-error-after-limit");
+      yield* createFailure(
+        sessionErrorAfterLimit,
+        yield* createRun(sessionErrorAfterLimit, "failed"),
+        "usage_limit",
+      );
+      yield* attachSessionError(sessionErrorAfterLimit, "Session crashed.");
+      const providerFailed = yield* createThread("provider-failed");
+      yield* createFailure(
+        providerFailed,
+        yield* createRun(providerFailed, "failed"),
+        "provider_error",
+      );
       const background = yield* createThread("idle-background");
       yield* createItem(background, yield* createRun(background), "idle");
       const persistent = yield* createThread("persistent-monitor");
@@ -324,7 +433,17 @@ it.effect(
       );
       assert.deepEqual(
         new Set(eligible.map((thread) => thread.id)),
-        new Set([idle, completed, queued, woke, background, persistent, rolledBack]),
+        new Set([
+          idle,
+          completed,
+          queued,
+          woke,
+          providerFailed,
+          sessionErrorAfterLimit,
+          background,
+          persistent,
+          rolledBack,
+        ]),
       );
       assert.deepEqual(
         new Set(eligible.map((thread) => thread.id)),

apps/server/src/orchestration-v2/ProjectionStore.ts

@@ -215,6 +215,7 @@ export type ProjectionSettlementCandidate = Pick<
   | "latestRunCompletedAt"
   | "latestUserMessageAt"
   | "status"
+  | "lastErrorClass"
   | "activityRunStartedAt"
   | "activityRunStatus"
   | "pendingRuntimeRequest"
@@ -6581,6 +6582,54 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
                       )
                   )
               )
+              -- A usage limit interrupts work; settling would hide its recovery controls.
+              -- Matches the shell's blocked run, so a message queued behind the limit
+              -- does not hide it. The cheap EXISTS keeps the run sort off other threads.
+              AND NOT (
+                EXISTS (
+                  SELECT 1 FROM orchestration_v2_projection_turn_items failure
+                  WHERE failure.thread_id = t.thread_id
+                    AND failure.type = 'error' AND failure.status = 'failed'
+                    AND json_extract(failure.payload_json, '$.failure.class') = 'usage_limit'
+                )
+                AND EXISTS (
+                  SELECT 1 FROM orchestration_v2_projection_runs blocked
+                  WHERE blocked.run_id = (
+                    SELECT latest.run_id FROM orchestration_v2_projection_runs latest
+                    WHERE latest.thread_id = t.thread_id
+                      AND latest.status <> 'queued'
+                      AND NOT (
+                        latest.status = 'cancelled'
+                        AND json_extract(latest.payload_json, '$.startedAt') IS NULL
+                      )
+                    -- latestExecutedRun: the run that ended last (runRanAfter).
+                    ORDER BY latest.completed_at IS NULL DESC, latest.completed_at DESC,
+                      latest.ordinal DESC, latest.run_id DESC
+                    LIMIT 1
+                  )
+                    AND blocked.status = 'failed'
+                    -- threadErrorSummary: a differing session error replaces the failure.
+                    AND (
+                      SELECT json_extract(error.payload_json, '$.failure.class') = 'usage_limit'
+                        AND COALESCE((
+                          SELECT json_extract(session.payload_json, '$.lastError')
+                          FROM orchestration_v2_projection_provider_session_bindings binding
+                          CROSS JOIN orchestration_v2_projection_provider_sessions session
+                            ON session.provider_session_id = binding.provider_session_id
+                          WHERE binding.thread_id = t.thread_id
+                            AND session.provider_instance_id = t.provider_instance_id
+                          ORDER BY session.updated_at DESC, session.provider_session_id DESC
+                          LIMIT 1
+                        ) = json_extract(error.payload_json, '$.failure.message'), 1)
+                      FROM orchestration_v2_projection_turn_items error
+                      WHERE error.thread_id = t.thread_id AND error.run_id = blocked.run_id
+                        AND error.type = 'error' AND error.status = 'failed'
+                        AND error.node_id IS json_extract(blocked.payload_json, '$.rootNodeId')
+                      ORDER BY error.updated_at DESC, error.ordinal DESC, error.turn_item_id DESC
+                      LIMIT 1
+                    ) = 1
+                )
+              )
             ORDER BY t.updated_at ASC, t.thread_id ASC
           `;
             if (rows.length === 0) return [];
@@ -6628,6 +6677,8 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
                   activityRunStatus: null,
                   activityRunStartedAt: null,
                   pendingRuntimeRequest: null,
+                  // The query above drops usage-limited threads.
+                  lastErrorClass: null,
                   pendingGuestMessageCount: pendingGuestMessageCount(thread.guestShare),
                   pendingBackgroundTasks: derivePendingBackgroundWork({
                     latestRun:

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

@@ -161,6 +161,17 @@ describe("isAutoSettlementCandidate", () => {
     expect(ThreadSettlementService.isSnoozed(snoozed, NOW_MS)).toBe(false);
   });
 
+  it("never settles a thread stopped by a usage limit", () => {
+    const limited = shell({ status: "failed", lastErrorClass: "usage_limit" });
+    expect(ThreadSettlementService.isAutoSettlementCandidate(limited, NOW_MS)).toBe(false);
+    expect(
+      ThreadSettlementService.isAutoSettlementCandidate(
+        { ...limited, lastErrorClass: "provider_error" },
+        NOW_MS,
+      ),
+    ).toBe(true);
+  });
+
   it("keeps snoozed threads parked until a fresh failure or timer wake", () => {
     const snoozed = shell({
       snoozedUntil: at(60 * 60 * 1_000),

apps/server/src/orchestration-v2/ThreadSettlementService.ts

@@ -153,6 +153,8 @@ export function isAutoSettlementCandidate(
   // A guest is waiting on the owner; settling would hide the Request.
   if ((thread.pendingGuestMessageCount ?? 0) > 0) return false;
   if (thread.activityRunStatus != null) return false;
+  // A usage limit interrupts the work; settling would hide Resume at reset.
+  if (thread.status === "failed" && thread.lastErrorClass === "usage_limit") return false;
   if (backgroundWorkHoldsCompletion(thread.pendingBackgroundTasks ?? [])) return false;
   if (threadHasQueuedTurnStart(thread, nowMs)) return false;
   // Interrupted runs must keep the same snooze protection the sidebar shows.