diff --git a/apps/server/src/guest/GuestShareService.test.ts b/apps/server/src/guest/GuestShareService.test.ts index 1ef25744c8c55..ac5ffc76aae77 100644 --- a/apps/server/src/guest/GuestShareService.test.ts +++ b/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 = []; + 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; diff --git a/apps/server/src/guest/GuestShareService.ts b/apps/server/src/guest/GuestShareService.ts index 347bdc11ece5b..2a1bf853507ba 100644 --- a/apps/server/src/guest/GuestShareService.ts +++ b/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; + /** Older turns of the shared thread, as the owner's client pages them. */ + readonly history: ( + session: string, + cursor: string, + ) => Effect.Effect; /** 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, diff --git a/apps/server/src/guest/http.ts b/apps/server/src/guest/http.ts index bf368ac884d92..b3e856200c845 100644 --- a/apps/server/src/guest/http.ts +++ b/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, diff --git a/apps/web/src/components/guest/GuestPage.tsx b/apps/web/src/components/guest/GuestPage.tsx index 21c35447ddf8c..7e41e9a8ca297 100644 --- a/apps/web/src/components/guest/GuestPage.tsx +++ b/apps/web/src/components/guest/GuestPage.tsx @@ -1,6 +1,7 @@ import { EnvironmentId, GUEST_MESSAGE_MAX_LENGTH, + GUEST_HISTORY_ROUTE, GUEST_NAME_MAX_LENGTH, GUEST_ROUTE_PREFIX, GUEST_SHARE_PARAM, @@ -8,6 +9,7 @@ import { 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 { + 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} diff --git a/apps/web/src/components/guest/guestStream.logic.test.ts b/apps/web/src/components/guest/guestStream.logic.test.ts index fbdf9ce1927c2..674227dc171ff 100644 --- a/apps/web/src/components/guest/guestStream.logic.test.ts +++ b/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 }); diff --git a/apps/web/src/components/guest/guestStream.logic.ts b/apps/web/src/components/guest/guestStream.logic.ts index 0b37d66eeec84..03eafb6d348ee 100644 --- a/apps/web/src/components/guest/guestStream.logic.ts +++ b/apps/web/src/components/guest/guestStream.logic.ts @@ -9,11 +9,13 @@ import { type GuestMessageId, type GuestShareView, type OrchestrationV2ProjectedTurnItem, + type OrchestrationV2ThreadHistoryPage, type OrchestrationV2ThreadProjection, type OrchestrationV2TurnItem, type ThreadGuestMessage, } from "@t3tools/contracts"; import { applyOrchestrationV2ProjectionEvent } from "@t3tools/client-runtime/state/orchestration-v2-projection"; +import { mergeOlderHistoryIntoProjection } from "@t3tools/client-runtime/state/thread-history-merge"; import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; @@ -25,6 +27,8 @@ export interface GuestStreamState { readonly sequence: number; /** Set while the snapshot is a window of a longer history, as the owner's client tracks it. */ readonly partial: { readonly latestLocalTurnOrdinal: number | null } | null; + /** Where "Load earlier turns" continues, or null once the whole thread is loaded. */ + readonly historyCursor: string | null; readonly ended: boolean; /** * Messages this page saw waiting that have no transcript row yet. Survives reconnects, and a @@ -38,6 +42,7 @@ export const initialGuestStreamState: GuestStreamState = { projection: null, sequence: 0, partial: null, + historyCursor: null, ended: false, awaitingRowIds: new Set(), }; @@ -76,16 +81,32 @@ const decodeGuestStreamEvent = Schema.decodeUnknownOption( export const parseGuestStreamEvent = (data: string): GuestStreamEvent | null => Option.getOrNull(decodeGuestStreamEvent(data)); +/** A page of older turns, fetched from `cursor`. */ +export type GuestHistoryPageLoaded = { + readonly type: "history-page"; + readonly cursor: string; + readonly page: OrchestrationV2ThreadHistoryPage; +}; + /** * Applies the owner's thread stream the way the owner's client does (see * EnvironmentThreadState in client-runtime): a snapshot replaces the thread, and later events - * pass through the same projection reducer, once each. + * pass through the same projection reducer, once each. Older pages merge in front, as they do in + * the owner's view. */ export function applyGuestStreamEvent( state: GuestStreamState, - event: GuestStreamEvent, + event: GuestStreamEvent | GuestHistoryPageLoaded, ): GuestStreamState { switch (event.type) { + case "history-page": + // A snapshot since the request replaced the thread and its cursor. + if (state.projection === null || state.historyCursor !== event.cursor) return state; + return { + ...state, + projection: mergeOlderHistoryIntoProjection(state.projection, event.page.items), + historyCursor: event.page.hasMoreHistory ? event.page.nextCursor : null, + }; case "view": // The page follows the share it names; a view of another one means the link changed. if (state.view !== null && state.view.shareId !== event.view.shareId) { @@ -108,6 +129,7 @@ export function applyGuestStreamEvent( item.hasMoreHistory === true || (item.historyCursor ?? null) !== null ? { latestLocalTurnOrdinal: item.latestLocalTurnOrdinal ?? null } : null, + historyCursor: item.hasMoreHistory === true ? (item.historyCursor ?? null) : null, awaitingRowIds: withoutPlaced( state.awaitingRowIds, projection.visibleTurnItems.map((row) => row.item), diff --git a/packages/client-runtime/package.json b/packages/client-runtime/package.json index e1f81638ea3d4..9600ad7ad356d 100644 --- a/packages/client-runtime/package.json +++ b/packages/client-runtime/package.json @@ -203,6 +203,10 @@ "types": "./src/state/orchestration.ts", "default": "./src/state/orchestration.ts" }, + "./state/thread-history-merge": { + "types": "./src/state/threadHistoryMerge.ts", + "default": "./src/state/threadHistoryMerge.ts" + }, "./state/orchestration-v2-projection": { "types": "./src/state/orchestrationV2Projection.ts", "default": "./src/state/orchestrationV2Projection.ts" diff --git a/packages/contracts/src/guestShare.ts b/packages/contracts/src/guestShare.ts index fd991fe1e4c3b..715a837daeae3 100644 --- a/packages/contracts/src/guestShare.ts +++ b/packages/contracts/src/guestShare.ts @@ -129,6 +129,11 @@ export const guestSessionCookieName = (shareId: string) => `t3_guest_${shareId}` export const GUEST_RENDER_ROUTE = `${GUEST_ROUTE_PREFIX}/render`; /** One item of the shared thread, as an expanded row loads it: `?share=&item=`. */ export const GUEST_ITEM_ROUTE = `${GUEST_ROUTE_PREFIX}/item`; +/** + * Older turns of the shared thread, as the owner's "Load earlier turns" pages them: + * `?share=&cursor=`. + */ +export const GUEST_HISTORY_ROUTE = `${GUEST_ROUTE_PREFIX}/history`; /** Server-sent output frames of one command in the shared thread: `?share=&item=`. */ export const GUEST_COMMAND_OUTPUT_ROUTE = `${GUEST_ROUTE_PREFIX}/command-output`; /**