1import { assert, it } from "@effect/vitest";
2import {
3 AuthAdministrativeScopes,
4 AuthOrchestrationOperateScope,
5 AuthOrchestrationReadScope,
6 AuthSessionId,
7 CommandId,
8 EnvironmentId,
9 HubError,
10 HubOrganizationCommand,
11 ORCHESTRATION_PROTOCOL_VERSION,
12 OrchestrationV2ShellSnapshot,
13 OrchestrationV2ThreadShell,
14 ThreadId,
15 type AuthEnvironmentScope,
16 type ExecutionEnvironmentDescriptor,
17} from "@t3tools/contracts";
18import * as NodeServices from "@effect/platform-node/NodeServices";
19import * as ConfigProvider from "effect/ConfigProvider";
20import * as ArchiveCatalog from "../archive/ArchiveCatalog.ts";
21import * as ArchiveObjects from "../archive/ArchiveObjects.ts";
22import * as Context from "effect/Context";
23import * as DateTime from "effect/DateTime";
24import * as Deferred from "effect/Deferred";
25import * as Effect from "effect/Effect";
26import * as Fiber from "effect/Fiber";
27import * as Layer from "effect/Layer";
28import * as Schema from "effect/Schema";
29import * as TestClock from "effect/testing/TestClock";
30import { HttpClient, HttpClientResponse } from "effect/http";
31import * as SqlitePersistence from "../persistence/Sqlite.ts";
32import * as Hub from "./Hub.ts";
33import * as HubInbox from "./HubInbox.ts";
34import { hubInboxThread as thread } from "./HubInbox.testkit.ts";
35
36const encodeSnapshot = Schema.encodeEffect(Schema.toCodecJson(OrchestrationV2ShellSnapshot));
37const decodeCommand = Schema.decodeEffect(
38 Schema.fromJsonString(Schema.toCodecJson(HubOrganizationCommand)),
39);
40
41const environmentId = EnvironmentId.make("laptop");
42const threadId = ThreadId.make("chat");
43const session = {
44 sessionId: AuthSessionId.make("phone"),
45 subject: "phone",
46 method: "bearer-access-token" as const,
47 scopes: AuthAdministrativeScopes,
48};
49
50const fixture = Effect.gen(function* () {
51 let snapshot: OrchestrationV2ShellSnapshot = {
52 schemaVersion: 2,
53 snapshotSequence: 0,
54 projects: [],
55 threads: [thread],
56 archivedThreads: [],
57 };
58 let online = true;
59 let scopes: ReadonlyArray<AuthEnvironmentScope> = AuthAdministrativeScopes;
60 let rejection = false;
61 let beforeDispatch: Effect.Effect<void> = Effect.void;
62 let afterDispatch: Effect.Effect<void> = Effect.void;
63 const commands: Array<HubOrganizationCommand> = [];
64 const client = HttpClient.make((request) =>
65 Effect.gen(function* () {
66 if (!online) return HttpClientResponse.fromWeb(request, new Response(null, { status: 503 }));
67 assert.equal(request.headers.authorization, "Bearer delegated-token");
68 if (request.method === "GET")
69 return HttpClientResponse.fromWeb(request, Response.json(yield* encodeSnapshot(snapshot)));
70 if (rejection)
71 return HttpClientResponse.fromWeb(request, new Response(null, { status: 409 }));
72 assert.equal(request.url, "http://source.private/api/hub/organization");
73 if (request.body._tag !== "Uint8Array") return yield* Effect.die("Missing command body");
74 const command = yield* decodeCommand(new TextDecoder().decode(request.body.body));
75 commands.push(command);
76 yield* beforeDispatch;
77 const now = yield* DateTime.now;
78 snapshot = {
79 ...snapshot,
80 snapshotSequence: snapshot.snapshotSequence + 1,
81 threads: snapshot.threads.map((value) => {
82 switch (command.type) {
83 case "thread.settle":
84 return {
85 ...value,
86 settledOverride: "settled" as const,
87 settledAt: command.settledAt ?? now,
88 pinnedAt: null,
89 pinOrderKey: null,
90 activeOrderKey: null,
91 };
92 case "thread.unsettle":
93 return {
94 ...value,
95 settledOverride: "active" as const,
96 settledAt: null,
97 unsettledAt: now,
98 };
99 case "thread.pin":
100 return {
101 ...value,
102 pinnedAt: now,
103 pinOrderKey: command.orderKey ?? null,
104 snoozedUntil: null,
105 snoozedAt: null,
106 };
107 case "thread.unpin":
108 return { ...value, pinnedAt: null, pinOrderKey: null };
109 case "thread.snooze":
110 return {
111 ...value,
112 snoozedUntil: DateTime.makeUnsafe(command.snoozedUntil),
113 snoozedAt: now,
114 };
115 case "thread.unsnooze":
116 return { ...value, snoozedUntil: null, snoozedAt: null };
117 case "thread.pin.reorder":
118 return { ...value, pinOrderKey: command.orderKey };
119 case "thread.active.reorder":
120 return { ...value, activeOrderKey: command.orderKey };
121 case "thread.auto-settle.set":
122 return { ...value, autoSettleDisabledAt: command.enabled ? null : now };
123 }
124 }),
125 };
126 const sequence = snapshot.snapshotSequence;
127 yield* afterDispatch;
128 return HttpClientResponse.fromWeb(request, Response.json({ sequence }));
129 }).pipe(Effect.orDie),
130 );
131 const descriptor: ExecutionEnvironmentDescriptor = {
132 environmentId,
133 label: "Laptop",
134 platform: { os: "linux", arch: "x64" },
135 serverVersion: "test",
136 orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION,
137 capabilities: { repositoryIdentity: true },
138 };
139 const hub = Hub.Hub.of({
140 directory: (current, options) => {
141 if (current.subject !== "hub-inbox") assert.equal(options?.refresh, false);
142 return Effect.succeed({
143 hubEnvironmentId: EnvironmentId.make("hub"),
144 environments: [
145 {
146 environmentId,
147 label: "Laptop",
148 httpBaseUrl: "http://hub/environments/laptop/",
149 descriptor,
150 scopes: scopes.filter((scope) => current.scopes.includes(scope)),
151 },
152 ],
153 });
154 },
155 authorize: (id, current) =>
156 id === environmentId
157 ? Effect.succeed(scopes.filter((scope) => current.scopes.includes(scope)))
158 : Effect.fail(new HubError({ reason: "not-found" })),
159 connect: () =>
160 Effect.succeed({
161 url: "http://source.private/",
162 token: "delegated-token",
163 invalidated: Effect.never,
164 }),
165 descriptor: () => Effect.succeed(descriptor),
166 manage: () => Effect.die("Unused"),
167 });
168 const layer = HubInbox.layer.pipe(
169 Layer.provide(
170 ArchiveCatalog.layer.pipe(
171 Layer.provide(ArchiveObjects.layer),
172 Layer.provide(ConfigProvider.layer(ConfigProvider.fromUnknown({}))),
173 Layer.provide(NodeServices.layer),
174 ),
175 ),
176 Layer.provide(Layer.succeed(Hub.Hub, hub)),
177 Layer.provide(Layer.succeed(HttpClient.HttpClient, client)),
178 );
179 const restart = Layer.build(layer).pipe(
180 Effect.map((context) => Context.get(context, HubInbox.HubInbox)),
181 );
182 const inbox = yield* restart;
183 yield* inbox.refresh;
184 yield* inbox.refresh;
185 let sequence = 0;
186 type CommandInput = {
187 [T in HubOrganizationCommand["type"]]: Omit<
188 Extract<HubOrganizationCommand, { type: T }>,
189 "commandId" | "threadId"
190 >;
191 }[HubOrganizationCommand["type"]];
192 const organize = (command: CommandInput) =>
193 inbox.organize(
194 {
195 environmentId,
196 command: { commandId: CommandId.make(`command-${++sequence}`), threadId, ...command },
197 },
198 session,
199 );
200 const read = inbox.read(session).pipe(Effect.map((value) => value.environments[0]!));
201 return {
202 inbox,
203 restart,
204 organize,
205 read,
206 commands,
207 online: (value: boolean) => {
208 online = value;
209 },
210 reject: (value: boolean) => {
211 rejection = value;
212 },
213 scopes: (value: ReadonlyArray<AuthEnvironmentScope>) => {
214 scopes = value;
215 },
216 source: (update: Partial<OrchestrationV2ThreadShell>) => {
217 snapshot = {
218 ...snapshot,
219 snapshotSequence: snapshot.snapshotSequence + 1,
220 threads: [{ ...snapshot.threads[0]!, ...update }],
221 };
222 },
223 beforeDispatch: (effect: Effect.Effect<void>) => {
224 beforeDispatch = effect;
225 },
226 afterDispatch: (effect: Effect.Effect<void>) => {
227 afterDispatch = effect;
228 },
229 removeThread: () => {
230 snapshot = { ...snapshot, threads: [] };
231 },
232 restoreThread: () => {
233 snapshot = { ...snapshot, threads: [thread] };
234 },
235 };
236});
237
238it.effect("discovers a newly created source chat when organized before the next catalog poll", () =>
239 Effect.gen(function* () {
240 const f = yield* fixture;
241 f.removeThread();
242 yield* f.inbox.refresh;
243 assert.deepEqual((yield* f.read).snapshot.threads, []);
244 f.restoreThread();
245 const entry = yield* f.organize({ type: "thread.settle" });
246 assert.equal(entry.snapshot.threads[0]?.settledOverride, "settled");
247 assert.deepEqual(entry.pendingThreadIds, [threadId]);
248 }).pipe(Effect.provide(SqlitePersistence.layerMemory), Effect.scoped),
249);
250
251it.effect(
252 "supports the organization commands and reversals without changing execution metadata",
253 () =>
254 Effect.gen(function* () {
255 const f = yield* fixture;
256 const now = yield* DateTime.now;
257 f.online(false);
258 yield* f.organize({
259 type: "thread.snooze",
260 snoozedUntil: DateTime.formatIso(DateTime.add(now, { hours: 1 })),
261 });
262 const settled = yield* f.organize({ type: "thread.settle" });
263 assert.isNull(settled.snapshot.threads[0]?.snoozedUntil);
264 assert.isNull(settled.snapshot.threads[0]?.snoozedAt);
265 yield* f.organize({ type: "thread.unsettle", reason: "user" });
266 yield* f.organize({
267 type: "thread.snooze",
268 snoozedUntil: DateTime.formatIso(DateTime.add(now, { hours: 1 })),
269 });
270 yield* f.organize({ type: "thread.unsnooze", reason: "user" });
271 yield* f.organize({ type: "thread.pin", orderKey: "a0" });
272 const reordered = yield* f.organize({ type: "thread.pin.reorder", orderKey: "a1" });
273 assert.equal(reordered.snapshot.threads[0]?.pinOrderKey, "a1");
274 yield* f.organize({ type: "thread.unpin" });
275 yield* f.organize({ type: "thread.active.reorder", orderKey: "a2" });
276 yield* f.organize({ type: "thread.auto-settle.set", enabled: false });
277 const enabled = yield* f.organize({ type: "thread.auto-settle.set", enabled: true });
278 assert.isNull(enabled.snapshot.threads[0]?.autoSettleDisabledAt);
279 assert.isNull(enabled.snapshot.threads[0]?.pinnedAt);
280 assert.equal(enabled.snapshot.threads[0]?.activeOrderKey, "a2");
281 assert.equal(enabled.snapshot.threads[0]?.status, "idle");
282 f.online(true);
283 yield* f.inbox.refresh;
284 assert.deepEqual((yield* f.read).pendingThreadIds, []);
285 assert.equal((yield* f.read).snapshot.threads[0]?.activeOrderKey, "a2");
286 }).pipe(Effect.provide(SqlitePersistence.layerMemory), Effect.scoped),
287);
288
289it.effect(
290 "removes pending organization when a fresh source shell no longer contains the chat",
291 () =>
292 Effect.gen(function* () {
293 const f = yield* fixture;
294 f.online(false);
295 yield* f.organize({ type: "thread.settle" });
296 f.removeThread();
297 f.online(true);
298 yield* f.inbox.refresh;
299 const entry = yield* f.read;
300 assert.deepEqual(entry.pendingThreadIds, []);
301 assert.deepEqual(entry.snapshot.threads, []);
302 assert.deepEqual(f.commands, []);
303 }).pipe(Effect.provide(SqlitePersistence.layerMemory), Effect.scoped),
304);
305
306it.effect("accepts a newer direct source change after the central command committed", () =>
307 Effect.gen(function* () {
308 const f = yield* fixture;
309 yield* f.organize({ type: "thread.settle" });
310 f.afterDispatch(Effect.sync(() => f.source({ settledOverride: "active", settledAt: null })));
311 yield* f.inbox.refresh;
312 const entry = yield* f.read;
313 assert.deepEqual(entry.pendingThreadIds, []);
314 assert.equal(entry.snapshot.threads[0]?.settledOverride, "active");
315 assert.equal(f.commands.filter((value) => value.type === "thread.settle").length, 1);
316 }).pipe(Effect.provide(SqlitePersistence.layerMemory), Effect.scoped),
317);
318
319it.effect("recognizes new source-authored work without comparing clocks across machines", () =>
320 Effect.gen(function* () {
321 const f = yield* fixture;
322 f.source({ latestUserAuthoredMessageAt: DateTime.makeUnsafe("2020-01-01T00:00:00Z") });
323 yield* f.inbox.refresh;
324 f.online(false);
325 yield* f.organize({ type: "thread.settle" });
326 f.source({
327 latestUserAuthoredMessageAt: DateTime.makeUnsafe("2020-01-01T00:00:01Z"),
328 status: "running",
329 });
330 f.online(true);
331 yield* f.inbox.refresh;
332 assert.equal((yield* f.read).snapshot.threads[0]?.settledOverride, "active");
333 assert.isFalse(f.commands.some((value) => value.type === "thread.settle"));
334 }).pipe(Effect.provide(SqlitePersistence.layerMemory), Effect.scoped),
335);
336
337it.effect(
338 "persists offline organization, shares it with another client, and restores it after restart",
339 () =>
340 Effect.gen(function* () {
341 const f = yield* fixture;
342 f.online(false);
343 yield* f.inbox.refresh;
344 const settled = yield* f.organize({ type: "thread.settle" });
345 assert.equal(settled.snapshot.threads[0]?.settledOverride, "settled");
346 assert.deepEqual(settled.pendingThreadIds, [threadId]);
347 const second = { ...session, sessionId: AuthSessionId.make("second-phone") };
348 assert.equal(
349 (yield* f.inbox.read(second)).environments[0]?.snapshot.threads[0]?.settledOverride,
350 "settled",
351 );
352 const restarted = yield* f.restart;
353 assert.equal(
354 (yield* restarted.read(session)).environments[0]?.snapshot.threads[0]?.settledOverride,
355 "settled",
356 );
357 assert.deepEqual(f.commands, []);
358 }).pipe(Effect.provide(SqlitePersistence.layerMemory), Effect.scoped),
359);
360
361it.effect(
362 "wakes a settle when a new source message arrives during dispatch and reconciles it active",
363 () =>
364 Effect.gen(function* () {
365 const f = yield* fixture;
366 yield* f.organize({ type: "thread.settle" });
367 const now = yield* DateTime.now;
368 f.beforeDispatch(
369 Effect.sync(() => f.source({ latestUserMessageAt: DateTime.add(now, { seconds: 1 }) })),
370 );
371 yield* f.inbox.refresh;
372 const raced = yield* f.read;
373 assert.equal(raced.snapshot.threads[0]?.settledOverride, "active");
374 assert.deepEqual(raced.pendingThreadIds, [threadId]);
375 assert.equal(f.commands.filter((value) => value.type === "thread.settle").length, 1);
376 f.beforeDispatch(Effect.void);
377 yield* f.inbox.refresh;
378 const confirmed = yield* f.read;
379 assert.equal(confirmed.snapshot.threads[0]?.settledOverride, "active");
380 assert.deepEqual(confirmed.pendingThreadIds, []);
381 assert.equal(f.commands.filter((value) => value.type === "thread.unsettle").length, 1);
382 }).pipe(Effect.provide(SqlitePersistence.layerMemory), Effect.scoped),
383);
384
385it.effect("reconciles current intent after reconnect without replaying an undone settle", () =>
386 Effect.gen(function* () {
387 const f = yield* fixture;
388 f.online(false);
389 yield* f.organize({ type: "thread.settle" });
390 yield* f.organize({ type: "thread.unsettle", reason: "user" });
391 yield* f.organize({ type: "thread.pin", orderKey: "a0" });
392 f.online(true);
393 yield* f.inbox.refresh;
394 assert.isFalse(f.commands.some((value) => value.type === "thread.settle"));
395 const result = yield* f.read;
396 assert.deepEqual(result.pendingThreadIds, []);
397 assert.isNotNull(result.snapshot.threads[0]?.pinnedAt);
398 assert.equal(result.snapshot.threads[0]?.settledOverride, "active");
399 f.source({ pinnedAt: null, pinOrderKey: null });
400 yield* f.inbox.refresh;
401 assert.isNull((yield* f.read).snapshot.threads[0]?.pinnedAt);
402 }).pipe(Effect.provide(SqlitePersistence.layerMemory), Effect.scoped),
403);
404
405it.effect("keeps execution state and pending intent when source settlement is blocked", () =>
406 Effect.gen(function* () {
407 const f = yield* fixture;
408 f.source({ status: "running" });
409 yield* f.inbox.refresh;
410 f.reject(true);
411 yield* f.organize({ type: "thread.settle" });
412 yield* f.inbox.refresh;
413 const blocked = yield* f.read;
414 assert.equal(blocked.snapshot.threads[0]?.status, "running");
415 assert.equal(blocked.snapshot.threads[0]?.settledOverride, "settled");
416 assert.deepEqual(blocked.pendingThreadIds, [threadId]);
417 f.source({ status: "completed" });
418 f.reject(false);
419 yield* f.inbox.refresh;
420 assert.deepEqual((yield* f.read).pendingThreadIds, []);
421 }).pipe(Effect.provide(SqlitePersistence.layerMemory), Effect.scoped),
422);
423
424it.effect("expires snooze while the source is offline and never resends an elapsed snooze", () =>
425 Effect.gen(function* () {
426 const f = yield* fixture;
427 const now = yield* DateTime.now;
428 f.online(false);
429 yield* f.organize({
430 type: "thread.snooze",
431 snoozedUntil: DateTime.formatIso(DateTime.add(now, { minutes: 1 })),
432 });
433 yield* TestClock.adjust("2 minutes");
434 assert.isNull((yield* f.read).snapshot.threads[0]?.snoozedUntil);
435 f.online(true);
436 yield* f.inbox.refresh;
437 assert.isFalse(f.commands.some((value) => value.type === "thread.snooze"));
438 assert.deepEqual((yield* f.read).pendingThreadIds, []);
439 }).pipe(Effect.provide(SqlitePersistence.layerMemory), Effect.scoped),
440);
441
442it.effect("checks current hub and enrollment grants before accepting offline changes", () =>
443 Effect.gen(function* () {
444 const f = yield* fixture;
445 f.online(false);
446 f.scopes([AuthOrchestrationReadScope]);
447 assert.equal((yield* f.organize({ type: "thread.settle" }).pipe(Effect.flip))._tag, "HubError");
448 f.scopes(AuthAdministrativeScopes);
449 assert.equal(
450 (yield* f.inbox
451 .organize(
452 {
453 environmentId,
454 command: { type: "thread.settle", commandId: CommandId.make("denied"), threadId },
455 },
456 { ...session, scopes: [AuthOrchestrationReadScope] },
457 )
458 .pipe(Effect.flip))._tag,
459 "HubError",
460 );
461 assert.deepEqual(
462 (yield* f.inbox.read({ ...session, scopes: [AuthOrchestrationOperateScope] })).environments,
463 [],
464 );
465 assert.isNull((yield* f.read).snapshot.threads[0]?.settledOverride);
466 }).pipe(Effect.provide(SqlitePersistence.layerMemory), Effect.scoped),
467);
468
469it.effect("does not acknowledge a newer command while an older source request is in flight", () =>
470 Effect.gen(function* () {
471 const f = yield* fixture;
472 yield* f.organize({ type: "thread.settle" });
473 const started = yield* Deferred.make<void>();
474 const release = yield* Deferred.make<void>();
475 f.beforeDispatch(
476 Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(release))),
477 );
478 const refresh = yield* f.inbox.refresh.pipe(Effect.forkScoped);
479 yield* Deferred.await(started);
480 yield* f.organize({ type: "thread.unsettle", reason: "user" });
481 const releaseReversal = yield* Deferred.make<void>();
482 f.beforeDispatch(Deferred.await(releaseReversal));
483 yield* Deferred.succeed(release, undefined);
484 const pending = yield* f.read;
485 assert.equal(pending.snapshot.threads[0]?.settledOverride, "active");
486 assert.deepEqual(pending.pendingThreadIds, [threadId]);
487 yield* Deferred.succeed(releaseReversal, undefined);
488 yield* Fiber.join(refresh);
489 yield* f.inbox.refresh;
490 assert.deepEqual((yield* f.read).pendingThreadIds, []);
491 assert.equal((yield* f.read).snapshot.threads[0]?.settledOverride, "active");
492 }).pipe(Effect.provide(SqlitePersistence.layerMemory), Effect.scoped),
493);
494
495it.effect(
496 "deduplicates retried commands after later organization changes and acknowledgement",
497 () =>
498 Effect.gen(function* () {
499 const f = yield* fixture;
500 const command = {
501 type: "thread.settle" as const,
502 threadId,
503 commandId: CommandId.make("retry-settle"),
504 };
505 yield* f.inbox.organize({ environmentId, command }, session);
506 yield* f.inbox.refresh;
507 yield* f.organize({ type: "thread.unsettle", reason: "user" });
508 yield* f.inbox.refresh;
509 const retry = yield* f.inbox.organize({ environmentId, command }, session);
510 assert.equal(retry.snapshot.threads[0]?.settledOverride, "active");
511 assert.deepEqual(retry.pendingThreadIds, []);
512 assert.equal(f.commands.filter((value) => value.type === "thread.settle").length, 1);
513 }).pipe(Effect.provide(SqlitePersistence.layerMemory), Effect.scoped),
514);
515
516it.effect(
517 "wakes a pending snooze on fresh source completion and preserves its execution status",
518 () =>
519 Effect.gen(function* () {
520 const f = yield* fixture;
521 const now = yield* DateTime.now;
522 f.online(false);
523 yield* f.organize({
524 type: "thread.snooze",
525 snoozedUntil: DateTime.formatIso(DateTime.add(now, { hours: 1 })),
526 });
527 f.source({ status: "completed", latestRunCompletedAt: DateTime.add(now, { minutes: 1 }) });
528 f.online(true);
529 yield* f.inbox.refresh;
530 const entry = yield* f.read;
531 assert.isNull(entry.snapshot.threads[0]?.snoozedUntil);
532 assert.equal(entry.snapshot.threads[0]?.status, "completed");
533 assert.deepEqual(entry.pendingThreadIds, []);
534 assert.isFalse(f.commands.some((value) => value.type === "thread.snooze"));
535 }).pipe(Effect.provide(SqlitePersistence.layerMemory), Effect.scoped),
536);