diff --git a/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts b/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts index 825bc5d7ce53c..414f7c3ec6689 100644 --- a/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts +++ b/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)), diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 23874c6014fe0..56c027864a497 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/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 = ) ) ) + -- 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 = activityRunStatus: null, activityRunStartedAt: null, pendingRuntimeRequest: null, + // The query above drops usage-limited threads. + lastErrorClass: null, pendingGuestMessageCount: pendingGuestMessageCount(thread.guestShare), pendingBackgroundTasks: derivePendingBackgroundWork({ latestRun: diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts index ff5caf6a591fa..db86da920fd46 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts +++ b/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), diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.ts index 40c6bf7d057f5..9e84902cb5c80 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.ts +++ b/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.