1import {
2 ProviderInstanceId,
3 RunId,
4 ThreadId,
5 TurnItemId,
6 type OrchestrationV2ThreadProjection,
7} from "@t3tools/contracts";
8import * as Effect from "effect/Effect";
9import * as Layer from "effect/Layer";
10import { expect, it } from "vite-plus/test";
11
12import * as EffectWorker from "./EffectWorker.ts";
13import * as EffectOutbox from "./EffectOutbox.ts";
14import * as EventSink from "./EventSink.ts";
15import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
16import * as ProjectionStore from "./ProjectionStore.ts";
17import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts";
18import * as ServerSettings from "../serverSettings.ts";
19
20it("uses the thread provider for stale background work without provider threads", async () => {
21 const threadId = ThreadId.make("thread_recovery_background_no_provider_threads");
22 const runId = RunId.make("run_recovery_background_no_provider_threads");
23 const itemId = TurnItemId.make("turn_item_recovery_background_no_provider_threads");
24 const providerInstanceId = ProviderInstanceId.make("codex");
25 let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =
26 null;
27 const projection = {
28 thread: { id: threadId, providerInstanceId },
29 runtimeRequests: [],
30 providerSessions: [],
31 providerThreads: [],
32 providerTurns: [],
33 runs: [{ id: runId, status: "completed", providerInstanceId }],
34 attempts: [],
35 nodes: [],
36 subagents: [],
37 messages: [],
38 turnItems: [
39 {
40 id: itemId,
41 runId: null,
42 nodeId: null,
43 providerThreadId: null,
44 type: "command_execution",
45 status: "running",
46 },
47 ],
48 } as unknown as OrchestrationV2ThreadProjection;
49 const layer = ProviderRuntimeRecovery.layer.pipe(
50 Layer.provide(ServerSettings.layerTest()),
51 Layer.provide(
52 Layer.mergeAll(
53 Layer.mock(ProjectionStore.ProjectionStoreV2)({
54 getRecoveryThreadIds: () => Effect.succeed([threadId]),
55 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
56 }),
57 Layer.mock(EventSink.EventSinkV2)({
58 commitCommand: (input) => {
59 committedInput = input;
60 return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);
61 },
62 }),
63 IdAllocator.layer,
64 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
65 runRecoveryOnce: Effect.succeed(false),
66 }),
67 Layer.mock(EffectOutbox.EffectOutboxV2)({
68 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
69 }),
70 ),
71 ),
72 );
73
74 await Effect.gen(function* () {
75 yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup");
76 const event = committedInput?.events.find(
77 (candidate) => candidate.type === "turn-item.updated",
78 );
79 expect(event?.providerInstanceId).toBe(providerInstanceId);
80 expect(event?.type === "turn-item.updated" ? event.payload.status : null).toBe("cancelled");
81 }).pipe(Effect.provide(layer), Effect.runPromise);
82});