1import { CommandId, MessageId, type OrchestrationV2Command } from "@t3tools/contracts";
2import * as DateTime from "effect/DateTime";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Scheduler from "../scheduling/Scheduler.ts";
6import * as ServerSettings from "../serverSettings.ts";
7import * as ProjectionStore from "./ProjectionStore.ts";
8import * as ThreadManagement from "./ThreadManagementService.ts";
9
10/** The persisted run and reset form the identity of one recovery opportunity. */
11export function limitRecoveryCommand(
12 thread: ProjectionStore.ProjectionLimitRecoveryCandidate,
13 autoResume: boolean,
14 nowMs: number,
15 snooze = false,
16): OrchestrationV2Command | null {
17 if (
18 thread.status !== "failed" ||
19 thread.lastErrorClass !== "usage_limit" ||
20 !thread.latestRunId ||
21 !thread.usageLimitResetAt ||
22 thread.archivedAt !== null ||
23 thread.settledOverride === "settled" ||
24 thread.pendingRuntimeRequest !== null
25 )
26 return null;
27 const resetMs = Date.parse(thread.usageLimitResetAt);
28 // An already-expired window reported with a fresh failure cannot start a retry loop.
29 if (
30 !Number.isFinite(resetMs) ||
31 resetMs <= DateTime.toEpochMillis(thread.latestRunCompletedAt ?? thread.updatedAt)
32 )
33 return null;
34 const identity = `${thread.id}:${thread.latestRunId}:${resetMs}`;
35 const recovery = thread.limitRecovery;
36 if (recovery?.runId !== thread.latestRunId || recovery.resetAt !== thread.usageLimitResetAt) {
37 if (!autoResume && (!snooze || resetMs <= nowMs)) return null;
38 return {
39 type: "thread.metadata.update",
40 commandId: CommandId.make(`limit-arm:${identity}`),
41 threadId: thread.id,
42 limitRecovery: {
43 runId: thread.latestRunId,
44 resetAt: thread.usageLimitResetAt,
45 autoResume,
46 snooze: snooze && resetMs > nowMs,
47 },
48 };
49 }
50 if (
51 !recovery.autoResume ||
52 resetMs > nowMs ||
53 (thread.snoozedUntil != null && DateTime.toEpochMillis(thread.snoozedUntil) > nowMs)
54 )
55 return null;
56 const deliveryIdentity = `${identity}:${recovery.requestId ?? "legacy"}`;
57 return {
58 type: "message.dispatch",
59 commandId: CommandId.make(`limit-resume:${deliveryIdentity}`),
60 messageId: MessageId.make(`limit-resume:${deliveryIdentity}`),
61 threadId: thread.id,
62 usageLimitContinuationOfRunId: thread.latestRunId,
63 ...(recovery.requestId === undefined
64 ? {}
65 : { usageLimitRecoveryRequestId: recovery.requestId }),
66 text: "Continue where you left off.",
67 attachments: [],
68 dispatchMode: { type: "start_immediately" },
69 createdBy: "user",
70 creationSource: "server",
71 };
72}
73
74const makeSweep = Effect.gen(function* () {
75 const projections = yield* ProjectionStore.ProjectionStoreV2;
76 const threads = yield* ThreadManagement.ThreadManagementService;
77 const settings = yield* ServerSettings.ServerSettingsService;
78 return Effect.fn("UsageLimitRecoveryWorker.sweep")(function* () {
79 const preferences = yield* settings.getSettings;
80 const now = yield* DateTime.now;
81 const candidates = yield* projections.getLimitRecoveryCandidates({
82 now,
83 autoResume: preferences.autoResumeLimitedThreads,
84 snooze: preferences.snoozeLimitedThreads,
85 });
86 const nowMs = DateTime.toEpochMillis(now);
87 for (const thread of candidates) {
88 const command = limitRecoveryCommand(
89 thread,
90 preferences.autoResumeLimitedThreads,
91 nowMs,
92 preferences.snoozeLimitedThreads,
93 );
94 if (command === null) continue;
95 yield* threads.dispatch(command).pipe(
96 Effect.catchCause((cause) =>
97 Effect.logWarning("orchestration-v2.limit-recovery.dispatch-failed", {
98 threadId: thread.id,
99 cause,
100 }),
101 ),
102 );
103 }
104 });
105});
106
107// The shared scheduler derives due work from persisted failures and recovery
108// choices, so restarts need no timer restoration or connected client.
109export const layer = Layer.effectDiscard(
110 Effect.gen(function* () {
111 const sweep = yield* makeSweep;
112 const scheduler = yield* Scheduler.Scheduler;
113 yield* scheduler.register("usage-limit-recovery", sweep());
114 }),
115);