11import { assert , describe , it } from "@effect/vitest" ;
2- import { RunId , ThreadId , type OrchestrationV2DomainEvent } from "@t3tools/contracts" ;
2+ import {
3+ CommandId ,
4+ ProviderInstanceId ,
5+ RunId ,
6+ ThreadId ,
7+ type ServerProviderUsageLimits ,
8+ type OrchestrationV2DomainEvent ,
9+ } from "@t3tools/contracts" ;
310import * as Clock from "effect/Clock" ;
411import * as DateTime from "effect/DateTime" ;
512import * as Effect from "effect/Effect" ;
@@ -8,6 +15,7 @@ import * as Queue from "effect/Queue";
815import * as Stream from "effect/Stream" ;
916import * as TestClock from "effect/testing/TestClock" ;
1017
18+ import * as ProviderRegistry from "../provider/ProviderRegistry.ts" ;
1119import * as Scheduler from "../scheduling/Scheduler.ts" ;
1220import * as ServerSettings from "../serverSettings.ts" ;
1321import * as Orchestrator from "./Orchestrator.ts" ;
@@ -22,10 +30,15 @@ it.effect("idle sweeps skip SQL until the next wake time or a relevant event", (
2230 const startMs = Date . parse ( "2026-07-01T09:00:00.000Z" ) ;
2331 const queries = yield * Queue . unbounded < number > ( ) ;
2432 const events = yield * Queue . unbounded < OrchestrationV2DomainEvent > ( ) ;
33+ const providerChanges = yield * Queue . unbounded < never [ ] > ( ) ;
2534 const layer = UsageLimitRecoveryWorker . layer . pipe (
2635 Layer . provide (
2736 Layer . mergeAll (
2837 Scheduler . layer ,
38+ Layer . mock ( ProviderRegistry . ProviderRegistry ) ( {
39+ getProviders : Effect . succeed ( [ ] ) ,
40+ streamChanges : Stream . fromQueue ( providerChanges ) ,
41+ } ) ,
2942 Layer . mock ( ProjectionStore . ProjectionStoreV2 ) ( {
3043 getLimitRecoveryCandidates : ( ) =>
3144 Clock . currentTimeMillis . pipe (
@@ -66,6 +79,9 @@ it.effect("idle sweeps skip SQL until the next wake time or a relevant event", (
6679 yield * TestClock . adjust ( "5 seconds" ) ;
6780 // A streaming item at 90s does not wake it; the failed run does on the 95s tick.
6881 assert . equal ( yield * Queue . take ( queries ) , 95_000 ) ;
82+ yield * Queue . offer ( providerChanges , [ ] ) ;
83+ yield * TestClock . adjust ( "5 seconds" ) ;
84+ assert . equal ( yield * Queue . take ( queries ) , 100_000 ) ;
6985 } ) . pipe ( Effect . provide ( layer ) ) ;
7086 } ) ,
7187) ;
@@ -77,12 +93,13 @@ const failedThread = (failedAgoMs: number, resetAgoMs: number) => {
7793 const failedAt = DateTime . makeUnsafe ( nowMs - failedAgoMs ) ;
7894 const candidate : ProjectionStore . ProjectionLimitRecoveryCandidate = {
7995 id : ThreadId . make ( "thread:limited" ) ,
96+ providerInstanceId : ProviderInstanceId . make ( "codex" ) ,
8097 status : "failed" ,
8198 lastErrorClass : "usage_limit" ,
8299 usageLimitResetAt : DateTime . formatIso ( DateTime . makeUnsafe ( nowMs - resetAgoMs ) ) ,
83100 latestRunId : RunId . make ( "run:limited" ) ,
84101 latestRunCompletedAt : failedAt ,
85- updatedAt : failedAt ,
102+ latestRunRequestedAt : failedAt ,
86103 archivedAt : null ,
87104 settledOverride : null ,
88105 pendingRuntimeRequest : null ,
@@ -116,3 +133,73 @@ describe("limitRecoveryCommand arming", () => {
116133 assert . isNull ( limitRecoveryCommand ( thread , true , nowMs ) ) ;
117134 } ) ;
118135} ) ;
136+
137+ describe ( "usage-based recovery" , ( ) => {
138+ const reading = ( checkedAt : number , usedPercent : number ) : ServerProviderUsageLimits => ( {
139+ checkedAt : DateTime . formatIso ( DateTime . makeUnsafe ( checkedAt ) ) ,
140+ windows : [ { id : "weekly" , label : "Weekly" , kind : "weekly" , usedPercent } ] ,
141+ } ) ;
142+ const armedThread = ( autoResume : boolean ) => {
143+ const thread = failedThread ( HOUR_MS , - 48 * HOUR_MS ) ;
144+ return {
145+ ...thread ,
146+ limitRecovery : {
147+ runId : thread . latestRunId ! ,
148+ resetAt : thread . usageLimitResetAt ! ,
149+ autoResume,
150+ snooze : true ,
151+ requestId : CommandId . make ( "recovery:request" ) ,
152+ } ,
153+ snoozedUntil : DateTime . makeUnsafe ( thread . usageLimitResetAt ! ) ,
154+ } ;
155+ } ;
156+ it ( "resumes early with fresh headroom and preserves the continuation identity" , ( ) => {
157+ const thread = armedThread ( true ) ;
158+ const command = limitRecoveryCommand ( thread , true , nowMs , true , reading ( nowMs , 2 ) ) ;
159+ assert . equal ( command ?. type , "message.dispatch" ) ;
160+ if ( command ?. type !== "message.dispatch" ) return ;
161+ assert . equal ( command . text , "Continue where you left off." ) ;
162+ const withoutCompletion = {
163+ ...thread ,
164+ latestRunCompletedAt : null ,
165+ updatedAt : DateTime . makeUnsafe ( nowMs + HOUR_MS ) ,
166+ } ;
167+ assert . equal (
168+ limitRecoveryCommand ( withoutCompletion , true , nowMs , true , reading ( nowMs , 2 ) ) ?. type ,
169+ "message.dispatch" ,
170+ ) ;
171+
172+ assert . equal ( command . usageLimitContinuationOfRunId , thread . latestRunId ) ;
173+ assert . equal (
174+ command . commandId ,
175+ limitRecoveryCommand (
176+ { ...thread , snoozedUntil : null } ,
177+ true ,
178+ Date . parse ( thread . usageLimitResetAt ! ) ,
179+ true ,
180+ ) ?. commandId ,
181+ ) ;
182+ } ) ;
183+ it ( "releases a snooze-only recovery with fresh headroom" , ( ) => {
184+ const command = limitRecoveryCommand ( armedThread ( false ) , false , nowMs , true , reading ( nowMs , 2 ) ) ;
185+ assert . equal ( command ?. type , "thread.metadata.update" ) ;
186+ if ( command ?. type !== "thread.metadata.update" ) return ;
187+ assert . equal ( command . limitRecovery ?. snooze , false ) ;
188+ assert . equal ( command . limitRecovery ?. autoResume , false ) ;
189+ } ) ;
190+ it . each ( [ true , false ] ) (
191+ "keeps autoResume=%s blocked for stale or exhausted readings" ,
192+ ( autoResume ) => {
193+ const thread = armedThread ( autoResume ) ;
194+ for ( const limits of [
195+ reading ( nowMs - 2 * HOUR_MS , 2 ) ,
196+ reading ( nowMs - HOUR_MS , 2 ) ,
197+ reading ( nowMs , 100 ) ,
198+ reading ( nowMs , 101 ) ,
199+ { checkedAt : DateTime . formatIso ( DateTime . makeUnsafe ( nowMs ) ) , windows : [ ] } ,
200+ ] ) {
201+ assert . isNull ( limitRecoveryCommand ( thread , autoResume , nowMs , true , limits ) ) ;
202+ }
203+ } ,
204+ ) ;
205+ } ) ;
0 commit comments