Open History to see the last commit
1import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import { assert, it } from "@effect/vitest";
4import {
5 ChatAttachmentId,
6 CommandId,
7 EventId,
8 GuestShareId,
9 type GuestStreamEvent,
10 MessageId,
11 type OrchestrationV2ThreadHistoryPage,
12 type OrchestrationV2TurnItem,
13 ProjectId,
14 ProviderDriverKind,
15 ProviderInstanceId,
16 RunId,
17 ThreadId,
18 TurnItemId,
19} from "@t3tools/contracts";
20import * as Context from "effect/Context";
21import * as DateTime from "effect/DateTime";
22import * as Deferred from "effect/Deferred";
23import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
24import * as Effect from "effect/Effect";
25import * as Exit from "effect/Exit";
26import * as FileSystem from "effect/FileSystem";
27import * as Fiber from "effect/Fiber";
28import * as Layer from "effect/Layer";
29import * as Queue from "effect/Queue";
30import * as Ref from "effect/Ref";
31import * as Stream from "effect/Stream";
32import * as TestClock from "effect/testing/TestClock";
33
34import {
35 createAttachmentId,
36 resolveAttachmentPath,
37 resolveAttachmentPathById,
38 sweepStalePendingAttachments,
39} from "../attachmentStore.ts";
40import { resolveAttachmentRelativePath } from "../attachmentPaths.ts";
41import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
42import { CloudManagedEndpointRuntime } from "../cloud/ManagedEndpointRuntime.ts";
43import * as ServerConfig from "../config.ts";
44import * as SqlitePersistence from "../persistence/Sqlite.ts";
45import { CodexProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/CodexAdapterV2.ts";
46import * as CommandOutputHub from "../orchestration-v2/CommandOutputHub.ts";
47import * as EffectOutbox from "../orchestration-v2/EffectOutbox.ts";
48import * as EffectWorker from "../orchestration-v2/EffectWorker.ts";
49import * as EventSink from "../orchestration-v2/EventSink.ts";
50import * as Orchestrator from "../orchestration-v2/Orchestrator.ts";
51import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
52import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts";
53import * as ResourceCleanupService from "../orchestration-v2/ResourceCleanupService.ts";
54import * as ProviderReplayHarness from "../orchestration-v2/testkit/ProviderReplayHarness.ts";
55import * as DocumentText from "../provider/DocumentText.ts";
56import * as ProviderRegistry from "../provider/ProviderRegistry.ts";
57import * as ServerSettings from "../serverSettings.ts";
58import * as AnalyticsService from "../telemetry/AnalyticsService.ts";
59import * as GuestShareService from "./GuestShareService.ts";
60
61const instanceId = ProviderInstanceId.make("codex");
62const adapter = {
63 instanceId,
64 driver: ProviderDriverKind.make("codex"),
65 getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2),
66 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }),
67 openSession: () => Effect.die("No provider process needed for guest shares"),
68} as ProviderAdapter.ProviderAdapterV2["Service"];
69const layerDatabase = SqlitePersistence.layerMemory;
70const layerOrchestration = Layer.mergeAll(
71 layerDatabase,
72 ProjectionStore.layer.pipe(Layer.provide(layerDatabase)),
73 EffectOutbox.layer.pipe(Layer.provide(layerDatabase)),
74 ProviderReplayHarness.layerWithRegistry(
75 { name: "guest-share-service" },
76 ProviderAdapterRegistry.layerFromAdapters([adapter]),
77 { databaseLayer: layerDatabase, runEffectWorker: false },
78 ),
79);
80
81// The Connect hostname the fake tunnel reports. Tests set it to null to simulate an unlink.
82const hostname = Ref.makeUnsafe<string | null>("env.t3.test");
83const layerConnect = Layer.mock(CloudManagedEndpointRuntime)({
84 inspectConnector: Effect.map(Ref.get(hostname), (value) => ({
85 pid: 1,
86 readyConnections: 1,
87 hostname: value,
88 origin: null,
89 protocol: "http2",
90 })),
91});
92const layerProviders = Layer.mock(ProviderRegistry.ProviderRegistry)({
93 getProviders: Effect.succeed([]),
94});
95const layerConfig = ServerConfig.layerTest(process.cwd(), { prefix: "t3-guest-share-" });
96// The real cleanup also closes terminals, previews and devices; these tests only need the files gone.
97const layerCleanup = Layer.effect(
98 ResourceCleanupService.ResourceCleanupService,
99 Effect.gen(function* () {
100 const { attachmentsDir } = yield* ServerConfig.ServerConfig;
101 const fileSystem = yield* FileSystem.FileSystem;
102 return {
103 cleanupScheduledTasks: () => Effect.void,
104 cleanupTerminals: () => Effect.void,
105 cleanupPreviews: () => Effect.void,
106 cleanupDevices: () => Effect.void,
107 cleanupAttachments: (attachmentIds) =>
108 Effect.forEach(
109 attachmentIds,
110 (attachmentId) => {
111 const path = resolveAttachmentPathById({ attachmentsDir, attachmentId });
112 return path === null ? Effect.void : fileSystem.remove(path).pipe(Effect.orDie);
113 },
114 { discard: true },
115 ),
116 };
117 }),
118).pipe(
119 Layer.provide(layerConfig),
120 Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
121);
122class RecordedAnalytics extends Context.Service<
123 RecordedAnalytics,
124 Ref.Ref<ReadonlyArray<Readonly<Record<string, unknown>>>>
125>()("t3/guest/GuestShareService.test/RecordedAnalytics") {}
126const layerAnalytics = Layer.effectContext(
127 Effect.gen(function* () {
128 const recorded = yield* Ref.make<ReadonlyArray<Readonly<Record<string, unknown>>>>([]);
129 return Context.make(RecordedAnalytics, recorded).pipe(
130 Context.add(
131 AnalyticsService.AnalyticsService,
132 AnalyticsService.AnalyticsService.of({
133 record: (event, properties) =>
134 Ref.update(recorded, (events) => [...events, { event, ...properties }]),
135 flush: Effect.void,
136 recordOwners: () => Effect.void,
137 }),
138 ),
139 );
140 }),
141);
142const layerTest = GuestShareService.layer.pipe(
143 Layer.provideMerge(
144 CommandOutputHub.layer.pipe(
145 Layer.provideMerge(layerOrchestration.pipe(Layer.provide(layerCleanup))),
146 ),
147 ),
148 Layer.provide(layerProviders),
149 Layer.provideMerge(ServerSettings.layerTest({ guestSharingEnabled: true })),
150 Layer.provide(layerConnect),
151 Layer.provideMerge(layerAnalytics),
152 Layer.provide(ServerSecretStore.layer),
153 Layer.provide(DocumentText.layer),
154 Layer.provideMerge(layerConfig),
155 Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
156);
157
158const threadId = ThreadId.make("thread:guest-share-service");
159
160const shareThreadFor = Effect.fn(function* (threadId: ThreadId) {
161 yield* Ref.set(hostname, "env.t3.test");
162 const orchestrator = yield* Orchestrator.OrchestratorV2;
163 yield* orchestrator.dispatch({
164 type: "thread.create",
165 commandId: CommandId.make(`create:${threadId}`),
166 threadId,
167 projectId: ProjectId.make("project:guest-share-service"),
168 title: "Review me",
169 modelSelection: { instanceId, model: "gpt-5.1-codex" },
170 runtimeMode: "full-access",
171 interactionMode: "default",
172 branch: null,
173 worktreePath: null,
174 createdBy: "user",
175 creationSource: "web",
176 });
177 const guestShares = yield* GuestShareService.GuestShareService;
178 const { url } = yield* guestShares.create({ threadId });
179 assert.isTrue(url.startsWith("https://env.t3.test/guest#"));
180 const credential = url.slice(url.indexOf("#") + 1);
181 return { orchestrator, guestShares, credential };
182});
183const shareThread = shareThreadFor(threadId);
184
185it.effect("spends an invite once and ends the guest stream when the owner stops sharing", () =>
186 Effect.gen(function* () {
187 const { orchestrator, guestShares, credential } = yield* shareThread;
188 const { token: session } = yield* guestShares.redeem({ credential, name: "Sam" });
189 const again = yield* Effect.exit(guestShares.redeem({ credential, name: "Eve" }));
190 assert.isTrue(Exit.isFailure(again));
191 // The browser that joined can come back without the spent invite.
192 assert.isTrue(Exit.isSuccess(yield* Effect.exit(guestShares.check(session))));
193 // An invite token is not a session token.
194 assert.isTrue(Exit.isFailure(yield* Effect.exit(guestShares.stream(credential))));
195
196 const events = yield* guestShares.stream(session);
197 yield* orchestrator.dispatch({
198 type: "thread.guest-share.close",
199 commandId: CommandId.make("close-guest-share"),
200 threadId,
201 });
202 const received = Array.from(yield* Stream.runCollect(events));
203 assert.deepEqual(
204 received.map((event) => (event.type === "thread" ? event.item.kind : event.type)),
205 ["view", "snapshot", "ended"],
206 );
207 assert.isTrue(Exit.isFailure(yield* Effect.exit(guestShares.stream(session))));
208 assert.isTrue(Exit.isFailure(yield* Effect.exit(guestShares.check(session))));
209 }).pipe(Effect.provide(layerTest)),
210);
211
212it.effect("counts a guest joining and sending, without their name or text", () =>
213 Effect.gen(function* () {
214 const { guestShares, credential } = yield* shareThread;
215 const { token: session, shareId } = yield* guestShares.redeem({ credential, name: "Sam" });
216 yield* guestShares.submit(session, { shareId, text: "Hello" }, []);
217 assert.deepEqual(yield* Ref.get(yield* RecordedAnalytics), [
218 { event: "feature.used", feature: "guest_share.join" },
219 { event: "feature.used", feature: "guest_message.submit", variant: "text_only" },
220 ]);
221 }).pipe(Effect.provide(layerTest)),
222);
223
224it.effect("streams the owner's thread events after the share and a snapshot", () =>
225 Effect.gen(function* () {
226 const { orchestrator, guestShares, credential } = yield* shareThread;
227 const { token: session } = yield* guestShares.redeem({ credential, name: "Sam" });
228 const events = yield* guestShares.stream(session);
229 yield* orchestrator.dispatch({
230 type: "message.dispatch",
231 commandId: CommandId.make("owner-message"),
232 threadId,
233 messageId: MessageId.make("message:owner"),
234 text: "Fix the login bug",
235 attachments: [],
236 dispatchMode: { type: "defer_start" },
237 createdBy: "user",
238 creationSource: "web",
239 });
240 yield* orchestrator.dispatch({
241 type: "thread.guest-share.close",
242 commandId: CommandId.make("close-after-message"),
243 threadId,
244 });
245 const received = Array.from(yield* Stream.runCollect(events));
246 assert.equal(received.at(-1)?.type, "ended");
247 const turnItems = received.flatMap((event) =>
248 event.type === "thread" &&
249 event.item.kind === "event" &&
250 event.item.event.type === "turn-item.updated"
251 ? [event.item.event.payload]
252 : [],
253 );
254 assert.deepEqual(
255 turnItems.flatMap((item) => (item.type === "user_message" ? [item.text] : [])),
256 ["Fix the login bug"],
257 );
258 }).pipe(Effect.provide(layerTest)),
259);
260
261it.effect("starts a stream with the runs and items already in the thread", () =>
262 Effect.gen(function* () {
263 const { orchestrator, guestShares, credential } = yield* shareThread;
264 const { token: session } = yield* guestShares.redeem({ credential, name: "Sam" });
265 yield* orchestrator.dispatch({
266 type: "message.dispatch",
267 commandId: CommandId.make("owner-message-before"),
268 threadId,
269 messageId: MessageId.make("message:before"),
270 text: "Fix the login bug",
271 attachments: [],
272 dispatchMode: { type: "defer_start" },
273 createdBy: "user",
274 creationSource: "web",
275 });
276 const [view, snapshot] = Array.from(
277 yield* Stream.runCollect(Stream.take(yield* guestShares.stream(session), 2)),
278 );
279 assert.equal(view?.type, "view");
280 if (snapshot?.type !== "thread" || snapshot.item.kind !== "snapshot") {
281 return assert.fail("Expected the thread snapshot second.");
282 }
283 assert.equal(snapshot.item.projection.runs.length, 1);
284 assert.include(
285 snapshot.item.projection.visibleTurnItems.map((row) => row.item.type),
286 "user_message",
287 );
288 }).pipe(Effect.provide(layerTest)),
289);
290
291it.effect("redeems only a short invite whose signature and share id are both current", () =>
292 Effect.gen(function* () {
293 const { guestShares, credential: replaced } = yield* shareThread;
294 const { url } = yield* guestShares.create({ threadId });
295 const credential = url.slice(url.indexOf("#") + 1);
296 assert.match(credential, /^[\w-]{22}\.[\w-]{22}$/);
297
298 const [shareId, signature] = credential.split(".");
299 const tampered = `${shareId}.${signature!.startsWith("A") ? "B" : "A"}${signature!.slice(1)}`;
300 for (const rejected of [tampered, replaced]) {
301 const error = yield* Effect.flip(guestShares.redeem({ credential: rejected, name: "Eve" }));
302 assert.equal(error.reason, "invalid");
303 }
304 assert.isTrue(
305 Exit.isSuccess(yield* Effect.exit(guestShares.redeem({ credential, name: "Sam" }))),
306 );
307 }).pipe(Effect.provide(layerTest)),
308);
309
310it.effect("refuses guest requests once T3 Connect no longer serves the link's host", () =>
311 Effect.gen(function* () {
312 const { guestShares, credential } = yield* shareThread;
313 const { token: session } = yield* guestShares.redeem({ credential, name: "Sam" });
314 const { id: shareId } = (yield* Orchestrator.OrchestratorV2.use((orchestrator) =>
315 orchestrator.getThreadRecords(threadId, []),
316 )).thread.guestShare!;
317 yield* Ref.set(hostname, null);
318 const submitted = yield* Effect.exit(
319 guestShares.submit(session, { shareId, text: "Hello" }, []),
320 );
321 assert.isTrue(Exit.isFailure(submitted));
322 }).pipe(Effect.provide(layerTest)),
323);
324
325it.effect("closes an open guest stream when T3 Connect stops serving the link's host", () =>
326 Effect.gen(function* () {
327 const { guestShares, credential } = yield* shareThread;
328 const { token: session } = yield* guestShares.redeem({ credential, name: "Sam" });
329 const collecting = yield* Stream.runCollect(yield* guestShares.stream(session)).pipe(
330 Effect.forkChild,
331 );
332 yield* Ref.set(hostname, null);
333 yield* TestClock.adjust("30 seconds");
334 // No `ended`: the browser reconnects, and the host check refuses it while Connect is gone.
335 const received = Array.from(yield* Fiber.join(collecting));
336 assert.deepEqual(
337 received.map((event) => (event.type === "thread" ? event.item.kind : event.type)),
338 ["view", "snapshot"],
339 );
340 assert.isTrue(Exit.isFailure(yield* Effect.exit(guestShares.stream(session))));
341 yield* Ref.set(hostname, "env.t3.test");
342 assert.isTrue(Exit.isSuccess(yield* Effect.exit(guestShares.stream(session))));
343 }).pipe(Effect.provide(layerTest)),
344);
345
346it.effect("refuses new links and every guest request while the beta is off", () =>
347 Effect.gen(function* () {
348 const { guestShares, credential } = yield* shareThread;
349 const { token: session } = yield* guestShares.redeem({ credential, name: "Sam" });
350 const collecting = yield* Stream.runCollect(yield* guestShares.stream(session)).pipe(
351 Effect.forkChild,
352 );
353 const settings = yield* ServerSettings.ServerSettingsService;
354 yield* settings.updateSettings({ guestSharingEnabled: false });
355 // An open stream closes at the next access check.
356 yield* TestClock.adjust("30 seconds");
357 yield* Fiber.join(collecting);
358 assert.equal((yield* Effect.flip(guestShares.create({ threadId }))).reason, "disabled");
359 assert.equal((yield* Effect.flip(guestShares.check(session))).reason, "inactive");
360 // Turning it back on lets the same guest back in.
361 yield* settings.updateSettings({ guestSharingEnabled: true });
362 assert.isTrue(Exit.isSuccess(yield* Effect.exit(guestShares.check(session))));
363 }).pipe(Effect.provide(layerTest)),
364);
365
366it.effect("ends an open guest stream when the link expires", () =>
367 Effect.gen(function* () {
368 const { guestShares, credential } = yield* shareThread;
369 const { token: session } = yield* guestShares.redeem({ credential, name: "Sam" });
370 const collecting = yield* Stream.runCollect(yield* guestShares.stream(session)).pipe(
371 Effect.forkChild,
372 );
373 yield* TestClock.adjust("25 hours");
374 const received = Array.from(yield* Fiber.join(collecting));
375 assert.equal(received.at(-1)?.type, "ended");
376 }).pipe(Effect.provide(layerTest)),
377);
378
379it.effect("shows the guest the thread's new model", () =>
380 Effect.gen(function* () {
381 const { orchestrator, guestShares, credential } = yield* shareThread;
382 const { token: session } = yield* guestShares.redeem({ credential, name: "Sam" });
383 const events = yield* guestShares.stream(session);
384 yield* orchestrator.dispatch({
385 type: "thread.model-selection.set",
386 commandId: CommandId.make("switch-model"),
387 threadId,
388 modelSelection: { instanceId, model: "gpt-5.2-codex" },
389 });
390 yield* orchestrator.dispatch({
391 type: "thread.guest-share.close",
392 commandId: CommandId.make("close-after-switch"),
393 threadId,
394 });
395 const views = Array.from(yield* Stream.runCollect(events)).flatMap((event) =>
396 event.type === "view" ? [event.view.model.name] : [],
397 );
398 assert.include(views, "gpt-5.2-codex");
399 }).pipe(Effect.provide(layerTest)),
400);
401
402it.effect("refuses a message a page sends for a share the cookie no longer opens", () =>
403 Effect.gen(function* () {
404 const { guestShares, credential } = yield* shareThread;
405 const { token: session } = yield* guestShares.redeem({ credential, name: "Sam" });
406 const submitted = yield* Effect.exit(
407 guestShares.submit(session, { shareId: GuestShareId.make("share:older"), text: "Hello" }, []),
408 );
409 assert.isTrue(Exit.isFailure(submitted));
410 }).pipe(Effect.provide(layerTest)),
411);
412
413it.effect("serves no render for an item outside the shared thread", () =>
414 Effect.gen(function* () {
415 const { guestShares, credential } = yield* shareThread;
416 const { token: session } = yield* guestShares.redeem({ credential, name: "Sam" });
417 assert.isTrue(Exit.isFailure(yield* Effect.exit(guestShares.render(session, "item:missing"))));
418 // The invite itself is not a session.
419 assert.isTrue(
420 Exit.isFailure(yield* Effect.exit(guestShares.render(credential, "item:missing"))),
421 );
422 }).pipe(Effect.provide(layerTest)),
423);
424
425const ownerRun = Effect.gen(function* () {
426 const orchestrator = yield* Orchestrator.OrchestratorV2;
427 yield* orchestrator.dispatch({
428 type: "message.dispatch",
429 commandId: CommandId.make("owner-message-for-command"),
430 threadId,
431 messageId: MessageId.make("message:command"),
432 text: "Build it",
433 attachments: [],
434 dispatchMode: { type: "defer_start" },
435 createdBy: "user",
436 creationSource: "web",
437 });
438 const { runs } = (yield* orchestrator.getThreadSnapshotWindow(threadId, { rowLimit: 10 }))
439 .projection;
440 return runs[0]!.id;
441});
442
443const writeItem = (
444 id: string,
445 fields:
446 | Pick<
447 Extract<OrchestrationV2TurnItem, { type: "command_execution" }>,
448 "type" | "status" | "input" | "output"
449 >
450 | Pick<
451 Extract<OrchestrationV2TurnItem, { type: "file_change" }>,
452 "type" | "status" | "fileName" | "diffStr" | "oldStr" | "newStr"
453 >
454 | Pick<
455 Extract<OrchestrationV2TurnItem, { type: "dynamic_tool" }>,
456 "type" | "status" | "toolName" | "input" | "output"
457 >,
458) =>
459 Effect.gen(function* () {
460 const runId = yield* ownerRun;
461 const eventSink = yield* EventSink.EventSinkV2;
462 const now = yield* DateTime.now;
463 yield* eventSink.write({
464 events: [
465 {
466 id: EventId.make(`event:${id}`),
467 type: "turn-item.updated",
468 threadId,
469 driver: ProviderDriverKind.make("codex"),
470 providerInstanceId: instanceId,
471 occurredAt: now,
472 payload: {
473 id: TurnItemId.make(id),
474 threadId,
475 runId,
476 nodeId: null,
477 providerThreadId: null,
478 providerTurnId: null,
479 nativeItemRef: null,
480 parentItemId: null,
481 ordinal: 1,
482 title: null,
483 startedAt: now,
484 completedAt: fields.status === "running" ? null : now,
485 updatedAt: now,
486 ...fields,
487 },
488 },
489 ],
490 });
491 return TurnItemId.make(id);
492 });
493
494it.effect("gives the guest a command's detail and output, but no file diffs", () =>
495 Effect.gen(function* () {
496 const { guestShares, credential } = yield* shareThread;
497 const { token: session } = yield* guestShares.redeem({ credential, name: "Sam" });
498 const command = yield* writeItem("item:command", {
499 type: "command_execution",
500 status: "completed",
501 input: "ls /work/src",
502 output: "/work/src/auth.ts\n",
503 });
504
505 const detail = (yield* guestShares.item(session, command)).item;
506 assert.deepInclude(detail, { input: "ls /work/src", output: "/work/src/auth.ts\n" });
507 const edit = yield* writeItem("item:edit", {
508 type: "file_change",
509 status: "completed",
510 fileName: "src/auth.ts",
511 diffStr: "-const token = 1;\n+const token = 2;\n",
512 });
513 const write = yield* writeItem("item:write", {
514 type: "file_change",
515 status: "completed",
516 fileName: "src/config.ts",
517 oldStr: "",
518 newStr: "export const key = 1;\n",
519 });
520 for (const [id, fileName] of [
521 [edit, "src/auth.ts"],
522 [write, "src/config.ts"],
523 ] as const) {
524 const change = (yield* guestShares.item(session, id)).item;
525 assert.deepInclude(change, { type: "file_change", fileName });
526 assert.doesNotHaveAnyKeys(change, ["diffStr", "oldStr", "newStr"]);
527 }
528 const frames = Array.from(
529 yield* Stream.runCollect(yield* guestShares.commandOutput(session, command)),
530 );
531 assert.deepEqual(frames, [
532 { kind: "replace", text: "/work/src/auth.ts\n", truncated: false, running: false },
533 ]);
534 // Only with a session, and only for an item of this thread.
535 const refused: ReadonlyArray<Effect.Effect<unknown, GuestShareService.GuestShareAccessError>> =
536 [
537 guestShares.item(credential, command),
538 guestShares.commandOutput(credential, command),
539 guestShares.item(session, "item:elsewhere"),
540 ];
541 for (const request of refused) {
542 assert.isTrue(Exit.isFailure(yield* Effect.exit(request)));
543 }
544 }).pipe(Effect.provide(layerTest)),
545);
546
547it.effect("gives a fork's guest the commands the fork shows from its source thread", () =>
548 Effect.gen(function* () {
549 const { orchestrator, guestShares } = yield* shareThread;
550 const command = yield* writeItem("item:command", {
551 type: "command_execution",
552 status: "completed",
553 input: "ls /work/src",
554 output: "/work/src/auth.ts\n",
555 });
556 const { attachmentsDir } = yield* ServerConfig.ServerConfig;
557 const page = createAttachmentId(threadId, "html")!;
558 yield* (yield* FileSystem.FileSystem).writeFileString(
559 resolveAttachmentRelativePath({ attachmentsDir, relativePath: `${page}.html` })!,
560 "<p>Plan</p>",
561 );
562 const render = yield* writeItem("item:render", {
563 type: "dynamic_tool",
564 status: "completed",
565 toolName: "mcp__t3-code__html_render",
566 input: {},
567 output: [
568 {
569 type: "text",
570 text: JSON.stringify({ htmlRender: { attachmentId: page, title: "Plan", height: 300 } }),
571 },
572 ],
573 });
574 const eventSink = yield* EventSink.EventSinkV2;
575 const now = yield* DateTime.now;
576 const run = (yield* orchestrator.getThreadRecords(threadId, ["runs"])).runs[0]!;
577 yield* eventSink.write({
578 events: [
579 {
580 id: EventId.make("event:run-done"),
581 type: "run.updated",
582 threadId,
583 runId: run.id,
584 providerInstanceId: instanceId,
585 occurredAt: now,
586 payload: { ...run, status: "completed", completedAt: now },
587 },
588 ],
589 });
590 const forkId = ThreadId.make("thread:guest-share-fork");
591 yield* orchestrator.dispatch({
592 type: "thread.fork",
593 commandId: CommandId.make("fork"),
594 sourceThreadId: threadId,
595 targetThreadId: forkId,
596 sourcePoint: { type: "run", runId: run.id },
597 createdBy: "user",
598 creationSource: "web",
599 });
600 const { url } = yield* guestShares.create({ threadId: forkId });
601 const { token: session } = yield* guestShares.redeem({
602 credential: url.slice(url.indexOf("#") + 1),
603 name: "Sam",
604 });
605
606 const detail = (yield* guestShares.item(session, command)).item;
607 assert.deepInclude(detail, { input: "ls /work/src", output: "/work/src/auth.ts\n" });
608 const frames = Array.from(
609 yield* Stream.runCollect(yield* guestShares.commandOutput(session, command)),
610 );
611 assert.deepEqual(frames, [
612 { kind: "replace", text: "/work/src/auth.ts\n", truncated: false, running: false },
613 ]);
614 assert.deepEqual(yield* guestShares.render(session, render), {
615 path: resolveAttachmentPathById({ attachmentsDir, attachmentId: page })!,
616 title: "Plan",
617 });
618 }).pipe(Effect.provide(layerTest)),
619);
620
621it.effect("pages the guest back through turns older than the stream's snapshot", () =>
622 Effect.gen(function* () {
623 const { guestShares, credential } = yield* shareThread;
624 const { token: session } = yield* guestShares.redeem({ credential, name: "Sam" });
625 const runId = yield* ownerRun;
626 const fileSystem = yield* FileSystem.FileSystem;
627 const directory = yield* fileSystem.makeTempDirectoryScoped();
628 const shot = `${directory}/old.png`;
629 yield* fileSystem.writeFile(shot, PNG);
630 const eventSink = yield* EventSink.EventSinkV2;
631 const now = yield* DateTime.now;
632 const base = { threadId, runId, nodeId: null, providerThreadId: null, providerTurnId: null };
633 const turnItem = (id: string, ordinal: number, payload: object) => ({
634 id: EventId.make(`event:${id}`),
635 type: "turn-item.updated" as const,
636 threadId,
637 driver: ProviderDriverKind.make("codex"),
638 providerInstanceId: instanceId,
639 occurredAt: now,
640 payload: {
641 ...base,
642 id: TurnItemId.make(id),
643 nativeItemRef: null,
644 parentItemId: null,
645 ordinal,
646 title: null,
647 status: "completed" as const,
648 startedAt: now,
649 completedAt: now,
650 updatedAt: now,
651 messageId: MessageId.make(id),
652 ...payload,
653 } as OrchestrationV2TurnItem,
654 });
655 yield* eventSink.write({
656 events: [
657 turnItem("item:old-image", 1, {
658 type: "assistant_message",
659 text: ``,
660 streaming: false,
661 }),
662 ...Array.from({ length: 30 }, (_, index) =>
663 turnItem(`item:ask-${index}`, index + 2, {
664 type: "user_message",
665 text: `Ask ${index}`,
666 createdBy: "user",
667 creationSource: "web",
668 inputIntent: "turn_start",
669 attachments: [],
670 }),
671 ),
672 ],
673 });
674
675 const [, opening] = Array.from(
676 yield* Stream.runCollect(Stream.take(yield* guestShares.stream(session), 2)),
677 );
678 if (opening?.type !== "thread" || opening.item.kind !== "snapshot") {
679 return assert.fail("Expected the thread snapshot second.");
680 }
681 const ids = (rows: ReadonlyArray<{ readonly sourceItemId: string }>) =>
682 rows.map((row) => row.sourceItemId);
683 assert.notInclude(ids(opening.item.projection.visibleTurnItems), "item:old-image");
684 let cursor = opening.item.historyCursor ?? null;
685 assert.isNotNull(cursor);
686 // The page has not shown the old reply yet, so its image does not load.
687 assert.equal((yield* Effect.flip(guestShares.media(session, shot))).reason, "unavailable");
688 // Only a session pages.
689 assert.isTrue(Exit.isFailure(yield* Effect.exit(guestShares.history(credential, cursor!))));
690
691 const older: Array<string> = [];
692 while (cursor !== null) {
693 const page: OrchestrationV2ThreadHistoryPage = yield* guestShares.history(session, cursor);
694 older.push(...ids(page.items));
695 cursor = page.hasMoreHistory ? page.nextCursor : null;
696 }
697 assert.include(older, "item:old-image");
698 assert.deepStrictEqual(yield* guestShares.media(session, shot), {
699 path: shot,
700 mimeType: "image/png",
701 });
702 }).pipe(Effect.scoped, Effect.provide(layerTest)),
703);
704
705it.effect("ends a running command's output once the owner stops sharing", () =>
706 Effect.gen(function* () {
707 const { orchestrator, guestShares, credential } = yield* shareThread;
708 const { token: session } = yield* guestShares.redeem({ credential, name: "Sam" });
709 const command = yield* writeItem("item:running", {
710 type: "command_execution",
711 status: "running",
712 input: "ls /work/src",
713 });
714 const collecting = yield* Stream.runCollect(
715 yield* guestShares.commandOutput(session, command),
716 ).pipe(Effect.forkChild);
717 yield* orchestrator.dispatch({
718 type: "thread.guest-share.close",
719 commandId: CommandId.make("close-during-command"),
720 threadId,
721 });
722 yield* TestClock.adjust("30 seconds");
723 const frames = Array.from(yield* Fiber.join(collecting));
724 assert.isTrue(frames.every((frame) => frame.running));
725 assert.isTrue(Exit.isFailure(yield* Effect.exit(guestShares.commandOutput(session, command))));
726 }).pipe(Effect.provide(layerTest)),
727);
728
729const PNG = Uint8Array.of(0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 0, 0, 0, 13);
730const MIB = 1024 * 1024;
731
732/** Files as the multipart parser leaves them: on disk, with the browser's name and type. */
733const uploadsOf = Effect.fn(function* (
734 files: ReadonlyArray<{
735 readonly name: string;
736 readonly contentType: string;
737 readonly bytes: Uint8Array;
738 }>,
739) {
740 const fileSystem = yield* FileSystem.FileSystem;
741 const directory = yield* fileSystem.makeTempDirectoryScoped();
742 return yield* Effect.forEach(files, (file, index) => {
743 const path = `${directory}/${index}`;
744 return fileSystem
745 .writeFile(path, file.bytes)
746 .pipe(Effect.as({ name: file.name, contentType: file.contentType, path }));
747 });
748});
749const png = (name: string, sizeBytes = PNG.length) => {
750 const bytes = new Uint8Array(sizeBytes);
751 bytes.set(PNG);
752 return { name, contentType: "image/png", bytes };
753};
754
755/** Every stored attachment file. */
756const storedFiles = Effect.gen(function* () {
757 const { attachmentsDir } = yield* ServerConfig.ServerConfig;
758 const fileSystem = yield* FileSystem.FileSystem;
759 return yield* fileSystem.readDirectory(attachmentsDir).pipe(Effect.orElseSucceed(() => []));
760});
761
762const joinedGuest = Effect.gen(function* () {
763 const shared = yield* shareThread;
764 const { token: session, shareId } = yield* shared.guestShares.redeem({
765 credential: shared.credential,
766 name: "Sam",
767 });
768 return { ...shared, session, shareId };
769});
770
771const waitingMessages = Effect.gen(function* () {
772 const orchestrator = yield* Orchestrator.OrchestratorV2;
773 const { thread } = yield* orchestrator.getThreadRecords(threadId, []);
774 return thread.guestShare?.messages ?? [];
775});
776
777it.effect("refuses too many files, too many bytes, and images that are not what they say", () =>
778 Effect.gen(function* () {
779 const { guestShares, session, shareId } = yield* joinedGuest;
780 const before = yield* storedFiles;
781 const refused = [
782 ["too_many_files", [1, 2, 3, 4, 5].map((index) => png(`${index}.png`))],
783 ["too_large", [png("big.png", 10 * MIB + 1)]],
784 ["unsupported_file", [{ name: "logo.svg", contentType: "image/svg+xml", bytes: PNG }]],
785 [
786 "unsupported_file",
787 [{ name: "page.png", contentType: "image/png", bytes: new TextEncoder().encode("<html>") }],
788 ],
789 [
790 "unsupported_file",
791 [{ name: "empty.txt", contentType: "text/plain", bytes: new Uint8Array() }],
792 ],
793 ] as const;
794 for (const [reason, files] of refused) {
795 const error = yield* Effect.flip(
796 guestShares.submit(session, { shareId, text: "Look" }, yield* uploadsOf(files)),
797 );
798 assert.equal(error.reason, reason);
799 }
800 // Nothing is kept from a refused message, and no message waits.
801 assert.deepEqual(yield* storedFiles, before);
802 assert.deepEqual(yield* waitingMessages, []);
803
804 yield* guestShares.submit(
805 session,
806 { shareId, text: "" },
807 yield* uploadsOf([
808 png("screen.png"),
809 // A direction override would make this name read as a PDF.
810 {
811 name: "notes\u202Efdp.html",
812 contentType: "text/html",
813 bytes: new TextEncoder().encode("<b>"),
814 },
815 {
816 name: "notes.rtf",
817 contentType: "application/rtf",
818 bytes: new TextEncoder().encode(String.raw`{\rtf1 Available}`),
819 },
820 ]),
821 );
822 const [message] = yield* waitingMessages;
823 assert.deepEqual(
824 message?.attachments?.map(({ type, name, mimeType, sizeBytes }) => ({
825 type,
826 name,
827 mimeType,
828 sizeBytes,
829 })),
830 [
831 { type: "image", name: "screen.png", mimeType: "image/png", sizeBytes: PNG.length },
832 { type: "file", name: "notes_fdp.html", mimeType: "text/html", sizeBytes: 3 },
833 { type: "file", name: "notes.rtf", mimeType: "application/rtf", sizeBytes: 17 },
834 ],
835 );
836 // Agents read a guest's documents through the same extracted text as the owner's uploads.
837 const { attachmentsDir } = yield* ServerConfig.ServerConfig;
838 const document = resolveAttachmentPath({
839 attachmentsDir,
840 attachment: message!.attachments![2]!,
841 });
842 assert.equal(
843 yield* (yield* FileSystem.FileSystem).readFileString(`${document}.txt`),
844 "Available",
845 );
846 }).pipe(Effect.scoped, Effect.provide(layerTest)),
847);
848
849it.effect("caps the bytes of files waiting in one share", () =>
850 Effect.gen(function* () {
851 const { guestShares, session, shareId } = yield* joinedGuest;
852 const submit = (sizeBytes: number, ...more: Parameters<typeof uploadsOf>[0]) =>
853 Effect.gen(function* () {
854 return guestShares.submit(
855 session,
856 { shareId, text: "Look" },
857 yield* uploadsOf([png("big.png", sizeBytes), ...more]),
858 );
859 }).pipe(Effect.flatten);
860 yield* submit(9 * MIB);
861 yield* submit(9 * MIB);
862 const before = yield* storedFiles;
863 const error = yield* Effect.flip(
864 submit(9 * MIB, {
865 name: "notes.rtf",
866 contentType: "application/rtf",
867 bytes: new TextEncoder().encode(String.raw`{\rtf1 Available}`),
868 }),
869 );
870 assert.equal(error.reason, "files_busy");
871 // The refused message keeps neither its files nor their extracted text.
872 assert.deepEqual(yield* storedFiles, before);
873 // Text alone still goes through.
874 yield* guestShares.submit(session, { shareId, text: "One more thing" }, []);
875 assert.equal((yield* waitingMessages).length, 3);
876 }).pipe(Effect.scoped, Effect.provide(layerTest)),
877);
878
879it.effect("deletes a waiting message's files when the owner dismisses it or stops sharing", () =>
880 Effect.gen(function* () {
881 const { orchestrator, guestShares, session, shareId } = yield* joinedGuest;
882 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
883 yield* guestShares.submit(
884 session,
885 { shareId, text: "First" },
886 yield* uploadsOf([png("a.png")]),
887 );
888 yield* guestShares.submit(
889 session,
890 { shareId, text: "Second" },
891 yield* uploadsOf([png("b.png")]),
892 );
893 const [first, second] = yield* waitingMessages;
894 const fileOf = (message: typeof first) => message!.attachments![0]!.id;
895 const stored = yield* storedFiles;
896 assert.isTrue(stored.some((entry) => entry.startsWith(fileOf(first))));
897
898 yield* orchestrator.dispatch({
899 type: "thread.guest-message.dismiss",
900 commandId: CommandId.make("dismiss-first"),
901 threadId,
902 guestMessageId: first!.id,
903 });
904 yield* worker.drain();
905 const afterDismiss = yield* storedFiles;
906 assert.isFalse(afterDismiss.some((entry) => entry.startsWith(fileOf(first))));
907 assert.isTrue(afterDismiss.some((entry) => entry.startsWith(fileOf(second))));
908 // A dismissed message's file is no longer the guest's to load.
909 assert.isTrue(
910 Exit.isFailure(yield* Effect.exit(guestShares.attachment(session, fileOf(first)))),
911 );
912
913 yield* orchestrator.dispatch({
914 type: "thread.guest-share.close",
915 commandId: CommandId.make("close-with-files"),
916 threadId,
917 });
918 yield* worker.drain();
919 assert.isFalse((yield* storedFiles).some((entry) => entry.startsWith(fileOf(second))));
920 }).pipe(Effect.scoped, Effect.provide(layerTest)),
921);
922
923it.effect.each(["thread.archive", "thread.delete"] as const)(
924 "deletes a waiting message's files on %s",
925 (type) =>
926 Effect.gen(function* () {
927 const { orchestrator, guestShares, session, shareId } = yield* joinedGuest;
928 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
929 const before = yield* storedFiles;
930 yield* guestShares.submit(
931 session,
932 { shareId, text: "Look" },
933 yield* uploadsOf([png("a.png")]),
934 );
935 assert.equal((yield* storedFiles).length, before.length + 1);
936 yield* orchestrator.dispatch({
937 type,
938 commandId: CommandId.make(`${type}-with-files`),
939 threadId,
940 });
941 yield* worker.drain();
942 assert.deepEqual(yield* storedFiles, before);
943 }).pipe(Effect.scoped, Effect.provide(layerTest)),
944);
945
946it.effect("sends a guest's files as the guest's message and keeps them the guest's to load", () =>
947 Effect.gen(function* () {
948 const { orchestrator, guestShares, session, shareId } = yield* joinedGuest;
949 yield* guestShares.submit(
950 session,
951 { shareId, text: "See this" },
952 yield* uploadsOf([png("a.png")]),
953 );
954 const [message] = yield* waitingMessages;
955 const pending = message!.attachments!;
956 // Served while it waits.
957 const served = yield* guestShares.attachment(session, pending[0]!.id);
958 assert.equal(served.attachment.mimeType, "image/png");
959
960 // The day-old pending-upload sweep leaves a waiting guest's files alone.
961 const { attachmentsDir } = yield* ServerConfig.ServerConfig;
962 // Any pending upload counts as stale at the end of time.
963 sweepStalePendingAttachments({ attachmentsDir, nowMs: Number.MAX_SAFE_INTEGER });
964 const copied = pending[0]!;
965 yield* orchestrator.dispatch({
966 type: "message.dispatch",
967 commandId: CommandId.make("send-guest-files"),
968 threadId,
969 messageId: MessageId.make("message:guest-files"),
970 guestMessageId: message!.id,
971 text: "",
972 attachments: [copied],
973 dispatchMode: { type: "defer_start" },
974 createdBy: "user",
975 creationSource: "web",
976 });
977 const items = (yield* orchestrator.getThreadSnapshotWindow(threadId, {
978 rowLimit: 20,
979 })).projection.visibleTurnItems.map((entry) => entry.item);
980 const sent = items.find(
981 (item) => item.type === "user_message" && item.context?.guest?.guestMessageId === message!.id,
982 );
983 assert.deepInclude(sent, { attachments: [copied] });
984 if (sent?.type !== "user_message") return;
985 assert.deepEqual(sent.context?.guest, {
986 guestMessageId: message!.id,
987 name: "Sam",
988 text: "See this",
989 shareId,
990 });
991 // The guest loads the thread's copy from their own row.
992 const fromRow = yield* guestShares.attachment(session, copied.id);
993 assert.equal(fromRow.attachment.id, copied.id);
994
995 // Still theirs after their share's message list has trimmed it away.
996 for (let index = 0; index < 21; index += 1) {
997 yield* guestShares.submit(session, { shareId, text: `note ${index}` }, []);
998 const [waiting] = (yield* waitingMessages).filter((next) => next.status === "pending");
999 yield* orchestrator.dispatch({
1000 type: "thread.guest-message.dismiss",
1001 commandId: CommandId.make(`dismiss-note-${index}`),
1002 threadId,
1003 guestMessageId: waiting!.id,
1004 });
1005 }
1006 assert.isFalse((yield* waitingMessages).some((next) => next.id === message!.id));
1007 assert.equal((yield* guestShares.attachment(session, copied.id)).attachment.id, copied.id);
1008 }).pipe(Effect.scoped, Effect.provide(layerTest)),
1009);
1010
1011it.effect("serves the shared thread's files to its guest, never another thread's", () =>
1012 Effect.gen(function* () {
1013 const { orchestrator, guestShares, session, shareId } = yield* joinedGuest;
1014 // Another thread's guest attaches a file.
1015 const other = yield* shareThreadFor(ThreadId.make("thread:other-share"));
1016 const otherGuest = yield* guestShares.redeem({ credential: other.credential, name: "Eve" });
1017 yield* guestShares.submit(
1018 otherGuest.token,
1019 { shareId: otherGuest.shareId, text: "Mine" },
1020 yield* uploadsOf([png("eve.png")]),
1021 );
1022 const { thread: otherThread } = yield* orchestrator.getThreadRecords(
1023 ThreadId.make("thread:other-share"),
1024 [],
1025 );
1026 const othersFile = otherThread.guestShare!.messages[0]!.attachments![0]!.id;
1027 assert.isTrue(
1028 Exit.isSuccess(yield* Effect.exit(guestShares.attachment(otherGuest.token, othersFile))),
1029 );
1030
1031 // The owner sends a message with an image of their own.
1032 const { attachmentsDir } = yield* ServerConfig.ServerConfig;
1033 const fileSystem = yield* FileSystem.FileSystem;
1034 const ownerImage = {
1035 type: "image" as const,
1036 id: ChatAttachmentId.make(createAttachmentId(threadId)!),
1037 name: "secret.png",
1038 mimeType: "image/png",
1039 sizeBytes: PNG.length,
1040 };
1041 yield* fileSystem.writeFile(
1042 resolveAttachmentPath({ attachmentsDir, attachment: ownerImage })!,
1043 PNG,
1044 );
1045 yield* orchestrator.dispatch({
1046 type: "message.dispatch",
1047 commandId: CommandId.make("owner-image"),
1048 threadId,
1049 messageId: MessageId.make("message:owner-image"),
1050 text: "My screen",
1051 attachments: [ownerImage],
1052 dispatchMode: { type: "defer_start" },
1053 createdBy: "user",
1054 creationSource: "web",
1055 });
1056
1057 // The guest sees the thread as the owner does, files included.
1058 assert.equal(
1059 (yield* guestShares.attachment(session, ownerImage.id)).attachment.id,
1060 ownerImage.id,
1061 );
1062 for (const id of [othersFile, "pending-missing"]) {
1063 const error = yield* Effect.flip(guestShares.attachment(session, id));
1064 assert.equal(error.reason, "unavailable");
1065 }
1066 // Nor with the other share's id on this page.
1067 assert.equal(
1068 (yield* Effect.flip(
1069 guestShares.submit(session, { shareId: otherGuest.shareId, text: "x" }, []),
1070 )).reason,
1071 "stale_page",
1072 );
1073 assert.isDefined(shareId);
1074 }).pipe(Effect.scoped, Effect.provide(layerTest)),
1075);
1076
1077it.effect("gives a fork's guest the files on messages the fork shows from its source", () =>
1078 Effect.gen(function* () {
1079 const { orchestrator } = yield* shareThread;
1080 const guestShares = yield* GuestShareService.GuestShareService;
1081 const { attachmentsDir } = yield* ServerConfig.ServerConfig;
1082 const fileSystem = yield* FileSystem.FileSystem;
1083 const ownerImage = Effect.fn(function* (name: string) {
1084 const image = {
1085 type: "image" as const,
1086 id: ChatAttachmentId.make(createAttachmentId(threadId)!),
1087 name,
1088 mimeType: "image/png",
1089 sizeBytes: PNG.length,
1090 };
1091 yield* fileSystem.writeFile(
1092 resolveAttachmentPath({ attachmentsDir, attachment: image })!,
1093 PNG,
1094 );
1095 yield* orchestrator.dispatch({
1096 type: "message.dispatch",
1097 commandId: CommandId.make(`owner:${name}`),
1098 threadId,
1099 messageId: MessageId.make(`message:${name}`),
1100 text: name,
1101 attachments: [image],
1102 dispatchMode: { type: "defer_start" },
1103 createdBy: "user",
1104 creationSource: "web",
1105 });
1106 return image;
1107 });
1108 const inherited = yield* ownerImage("before.png");
1109 const run = (yield* orchestrator.getThreadRecords(threadId, ["runs"])).runs[0]!;
1110 const now = yield* DateTime.now;
1111 yield* (yield* EventSink.EventSinkV2).write({
1112 events: [
1113 {
1114 id: EventId.make("event:run-done"),
1115 type: "run.updated",
1116 threadId,
1117 runId: run.id,
1118 providerInstanceId: instanceId,
1119 occurredAt: now,
1120 payload: { ...run, status: "completed", completedAt: now },
1121 },
1122 ],
1123 });
1124 const forkId = ThreadId.make("thread:guest-share-fork");
1125 yield* orchestrator.dispatch({
1126 type: "thread.fork",
1127 commandId: CommandId.make("fork"),
1128 sourceThreadId: threadId,
1129 targetThreadId: forkId,
1130 sourcePoint: { type: "run", runId: run.id },
1131 createdBy: "user",
1132 creationSource: "web",
1133 });
1134 const later = yield* ownerImage("after.png");
1135 const { url } = yield* guestShares.create({ threadId: forkId });
1136 const { token: session } = yield* guestShares.redeem({
1137 credential: url.slice(url.indexOf("#") + 1),
1138 name: "Sam",
1139 });
1140
1141 assert.equal(
1142 (yield* guestShares.attachment(session, inherited.id)).attachment.id,
1143 inherited.id,
1144 );
1145 // The source's messages after the fork point are not part of the fork.
1146 assert.equal(
1147 (yield* Effect.flip(guestShares.attachment(session, later.id))).reason,
1148 "unavailable",
1149 );
1150 }).pipe(Effect.provide(layerTest)),
1151);
1152
1153it.effect("serves an image on the host only when an agent reply the guest sees shows it", () =>
1154 Effect.gen(function* () {
1155 const { orchestrator, guestShares } = yield* shareThread;
1156 const runId = yield* ownerRun;
1157 const fileSystem = yield* FileSystem.FileSystem;
1158 const directory = yield* fileSystem.makeTempDirectoryScoped();
1159 const shot = `${directory}/my shot.png`;
1160 const fileUrl = `${directory}/file-url.png`;
1161 const page = `${directory}/page.png`;
1162 const fromUser = `${directory}/from-user.png`;
1163 const unnamed = `${directory}/unnamed.png`;
1164 const later = `${directory}/later.png`;
1165 for (const path of [shot, fileUrl, fromUser, unnamed, later]) {
1166 yield* fileSystem.writeFile(path, PNG);
1167 }
1168 yield* fileSystem.writeFileString(page, "<script>alert(1)</script>");
1169 const eventSink = yield* EventSink.EventSinkV2;
1170 const now = yield* DateTime.now;
1171 const run = (yield* orchestrator.getThreadRecords(threadId, ["runs"])).runs[0];
1172 const item = (
1173 id: string,
1174 ordinal: number,
1175 fields:
1176 | { readonly type: "assistant_message"; readonly text: string }
1177 | { readonly type: "user_message"; readonly text: string },
1178 target: { readonly threadId: ThreadId; readonly runId: RunId } = { threadId, runId },
1179 ) => ({
1180 id: EventId.make(`event:${id}`),
1181 type: "turn-item.updated" as const,
1182 threadId: target.threadId,
1183 driver: ProviderDriverKind.make("codex"),
1184 providerInstanceId: instanceId,
1185 occurredAt: now,
1186 payload: {
1187 id: TurnItemId.make(id),
1188 threadId: target.threadId,
1189 runId: target.runId,
1190 nodeId: null,
1191 providerThreadId: null,
1192 providerTurnId: null,
1193 nativeItemRef: null,
1194 parentItemId: null,
1195 ordinal,
1196 title: null,
1197 status: "completed" as const,
1198 startedAt: now,
1199 completedAt: now,
1200 updatedAt: now,
1201 messageId: MessageId.make(id),
1202 ...(fields.type === "assistant_message"
1203 ? { ...fields, streaming: false }
1204 : {
1205 ...fields,
1206 createdBy: "user" as const,
1207 creationSource: "web" as const,
1208 inputIntent: "turn_start" as const,
1209 attachments: [],
1210 }),
1211 },
1212 });
1213 yield* eventSink.write({
1214 events: [
1215 item("item:agent-images", 1, {
1216 type: "assistant_message",
1217 text: `Done:  }) `,
1218 }),
1219 // A guest's sent message is a user message; it must not open files on the host.
1220 item("item:guest-image", 2, { type: "user_message", text: `` }),
1221 {
1222 id: EventId.make("event:run-done"),
1223 type: "run.updated" as const,
1224 threadId,
1225 runId,
1226 providerInstanceId: instanceId,
1227 occurredAt: now,
1228 payload: { ...run!, status: "completed" as const, completedAt: now },
1229 },
1230 ],
1231 });
1232 // The guest watches a fork, which shows these replies as inherited history.
1233 const forkId = ThreadId.make("thread:guest-share-fork");
1234 yield* orchestrator.dispatch({
1235 type: "thread.fork",
1236 commandId: CommandId.make("fork"),
1237 sourceThreadId: threadId,
1238 targetThreadId: forkId,
1239 sourcePoint: { type: "run", runId },
1240 createdBy: "user",
1241 creationSource: "web",
1242 });
1243 const { url } = yield* guestShares.create({ threadId: forkId });
1244 const { token: session } = yield* guestShares.redeem({
1245 credential: url.slice(url.indexOf("#") + 1),
1246 name: "Sam",
1247 });
1248
1249 // The guest's page opens, and its stream sends the replies the fork inherits.
1250 const sent = yield* Queue.unbounded<GuestStreamEvent>();
1251 yield* Stream.runForEach(yield* guestShares.stream(session), (event) =>
1252 Queue.offer(sent, event),
1253 ).pipe(Effect.forkScoped);
1254 const sentItem = Effect.fn(function* (id: string) {
1255 for (;;) {
1256 const event = yield* Queue.take(sent);
1257 if (
1258 event.type === "thread" &&
1259 event.item.kind === "event" &&
1260 event.item.event.type === "turn-item.updated" &&
1261 event.item.event.payload.id === id
1262 ) {
1263 return;
1264 }
1265 }
1266 });
1267
1268 // The agent shows another screenshot live, then enough rows to push the first reply out of
1269 // the window, before the page asks for any image.
1270 yield* orchestrator.dispatch({
1271 type: "message.dispatch",
1272 commandId: CommandId.make("fork-message"),
1273 threadId: forkId,
1274 messageId: MessageId.make("message:fork"),
1275 text: "One more",
1276 attachments: [],
1277 dispatchMode: { type: "defer_start" },
1278 createdBy: "user",
1279 creationSource: "web",
1280 });
1281 const forkRun = (yield* orchestrator.getThreadRecords(forkId, ["runs"])).runs.at(-1)!;
1282 const inFork = { threadId: forkId, runId: forkRun.id };
1283 yield* eventSink.write({
1284 events: [
1285 item(
1286 "item:later-image",
1287 1,
1288 { type: "assistant_message", text: `` },
1289 inFork,
1290 ),
1291 ...Array.from({ length: 600 }, (_, index) =>
1292 item(
1293 `item:filler-${index}`,
1294 index + 2,
1295 { type: "assistant_message", text: "Working." },
1296 inFork,
1297 ),
1298 ),
1299 ],
1300 });
1301 yield* sentItem("item:later-image");
1302
1303 // The page still shows every reply it was sent, so each image loads.
1304 for (const path of [shot, fileUrl, later]) {
1305 assert.deepStrictEqual(yield* guestShares.media(session, path), {
1306 path,
1307 mimeType: "image/png",
1308 });
1309 }
1310 for (const path of [page, fromUser, unnamed, `${directory}/missing.png`]) {
1311 assert.equal((yield* Effect.flip(guestShares.media(session, path))).reason, "unavailable");
1312 }
1313
1314 // A reply edited to drop its image stops allowing it.
1315 yield* eventSink.write({
1316 events: [
1317 {
1318 ...item("item:later-image", 1, { type: "assistant_message", text: "Gone." }, inFork),
1319 id: EventId.make("event:later-image-edited"),
1320 },
1321 ],
1322 });
1323 yield* sentItem("item:later-image");
1324 assert.equal((yield* Effect.flip(guestShares.media(session, later))).reason, "unavailable");
1325 }).pipe(Effect.scoped, Effect.provide(layerTest)),
1326);
1327
1328it.effect("keeps a connected guest's images while other shares come and go", () =>
1329 Effect.gen(function* () {
1330 const fileSystem = yield* FileSystem.FileSystem;
1331 const directory = yield* fileSystem.makeTempDirectoryScoped();
1332 const shot = `${directory}/shot.png`;
1333 yield* fileSystem.writeFile(shot, PNG);
1334 const eventSink = yield* EventSink.EventSinkV2;
1335 const now = yield* DateTime.now;
1336 /** A shared thread whose agent reply shows the screenshot, and its guest's session. */
1337 const sharedWithImage = Effect.fn(function* (id: ThreadId) {
1338 const { orchestrator, guestShares, credential } = yield* shareThreadFor(id);
1339 yield* orchestrator.dispatch({
1340 type: "message.dispatch",
1341 commandId: CommandId.make(`message:${id}`),
1342 threadId: id,
1343 messageId: MessageId.make(`message:${id}`),
1344 text: "Show me",
1345 attachments: [],
1346 dispatchMode: { type: "defer_start" },
1347 createdBy: "user",
1348 creationSource: "web",
1349 });
1350 const runId = (yield* orchestrator.getThreadRecords(id, ["runs"])).runs[0]!.id;
1351 yield* eventSink.write({
1352 events: [
1353 {
1354 id: EventId.make(`event:reply:${id}`),
1355 type: "turn-item.updated",
1356 threadId: id,
1357 driver: ProviderDriverKind.make("codex"),
1358 providerInstanceId: instanceId,
1359 occurredAt: now,
1360 payload: {
1361 id: TurnItemId.make(`item:reply:${id}`),
1362 threadId: id,
1363 runId,
1364 nodeId: null,
1365 providerThreadId: null,
1366 providerTurnId: null,
1367 nativeItemRef: null,
1368 parentItemId: null,
1369 ordinal: 1,
1370 title: null,
1371 status: "completed",
1372 startedAt: now,
1373 completedAt: now,
1374 updatedAt: now,
1375 type: "assistant_message",
1376 messageId: MessageId.make(`item:reply:${id}`),
1377 text: ``,
1378 streaming: false,
1379 },
1380 },
1381 ],
1382 });
1383 const { token } = yield* guestShares.redeem({ credential, name: "Sam" });
1384 return { guestShares, token };
1385 });
1386
1387 // This guest's page stays open while more guests than the record keeps open and close theirs.
1388 const { guestShares, token } = yield* sharedWithImage(threadId);
1389 const opened = yield* Deferred.make<void>();
1390 yield* Stream.runForEach(yield* guestShares.stream(token), () =>
1391 Deferred.succeed(opened, undefined),
1392 ).pipe(Effect.forkScoped);
1393 yield* Deferred.await(opened);
1394 const closedTokens: Array<string> = [];
1395 for (let index = 0; index < 65; index++) {
1396 const other = yield* sharedWithImage(ThreadId.make(`thread:other-${index}`));
1397 const events = yield* guestShares.stream(other.token);
1398 // The first is built but never read, as for a HEAD request; the rest open and close.
1399 if (index > 0) yield* Stream.runDrain(Stream.take(events, 2));
1400 closedTokens.push(other.token);
1401 }
1402 assert.equal((yield* guestShares.media(token, shot)).mimeType, "image/png");
1403 // Records of guests with no open page were dropped to make room.
1404 for (const closed of closedTokens.slice(0, 2)) {
1405 assert.equal((yield* Effect.flip(guestShares.media(closed, shot))).reason, "unavailable");
1406 }
1407
1408 // A new link on the thread starts empty: its guest gets nothing the last guest was sent.
1409 const { url } = yield* guestShares.create({ threadId });
1410 const next = yield* guestShares.redeem({
1411 credential: url.slice(url.indexOf("#") + 1),
1412 name: "Kim",
1413 });
1414 assert.equal((yield* Effect.flip(guestShares.media(next.token, shot))).reason, "unavailable");
1415 yield* Stream.runDrain(Stream.take(yield* guestShares.stream(next.token), 2));
1416 assert.equal((yield* guestShares.media(next.token, shot)).mimeType, "image/png");
1417 }).pipe(Effect.scoped, Effect.provide(layerTest)),
1418);