fix: guest links can load earlier turns

apps/server/src/guest/GuestShareService.test.ts

@@ -7,6 +7,7 @@ import {
   GuestShareId,
   type GuestStreamEvent,
   MessageId,
+  type OrchestrationV2ThreadHistoryPage,
   type OrchestrationV2TurnItem,
   ProjectId,
   ProviderDriverKind,
@@ -529,6 +530,90 @@ it.effect("gives the guest a command's detail and output, but no file diffs", ()
   }).pipe(Effect.provide(layerTest)),
 );
 
+it.effect("pages the guest back through turns older than the stream's snapshot", () =>
+  Effect.gen(function* () {
+    const { guestShares, credential } = yield* shareThread;
+    const { token: session } = yield* guestShares.redeem({ credential, name: "Sam" });
+    const runId = yield* ownerRun;
+    const fileSystem = yield* FileSystem.FileSystem;
+    const directory = yield* fileSystem.makeTempDirectoryScoped();
+    const shot = `${directory}/old.png`;
+    yield* fileSystem.writeFile(shot, PNG);
+    const eventSink = yield* EventSink.EventSinkV2;
+    const now = yield* DateTime.now;
+    const base = { threadId, runId, nodeId: null, providerThreadId: null, providerTurnId: null };
+    const turnItem = (id: string, ordinal: number, payload: object) => ({
+      id: EventId.make(`event:${id}`),
+      type: "turn-item.updated" as const,
+      threadId,
+      driver: ProviderDriverKind.make("codex"),
+      providerInstanceId: instanceId,
+      occurredAt: now,
+      payload: {
+        ...base,
+        id: TurnItemId.make(id),
+        nativeItemRef: null,
+        parentItemId: null,
+        ordinal,
+        title: null,
+        status: "completed" as const,
+        startedAt: now,
+        completedAt: now,
+        updatedAt: now,
+        messageId: MessageId.make(id),
+        ...payload,
+      } as OrchestrationV2TurnItem,
+    });
+    yield* eventSink.write({
+      events: [
+        turnItem("item:old-image", 1, {
+          type: "assistant_message",
+          text: `![Old](${shot})`,
+          streaming: false,
+        }),
+        ...Array.from({ length: 30 }, (_, index) =>
+          turnItem(`item:ask-${index}`, index + 2, {
+            type: "user_message",
+            text: `Ask ${index}`,
+            createdBy: "user",
+            creationSource: "web",
+            inputIntent: "turn_start",
+            attachments: [],
+          }),
+        ),
+      ],
+    });
+
+    const [, opening] = Array.from(
+      yield* Stream.runCollect(Stream.take(yield* guestShares.stream(session), 2)),
+    );
+    if (opening?.type !== "thread" || opening.item.kind !== "snapshot") {
+      return assert.fail("Expected the thread snapshot second.");
+    }
+    const ids = (rows: ReadonlyArray<{ readonly sourceItemId: string }>) =>
+      rows.map((row) => row.sourceItemId);
+    assert.notInclude(ids(opening.item.projection.visibleTurnItems), "item:old-image");
+    let cursor = opening.item.historyCursor ?? null;
+    assert.isNotNull(cursor);
+    // The page has not shown the old reply yet, so its image does not load.
+    assert.equal((yield* Effect.flip(guestShares.media(session, shot))).reason, "unavailable");
+    // Only a session pages.
+    assert.isTrue(Exit.isFailure(yield* Effect.exit(guestShares.history(credential, cursor!))));
+
+    const older: Array<string> = [];
+    while (cursor !== null) {
+      const page: OrchestrationV2ThreadHistoryPage = yield* guestShares.history(session, cursor);
+      older.push(...ids(page.items));
+      cursor = page.hasMoreHistory ? page.nextCursor : null;
+    }
+    assert.include(older, "item:old-image");
+    assert.deepStrictEqual(yield* guestShares.media(session, shot), {
+      path: shot,
+      mimeType: "image/png",
+    });
+  }).pipe(Effect.scoped, Effect.provide(layerTest)),
+);
+
 it.effect("ends a running command's output once the owner stops sharing", () =>
   Effect.gen(function* () {
     const { orchestrator, guestShares, credential } = yield* shareThread;

apps/server/src/guest/GuestShareService.ts

@@ -15,6 +15,7 @@ import {
   type OrchestrationV2CommandOutputFrame,
   type OrchestrationV2DomainEvent,
   type OrchestrationV2GetTurnItemResult,
+  type OrchestrationV2ThreadHistoryPage,
   type OrchestrationV2TurnItem,
   ThreadId,
   TurnItemId,
@@ -185,6 +186,11 @@ export class GuestShareService extends Context.Service<
       session: string,
       itemId: string,
     ) => Effect.Effect<OrchestrationV2GetTurnItemResult, GuestShareAccessError>;
+    /** Older turns of the shared thread, as the owner's client pages them. */
+    readonly history: (
+      session: string,
+      cursor: string,
+    ) => Effect.Effect<OrchestrationV2ThreadHistoryPage, GuestShareAccessError>;
     /** A command's output frames, as the owner's expanded row gets them, until the link ends. */
     readonly commandOutput: (
       session: string,
@@ -665,6 +671,21 @@ const make = Effect.gen(function* () {
     };
   });
 
+  const history = Effect.fn("GuestShareService.history")(function* (
+    session: string,
+    cursor: string,
+  ) {
+    const { thread, share } = yield* authorize(session, "session");
+    const page = yield* orchestrator
+      .getThreadHistoryPage(thread.id, cursor)
+      .pipe(Effect.mapError(() => unavailable));
+    rememberSent(
+      share.id,
+      page.items.map((row) => row.item),
+    );
+    return page;
+  });
+
   const commandOutput = Effect.fn("GuestShareService.commandOutput")(function* (
     session: string,
     itemId: string,
@@ -739,6 +760,7 @@ const make = Effect.gen(function* () {
     stream,
     render,
     item,
+    history,
     commandOutput,
     attachment,
     media,

apps/server/src/guest/http.ts

@@ -1,6 +1,7 @@
 import {
   GUEST_ATTACHMENT_ROUTE,
   GUEST_COMMAND_OUTPUT_ROUTE,
+  GUEST_HISTORY_ROUTE,
   GUEST_ITEM_ROUTE,
   GUEST_MEDIA_ROUTE,
   GUEST_RENDER_ROUTE,
@@ -16,6 +17,7 @@ import {
   isProviderSendTurnSupportedImageMimeType,
   OrchestrationV2CommandOutputFrame,
   OrchestrationV2GetTurnItemResult,
+  OrchestrationV2ThreadHistoryPage,
 } from "@t3tools/contracts";
 import * as ByteSize from "effect/ByteSize";
 import * as Duration from "effect/Duration";
@@ -52,6 +54,7 @@ const encodeOutputFrame = Schema.encodeEffect(
   Schema.fromJsonString(OrchestrationV2CommandOutputFrame),
 );
 const encodeItem = Schema.encodeEffect(Schema.toCodecJson(OrchestrationV2GetTurnItemResult));
+const encodeHistoryPage = Schema.encodeEffect(Schema.toCodecJson(OrchestrationV2ThreadHistoryPage));
 
 const errorResponse = (error: GuestShareAccessError) =>
   HttpServerResponse.json(
@@ -292,6 +295,26 @@ const itemRoute = HttpRouter.add(
   }),
 );
 
+const historyRoute = HttpRouter.add(
+  "GET",
+  GUEST_HISTORY_ROUTE,
+  Effect.gen(function* () {
+    const request = yield* HttpServerRequest.HttpServerRequest;
+    const cursor = new URLSearchParams(request.url.split("?", 2)[1] ?? "").get("cursor");
+    if (!cursor) return yield* badRequest;
+    const guestShares = yield* GuestShareService;
+    return yield* guestShares.history(yield* sessionToken, cursor).pipe(
+      Effect.flatMap((page) =>
+        encodeHistoryPage(page).pipe(
+          Effect.mapError(() => new GuestShareAccessError({ reason: "unavailable" })),
+        ),
+      ),
+      Effect.map((body) => HttpServerResponse.jsonUnsafe(body, { headers: NO_STORE })),
+      Effect.catchTags({ GuestShareAccessError: errorResponse }),
+    );
+  }),
+);
+
 // Ends when the command settles; the page closes its EventSource on that last frame, so the
 // browser does not reconnect.
 const commandOutputRoute = HttpRouter.add(
@@ -365,6 +388,7 @@ export const layer = Layer.mergeAll(
   streamRoute,
   renderRoute,
   itemRoute,
+  historyRoute,
   commandOutputRoute,
   attachmentRoute,
   mediaRoute,

apps/web/src/components/guest/GuestPage.tsx

@@ -1,13 +1,15 @@
 import {
   EnvironmentId,
   GUEST_MESSAGE_MAX_LENGTH,
+  GUEST_HISTORY_ROUTE,
   GUEST_NAME_MAX_LENGTH,
   GUEST_ROUTE_PREFIX,
   GUEST_SHARE_PARAM,
   GuestMessageId,
   type GuestRedeemRequest,
   type GuestShareView,
   type GuestThreadModel,
+  OrchestrationV2ThreadHistoryPage,
   type OrchestrationV2ThreadProjection,
   type ThreadGuestMessage,
   type ThreadGuestMessageStatus,
@@ -46,7 +48,7 @@ import { GuestMessageAttachments } from "../chat/GuestMessageAttachments";
 import { HostImageUrlContext } from "../chat/hostImageUrl";
 import { PierreEntryIcon } from "../chat/PierreEntryIcon";
 import { GuestMessageName, guestBubbleClassName } from "../chat/guestMessageStyle";
-import { MessagesTimeline } from "../chat/MessagesTimeline";
+import { MessagesTimeline, type MessagesTimelineHistoryControls } from "../chat/MessagesTimeline";
 import { ProviderInstanceIcon } from "../chat/ProviderInstanceIcon";
 import { getDisplayModelName, getTriggerDisplayModelName } from "../chat/providerIconUtils";
 import { ScrollToEndButton } from "../chat/TimelineOverlays";
@@ -131,6 +133,32 @@ function takeGuestEntry(): GuestEntry {
 }
 
 const decodeErrorBody = Schema.decodeUnknownOption(Schema.Struct({ message: Schema.String }));
+const decodeHistoryPage = Schema.decodeUnknownOption(
+  Schema.toCodecJson(OrchestrationV2ThreadHistoryPage),
+);
+
+/** Resolves to the page, or to a message the guest can read. */
+async function fetchGuestHistoryPage(
+  shareId: string,
+  cursor: string,
+): Promise<OrchestrationV2ThreadHistoryPage | string> {
+  const failed = "Could not load earlier turns. Try again.";
+  try {
+    const response = await fetch(
+      `${GUEST_HISTORY_ROUTE}?${GUEST_SHARE_PARAM}=${encodeURIComponent(shareId)}&cursor=${encodeURIComponent(cursor)}`,
+    );
+    const json: unknown = await response.json().catch(() => null);
+    if (!response.ok) {
+      return Option.match(decodeErrorBody(json), {
+        onNone: () => failed,
+        onSome: ({ message }) => message,
+      });
+    }
+    return Option.getOrElse(decodeHistoryPage(json), () => failed);
+  } catch {
+    return "Could not reach the server.";
+  }
+}
 
 /** Resolves to null on success, or to a message the guest can read. */
 async function postGuest(
@@ -241,6 +269,32 @@ function GuestSession({ shareId }: { readonly shareId: string }) {
   // missing, invalid, or ended session. Network drops reconnect on their own.
   const [rejected, setRejected] = useState(false);
 
+  const [historyLoad, setHistoryLoad] = useState<{
+    readonly cursor: string;
+    readonly error: string | null;
+  } | null>(null);
+  const historyCursor = stream.historyCursor;
+  // A load for an older cursor is stale: a page or a reconnect snapshot moved the cursor on.
+  const currentLoad = historyLoad?.cursor === historyCursor ? historyLoad : null;
+  const loadingHistory = currentLoad !== null && currentLoad.error === null;
+  const historyError = currentLoad?.error ?? null;
+  const loadEarlier = async () => {
+    if (historyCursor === null || loadingHistory) return;
+    setHistoryLoad({ cursor: historyCursor, error: null });
+    const result = await fetchGuestHistoryPage(shareId, historyCursor);
+    if (typeof result === "string") setHistoryLoad({ cursor: historyCursor, error: result });
+    else dispatch({ type: "history-page", cursor: historyCursor, page: result });
+  };
+  const historyControls: MessagesTimelineHistoryControls | undefined =
+    historyCursor === null
+      ? undefined
+      : {
+          hasMoreHistory: true,
+          loading: loadingHistory,
+          error: historyError,
+          onLoadEarlier: () => void loadEarlier(),
+        };
+
   // oxlint-disable-next-line t3code/no-unreasoned-state-sync -- sets document.title to the shared thread's title
   useEffect(() => {
     if (stream.view) document.title = stream.view.title;
@@ -273,6 +327,7 @@ function GuestSession({ shareId }: { readonly shareId: string }) {
         view={stream.view}
         projection={stream.projection}
         awaitingRowIds={stream.awaitingRowIds}
+        historyControls={historyControls}
       />
     );
   }
@@ -358,10 +413,12 @@ function GuestThread({
   view: { shareId, title, expiresAt, messages, model },
   projection,
   awaitingRowIds,
+  historyControls,
 }: {
   readonly view: GuestShareView;
   readonly projection: OrchestrationV2ThreadProjection;
   readonly awaitingRowIds: GuestStreamState["awaitingRowIds"];
+  readonly historyControls: MessagesTimelineHistoryControls | undefined;
 }) {
   // No ":" so the timeline finds no environment thread behind the key.
   const threadKey = `guest-${shareId}`;
@@ -467,6 +524,7 @@ function GuestThread({
             onFileDownload={openFile}
             renderHtmlRender={renderHtmlRender}
             turnItemSource={turnItemSource}
+            historyControls={historyControls}
             activeThreadEnvironmentId={GUEST_ENVIRONMENT_ID}
             markdownCwd={undefined}
             resolvedTheme={resolvedTheme}

apps/web/src/components/guest/guestStream.logic.test.ts

@@ -112,6 +112,45 @@ describe("applyGuestStreamEvent", () => {
     expect(apply([itemEvent(2, reply("a1", "Changed"))], once)).toBe(once);
   });
 
+  it("merges older pages in front, only for the cursor the thread is at", () => {
+    const windowed: GuestStreamEvent = {
+      type: "thread",
+      item: {
+        kind: "snapshot",
+        snapshotSequence: 1,
+        projection,
+        historyCursor: "cursor-1",
+        hasMoreHistory: true,
+      },
+    };
+    const page = (id: string, nextCursor: string | null) => ({
+      snapshotSequence: 1,
+      items: [
+        {
+          position: 0,
+          visibility: "local" as const,
+          sourceThreadId: projection.thread.id,
+          sourceItemId: TurnItemId.make(id),
+          item: reply(id, "Earlier"),
+        },
+      ],
+      nextCursor,
+      hasMoreHistory: nextCursor !== null,
+    });
+    const live = apply([windowed, itemEvent(2, reply("a1", "Hi"))]);
+    expect(live.historyCursor).toBe("cursor-1");
+    // A page fetched before a reconnect moved the cursor changes nothing.
+    const stale = { type: "history-page" as const, cursor: "cursor-0", page: page("old", null) };
+    expect(applyGuestStreamEvent(live, stale)).toBe(live);
+    const merged = applyGuestStreamEvent(live, {
+      type: "history-page",
+      cursor: "cursor-1",
+      page: page("o1", null),
+    });
+    expect(itemIds(merged)).toEqual(["o1", "a1"]);
+    expect(merged.historyCursor).toBeNull();
+  });
+
   it("replaces the thread on a new snapshot, as after a reconnect", () => {
     const before = apply([snapshot(1), itemEvent(2, reply("a1", "Hi")), { type: "ended" }]);
     const after = apply([snapshot(1)], { ...before, ended: false });