1import { PreviewAutomationControlReason } from "@t3tools/contracts";
2import * as Effect from "effect/Effect";
3import * as Schema from "effect/Schema";
4
5export class BrowserControlInterrupted extends Schema.TaggedError<BrowserControlInterrupted>()(
6 "BrowserControlInterrupted",
7 {
8 message: Schema.String.pipe(
9 Schema.withConstructorDefault(
10 Effect.succeed("Browser control changed. Refresh the snapshot before trying again."),
11 ),
12 ),
13 reason: PreviewAutomationControlReason.pipe(
14 Schema.withConstructorDefault(Effect.succeed("interrupted" as const)),
15 ),
16 },
17) {}
18
19const settle = (work: Promise<unknown>) =>
20 work.then(
21 () => undefined,
22 () => undefined,
23 );
24
25/** Reserves control immediately, but drains running work before the new owner can act. */
26export class SessionControl {
27 readonly agentId: string | null;
28 private readonly onGenerationChange: () => void;
29 private tail: Promise<unknown> = Promise.resolve();
30 private latestHuman: {
31 entry: { viewerId: string; key: string; input: unknown; run: (input: never) => Promise<void> };
32 result: Promise<void>;
33 } | null = null;
34 private pending: Promise<void> = Promise.resolve();
35 private readonly handoff = new Set<Promise<void>>();
36 private owner: string | null = null;
37 private epoch = 0;
38 private closed = false;
39
40 constructor(agentId: string | null, onGenerationChange: () => void = () => {}) {
41 this.agentId = agentId;
42 this.onGenerationChange = onGenerationChange;
43 }
44
45 get controller() {
46 return this.owner;
47 }
48
49 get generation() {
50 return this.epoch;
51 }
52
53 private enqueue<A>(run: () => Promise<A>): Promise<A> {
54 this.latestHuman = null;
55 const result = this.tail.then(run);
56 this.tail = result.then(
57 () => this.pending,
58 () => this.pending,
59 );
60 return result;
61 }
62
63 /** An action can respond before its navigation commits, while later actions still wait. */
64 track(work: Promise<unknown>) {
65 this.pending = Promise.all([this.pending, settle(work)]).then(() => undefined);
66 }
67
68 /** Viewer input stays responsive, but control handoffs still drain its navigations. */
69 trackUntilHandoff(work: Promise<unknown>) {
70 const settled = settle(work).finally(() => {
71 this.handoff.delete(settled);
72 });
73 this.handoff.add(settled);
74 }
75
76 private assertOpen() {
77 if (this.closed)
78 throw new BrowserControlInterrupted({
79 message: "This browser tab is closed.",
80 reason: "closed",
81 });
82 }
83
84 private changeOwner(owner: string | null) {
85 this.latestHuman = null;
86 this.owner = owner;
87 this.epoch++;
88 this.onGenerationChange();
89 }
90
91 private async action<A>(allowed: () => boolean, run: () => Promise<A>, deadlineMs?: number) {
92 this.assertOpen();
93 if (!allowed())
94 throw new BrowserControlInterrupted({ message: "You do not control this browser tab." });
95 const epoch = this.epoch;
96 return this.enqueue(async () => {
97 this.assertOpen();
98 if (epoch !== this.epoch || !allowed()) throw new BrowserControlInterrupted();
99 if (deadlineMs !== undefined && performance.now() >= deadlineMs) {
100 throw new DOMException(
101 "This browser request expired before it could start.",
102 "TimeoutError",
103 );
104 }
105 return run();
106 });
107 }
108
109 agentMayAct(agentId: string) {
110 return this.agentId === null || this.agentId === agentId;
111 }
112
113 agent<A>(agentId: string, run: () => Promise<A>, deadlineMs?: number) {
114 if (!this.agentMayAct(agentId))
115 return Promise.reject(
116 new BrowserControlInterrupted({
117 message: "This tab belongs to another agent.",
118 reason: "agentMismatch",
119 }),
120 );
121 if (this.owner !== null)
122 return Promise.reject(
123 new BrowserControlInterrupted({
124 message: "A human controls this tab.",
125 reason: "humanControl",
126 }),
127 );
128 return this.action(() => this.agentMayAct(agentId) && this.owner === null, run, deadlineMs);
129 }
130
131 agentCanActNow(agentId: string) {
132 return this.agentMayAct(agentId) && this.owner === null;
133 }
134
135 /** Read without control or queueing so agents can observe without delaying human input. */
136 observe<A>(run: () => Promise<A>) {
137 this.assertOpen();
138 return run();
139 }
140
141 system<A>(run: () => Promise<A>) {
142 this.assertOpen();
143 return this.enqueue(async () => {
144 this.assertOpen();
145 return run();
146 });
147 }
148
149 human<A>(viewerId: string, run: () => Promise<A>) {
150 return this.action(() => this.owner === viewerId, run);
151 }
152
153 /**
154 * Merge into only adjacent waiting work with the same key; the running action and
155 * intervening actions keep their order. Work sharing a key must share its input type.
156 */
157 humanLatest<A>(
158 viewerId: string,
159 key: string,
160 input: A,
161 run: (input: A) => Promise<void>,
162 merge: (waiting: A, next: A) => A = (_waiting, next) => next,
163 ): Promise<void> {
164 if (this.closed || this.owner !== viewerId) return this.human(viewerId, () => run(input));
165 const latest = this.latestHuman;
166 if (latest?.entry.viewerId === viewerId && latest.entry.key === key) {
167 latest.entry.input = merge(latest.entry.input as A, input);
168 latest.entry.run = run;
169 return latest.result;
170 }
171 const entry = { viewerId, key, input, run };
172 const result = this.human(viewerId, async () => {
173 if (this.latestHuman?.entry === entry) this.latestHuman = null;
174 await entry.run(entry.input);
175 });
176 this.latestHuman = { entry, result };
177 return result;
178 }
179
180 async take(viewerId: string) {
181 this.assertOpen();
182 if (this.owner !== null && this.owner !== viewerId) {
183 throw new BrowserControlInterrupted({ message: "Another viewer controls this browser tab." });
184 }
185 if (this.owner !== viewerId) this.changeOwner(viewerId);
186 return this.human(viewerId, async () => {});
187 }
188
189 async release(viewerId: string, afterDrain: () => Promise<void> = async () => {}) {
190 this.assertOpen();
191 if (this.owner !== viewerId) {
192 throw new BrowserControlInterrupted({
193 message: "Only the controlling viewer can release this browser tab.",
194 });
195 }
196 this.changeOwner(null);
197 await this.enqueue(async () => {
198 await Promise.all(this.handoff);
199 await afterDrain();
200 });
201 }
202
203 async disconnect(viewerId: string, afterDrain?: () => Promise<void>) {
204 if (!this.closed && this.owner === viewerId) await this.release(viewerId, afterDrain);
205 }
206
207 async close() {
208 if (!this.closed) {
209 this.closed = true;
210 this.changeOwner(null);
211 }
212 await this.tail;
213 }
214}