1import { describe, expect, it } from "@effect/vitest";
2import {
3 CommandId,
4 EnvironmentId,
5 MessageId,
6 EventId,
7 TurnItemId,
8 RuntimeRequestId,
9 type OrchestrationV2ShellSnapshot,
10} from "@t3tools/contracts";
11import * as DateTime from "effect/DateTime";
12import * as Effect from "effect/Effect";
13import * as Option from "effect/Option";
14import * as Schema from "effect/Schema";
15
16import {
17 ORCHESTRATION_CACHE_SCHEMA_VERSION,
18 ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
19 StoredOrchestrationShellSnapshot,
20 StoredOrchestrationThreadSnapshot,
21 decodeOrDiscardOrchestrationCache,
22} from "./orchestrationCache.ts";
23import {
24 v2Now,
25 v2Projection,
26 v2ShellSnapshot,
27 v2ThreadShell,
28 v2ThreadId,
29} from "../state/orchestrationV2TestFixtures.ts";
30
31import { applyOrchestrationV2ProjectionEvent } from "../state/orchestrationV2Projection.ts";
32
33const environmentId = EnvironmentId.make("environment-cache-test");
34const StoredShellSnapshotJson = Schema.fromJsonString(StoredOrchestrationShellSnapshot);
35const StoredThreadSnapshotJson = Schema.fromJsonString(StoredOrchestrationThreadSnapshot);
36const encodeStoredShellSnapshot = Schema.encodeSync(StoredOrchestrationShellSnapshot);
37const decodeStoredShellSnapshotSync = Schema.decodeUnknownSync(StoredOrchestrationShellSnapshot);
38const encodeStoredShellSnapshotJson = Schema.encodeSync(StoredShellSnapshotJson);
39const decodeStoredShellSnapshotJson = Schema.decodeUnknownSync(StoredShellSnapshotJson);
40const encodeStoredThreadSnapshotJson = Schema.encodeSync(StoredThreadSnapshotJson);
41const decodeStoredThreadSnapshotJson = Schema.decodeUnknownSync(StoredThreadSnapshotJson);
42
43class TestCacheDecodeError extends Schema.TaggedError<TestCacheDecodeError>()(
44 "TestCacheDecodeError",
45 {
46 message: Schema.String,
47 },
48) {}
49
50const shellSnapshotWithSummaries: OrchestrationV2ShellSnapshot = {
51 ...v2ShellSnapshot,
52 threads: [
53 {
54 ...v2ThreadShell,
55 pendingRuntimeRequest: {
56 id: RuntimeRequestId.make("runtime-request-v2"),
57 kind: "command",
58 createdAt: v2Now,
59 },
60 latestVisibleMessage: {
61 id: MessageId.make("message-v2"),
62 role: "assistant",
63 text: "Done",
64 updatedAt: v2Now,
65 },
66 latestUserMessageAt: v2Now,
67 settledAt: v2Now,
68 titleRegeneration: {
69 requestId: CommandId.make("title-regeneration-v2"),
70 startedAt: v2Now,
71 },
72 },
73 ],
74};
75
76describe("orchestration cache envelopes", () => {
77 it("preserves live provider notices through persisted thread snapshots", () => {
78 const item = {
79 id: TurnItemId.make("notice-cache"),
80 threadId: v2ThreadId,
81 runId: null,
82 nodeId: null,
83 providerThreadId: null,
84 providerTurnId: null,
85 nativeItemRef: null,
86 parentItemId: null,
87 ordinal: 1,
88 status: "completed" as const,
89 title: null,
90 startedAt: v2Now,
91 completedAt: v2Now,
92 updatedAt: v2Now,
93 type: "system_notice" as const,
94 message: "Safeguards flagged this message. Switched to Opus 4.8.",
95 };
96 const projection = applyOrchestrationV2ProjectionEvent(v2Projection, {
97 id: EventId.make("notice-cache-event"),
98 type: "turn-item.updated",
99 threadId: v2ThreadId,
100 occurredAt: v2Now,
101 payload: item,
102 });
103 expect(projection).not.toBeNull();
104 if (projection === null) throw new Error("Expected live notice projection");
105 const encoded = encodeStoredThreadSnapshotJson({
106 schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
107 environmentId,
108 threadId: v2ThreadId,
109 snapshot: { snapshotSequence: 5, projection },
110 });
111 const decoded = decodeStoredThreadSnapshotJson(encoded).snapshot.projection;
112 expect(decoded.turnItems.find((entry) => entry.id === item.id)).toEqual(item);
113 expect(decoded.visibleTurnItems.find((entry) => entry.item.id === item.id)?.item).toEqual(item);
114 });
115
116 it.effect("round-trips V2 shell and thread cache envelopes as JSON", () =>
117 Effect.sync(() => {
118 const encodedShell = encodeStoredShellSnapshotJson({
119 schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION,
120 environmentId,
121 snapshot: shellSnapshotWithSummaries,
122 });
123 const shell = decodeStoredShellSnapshotJson(encodedShell);
124 const encodedThread = encodeStoredThreadSnapshotJson({
125 schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
126 environmentId,
127 threadId: v2ThreadId,
128 snapshot: { snapshotSequence: 4, projection: v2Projection },
129 });
130 const thread = decodeStoredThreadSnapshotJson(encodedThread);
131 const [threadShell] = shell.snapshot.threads;
132
133 expect(threadShell?.latestVisibleMessage?.text).toBe("Done");
134 expect(
135 threadShell?.latestVisibleMessage === null ||
136 threadShell?.latestVisibleMessage === undefined
137 ? null
138 : DateTime.formatIso(threadShell.latestVisibleMessage.updatedAt),
139 ).toBe("2026-06-20T00:00:00.000Z");
140 expect(
141 threadShell?.latestUserMessageAt === null || threadShell?.latestUserMessageAt === undefined
142 ? null
143 : DateTime.formatIso(threadShell.latestUserMessageAt),
144 ).toBe("2026-06-20T00:00:00.000Z");
145 expect(
146 threadShell?.settledAt === null || threadShell?.settledAt === undefined
147 ? null
148 : DateTime.formatIso(threadShell.settledAt),
149 ).toBe("2026-06-20T00:00:00.000Z");
150 expect(
151 threadShell?.titleRegeneration === null || threadShell?.titleRegeneration === undefined
152 ? null
153 : DateTime.formatIso(threadShell.titleRegeneration.startedAt),
154 ).toBe("2026-06-20T00:00:00.000Z");
155 expect(DateTime.formatIso(thread.snapshot.projection.thread.updatedAt)).toBe(
156 "2026-06-20T00:00:00.000Z",
157 );
158 expect(thread.snapshot.projection).toEqual(v2Projection);
159 expect(thread.snapshot.snapshotSequence).toBe(4);
160 }),
161 );
162
163 it("rejects thread snapshots saved before queued-run items were kept, but not shells", () => {
164 // Version 3 thread windows may have dropped a queued run's items (#16987).
165 const staleThread = encodeStoredThreadSnapshotJson({
166 schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
167 environmentId,
168 threadId: v2ThreadId,
169 snapshot: { snapshotSequence: 4, projection: v2Projection },
170 }).replace(`"schemaVersion":${ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION}`, `"schemaVersion":3`);
171 expect(() => decodeStoredThreadSnapshotJson(staleThread)).toThrow();
172 const shell = encodeStoredShellSnapshot({
173 schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION,
174 environmentId,
175 snapshot: v2ShellSnapshot,
176 });
177 expect(decodeStoredShellSnapshotSync(shell).schemaVersion).toBe(3);
178 });
179
180 it.effect("discards V1-versioned cache envelopes after a decode failure", () =>
181 Effect.gen(function* () {
182 let discardCount = 0;
183 const encodedShell = encodeStoredShellSnapshot({
184 schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION,
185 environmentId,
186 snapshot: v2ShellSnapshot,
187 });
188 const decoded = Effect.try({
189 try: () => decodeStoredShellSnapshotSync({ ...encodedShell, schemaVersion: 1 }),
190 catch: (cause) => new TestCacheDecodeError({ message: String(cause) }),
191 }).pipe(Effect.map(Option.some));
192
193 const result = yield* decodeOrDiscardOrchestrationCache(
194 decoded,
195 Effect.sync(() => {
196 discardCount += 1;
197 }),
198 );
199
200 expect(Option.isNone(result)).toBe(true);
201 expect(discardCount).toBe(1);
202 }),
203 );
204});