1import {
2 type ChatAttachment,
3 ChatAttachmentId,
4 CommandId,
5 GUEST_SHARE_TTL_MS,
6 GuestMessageId,
7 GuestShareCreateError,
8 type GuestShareCreateInput,
9 type GuestShareCreateResult,
10 GuestShareId,
11 type GuestRedeemRequest,
12 type GuestStreamEvent,
13 type GuestSubmitRequest,
14 type OrchestrationV2AppThread,
15 type OrchestrationV2CommandOutputFrame,
16 type OrchestrationV2DomainEvent,
17 type OrchestrationV2GetTurnItemResult,
18 type OrchestrationV2ThreadHistoryPage,
19 type OrchestrationV2TurnItem,
20 ThreadId,
21 TurnItemId,
22} from "@t3tools/contracts";
23import { chatMarkdownImageSources } from "@t3tools/shared/markdownPipeline";
24import { htmlRenderFromToolItem } from "@t3tools/shared/toolOutput";
25import * as Context from "effect/Context";
26import * as SecureRandom from "effect/crypto/SecureRandom";
27import * as DateTime from "effect/DateTime";
28import * as Duration from "effect/Duration";
29import * as Effect from "effect/Effect";
30import * as FileSystem from "effect/FileSystem";
31import * as Layer from "effect/Layer";
32import * as Option from "effect/Option";
33import * as Schedule from "effect/Schedule";
34import * as Schema from "effect/Schema";
35import * as Stream from "effect/Stream";
36
37import { base64UrlDecodeUtf8, base64UrlEncode, timingSafeEqualBase64Url } from "../auth/utils.ts";
38import * as PayloadSigner from "../auth/PayloadSigner.ts";
39import {
40 attachmentFileExtension,
41 createAttachmentId,
42 resolveAttachmentPath,
43 resolveAttachmentPathById,
44 threadHtmlRenderAttachmentIds,
45} from "../attachmentStore.ts";
46import * as ServerConfig from "../config.ts";
47import { CloudManagedEndpointRuntime } from "../cloud/ManagedEndpointRuntime.ts";
48import * as CommandOutputHub from "../orchestration-v2/CommandOutputHub.ts";
49import { GUEST_FAILURE } from "../orchestration-v2/GuestShareDecider.ts";
50import * as Orchestrator from "../orchestration-v2/Orchestrator.ts";
51import { coalesceThreadLiveStream } from "../orchestration-v2/ThreadLiveEventCoalescer.ts";
52import { buildBoundedThreadStreamSnapshot } from "../orchestration-v2/ThreadStream.ts";
53import {
54 projectTurnItemForDetail,
55 projectTurnItemForWire,
56 threadStreamEventItem,
57} from "../orchestration-v2/WireProjection.ts";
58import * as ProviderRegistry from "../provider/ProviderRegistry.ts";
59import * as ServerSettings from "../serverSettings.ts";
60import * as AnalyticsService from "../telemetry/AnalyticsService.ts";
61import { guestShareView, guestThreadModel } from "./guestView.ts";
62import {
63 classifyGuestUploads,
64 GUEST_UPLOAD_HEADER_BYTES,
65 guestHostImageType,
66 type GuestUpload,
67 matchesImageSignature,
68 repliesShowImage,
69} from "./guestAttachments.ts";
70
71const SIGNING_SECRET_NAME = "guest-share-signing-key";
72const SHARE_ID_BYTES = 16;
73const INVITE_SIGNATURE_BYTES = 16;
74const SNAPSHOT_ROW_LIMIT = 600;
75const SNAPSHOT_USER_TURN_LIMIT = 20;
76// Shares with no open guest stream whose sent image replies are still kept.
77const SHARED_THREAD_REPLIES_LIMIT = 64;
78// How soon an open read notices that its link stopped working, such as Connect no longer serving
79// the host. Closing or replacing the link ends the main stream at once.
80const ACCESS_RECHECK_INTERVAL = Duration.seconds(30);
81
82const GUEST_ACCESS_MESSAGES = {
83 ...GUEST_FAILURE,
84 invalid: "This guest link is not valid.",
85 stale_page: "This page is for an older guest link. Reload it to continue.",
86 unavailable: "This item is not available.",
87 rejected: "The owner's server rejected this request.",
88 empty: "Write a message or attach a file.",
89 too_many_files: "Attach up to 4 files to a message.",
90 too_large: "Attached files can total up to 10 MB per message.",
91 unsupported_file: "This file cannot be attached. Images must be PNG, JPEG, GIF, or WebP.",
92 files_busy: GUEST_FAILURE.filesBusy,
93} as const;
94
95/** What a guest is told. The message comes from `reason`, never from a cause. */
96export class GuestShareAccessError extends Schema.TaggedError<GuestShareAccessError>()(
97 "GuestShareAccessError",
98 {
99 reason: Schema.Literals([
100 "invalid",
101 "inactive",
102 "used",
103 "busy",
104 "stale_page",
105 "unavailable",
106 "rejected",
107 "empty",
108 "too_many_files",
109 "too_large",
110 "unsupported_file",
111 "files_busy",
112 ]),
113 },
114) {
115 override get message(): string {
116 return GUEST_ACCESS_MESSAGES[this.reason];
117 }
118}
119
120const guestFailureReason = (cause: unknown): GuestShareAccessError["reason"] =>
121 cause === GUEST_FAILURE.inactive
122 ? "inactive"
123 : cause === GUEST_FAILURE.used
124 ? "used"
125 : cause === GUEST_FAILURE.busy
126 ? "busy"
127 : cause === GUEST_FAILURE.filesBusy
128 ? "files_busy"
129 : "rejected";
130
131const SessionClaims = Schema.Struct({
132 kind: Schema.Literal("session"),
133 threadId: ThreadId,
134 shareId: GuestShareId,
135});
136type SessionClaims = typeof SessionClaims.Type;
137const decodeSessionClaims = Schema.decodeUnknownOption(Schema.fromJsonString(SessionClaims));
138
139const inactive = new GuestShareAccessError({ reason: "inactive" });
140
141export class GuestShareService extends Context.Service<
142 GuestShareService,
143 {
144 readonly create: (
145 input: GuestShareCreateInput,
146 ) => Effect.Effect<GuestShareCreateResult, GuestShareCreateError>;
147 /** Spends an invite and returns the session token for the guest's cookie. */
148 readonly redeem: (
149 request: GuestRedeemRequest,
150 ) => Effect.Effect<
151 { readonly token: string; readonly shareId: GuestShareId },
152 GuestShareAccessError
153 >;
154 /** Succeeds while the session can still open its share. */
155 readonly check: (session: string) => Effect.Effect<void, GuestShareAccessError>;
156 /** Stores the guest's files as the thread's attachments and records the message with them. */
157 readonly submit: (
158 session: string,
159 request: GuestSubmitRequest,
160 uploads: ReadonlyArray<GuestUpload>,
161 ) => Effect.Effect<void, GuestShareAccessError>;
162 readonly stream: (
163 session: string,
164 ) => Effect.Effect<Stream.Stream<GuestStreamEvent>, GuestShareAccessError>;
165 /** The file of an HTML render in the shared thread, and its title. */
166 readonly render: (
167 session: string,
168 itemId: string,
169 ) => Effect.Effect<{ readonly path: string; readonly title: string }, GuestShareAccessError>;
170 /** One item of the shared thread, as the owner's expanded row loads it, without file diffs. */
171 readonly item: (
172 session: string,
173 itemId: string,
174 ) => Effect.Effect<OrchestrationV2GetTurnItemResult, GuestShareAccessError>;
175 /** Older turns of the shared thread, as the owner's client pages them. */
176 readonly history: (
177 session: string,
178 cursor: string,
179 ) => Effect.Effect<OrchestrationV2ThreadHistoryPage, GuestShareAccessError>;
180 /** A command's output frames, as the owner's expanded row gets them, until the link ends. */
181 readonly commandOutput: (
182 session: string,
183 itemId: string,
184 ) => Effect.Effect<Stream.Stream<OrchestrationV2CommandOutputFrame>, GuestShareAccessError>;
185 /** A file in the shared thread or one of the share's messages, never a dismissed one's. */
186 readonly attachment: (
187 session: string,
188 attachmentId: string,
189 ) => Effect.Effect<
190 { readonly path: string; readonly attachment: ChatAttachment },
191 GuestShareAccessError
192 >;
193 /** An image on the host that an agent reply the guest sees shows, and its type. */
194 readonly media: (
195 session: string,
196 path: string,
197 ) => Effect.Effect<{ readonly path: string; readonly mimeType: string }, GuestShareAccessError>;
198 }
199>()("t3/guest/GuestShareService") {}
200
201/** A reply a guest's stream sent, with its image sources once a media request parsed them. */
202interface SentReply {
203 readonly text: string;
204 sources?: ReadonlyArray<string>;
205}
206
207/** The thread record an event carries, for the share and the view. */
208const threadOf = (event: OrchestrationV2DomainEvent): OrchestrationV2AppThread | null => {
209 switch (event.type) {
210 case "thread.guest-share-updated":
211 case "thread.metadata-updated":
212 case "thread.model-selection-updated":
213 case "thread.provider-switched":
214 return event.payload;
215 default:
216 return null;
217 }
218};
219
220const make = Effect.gen(function* () {
221 const orchestrator = yield* Orchestrator.OrchestratorV2;
222 const commandOutputs = yield* CommandOutputHub.CommandOutputHub;
223 const providerRegistry = yield* ProviderRegistry.ProviderRegistry;
224 const serverSettings = yield* ServerSettings.ServerSettingsService;
225 const managedEndpoint = yield* CloudManagedEndpointRuntime;
226 const { attachmentsDir } = yield* ServerConfig.ServerConfig;
227 const fileSystem = yield* FileSystem.FileSystem;
228 const random = yield* SecureRandom.SecureRandom;
229 const analytics = yield* AnalyticsService.AnalyticsService;
230 const signer = yield* PayloadSigner.make(SIGNING_SECRET_NAME);
231 /** A truncated HMAC keeps invite links short. 128 bits is still far beyond guessing range. */
232 const inviteSignature = (shareId: string) =>
233 signer.sign(`guest-invite:${shareId}`).pipe(
234 Effect.orDie,
235 Effect.map((signature) =>
236 base64UrlEncode(Buffer.from(signature, "base64url").subarray(0, INVITE_SIGNATURE_BYTES)),
237 ),
238 );
239
240 // Tokens are derived from the share id, so nothing secret is stored and replacing or closing
241 // the share in the thread projection revokes every token for it. An invite is
242 // `<shareId>.<signature>` and the thread is looked up by share id; a session token carries
243 // signed claims.
244 const signInvite = (shareId: GuestShareId) =>
245 inviteSignature(shareId).pipe(Effect.map((signature) => `${shareId}.${signature}`));
246 const signSession = (claims: SessionClaims) => {
247 const payload = base64UrlEncode(JSON.stringify(claims));
248 return signer.sign(payload).pipe(
249 Effect.orDie,
250 Effect.map((signature) => `${payload}.${signature}`),
251 );
252 };
253 const splitToken = (token: string) => {
254 const [payload, signature, extra] = token.split(".");
255 return payload && signature && extra === undefined ? { payload, signature } : null;
256 };
257 const verifyInvite = Effect.fn("GuestShareService.verifyInvite")(function* (token: string) {
258 const parts = splitToken(token);
259 if (parts === null) return Option.none<SessionClaims>();
260 const expected = yield* inviteSignature(parts.payload);
261 if (!timingSafeEqualBase64Url(parts.signature, expected)) {
262 return Option.none<SessionClaims>();
263 }
264 const shareId = GuestShareId.make(parts.payload);
265 const threadId = yield* orchestrator
266 .findGuestShareThreadId(shareId)
267 .pipe(Effect.mapError(() => inactive));
268 return threadId === null
269 ? Option.none<SessionClaims>()
270 : Option.some<SessionClaims>({ kind: "session", threadId, shareId });
271 });
272 // Untraced: runs on every guest request, like the Effect.map it replaced.
273 const verifySession = Effect.fnUntraced(function* (token: string) {
274 const parts = splitToken(token);
275 if (parts === null) return Option.none<SessionClaims>();
276 if (!(yield* signer.verify(parts.payload, parts.signature).pipe(Effect.orDie))) {
277 return Option.none<SessionClaims>();
278 }
279 return decodeSessionClaims(base64UrlDecodeUtf8(parts.payload));
280 });
281
282 const connectHostname = managedEndpoint.inspectConnector.pipe(
283 Effect.map((facts) => facts?.hostname ?? null),
284 );
285 // A beta: while it is off, no link opens and every guest request is refused.
286 const sharingEnabled = serverSettings.getSettings.pipe(
287 Effect.map((settings) => settings.guestSharingEnabled),
288 Effect.orElseSucceed(() => false),
289 );
290
291 const dispatchGuestCommand = (
292 command: Parameters<Orchestrator.OrchestratorV2["Service"]["dispatch"]>[0],
293 ) =>
294 orchestrator.dispatch(command).pipe(
295 Effect.mapError(
296 (error) =>
297 new GuestShareAccessError({
298 reason: guestFailureReason("cause" in error ? error.cause : undefined),
299 }),
300 ),
301 );
302
303 const commandId = (label: string) =>
304 random.uuidV4().pipe(
305 Effect.map((uuid) => CommandId.make(`server:guest-share:${label}:${uuid}`)),
306 Effect.orDie,
307 );
308
309 /** Checks a token against the live share: same id, not expired, and Connect still serving it. */
310 const authorize = Effect.fn("GuestShareService.authorize")(function* (
311 token: string,
312 kind: "invite" | "session",
313 ) {
314 const claims = yield* kind === "invite" ? verifyInvite(token) : verifySession(token);
315 if (Option.isNone(claims)) {
316 return yield* new GuestShareAccessError({ reason: "invalid" });
317 }
318 const { thread } = yield* orchestrator
319 .getThreadRecords(claims.value.threadId, [])
320 .pipe(Effect.mapError(() => inactive));
321 const share = thread.guestShare;
322 const now = yield* DateTime.now;
323 if (
324 !(yield* sharingEnabled) ||
325 !share ||
326 share.id !== claims.value.shareId ||
327 thread.archivedAt !== null ||
328 thread.deletedAt !== null ||
329 DateTime.toEpochMillis(now) >= Date.parse(share.expiresAt) ||
330 (kind === "session" && share.guestName === null) ||
331 (yield* connectHostname) !== share.host
332 ) {
333 return yield* inactive;
334 }
335 return { thread, share };
336 });
337
338 const create = Effect.fn("GuestShareService.create")(function* (input: GuestShareCreateInput) {
339 if (!(yield* sharingEnabled)) {
340 return yield* new GuestShareCreateError({ reason: "disabled" });
341 }
342 const host = yield* connectHostname;
343 if (host === null) {
344 return yield* new GuestShareCreateError({ reason: "connect_unavailable" });
345 }
346 const shareId = GuestShareId.make(base64UrlEncode(yield* random.randomBytes(SHARE_ID_BYTES)));
347 const now = yield* DateTime.now;
348 const expiresAt = DateTime.formatIso(DateTime.add(now, { milliseconds: GUEST_SHARE_TTL_MS }));
349 yield* orchestrator
350 .dispatch({
351 type: "thread.guest-share.open",
352 commandId: yield* commandId("open"),
353 threadId: input.threadId,
354 shareId,
355 host,
356 expiresAt,
357 })
358 .pipe(Effect.mapError(() => new GuestShareCreateError({ reason: "thread_unavailable" })));
359 const { thread } = yield* orchestrator
360 .getThreadRecords(input.threadId, [])
361 .pipe(Effect.mapError(() => new GuestShareCreateError({ reason: "read_failed" })));
362 if (!thread.guestShare || thread.guestShare.id !== shareId) {
363 return yield* new GuestShareCreateError({ reason: "changed" });
364 }
365 const credential = yield* signInvite(shareId);
366 return { url: `https://${host}/guest#${credential}`, share: thread.guestShare };
367 });
368
369 const redeem = Effect.fn("GuestShareService.redeem")(function* (request: GuestRedeemRequest) {
370 const { thread, share } = yield* authorize(request.credential, "invite");
371 yield* dispatchGuestCommand({
372 type: "thread.guest-share.redeem",
373 commandId: yield* commandId("redeem"),
374 threadId: thread.id,
375 shareId: share.id,
376 guestName: request.name,
377 });
378 // Guests have no client connection, so their events carry no surface.
379 yield* analytics.record("feature.used", { feature: "guest_share.join" });
380 const token = yield* signSession({ kind: "session", threadId: thread.id, shareId: share.id });
381 return { token, shareId: share.id };
382 });
383
384 const check = Effect.fn("GuestShareService.check")(function* (session: string) {
385 yield* authorize(session, "session");
386 });
387
388 const uploadFailed = new GuestShareAccessError({ reason: "rejected" });
389 const removeStored = (attachments: ReadonlyArray<ChatAttachment>) =>
390 Effect.forEach(
391 attachments,
392 (attachment) => {
393 const path = resolveAttachmentPath({ attachmentsDir, attachment });
394 return path === null
395 ? Effect.void
396 : fileSystem.remove(path, { force: true }).pipe(Effect.ignore);
397 },
398 { discard: true },
399 );
400
401 /**
402 * Checks the files and stores them as the thread's attachments, so a message can wait longer
403 * than the day the pending-upload sweep allows. Send uses them as they are; dismissing the
404 * message or ending the share deletes them.
405 */
406 const storeUploads = Effect.fn("GuestShareService.storeUploads")(function* (
407 threadId: ThreadId,
408 uploads: ReadonlyArray<GuestUpload>,
409 ) {
410 const inspected = yield* Effect.forEach(uploads, (upload) =>
411 Effect.gen(function* () {
412 const info = yield* fileSystem.stat(upload.path);
413 const file = yield* fileSystem.open(upload.path, { flag: "r" });
414 const header = Option.getOrElse(
415 yield* file.readAlloc(GUEST_UPLOAD_HEADER_BYTES),
416 () => new Uint8Array(),
417 );
418 return { ...upload, sizeBytes: Number(info.size), header };
419 }).pipe(
420 Effect.scoped,
421 Effect.mapError(() => uploadFailed),
422 ),
423 );
424 const classified = classifyGuestUploads(inspected);
425 if (!classified.ok) return yield* new GuestShareAccessError({ reason: classified.reason });
426 const stored: Array<ChatAttachment> = [];
427 yield* Effect.forEach(
428 classified.attachments,
429 (draft, index) =>
430 Effect.gen(function* () {
431 const id = createAttachmentId(
432 threadId,
433 draft.type === "file" ? attachmentFileExtension(draft.name) : undefined,
434 );
435 if (id === null) return yield* uploadFailed;
436 const attachment: ChatAttachment = { ...draft, id: ChatAttachmentId.make(id) };
437 const path = resolveAttachmentPath({ attachmentsDir, attachment });
438 if (path === null) return yield* uploadFailed;
439 yield* fileSystem.makeDirectory(attachmentsDir, { recursive: true });
440 yield* fileSystem.copyFile(uploads[index]!.path, path);
441 stored.push(attachment);
442 }).pipe(Effect.mapError(() => uploadFailed)),
443 { discard: true },
444 ).pipe(Effect.onError(() => removeStored(stored)));
445 return stored;
446 });
447
448 const submit = Effect.fn("GuestShareService.submit")(function* (
449 session: string,
450 request: GuestSubmitRequest,
451 uploads: ReadonlyArray<GuestUpload>,
452 ) {
453 const { thread, share } = yield* authorize(session, "session");
454 if (request.shareId !== share.id) {
455 return yield* new GuestShareAccessError({ reason: "stale_page" });
456 }
457 if (request.text === "" && uploads.length === 0) {
458 return yield* new GuestShareAccessError({ reason: "empty" });
459 }
460 const attachments = yield* storeUploads(thread.id, uploads);
461 yield* dispatchGuestCommand({
462 type: "thread.guest-message.submit",
463 commandId: yield* commandId("submit"),
464 threadId: thread.id,
465 shareId: share.id,
466 guestMessageId: GuestMessageId.make(yield* random.uuidV4()),
467 text: request.text,
468 ...(attachments.length === 0 ? {} : { attachments }),
469 }).pipe(Effect.onError(() => removeStored(attachments)));
470 yield* analytics.record("feature.used", {
471 feature: "guest_message.submit",
472 variant: attachments.length === 0 ? "text_only" : "with_files",
473 });
474 });
475
476 /** Completes once the session no longer opens its share, so open reads end with it. */
477 const untilRevoked = (session: string) =>
478 authorize(session, "session").pipe(
479 Effect.repeat(Schedule.spaced(ACCESS_RECHECK_INTERVAL)),
480 Effect.ignore,
481 );
482
483 // The image replies each share's guest stream has sent, in its opening snapshot or live, so an
484 // image loads while the guest's page can show its reply, inherited ones included. Keyed by
485 // share, so a new link on the thread starts empty. A share with an open stream keeps its
486 // record; past the limit, closed ones are dropped.
487 const sentImageReplies = new Map<
488 GuestShareId,
489 { openStreams: number; readonly replies: Map<string, SentReply> }
490 >();
491 const sentRecord = (shareId: GuestShareId) => {
492 let record = sentImageReplies.get(shareId);
493 if (record === undefined) {
494 record = { openStreams: 0, replies: new Map() };
495 sentImageReplies.set(shareId, record);
496 for (const [id, other] of sentImageReplies) {
497 if (sentImageReplies.size <= SHARED_THREAD_REPLIES_LIMIT) break;
498 if (other.openStreams === 0 && id !== shareId) sentImageReplies.delete(id);
499 }
500 }
501 return record;
502 };
503 const rememberSent = (shareId: GuestShareId, items: Iterable<OrchestrationV2TurnItem>) => {
504 for (const item of items) {
505 if (item.type !== "assistant_message") continue;
506 const key = `${item.threadId}\u0000${item.id}`;
507 // Only replies with image syntax can show an image. One edited to drop its images stops
508 // allowing them.
509 if (item.text.includes("![") || /<img/i.test(item.text)) {
510 sentRecord(shareId).replies.set(key, { text: item.text });
511 } else {
512 sentImageReplies.get(shareId)?.replies.delete(key);
513 }
514 }
515 };
516
517 /**
518 * The thread as the owner's thread subscription sends it: a bounded snapshot, then the same
519 * wire events, batched the same way. Only the share's own state is added.
520 */
521 const stream = Effect.fn("GuestShareService.stream")(function* (session: string) {
522 const { thread, share } = yield* authorize(session, "session");
523 const snapshot = yield* orchestrator
524 .getThreadSnapshotWindow(thread.id, {
525 rowLimit: SNAPSHOT_ROW_LIMIT,
526 userTurnLimit: SNAPSHOT_USER_TURN_LIMIT,
527 })
528 .pipe(Effect.mapError(() => inactive));
529 // The share can end between authorize and this read; the live tail starts after it.
530 if (
531 snapshot.projection.thread.guestShare?.id !== share.id ||
532 snapshot.projection.thread.archivedAt !== null ||
533 snapshot.projection.thread.deletedAt !== null
534 ) {
535 return yield* inactive;
536 }
537 // Read once; a provider configured mid-share shows after the guest reconnects.
538 const providers = yield* providerRegistry.getProviders;
539 const viewOf = (current: OrchestrationV2AppThread) =>
540 guestShareView(current, guestThreadModel(current.modelSelection, providers));
541 const initialView = viewOf(snapshot.projection.thread);
542 if (initialView === null) return yield* inactive;
543
544 const ended: GuestStreamEvent = { type: "ended" };
545 const endsShare = (event: OrchestrationV2DomainEvent) => {
546 if (event.type === "thread.archived" || event.type === "thread.deleted") return true;
547 const updated = threadOf(event);
548 return updated !== null && updated.guestShare?.id !== share.id;
549 };
550 const live = orchestrator
551 .streamStoredEventsFrom({ threadId: thread.id, afterSequence: snapshot.snapshotSequence })
552 .pipe(
553 Stream.map(threadStreamEventItem),
554 // Ending the source here flushes what the coalescer holds, so `ended` is not held back.
555 Stream.takeUntil((item) => endsShare(item.event)),
556 coalesceThreadLiveStream,
557 Stream.map((item): ReadonlyArray<GuestStreamEvent> => {
558 if (item.kind !== "event") return [];
559 if (endsShare(item.event)) return [ended];
560 if (item.event.type === "turn-item.updated") rememberSent(share.id, [item.event.payload]);
561 const updated = threadOf(item.event);
562 const view = updated === null ? null : viewOf(updated);
563 return view === null
564 ? [{ type: "thread", item }]
565 : [
566 { type: "thread", item },
567 { type: "view", view },
568 ];
569 }),
570 Stream.flattenIterable,
571 // A failed read closes the response without `ended`: the browser reconnects, and
572 // `authorize` decides whether the link still works.
573 Stream.catchCause(() => Stream.empty),
574 );
575 const untilExpiry = Effect.sleep(
576 Math.max(0, Date.parse(share.expiresAt) - DateTime.toEpochMillis(yield* DateTime.now)),
577 );
578 const openingSnapshot = buildBoundedThreadStreamSnapshot({
579 snapshotSequence: snapshot.snapshotSequence,
580 projection: snapshot.projection,
581 });
582 const opening: ReadonlyArray<GuestStreamEvent> = [
583 { type: "view", view: initialView },
584 { type: "thread", item: openingSnapshot },
585 ];
586 const events = Stream.fromIterable(opening).pipe(
587 Stream.concat(
588 Stream.merge(live, Stream.fromEffect(Effect.as(untilExpiry, ended)), {
589 haltStrategy: "either",
590 }),
591 ),
592 Stream.takeUntil((event) => event.type === "ended"),
593 // Unlinking T3 Connect closes it quietly, so a tunnel that comes back on the same host
594 // keeps the guest in. Turning the beta off closes it the same way. Closing and expiry end
595 // it above, with `ended`.
596 Stream.interruptWhen(
597 Effect.all([connectHostname, sharingEnabled]).pipe(
598 Effect.repeat({
599 schedule: Schedule.spaced(ACCESS_RECHECK_INTERVAL),
600 until: ([host, enabled]) => host !== share.host || !enabled,
601 }),
602 ),
603 ),
604 );
605 // The record is opened and counted when the stream runs, paired with its release, so a
606 // response whose body never runs (HEAD) touches nothing.
607 return Stream.unwrap(
608 Effect.acquireRelease(
609 Effect.sync(() => {
610 const record = sentRecord(share.id);
611 record.openStreams += 1;
612 rememberSent(
613 share.id,
614 openingSnapshot.projection.visibleTurnItems.map((row) => row.item),
615 );
616 return record;
617 }),
618 (record) =>
619 Effect.sync(() => {
620 record.openStreams -= 1;
621 }),
622 ).pipe(Effect.as(events)),
623 );
624 });
625
626 const unavailable = new GuestShareAccessError({ reason: "unavailable" });
627
628 /** An item of the shared thread, read after the session check. */
629 const threadItem = Effect.fn("GuestShareService.threadItem")(function* (
630 session: string,
631 itemId: string,
632 ) {
633 const { thread } = yield* authorize(session, "session");
634 const item = yield* orchestrator
635 .getTurnItem({ threadId: thread.id, itemId: TurnItemId.make(itemId) })
636 .pipe(Effect.mapError(() => unavailable));
637 return item === null ? yield* unavailable : { thread, item };
638 });
639
640 const render = Effect.fn("GuestShareService.render")(function* (session: string, itemId: string) {
641 const { thread, item } = yield* threadItem(session, itemId);
642 if (item.type !== "dynamic_tool") return yield* unavailable;
643 const reference = htmlRenderFromToolItem(item);
644 const [attachmentId] = threadHtmlRenderAttachmentIds(thread.id, [item]);
645 const path =
646 attachmentId === undefined
647 ? null
648 : resolveAttachmentPathById({ attachmentsDir, attachmentId });
649 if (reference === undefined || path === null) return yield* unavailable;
650 return { path, title: reference.title };
651 });
652
653 const item = Effect.fn("GuestShareService.item")(function* (session: string, itemId: string) {
654 const { item } = yield* threadItem(session, itemId);
655 return {
656 item:
657 item.type === "file_change" ? projectTurnItemForWire(item) : projectTurnItemForDetail(item),
658 };
659 });
660
661 const history = Effect.fn("GuestShareService.history")(function* (
662 session: string,
663 cursor: string,
664 ) {
665 const { thread, share } = yield* authorize(session, "session");
666 const page = yield* orchestrator
667 .getThreadHistoryPage(thread.id, cursor)
668 .pipe(Effect.mapError(() => unavailable));
669 rememberSent(
670 share.id,
671 page.items.map((row) => row.item),
672 );
673 return page;
674 });
675
676 const commandOutput = Effect.fn("GuestShareService.commandOutput")(function* (
677 session: string,
678 itemId: string,
679 ) {
680 const { thread, item } = yield* threadItem(session, itemId);
681 if (item.type !== "command_execution") return yield* unavailable;
682 return commandOutputs.subscribe({ threadId: thread.id, itemId: item.id }).pipe(
683 // A lost hub stream ends the output; the row reloads it when reopened.
684 Stream.catchCause(() => Stream.empty),
685 Stream.interruptWhen(untilRevoked(session)),
686 );
687 });
688
689 const attachment = Effect.fn("GuestShareService.attachment")(function* (
690 session: string,
691 attachmentId: string,
692 ) {
693 const { thread, share } = yield* authorize(session, "session");
694 // A dismissed message's files are deleted with it.
695 const fromShare = share.messages
696 .flatMap((message) => (message.status === "dismissed" ? [] : (message.attachments ?? [])))
697 .find((candidate) => candidate.id === attachmentId);
698 const found =
699 fromShare ??
700 (yield* orchestrator.getThreadRecords(thread.id, ["messages"]).pipe(
701 Effect.map(({ messages }) =>
702 messages
703 .flatMap((message) => message.attachments ?? [])
704 .find((candidate) => candidate.id === attachmentId),
705 ),
706 Effect.mapError(() => unavailable),
707 )) ??
708 // A fork shows its source's earlier messages, which stay in the source thread.
709 (thread.forkedFrom === null
710 ? undefined
711 : yield* orchestrator.getTimelinePage(thread.id, { attachmentId, limit: 1 }).pipe(
712 Effect.map(({ items }) =>
713 items
714 .flatMap(({ item }) => (item.type === "user_message" ? item.attachments : []))
715 .find((candidate) => candidate.id === attachmentId),
716 ),
717 Effect.mapError(() => unavailable),
718 ));
719 const path =
720 found === undefined ? null : resolveAttachmentPath({ attachmentsDir, attachment: found });
721 if (found === undefined || path === null) return yield* unavailable;
722 const exists = yield* fileSystem.exists(path).pipe(Effect.orElseSucceed(() => false));
723 return exists ? { path, attachment: found } : yield* unavailable;
724 });
725
726 const media = Effect.fn("GuestShareService.media")(function* (session: string, path: string) {
727 const { share } = yield* authorize(session, "session");
728 const mimeType = guestHostImageType(path);
729 if (mimeType === null) return yield* unavailable;
730 const sent = sentImageReplies.get(share.id)?.replies.values() ?? [];
731 // Each sent reply is parsed once, on the first request that needs it.
732 if (
733 !repliesShowImage(
734 path,
735 [...sent],
736 (reply) => (reply.sources ??= chatMarkdownImageSources(reply.text)),
737 )
738 ) {
739 return yield* unavailable;
740 }
741 // The bytes must be the image the name says, as for a guest's own upload.
742 const header = yield* fileSystem.open(path, { flag: "r" }).pipe(
743 Effect.flatMap((file) => file.readAlloc(GUEST_UPLOAD_HEADER_BYTES)),
744 Effect.scoped,
745 Effect.mapError(() => unavailable),
746 );
747 if (Option.isNone(header) || !matchesImageSignature(mimeType, header.value)) {
748 return yield* unavailable;
749 }
750 return { path, mimeType };
751 });
752
753 return GuestShareService.of({
754 create,
755 redeem,
756 check,
757 submit,
758 stream,
759 render,
760 item,
761 history,
762 commandOutput,
763 attachment,
764 media,
765 });
766});
767
768export const layer = Layer.effect(GuestShareService, make);