1import { describe, expect, it } from "@effect/vitest";
2import {
3 AuthStandardClientScopes,
4 CommandId,
5 EnvironmentId,
6 HubInbox,
7 HubInboxCommand,
8 HubInboxEnvironment,
9 ThreadId,
10 ORCHESTRATION_PROTOCOL_VERSION,
11} from "@t3tools/contracts";
12import * as Effect from "effect/Effect";
13import * as DateTime from "effect/DateTime";
14import * as Layer from "effect/Layer";
15import * as Option from "effect/Option";
16import * as Schema from "effect/Schema";
17import * as Stream from "effect/Stream";
18import * as SubscriptionRef from "effect/SubscriptionRef";
19
20import * as Hub from "./hub.ts";
21import * as Registry from "./registry.ts";
22import * as Driver from "./driver.ts";
23import * as Connectivity from "./connectivity.ts";
24import * as Wakeups from "./wakeups.ts";
25import * as Profiles from "./profileStore.ts";
26import * as Credentials from "./credentialStore.ts";
27import * as Capabilities from "../platform/capabilities.ts";
28import * as Persistence from "../platform/persistence.ts";
29import * as Http from "../rpc/http.ts";
30import {
31 BearerConnectionCredential,
32 HostedConnectionCredential,
33 BearerConnectionProfile,
34 BearerConnectionRegistration,
35 RelayConnectionRegistration,
36 type ConnectionCredential,
37 type ConnectionProfile,
38} from "./catalog.ts";
39import { BearerConnectionTarget, RelayConnectionTarget, ConnectionBlockedError } from "./model.ts";
40import { connectionRoutes, connectionRouteId } from "./routes.ts";
41import { v2Project, v2ThreadShell } from "../state/orchestrationV2TestFixtures.ts";
42
43const HUB_ID = EnvironmentId.make("hub");
44const BASE = "https://hub.example/knife/";
45const encodeInbox = Schema.encodeSync(Schema.toCodecJson(HubInbox));
46const encodeInboxEnvironment = Schema.encodeSync(Schema.toCodecJson(HubInboxEnvironment));
47const decodeInboxCommand = Schema.decodeUnknownSync(Schema.toCodecJson(HubInboxCommand));
48const descriptor = (id: string) => ({
49 environmentId: EnvironmentId.make(id),
50 label: id,
51 platform: { os: "linux" as const, arch: "x64" as const },
52 serverVersion: "test",
53 orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION,
54 capabilities: { repositoryIdentity: true },
55});
56
57function harness() {
58 const profiles = new Map<string, ConnectionProfile>();
59 const credentials = new Map<string, ConnectionCredential>();
60 const requests: Array<{
61 url: string;
62 authorization: string | null;
63 method: string;
64 etag: string | null;
65 }> = [];
66 const writes: Array<string> = [];
67 const commands: Array<HubInboxCommand> = [];
68 const cleanups: Array<{ id: EnvironmentId; retainDrafts?: boolean }> = [];
69 const control = {
70 machines: ["laptop", "desktop"],
71 offline: false,
72 badAddress: false,
73 protocolVersion: ORCHESTRATION_PROTOCOL_VERSION,
74 inboxStatus: 200,
75 notModified: false,
76 inboxEtag: '"inbox-v1"',
77 inboxSequence: 5,
78 token: "hub-secret",
79 hostedAuthorization: "Bearer hosted-ephemeral" as string | null,
80 failProfileWrite: false,
81 failCredentialWrite: false,
82 };
83 const dependencies = Layer.mergeAll(
84 Layer.succeed(Capabilities.HostedAuthAccess, {
85 headers: () =>
86 Effect.sync(() =>
87 control.hostedAuthorization === null
88 ? { "x-csrf-token": "cookie-proof" }
89 : { authorization: control.hostedAuthorization },
90 ),
91 }),
92 Http.layerRemoteHttpClient((async (input, init) => {
93 const url = String(input);
94 requests.push({
95 url,
96 authorization: new Headers(init?.headers).get("authorization"),
97 method: init?.method ?? "GET",
98 etag: new Headers(init?.headers).get("if-none-match"),
99 });
100 if (url.endsWith("/.well-known/t3/environment")) return Response.json(descriptor(HUB_ID));
101 if (url.endsWith("/oauth/token"))
102 return Response.json({
103 access_token: control.token,
104 issued_token_type: "urn:ietf:params:oauth:token-type:access_token",
105 token_type: "Bearer",
106 expires_in: 3600,
107 scope: AuthStandardClientScopes.join(" "),
108 });
109 if (url === `${BASE}api/hub/environments`) {
110 if (control.offline) throw new Error("offline");
111 return Response.json({
112 hubEnvironmentId: HUB_ID,
113 environments: control.machines.map((id) => ({
114 ...descriptor(id),
115 descriptor: {
116 ...descriptor(id),
117 orchestrationProtocolVersion: control.protocolVersion,
118 },
119 scopes: AuthStandardClientScopes,
120 httpBaseUrl: `${control.badAddress ? "https://elsewhere.example/" : BASE}environments/${id}/`,
121 })),
122 });
123 }
124 if (url === `${BASE}api/hub/inbox`) {
125 if (control.offline) throw new Error("offline");
126 if (control.inboxStatus !== 200) return new Response(null, { status: control.inboxStatus });
127 if (
128 control.notModified &&
129 init?.method !== "POST" &&
130 new Headers(init?.headers).get("if-none-match") === control.inboxEtag
131 ) {
132 return new Response(null, { status: 304 });
133 }
134 const entry: HubInboxEnvironment = {
135 environmentId: EnvironmentId.make("laptop"),
136 snapshot: {
137 schemaVersion: 2,
138 snapshotSequence: control.inboxSequence,
139 projects: [v2Project],
140 threads: [
141 {
142 ...v2ThreadShell,
143 id: ThreadId.make("chat"),
144 lineage: {
145 rootThreadId: ThreadId.make("chat"),
146 parentThreadId: null,
147 relationshipToParent: null,
148 },
149 snoozedAt: v2ThreadShell.updatedAt,
150 snoozedUntil: DateTime.makeUnsafe("2030-01-01T00:00:00.000Z"),
151 },
152 ],
153 archivedThreads: [],
154 },
155 pendingThreadIds: [],
156 };
157 if (init?.method === "POST") {
158 commands.push(decodeInboxCommand(JSON.parse(await new Response(init.body).text())));
159 }
160 const body =
161 init?.method === "POST"
162 ? encodeInboxEnvironment(entry)
163 : encodeInbox({ environments: [entry] });
164 return Response.json(body, {
165 headers: { etag: control.inboxEtag },
166 });
167 }
168 throw new Error(`Unexpected URL ${url}`);
169 }) satisfies typeof fetch),
170 Layer.succeed(Capabilities.ClientPresentation, {
171 metadata: { label: "Phone", deviceType: "mobile", surface: "mobile" },
172 }),
173 Layer.succeed(Profiles.ConnectionProfileStore, {
174 get: (id) => Effect.succeed(Option.fromUndefinedOr(profiles.get(id))),
175 put: (profile) =>
176 Effect.suspend(() =>
177 control.failProfileWrite
178 ? Effect.fail(
179 new ConnectionBlockedError({
180 reason: "configuration",
181 detail: "Profile storage unavailable",
182 }),
183 )
184 : Effect.sync(() => {
185 profiles.set(profile.connectionId, profile);
186 }),
187 ),
188 remove: (id) =>
189 Effect.sync(() => {
190 profiles.delete(id);
191 }),
192 }),
193 Layer.succeed(Credentials.ConnectionCredentialStore, {
194 get: (id) => Effect.succeed(Option.fromUndefinedOr(credentials.get(id))),
195 put: (id, credential) =>
196 Effect.suspend(() =>
197 control.failCredentialWrite
198 ? Effect.fail(
199 new ConnectionBlockedError({
200 reason: "configuration",
201 detail: "Credential storage unavailable",
202 }),
203 )
204 : Effect.sync(() => {
205 credentials.set(id, credential);
206 }),
207 ),
208 remove: (id) =>
209 Effect.sync(() => {
210 credentials.delete(id);
211 }),
212 }),
213 Layer.mock(Persistence.ConnectionTargetStore)({
214 list: Effect.succeed([]),
215 listDisabled: Effect.succeed([]),
216 }),
217 Layer.mock(Persistence.ConnectionRegistrationStore)({
218 register: (registration) =>
219 Effect.sync(() => {
220 writes.push(registration.target.environmentId);
221 if (registration._tag === "BearerConnectionRegistration") {
222 profiles.set(registration.profile.connectionId, registration.profile);
223 credentials.set(registration.target.connectionId, registration.credential);
224 }
225 }),
226 setRoutes: () => Effect.void,
227 remove: () => Effect.void,
228 setEnabled: () => Effect.void,
229 }),
230 Layer.mock(Persistence.EnvironmentCacheStore)({ clear: () => Effect.void }),
231 Layer.succeed(Persistence.EnvironmentOwnedDataCleanup, {
232 clear: (id, options) =>
233 Effect.sync(() => {
234 cleanups.push({ id, ...options });
235 }),
236 }),
237 Layer.mock(Capabilities.SshEnvironmentGateway)({}),
238 Layer.succeed(Connectivity.Connectivity, {
239 status: Effect.succeed("online" as const),
240 changes: Stream.never,
241 }),
242 Layer.succeed(Wakeups.ConnectionWakeups, { changes: Stream.never }),
243 Layer.mock(Driver.ConnectionDriver)({ connect: () => Effect.never }),
244 );
245 const registry = Registry.layer.pipe(Layer.provideMerge(dependencies));
246 return {
247 profiles,
248 credentials,
249 requests,
250 writes,
251 commands,
252 cleanups,
253 control,
254 layer: Hub.layer.pipe(Layer.provideMerge(registry)),
255 };
256}
257
258describe("self-hosted server connections", () => {
259 it.effect(
260 "stores only a server marker for SSO and renews native headers through the broker",
261 () => {
262 const test = harness();
263 return Effect.gen(function* () {
264 const hub = yield* Hub.HubConnection;
265 yield* hub.connectAuthenticated({ httpBaseUrl: BASE });
266 expect(test.credentials.get(`hub-session:${HUB_ID}`)).toEqual(
267 new HostedConnectionCredential({ serverUrl: BASE }),
268 );
269 expect(test.requests.some((request) => request.url.endsWith("/oauth/token"))).toBe(false);
270 expect(yield* hub.headers).toEqual({ authorization: "Bearer hosted-ephemeral" });
271 test.control.hostedAuthorization = "Bearer renewed-ephemeral";
272 yield* hub.refreshInbox;
273 expect(test.requests.at(-1)?.authorization).toBe("Bearer renewed-ephemeral");
274 test.control.hostedAuthorization = null;
275 expect(yield* hub.headers).toEqual({ "x-csrf-token": "cookie-proof" });
276 yield* hub.refreshInbox;
277 expect(test.requests.at(-1)?.authorization).toBe(null);
278 }).pipe(Effect.provide(test.layer));
279 },
280 );
281
282 it.effect("preserves paired access if SSO connection validation or persistence fails", () => {
283 const test = harness();
284 return Effect.gen(function* () {
285 const hub = yield* Hub.HubConnection;
286 yield* hub.connect({ host: BASE, pairingCode: "code" });
287 const previous = test.credentials.get(`hub-session:${HUB_ID}`);
288 test.control.offline = true;
289 yield* Effect.option(hub.connectAuthenticated({ httpBaseUrl: BASE }));
290 expect(test.credentials.get(`hub-session:${HUB_ID}`)).toEqual(previous);
291 test.control.offline = false;
292 test.control.failProfileWrite = true;
293 yield* Effect.option(hub.connectAuthenticated({ httpBaseUrl: BASE }));
294 expect(test.credentials.get(`hub-session:${HUB_ID}`)).toEqual(previous);
295 }).pipe(Effect.provide(test.layer));
296 });
297
298 it.effect("round-trips timestamped inbox snapshots and organization commands as JSON", () => {
299 const test = harness();
300 return Effect.gen(function* () {
301 const hub = yield* Hub.HubConnection;
302 yield* hub.connect({ host: BASE, pairingCode: "code" });
303 const thread = (yield* SubscriptionRef.get(hub.state)).inbox.environments[0]?.snapshot
304 .threads[0];
305 expect(DateTime.formatIso(thread!.createdAt)).toBe("2026-06-20T00:00:00.000Z");
306 expect(DateTime.formatIso(thread!.snoozedUntil!)).toBe("2030-01-01T00:00:00.000Z");
307 const settledAt = DateTime.makeUnsafe("2026-06-21T12:34:56.000Z");
308 yield* hub.organize({
309 environmentId: EnvironmentId.make("laptop"),
310 command: {
311 type: "thread.settle",
312 commandId: CommandId.make("settle-with-timestamp"),
313 threadId: ThreadId.make("chat"),
314 settledAt,
315 },
316 });
317 expect(test.commands).toEqual([
318 {
319 environmentId: EnvironmentId.make("laptop"),
320 command: {
321 type: "thread.settle",
322 commandId: CommandId.make("settle-with-timestamp"),
323 threadId: ThreadId.make("chat"),
324 settledAt,
325 },
326 },
327 ]);
328 const updated = (yield* SubscriptionRef.get(hub.state)).inbox.environments[0]?.snapshot
329 .threads[0];
330 expect(DateTime.formatIso(updated!.updatedAt)).toBe("2026-06-20T00:00:00.000Z");
331 expect(DateTime.formatIso(updated!.snoozedAt!)).toBe("2026-06-20T00:00:00.000Z");
332 }).pipe(Effect.provide(test.layer), Effect.scoped);
333 });
334 it.effect("accepts organization through the hub while the source has no connection", () => {
335 const test = harness();
336 return Effect.gen(function* () {
337 const hub = yield* Hub.HubConnection;
338 yield* hub.connect({ host: BASE, pairingCode: "code" });
339 const result = yield* hub.organize({
340 environmentId: EnvironmentId.make("laptop"),
341 command: {
342 type: "thread.settle",
343 commandId: CommandId.make("settle-offline"),
344 threadId: ThreadId.make("chat"),
345 },
346 });
347 expect(Option.getOrNull(result)).toEqual({ sequence: 5 });
348 expect(test.requests.at(-1)).toEqual({
349 url: `${BASE}api/hub/inbox`,
350 authorization: "Bearer hub-secret",
351 method: "POST",
352 etag: null,
353 });
354 const direct = yield* hub.organize({
355 environmentId: EnvironmentId.make("direct-only"),
356 command: {
357 type: "thread.unpin",
358 commandId: CommandId.make("unpin-direct"),
359 threadId: ThreadId.make("chat"),
360 },
361 });
362 expect(Option.isNone(direct)).toBe(true);
363 test.control.inboxStatus = 403;
364 expect(
365 (yield* Effect.result(
366 hub.organize({
367 environmentId: EnvironmentId.make("laptop"),
368 command: {
369 type: "thread.pin",
370 commandId: CommandId.make("denied-pin"),
371 threadId: ThreadId.make("chat"),
372 },
373 }),
374 ))._tag,
375 ).toBe("Failure");
376 }).pipe(Effect.provide(test.layer), Effect.scoped);
377 });
378 it.effect("revalidates unchanged inbox metadata without replacing the cached snapshot", () => {
379 const test = harness();
380 return Effect.gen(function* () {
381 const hub = yield* Hub.HubConnection;
382 yield* hub.connect({ host: BASE, pairingCode: "code" });
383 const inbox = (yield* SubscriptionRef.get(hub.state)).inbox;
384 test.control.notModified = true;
385 yield* hub.refreshInbox;
386 expect(test.requests.at(-1)?.etag).toBe('"inbox-v1"');
387 expect((yield* SubscriptionRef.get(hub.state)).inbox).toBe(inbox);
388 }).pipe(Effect.provide(test.layer), Effect.scoped);
389 });
390 it.effect("reuses the validated body after a reconnect fails to persist its profile", () => {
391 const test = harness();
392 return Effect.gen(function* () {
393 const hub = yield* Hub.HubConnection;
394 yield* hub.connect({ host: BASE, pairingCode: "code" });
395 test.control.inboxSequence = 6;
396 test.control.inboxEtag = '"inbox-v2"';
397 test.control.token = "new-account-token";
398 test.control.failProfileWrite = true;
399 yield* hub.connect({ host: BASE, pairingCode: "new-code" }).pipe(Effect.flip);
400 expect(
401 (yield* SubscriptionRef.get(hub.state)).inbox.environments[0]?.snapshot.snapshotSequence,
402 ).toBe(5);
403 test.control.notModified = true;
404 yield* hub.refreshInbox;
405 expect(test.requests.at(-1)?.etag).toBe('"inbox-v2"');
406 expect(
407 (yield* SubscriptionRef.get(hub.state)).inbox.environments[0]?.snapshot.snapshotSequence,
408 ).toBe(6);
409 }).pipe(Effect.provide(test.layer), Effect.scoped);
410 });
411
412 it.effect(
413 "never sends a validator fetched with another credential after failed pairing persistence",
414 () => {
415 const test = harness();
416 return Effect.gen(function* () {
417 const hub = yield* Hub.HubConnection;
418 yield* hub.connect({ host: BASE, pairingCode: "code" });
419 test.control.inboxSequence = 6;
420 test.control.inboxEtag = '"inbox-v2"';
421 test.control.token = "new-account-token";
422 test.control.failCredentialWrite = true;
423 yield* hub.connect({ host: BASE, pairingCode: "new-code" }).pipe(Effect.flip);
424 test.control.notModified = true;
425 yield* hub.refreshInbox;
426 expect(test.requests.at(-1)?.authorization).toBe("Bearer hub-secret");
427 expect(test.requests.at(-1)?.etag).toBeNull();
428 }).pipe(Effect.provide(test.layer), Effect.scoped);
429 },
430 );
431 it.effect(
432 "clears an outdated machine's compatibility rejection after its advertised protocol updates",
433 () => {
434 const test = harness();
435 test.control.protocolVersion = ORCHESTRATION_PROTOCOL_VERSION - 1;
436 return Effect.gen(function* () {
437 const hub = yield* Hub.HubConnection;
438 const registry = yield* Registry.EnvironmentRegistry;
439 yield* hub.connect({ host: BASE, pairingCode: "code" });
440 const id = EnvironmentId.make("laptop");
441 expect(
442 (yield* SubscriptionRef.get(registry.entries)).get(id)?.unsupportedReason,
443 ).toBeDefined();
444 test.control.protocolVersion = ORCHESTRATION_PROTOCOL_VERSION;
445 yield* hub.refresh;
446 expect(
447 (yield* SubscriptionRef.get(registry.entries)).get(id)?.unsupportedReason,
448 ).toBeUndefined();
449 yield* registry.setEnabled(id, true);
450 expect((yield* SubscriptionRef.get(registry.entries)).get(id)?.enabled).toBe(true);
451 }).pipe(Effect.provide(test.layer), Effect.scoped);
452 },
453 );
454 it.effect(
455 "pairs once for two machines, links one secret, and leaves unchanged routes running on refresh",
456 () => {
457 const test = harness();
458 return Effect.gen(function* () {
459 const hub = yield* Hub.HubConnection;
460 const registry = yield* Registry.EnvironmentRegistry;
461 yield* hub.connect({ host: BASE, pairingCode: "one-use-code" });
462 const entries = yield* SubscriptionRef.get(registry.entries);
463 expect([...entries.keys()]).toEqual(["laptop", "desktop"]);
464 expect(
465 test.requests.filter((request) => request.url.endsWith("/oauth/token")),
466 ).toHaveLength(1);
467 expect(
468 [...test.credentials.values()].filter(
469 (value) => value._tag === "BearerConnectionCredential",
470 ),
471 ).toEqual([new BearerConnectionCredential({ token: "hub-secret" })]);
472 expect(test.credentials.get("hub:hub:laptop")).toMatchObject({
473 _tag: "HubConnectionCredential",
474 connectionId: "hub-session:hub",
475 });
476 expect(
477 test.requests.find((request) => request.url.endsWith("/api/hub/environments"))
478 ?.authorization,
479 ).toBe("Bearer hub-secret");
480 yield* hub.refresh;
481 expect(test.writes).toEqual(["laptop", "desktop"]);
482 expect(yield* SubscriptionRef.get(registry.entries)).toBe(entries);
483 const restored = yield* Hub.make;
484 expect((yield* SubscriptionRef.get(restored.state)).connection?.httpBaseUrl).toBe(BASE);
485 }).pipe(Effect.provide(test.layer), Effect.scoped);
486 },
487 );
488
489 it.effect(
490 "keeps saved machines during an outage and discovers later enrollment on refresh",
491 () => {
492 const test = harness();
493 return Effect.gen(function* () {
494 const hub = yield* Hub.HubConnection;
495 const registry = yield* Registry.EnvironmentRegistry;
496 yield* hub.connect({ host: BASE, pairingCode: "one-use-code" });
497 const entries = yield* SubscriptionRef.get(registry.entries);
498 test.control.offline = true;
499 expect((yield* Effect.result(hub.refresh))._tag).toBe("Failure");
500 expect(yield* SubscriptionRef.get(registry.entries)).toBe(entries);
501 expect((yield* SubscriptionRef.get(hub.state)).error).toContain("Couldn't reach");
502 test.control.offline = false;
503 test.control.machines.push("new-machine");
504 yield* hub.refresh;
505 expect([...(yield* SubscriptionRef.get(registry.entries)).keys()]).toEqual([
506 "laptop",
507 "desktop",
508 "new-machine",
509 ]);
510 }).pipe(Effect.provide(test.layer), Effect.scoped);
511 },
512 );
513
514 it.effect(
515 "rejects a directory that sends machine traffic outside the paired origin without saving credentials",
516 () => {
517 const test = harness();
518 test.control.badAddress = true;
519 return Effect.gen(function* () {
520 const hub = yield* Hub.HubConnection;
521 const registry = yield* Registry.EnvironmentRegistry;
522 expect((yield* Effect.result(hub.connect({ host: BASE, pairingCode: "code" })))._tag).toBe(
523 "Failure",
524 );
525 expect(test.credentials.size).toBe(0);
526 expect((yield* SubscriptionRef.get(registry.entries)).size).toBe(0);
527 }).pipe(Effect.provide(test.layer), Effect.scoped);
528 },
529 );
530
531 it.effect(
532 "disconnects only hub-owned routes, preserving directly paired and T3 Connect routes",
533 () => {
534 const test = harness();
535 return Effect.gen(function* () {
536 const hub = yield* Hub.HubConnection;
537 const registry = yield* Registry.EnvironmentRegistry;
538 yield* hub.connect({ host: BASE, pairingCode: "code" });
539 const id = EnvironmentId.make("laptop");
540 const connectionId = "direct-laptop";
541 yield* registry.register(
542 new BearerConnectionRegistration({
543 target: new BearerConnectionTarget({
544 environmentId: id,
545 label: "Laptop",
546 connectionId,
547 }),
548 profile: new BearerConnectionProfile({
549 environmentId: id,
550 label: "Laptop",
551 connectionId,
552 httpBaseUrl: "https://laptop.example/",
553 wsBaseUrl: "wss://laptop.example/",
554 }),
555 credential: new BearerConnectionCredential({ token: "direct-secret" }),
556 }),
557 );
558 yield* registry.register(
559 new RelayConnectionRegistration({
560 target: new RelayConnectionTarget({ environmentId: id, label: "Laptop" }),
561 }),
562 );
563 yield* hub.disconnect;
564 const entries = yield* SubscriptionRef.get(registry.entries);
565 expect([...entries.keys()]).toEqual([id]);
566 expect(
567 connectionRoutes(entries.get(id)!).map((route) => connectionRouteId(route.target)),
568 ).toEqual([connectionId, "relay"]);
569 expect(test.cleanups).toEqual([{ id: "desktop", retainDrafts: true }]);
570 expect(test.credentials.has("hub-session:hub")).toBe(false);
571 expect((yield* SubscriptionRef.get(hub.state)).connection).toBeNull();
572 }).pipe(Effect.provide(test.layer), Effect.scoped);
573 },
574 );
575});