1import { assert, describe, it } from "@effect/vitest";
2import { RunId, ThreadId, type OrchestrationV2DomainEvent } from "@t3tools/contracts";
3import * as Clock from "effect/Clock";
4import * as DateTime from "effect/DateTime";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import * as Queue from "effect/Queue";
8import * as Stream from "effect/Stream";
9import * as TestClock from "effect/testing/TestClock";
10
11import * as Scheduler from "../scheduling/Scheduler.ts";
12import * as ServerSettings from "../serverSettings.ts";
13import * as Orchestrator from "./Orchestrator.ts";
14import * as ProjectionStore from "./ProjectionStore.ts";
15import * as ThreadManagement from "./ThreadManagementService.ts";
16import * as UsageLimitRecoveryWorker from "./UsageLimitRecoveryWorker.ts";
17import { limitRecoveryCommand } from "./UsageLimitRecoveryWorker.ts";
18
19it.effect("idle sweeps skip SQL until the next wake time or a relevant event", () =>
20 Effect.gen(function* () {
21 yield* TestClock.setTime(Date.parse("2026-07-01T09:00:00.000Z"));
22 const startMs = Date.parse("2026-07-01T09:00:00.000Z");
23 const queries = yield* Queue.unbounded<number>();
24 const events = yield* Queue.unbounded<OrchestrationV2DomainEvent>();
25 const layer = UsageLimitRecoveryWorker.layer.pipe(
26 Layer.provide(
27 Layer.mergeAll(
28 Scheduler.layer,
29 Layer.mock(ProjectionStore.ProjectionStoreV2)({
30 getLimitRecoveryCandidates: () =>
31 Clock.currentTimeMillis.pipe(
32 Effect.flatMap((nowMs) => Queue.offer(queries, nowMs - startMs)),
33 Effect.as([]),
34 ),
35 // One armed thread resets at 60s.
36 getLimitRecoveryWakeAt: (now) =>
37 Effect.succeed(
38 DateTime.toEpochMillis(now) < startMs + 60_000 ? startMs + 60_000 : null,
39 ),
40 }),
41 Layer.mock(ThreadManagement.ThreadManagementService)({}),
42 Layer.mock(Orchestrator.OrchestratorV2)({
43 streamDomainEvents: Stream.fromQueue(events),
44 }),
45 Layer.mock(ServerSettings.ServerSettingsService)({
46 getSettings: Effect.succeed({
47 autoResumeLimitedThreads: true,
48 snoozeLimitedThreads: false,
49 } as never),
50 streamChanges: Stream.never,
51 }),
52 ),
53 ),
54 );
55 yield* Effect.gen(function* () {
56 assert.equal(yield* Queue.take(queries), 0);
57 // Eleven idle ticks before the recorded wake time, then the wake tick itself.
58 yield* TestClock.adjust("60 seconds");
59 assert.equal(yield* Queue.take(queries), 60_000);
60 yield* Queue.offer(events, {
61 type: "turn-item.updated",
62 payload: { type: "assistant_message" },
63 } as never);
64 yield* TestClock.adjust("30 seconds");
65 yield* Queue.offer(events, { type: "run.updated", payload: { status: "failed" } } as never);
66 yield* TestClock.adjust("5 seconds");
67 // A streaming item at 90s does not wake it; the failed run does on the 95s tick.
68 assert.equal(yield* Queue.take(queries), 95_000);
69 }).pipe(Effect.provide(layer));
70 }),
71);
72
73const HOUR_MS = 60 * 60 * 1000;
74const nowMs = Date.parse("2026-10-19T12:00:00.000Z");
75
76const failedThread = (failedAgoMs: number, resetAgoMs: number) => {
77 const failedAt = DateTime.makeUnsafe(nowMs - failedAgoMs);
78 const candidate: ProjectionStore.ProjectionLimitRecoveryCandidate = {
79 id: ThreadId.make("thread:limited"),
80 status: "failed",
81 lastErrorClass: "usage_limit",
82 usageLimitResetAt: DateTime.formatIso(DateTime.makeUnsafe(nowMs - resetAgoMs)),
83 latestRunId: RunId.make("run:limited"),
84 latestRunCompletedAt: failedAt,
85 updatedAt: failedAt,
86 archivedAt: null,
87 settledOverride: null,
88 pendingRuntimeRequest: null,
89 limitRecovery: null,
90 snoozedUntil: null,
91 };
92 return candidate;
93};
94
95describe("limitRecoveryCommand arming", () => {
96 it("does not freshly arm a failure whose reset passed days ago", () => {
97 const thread = failedThread(5 * 24 * HOUR_MS, 4 * 24 * HOUR_MS);
98 assert.isNull(limitRecoveryCommand(thread, true, nowMs));
99 assert.isNull(limitRecoveryCommand(thread, true, nowMs, true));
100 });
101
102 it("still arms a failure whose reset only just passed", () => {
103 const thread = failedThread(6 * HOUR_MS, HOUR_MS);
104 const arm = limitRecoveryCommand(thread, true, nowMs);
105 assert.equal(arm?.type, "thread.metadata.update");
106 });
107
108 it("arms a reset that is exactly 24 hours overdue", () => {
109 const thread = failedThread(30 * HOUR_MS, 24 * HOUR_MS);
110 const arm = limitRecoveryCommand(thread, true, nowMs);
111 assert.equal(arm?.type, "thread.metadata.update");
112 });
113
114 it("does not arm a reset that is 24 hours and 1ms overdue", () => {
115 const thread = failedThread(30 * HOUR_MS, 24 * HOUR_MS + 1);
116 assert.isNull(limitRecoveryCommand(thread, true, nowMs));
117 });
118});