1import { EnvironmentId, ThreadId } from "@t3tools/contracts";
2import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
3import { expect, it, vi } from "vite-plus/test";
4
5import type { VisitThreadInput } from "../operations/commands.ts";
6import { claimThreadOpen } from "./threadSearch.ts";
7import { createThreadVisitTracker, VISIT_DISPATCH_THROTTLE_MS } from "./threadVisits.ts";
8
9const environmentId = EnvironmentId.make("env-visits");
10
11interface Shell {
12 readonly environmentId: EnvironmentId;
13 readonly id: ThreadId;
14 readonly updatedAt: string;
15 readonly lastVisitedAt?: string | null;
16 readonly latestRun: { readonly completedAt: string | null } | null;
17 readonly wokeAt?: string | null;
18}
19
20function at(seconds: number) {
21 return `2026-01-01T00:00:${String(seconds).padStart(2, "0")}.000Z`;
22}
23
24// claimThreadOpen dedupes per thread in module state, so each test opens its own thread.
25function setup(threadName: string, initial: Omit<Shell, "environmentId" | "id"> | null) {
26 vi.useFakeTimers({ now: Date.UTC(2026, 0, 1) });
27 const threadId = ThreadId.make(threadName);
28 const ref = { environmentId, threadId };
29 const shellAtom = Atom.make<Shell | null>(
30 initial === null ? null : { environmentId, id: threadId, ...initial },
31 );
32 const visits: Array<VisitThreadInput> = [];
33 const localVisits: Array<string> = [];
34 const tracker = createThreadVisitTracker({
35 threadShellAtom: () => shellAtom,
36 visit: {
37 label: "test-visit",
38 run: async (_registry, value) => {
39 visits.push(value.input);
40 return AsyncResult.success(undefined);
41 },
42 },
43 markLocalVisit: (_ref, visitedAt) => localVisits.push(visitedAt),
44 });
45 const registry = AtomRegistry.make();
46 return {
47 ref,
48 visits,
49 localVisits,
50 mount: () => registry.mount(tracker(ref)),
51 update: (patch: Partial<Shell>) => {
52 const current = registry.get(shellAtom);
53 registry.set(shellAtom, {
54 environmentId,
55 id: threadId,
56 updatedAt: at(0),
57 latestRun: null,
58 ...current,
59 ...patch,
60 });
61 },
62 [Symbol.dispose]: () => vi.useRealTimers(),
63 };
64}
65
66it("records the open once, then throttles streaming visits", () => {
67 using t = setup("thread-throttle", {
68 updatedAt: at(10),
69 lastVisitedAt: at(0),
70 latestRun: null,
71 });
72 t.mount();
73 expect(t.visits).toEqual([{ threadId: t.ref.threadId, visitedAt: at(10), opened: true }]);
74
75 vi.advanceTimersByTime(1_000);
76 t.update({ updatedAt: at(11) });
77 t.update({ updatedAt: at(12) });
78 expect(t.visits).toHaveLength(1);
79
80 vi.advanceTimersByTime(VISIT_DISPATCH_THROTTLE_MS - 1_000);
81 expect(t.visits.slice(1)).toEqual([{ threadId: t.ref.threadId, visitedAt: at(12) }]);
82});
83
84it("sends an unseen completion without waiting for the throttle", () => {
85 using t = setup("thread-completion", {
86 updatedAt: at(10),
87 lastVisitedAt: at(0),
88 latestRun: null,
89 });
90 t.mount();
91 t.update({ updatedAt: at(20), latestRun: { completedAt: at(20) } });
92 expect(t.visits.slice(1)).toEqual([{ threadId: t.ref.threadId, visitedAt: at(20) }]);
93});
94
95it("does not resend the open a palette pick already sent", () => {
96 using t = setup("thread-palette", { updatedAt: at(10), lastVisitedAt: at(0), latestRun: null });
97 claimThreadOpen(environmentId, t.ref.threadId, Date.UTC(2026, 0, 1));
98 t.mount();
99 expect(t.visits).toEqual([{ threadId: t.ref.threadId, visitedAt: at(10) }]);
100});
101
102it("keeps a mark-unread on the open thread until new activity lands", () => {
103 using t = setup("thread-unread", { updatedAt: at(10), lastVisitedAt: at(0), latestRun: null });
104 t.mount();
105 t.update({ lastVisitedAt: at(10) });
106 t.update({ lastVisitedAt: at(5) });
107 vi.advanceTimersByTime(VISIT_DISPATCH_THROTTLE_MS);
108 expect(t.visits).toHaveLength(1);
109
110 t.update({ updatedAt: at(30) });
111 expect(t.visits.slice(1)).toEqual([{ threadId: t.ref.threadId, visitedAt: at(30) }]);
112});
113
114it("keeps a mark-unread on a read thread reopened within the open dedupe window", () => {
115 using t = setup("thread-unread-reopened", {
116 updatedAt: at(10),
117 lastVisitedAt: at(10),
118 latestRun: { completedAt: at(9) },
119 });
120 claimThreadOpen(environmentId, t.ref.threadId, Date.UTC(2026, 0, 1));
121 t.mount();
122 t.update({ lastVisitedAt: at(8) });
123 vi.advanceTimersByTime(VISIT_DISPATCH_THROTTLE_MS);
124 expect(t.visits).toEqual([]);
125
126 // An authoritative snapshot can briefly replace the row with an older one.
127 t.update({ updatedAt: at(5), lastVisitedAt: at(5), latestRun: { completedAt: at(4) } });
128 t.update({ updatedAt: at(10), lastVisitedAt: at(8), latestRun: { completedAt: at(9) } });
129 vi.advanceTimersByTime(VISIT_DISPATCH_THROTTLE_MS);
130 expect(t.visits).toEqual([]);
131});
132
133it("keeps mark unread sticky after an expired wake is retired", () => {
134 using t = setup("thread-unread-retired-wake", {
135 updatedAt: at(10),
136 lastVisitedAt: at(0),
137 latestRun: { completedAt: at(9) },
138 wokeAt: at(15),
139 });
140 t.mount();
141 expect(t.visits).toEqual([{ threadId: t.ref.threadId, visitedAt: at(15), opened: true }]);
142 t.update({ lastVisitedAt: at(15) });
143 t.update({ lastVisitedAt: at(8), wokeAt: null });
144 vi.advanceTimersByTime(VISIT_DISPATCH_THROTTLE_MS);
145 expect(t.visits).toHaveLength(1);
146
147 t.update({ updatedAt: at(30) });
148 expect(t.visits.slice(1)).toEqual([{ threadId: t.ref.threadId, visitedAt: at(30) }]);
149});
150
151it("waits for the shell and records local visits for servers without tracking", () => {
152 using t = setup("thread-local", null);
153 t.mount();
154 expect(t.visits).toEqual([]);
155 t.update({ updatedAt: at(10), latestRun: null });
156 t.update({ updatedAt: at(11) });
157 expect(t.visits).toEqual([]);
158 expect(t.localVisits).toEqual([at(10), at(11)]);
159});
160
161it("drops a pending visit when the thread is no longer viewed", () => {
162 using t = setup("thread-unmount", { updatedAt: at(10), lastVisitedAt: at(0), latestRun: null });
163 const unmount = t.mount();
164 t.update({ updatedAt: at(11) });
165 unmount();
166 vi.advanceTimersByTime(VISIT_DISPATCH_THROTTLE_MS);
167 expect(t.visits).toHaveLength(1);
168});
169
170it("sends the open with the first visit when the shell arrives after the thread is shown", () => {
171 using t = setup("thread-shell-missing", null);
172 t.mount();
173 expect(claimThreadOpen(environmentId, t.ref.threadId, Date.UTC(2026, 0, 1))).toBe(false);
174 expect(t.visits).toEqual([]);
175
176 t.update({ updatedAt: at(10), lastVisitedAt: at(0), latestRun: null });
177 expect(t.visits).toEqual([{ threadId: t.ref.threadId, visitedAt: at(10), opened: true }]);
178
179 t.update({ updatedAt: at(10), lastVisitedAt: at(10) });
180 vi.advanceTimersByTime(VISIT_DISPATCH_THROTTLE_MS);
181 expect(t.visits).toHaveLength(1);
182});
183
184it("advances the watermark to a wake without waiting for the throttle", () => {
185 using t = setup("thread-wake", { updatedAt: at(10), lastVisitedAt: at(0), latestRun: null });
186 t.mount();
187 t.update({ wokeAt: at(15) });
188 expect(t.visits.slice(1)).toEqual([{ threadId: t.ref.threadId, visitedAt: at(15) }]);
189
190 vi.advanceTimersByTime(1_000);
191 t.update({ updatedAt: at(16) });
192 expect(t.visits).toHaveLength(2);
193});