From be66f94baea206e4395c14399ba04cf923b7de17 Mon Sep 17 00:00:00 2001 From: Renfei Zhou Date: Fri, 13 Nov 2026 16:48:32 +0000 Subject: [PATCH 1/2] test(fork): display modes must open on shown messages in bot-heavy threads In Messages only, Messages + Dashboard and Chat + Replies, the server fills a thread's first window and its older pages by Normal rules. A thread where a bot ran many turns opens on rows the mode hides, so the pane shows only "Load earlier turns", and the Replies list drops replies whose first message is outside the window. These tests ask for a shown-only view: the subscription flag `shownOnly`, the HTTP bounded snapshot query `shownOnly=1` and the capability `threadShownHistory`, added here as optional contract fields. They check that the window holds the latest shown messages and the first messages of its replies, that older pages hold only shown rows, forks and targeted pages, and that a shown client state never uses the saved Normal snapshot and loads one page per request. Each fails on main. Co-Authored-By: Claude Opus 5.5 --- .../fork/shownHistory/ShownHistory.test.ts | 1023 +++++++++++++++++ .../src/fork/shownOnlyThread.test.ts | 279 +++++ packages/contracts/src/environment.ts | 2 + packages/contracts/src/environmentHttp.ts | 1 + packages/contracts/src/orchestrationV2.ts | 2 + 5 files changed, 1307 insertions(+) create mode 100644 apps/server/src/fork/shownHistory/ShownHistory.test.ts create mode 100644 packages/client-runtime/src/fork/shownOnlyThread.test.ts diff --git a/apps/server/src/fork/shownHistory/ShownHistory.test.ts b/apps/server/src/fork/shownHistory/ShownHistory.test.ts new file mode 100644 index 0000000000000..300916e148192 --- /dev/null +++ b/apps/server/src/fork/shownHistory/ShownHistory.test.ts @@ -0,0 +1,1023 @@ +import { assert, describe, it } from "@effect/vitest"; +import { + AuthOrchestrationReadScope, + AuthSessionId, + ComposerContextId, + EnvironmentAuthenticatedAuth, + EnvironmentAuthenticatedPrincipal, + EnvironmentHttpApi, + EnvironmentId, + EventId, + MessageId, + NodeId, + ORCHESTRATION_PROTOCOL_HEADER, + ORCHESTRATION_PROTOCOL_VERSION, + OrchestrationV2ThreadBoundedSnapshot, + OrchestrationV2ThreadHistoryPage, + OrchestrationV2ThreadStreamItem, + type OrchestrationV2DomainEvent, + type OrchestrationV2SubscribeThreadInput, + type OrchestrationV2ThreadProjection, + type OrchestrationV2TurnItem, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + RunId, + RuntimeRequestId, + ScheduledTaskId, + ThreadId, + TurnItemId, +} from "@t3tools/contracts"; +import { + displayEventFromTurnItem, + foldCores, + shortId, + SUB_THREAD_RECORD_KIND, +} from "@t3tools/shared/displayModes"; +import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot"; +import { resolveT3McpToolId } from "@t3tools/shared/t3McpToolPresentation"; +import * as NodeHttpPlatform from "@effect/platform-node/NodeHttpPlatform"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as HttpApi from "effect/http-api/HttpApi"; +import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder"; +import { Etag, HttpRouter } from "effect/http"; + +import * as ServerEnvironment from "../../environment/ServerEnvironment.ts"; +import * as OrchestrationHttp from "../../orchestration-v2/http.ts"; +import * as ProjectionStore from "../../orchestration-v2/ProjectionStore.ts"; +import * as ProjectStore from "../../orchestration-v2/ProjectStore.ts"; +import * as ThreadManagementService from "../../orchestration-v2/ThreadManagementService.ts"; +import * as OrchestrationEventStore from "../../persistence/OrchestrationEventStore.ts"; +import * as SqlitePersistence from "../../persistence/Sqlite.ts"; +import * as ProjectEnrichmentService from "../../project/ProjectEnrichmentService.ts"; +import { subscribeOrchestrationV2Thread } from "../../ws.ts"; + +const decodeBounded = Schema.decodeUnknownSync( + Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadBoundedSnapshot)), +); +const decodeHistoryPage = Schema.decodeUnknownSync( + Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadHistoryPage)), +); +const decodeStreamItem = Schema.decodeUnknownSync( + Schema.toCodecJson(OrchestrationV2ThreadStreamItem), +); +const encodeStreamItem = Schema.encodeSync(Schema.toCodecJson(OrchestrationV2ThreadStreamItem)); + +const at = DateTime.makeUnsafe("2026-11-13T00:00:00.000Z"); +const provider = ProviderInstanceId.make("codex"); +const modelSelection = { instanceId: provider, model: "gpt-5.4" }; +const projectId = ProjectId.make("project:shown-history"); + +const BOT = ThreadId.make("thread:shown:bot"); +const PAGED = ThreadId.make("thread:shown:paged"); +const REPLIES = ThreadId.make("thread:shown:replies"); +const PARENT = ThreadId.make("thread:shown:parent"); +const CHILD_SHORT = ThreadId.make("thread:shown:child-short"); +const CHILD_LONG = ThreadId.make("thread:shown:child-long"); +const CHILD_EDGE = ThreadId.make("thread:shown:child-edge"); +const NAMES = ThreadId.make("thread:shown:names"); +const GUARD = ThreadId.make("thread:shown:guard"); +const QUIET = ThreadId.make("thread:shown:quiet"); +const HIDDEN = ThreadId.make("thread:shown:hidden"); +const UNANSWERED = ThreadId.make("thread:shown:unanswered"); +const LARGE = ThreadId.make("thread:shown:large"); +const REWRITES = ThreadId.make("thread:shown:rewrites"); + +const runIdOf = (threadId: ThreadId, run: number) => RunId.make(`run:${threadId}:${run}`); +const nodeIdOf = (threadId: ThreadId, run: number) => NodeId.make(`node:${threadId}:${run}`); +const itemIdOf = (threadId: ThreadId, run: number, k: number) => + TurnItemId.make(`item:${threadId}:${run}:${k}`); +const messageIdOf = (threadId: ThreadId, run: number, k: number) => + MessageId.make(`message:${threadId}:${run}:${k}`); + +function threadCreated( + id: ThreadId, + forkOf?: { readonly threadId: ThreadId; readonly runId: RunId }, +): OrchestrationV2DomainEvent { + return { + id: EventId.make(`event:${id}:created`), + type: "thread.created", + threadId: id, + occurredAt: at, + payload: { + createdBy: "user", + creationSource: "web", + id, + projectId, + title: String(id), + providerInstanceId: provider, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: + forkOf === undefined + ? { parentThreadId: null, relationshipToParent: null, rootThreadId: id } + : { + parentThreadId: forkOf.threadId, + relationshipToParent: "fork", + rootThreadId: forkOf.threadId, + }, + forkedFrom: forkOf === undefined ? null : { type: "run", ...forkOf }, + createdAt: at, + updatedAt: at, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }; +} + +function runEvent( + threadId: ThreadId, + run: number, + status: "completed" | "failed" | "rolled_back" = "completed", +): OrchestrationV2DomainEvent { + const runId = runIdOf(threadId, run); + return { + id: EventId.make(`event:${runId}:${status}`), + type: status === "completed" ? "run.created" : "run.updated", + threadId, + runId, + nodeId: nodeIdOf(threadId, run), + driver: ProviderDriverKind.make("codex"), + occurredAt: at, + payload: { + id: runId, + threadId, + ordinal: run, + providerInstanceId: provider, + modelSelection, + providerThreadId: null, + userMessageId: messageIdOf(threadId, run, 1), + rootNodeId: nodeIdOf(threadId, run), + activeAttemptId: null, + status, + requestedAt: at, + startedAt: at, + completedAt: at, + checkpointId: null, + contextHandoffId: null, + }, + }; +} + +const base = ( + threadId: ThreadId, + run: number, + k: number, + status: OrchestrationV2TurnItem["status"] = "completed", +) => ({ + id: itemIdOf(threadId, run, k), + threadId, + runId: runIdOf(threadId, run), + nodeId: nodeIdOf(threadId, run), + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + // TurnItemPositionStore numbers a run's items from run * 1_000_000. + ordinal: run * 1_000_000 + k, + status, + title: null, + startedAt: at, + completedAt: at, + updatedAt: at, +}); + +type Origin = Pick< + Extract, + "createdBy" | "creationSource" | "scheduledTaskId" | "senderThreadId" | "idleCompaction" +>; +const TYPED: Origin = { createdBy: "user", creationSource: "web" }; + +const userMessage = ( + threadId: ThreadId, + run: number, + k: number, + text: string, + origin: Origin = TYPED, + under?: string, +): OrchestrationV2TurnItem => ({ + ...base(threadId, run, k), + ...origin, + type: "user_message", + messageId: messageIdOf(threadId, run, k), + inputIntent: "turn_start", + text, + attachments: [], + ...(under === undefined + ? {} + : { + context: { + version: 1 as const, + records: [ + { + version: 1, + contextId: ComposerContextId.make("fork-reply"), + kind: SUB_THREAD_RECORD_KIND, + label: "Reply", + payload: { first: under }, + }, + ], + }, + }), +}); + +const send = ( + threadId: ThreadId, + run: number, + k: number, + text: string, + options: { + readonly replyTo?: string; + readonly toolName?: string; + readonly status?: OrchestrationV2TurnItem["status"]; + readonly output?: unknown; + } = {}, +): OrchestrationV2TurnItem => ({ + ...base(threadId, run, k, options.status), + type: "dynamic_tool", + toolName: options.toolName ?? "mcp__t3-code__send_to_user", + input: { text, ...(options.replyTo === undefined ? {} : { reply_to: options.replyTo }) }, + ...(options.output === undefined ? {} : { output: options.output }), +}); + +const rewrite = ( + threadId: ThreadId, + run: number, + k: number, + toolName: string, + replyTo?: string, +): OrchestrationV2TurnItem => ({ + ...base(threadId, run, k), + type: "dynamic_tool", + toolName, + input: { + user_message: "", + old_text: "", + new_text: replyTo === undefined ? "Rewritten." : `Progress ${run}.`, + ...(replyTo === undefined ? {} : { reply_to: replyTo }), + }, +}); + +const hidden = ( + threadId: ThreadId, + run: number, + k: number, + status: OrchestrationV2TurnItem["status"] = "completed", +): OrchestrationV2TurnItem => ({ + ...base(threadId, run, k, status), + type: "command_execution", + input: `check ${k}`, + output: "ok", + exitCode: 0, +}); + +const itemEvent = (item: OrchestrationV2TurnItem): OrchestrationV2DomainEvent => ({ + id: EventId.make(`event:${item.id}`), + type: "turn-item.updated", + threadId: item.threadId, + ...(item.runId === null ? {} : { runId: item.runId }), + occurredAt: at, + payload: item, +}); + +const range = (from: number, to: number) => + Array.from({ length: to - from + 1 }, (_, index) => from + index); + +const TYPED_FIRST = itemIdOf(BOT, 1, 1); +const SCHEDULED = itemIdOf(BOT, 15, 23); + +/** A typed message, then 30 runs a bot started, with sends in three of them, then 200 hidden rows. */ +function botThread(): OrchestrationV2TurnItem[] { + const items = [ + userMessage(BOT, 1, 1, "Please run the checks and tell me when something needs me."), + ...range(2, 6).map((k) => hidden(BOT, 1, k)), + send(BOT, 1, 7, "On it."), + ]; + for (const run of range(2, 31)) { + items.push( + userMessage(BOT, run, 1, `Bot check ${run}.`, { createdBy: "agent", creationSource: "mcp" }), + ); + items.push(...range(2, 21).map((k) => hidden(BOT, run, k))); + if (run % 10 === 0) items.push(send(BOT, run, 22, `Check ${run} found nothing.`)); + if (run === 15) + items.push( + userMessage(BOT, run, 23, "Nightly task.", { + createdBy: "user", + creationSource: "web", + scheduledTaskId: ScheduledTaskId.make("task:nightly"), + }), + ); + } + items.push( + userMessage(BOT, 32, 1, "Bot check 32.", { createdBy: "agent", creationSource: "mcp" }), + ); + items.push(...range(2, 201).map((k) => hidden(BOT, 32, k))); + return items; +} + +/** A typed message, then 300 sends in 30 bot runs, three hidden rows after each. */ +function pagedThread(): OrchestrationV2TurnItem[] { + const items = [userMessage(PAGED, 1, 1, "Start the long job.")]; + for (const run of range(2, 31)) { + items.push( + userMessage(PAGED, run, 1, `Bot step ${run}.`, { createdBy: "agent", creationSource: "mcp" }), + ); + for (const j of range(0, 9)) { + items.push(send(PAGED, run, 2 + 4 * j, `Step ${run}.${j} done.`)); + items.push(...range(3 + 4 * j, 5 + 4 * j).map((k) => hidden(PAGED, run, k))); + } + } + return items; +} + +const FIRST = messageIdOf(REPLIES, 1, 1); +const REPLY_ONE = messageIdOf(REPLIES, 17, 1); +const LONG_TEXT = "The nightly job finished; ".repeat(35); + +/** + * A first message, 15 typed turns with 150 long sends, then two replies under the first message + * and an agent reply to the first of them. + */ +function repliesThread(): OrchestrationV2TurnItem[] { + const items = [userMessage(REPLIES, 1, 1, "What failed overnight?")]; + for (const run of range(2, 16)) { + items.push(userMessage(REPLIES, run, 1, `Check batch ${run}.`)); + for (const j of range(0, 9)) { + items.push(send(REPLIES, run, 2 + 2 * j, `${LONG_TEXT}${run}.${j}`)); + items.push(hidden(REPLIES, run, 3 + 2 * j)); + } + } + items.push(userMessage(REPLIES, 17, 1, "Any failures in the last run?", TYPED, FIRST)); + items.push(send(REPLIES, 17, 2, "None in the last run.", { replyTo: shortId(REPLY_ONE) })); + items.push(userMessage(REPLIES, 17, 3, "Also check the nightly job.", TYPED, FIRST)); + items.push(...range(4, 10).map((k) => hidden(REPLIES, 17, k))); + return items; +} + +const FORK_RUN = 8; +const ROLLED_BACK_RUN = 5; + +/** Ten sends in each of runs 1 to 8, the fork run, and three in each of runs 9 and 10 after it. */ +function parentThread(): OrchestrationV2TurnItem[] { + return range(1, 10).flatMap((run) => [ + ...range(1, run <= FORK_RUN ? 10 : 3).map((k) => send(PARENT, run, k, `Parent ${run}.${k}`)), + hidden(PARENT, run, 11), + ]); +} + +const childThread = (threadId: ThreadId, runs: number, sends: number) => + range(1, runs).flatMap((run) => [ + ...range(1, sends).map((k) => send(threadId, run, k, `Child ${run}.${k}`)), + hidden(threadId, run, sends + 1), + ]); + +const SEND_NAMES = [ + "send_to_user", + "replace_send_to_user", + "mcp__t3-code__send_to_user", + "mcp__t3_code__replace_send_to_user", + "mcp__t3code__send_to_user", + "t3-code.send_to_user", + "t3_code:send_to_user", + "t3code/replace_send_to_user", + "t3-code · send_to_user", + "t3-code-direct.send_to_user", + "mcp-t3-code-send_to_user", + "t3-code-thread_7_send_to_user", + "send_to_user completed", + "mcp__t3-code__replace_send_to_user complete", +] as const; + +const UNTYPED: ReadonlyArray = [ + { createdBy: "agent", creationSource: "mcp" }, + { createdBy: "user", creationSource: "server" }, + { createdBy: "user", creationSource: "web", scheduledTaskId: ScheduledTaskId.make("task:n") }, + { createdBy: "user", creationSource: "web", senderThreadId: ThreadId.make("thread:sender") }, + { createdBy: "user", creationSource: "mobile", idleCompaction: "onReturn" }, +]; + +/** Every send and rewrite name form, then user messages nobody typed, then 100 hidden rows. */ +function namesThread(): OrchestrationV2TurnItem[] { + return [ + userMessage(NAMES, 1, 1, "Report in every way you can."), + ...SEND_NAMES.flatMap((toolName, index) => [ + toolName.includes("replace") + ? rewrite(NAMES, 2, 2 * index + 1, toolName) + : send(NAMES, 2, 2 * index + 1, `Report ${index}.`, { toolName }), + hidden(NAMES, 2, 2 * index + 2), + ]), + ...UNTYPED.flatMap((origin, index) => [ + userMessage(NAMES, 3, 2 * index + 1, `Untyped ${index}.`, origin), + hidden(NAMES, 3, 2 * index + 2), + ]), + ...range(1, 100).map((k) => hidden(NAMES, 4, k)), + ]; +} + +const QUESTION = RuntimeRequestId.make("request:shown:question"); + +/** + * A pending question in run 1, a typed message and a send in run 2, and in the failed run 3 an + * error, a running command and more hidden rows after them. + */ +function guardThread(): OrchestrationV2TurnItem[] { + return [ + { + ...base(GUARD, 1, 1, "running"), + type: "user_input_request", + requestId: QUESTION, + questions: [{ id: "scope", header: "Scope", question: "Which scope?", options: [] }], + }, + userMessage(GUARD, 2, 1, "Deploy when ready."), + send(GUARD, 2, 2, "Deploying."), + { + ...base(GUARD, 3, 1, "failed"), + type: "error", + failure: { + class: "provider_error", + message: "Provider crashed.", + code: null, + retryable: false, + }, + }, + hidden(GUARD, 3, 2, "running"), + ...range(3, 5).map((k) => hidden(GUARD, 3, k)), + ]; +} + +/** Three failed runs, each with an error and work rows, and no message to show. */ +const quietThread = () => + range(1, 3).flatMap((run): OrchestrationV2TurnItem[] => [ + { + ...base(QUIET, run, 1, "failed"), + type: "error", + failure: { + class: "provider_error", + message: `Run ${run} failed.`, + code: null, + retryable: false, + }, + }, + ...range(2, 4).map((k) => hidden(QUIET, run, k)), + ]); + +const REFUSED = { content: [{ type: "text", text: "Too long" }], isError: true }; +const UNSENT = ["failed", "cancelled", "interrupted"] as const; +const RUNNING_SEND = itemIdOf(HIDDEN, 3, 641); +const QUEUED = itemIdOf(HIDDEN, 3, 642); + +/** + * One exchange between sends that show nothing: failed ones before it, then 640 refused, failed, + * cancelled or interrupted ones, a send still running and a queued message that has not started. + */ +const hiddenThread = (): OrchestrationV2TurnItem[] => [ + ...range(1, 3).map((k) => send(HIDDEN, 1, k, `Lost ${k}.`, { status: "failed" })), + userMessage(HIDDEN, 2, 1, "Is the deploy done?"), + send(HIDDEN, 2, 2, "Yes, it is live."), + ...range(1, 640).map((k) => + k % 4 === 0 + ? send(HIDDEN, 3, k, `Refused ${k}.`, { output: REFUSED }) + : send(HIDDEN, 3, k, `Unsent ${k}.`, { status: UNSENT[(k % 4) - 1]! }), + ), + send(HIDDEN, 3, 641, "Still sending.", { status: "running" }), + { + ...base(HIDDEN, 3, 642, "pending"), + ...TYPED, + type: "user_message", + messageId: messageIdOf(HIDDEN, 3, 642), + inputIntent: "queued_turn", + text: "Then roll back the canary.", + attachments: [], + }, +]; + +/** A first message, 300 sends, then an agent reply whose reply_to names no message. */ +const unansweredThread = () => [ + userMessage(UNANSWERED, 1, 1, "Watch the queue."), + ...range(2, 31).flatMap((run) => + range(1, 10).map((k) => send(UNANSWERED, run, k, `Queue ${run}.${k} is fine.`)), + ), + send(UNANSWERED, 32, 1, "Answering a message that is not there.", { replyTo: "zzzzzz" }), +]; + +const REWRITTEN_ANSWER = shortId(messageIdOf(REWRITES, 1, 1)); + +/** A typed message and its answer, then 200 bot runs that each rewrite that answer. */ +const rewritesThread = () => [ + userMessage(REWRITES, 1, 1, "Tell me how the long job goes."), + send(REWRITES, 1, 2, "Started.", { replyTo: REWRITTEN_ANSWER }), + ...range(2, 201).map((run) => + rewrite(REWRITES, run, 1, "mcp__t3-code__replace_send_to_user", REWRITTEN_ANSWER), + ), +]; + +const LARGE_TEXT = "A full report line. ".repeat(512); + +/** A typed message and 150 sends of about 10 KiB each, one per run. */ +const largeThread = () => [ + userMessage(LARGE, 1, 1, "Send the full reports."), + ...range(2, 151).map((run) => send(LARGE, run, 1, `${LARGE_TEXT}${run}`)), +]; + +/** Forks of PARENT: label, thread, runs, sends per run, and the thread the window's cursor is in. */ +const CHILDREN = [ + ["a child with fewer shown rows than the window holds", CHILD_SHORT, 9, 10, PARENT], + ["a child with more shown rows than the window holds", CHILD_LONG, 13, 10, CHILD_LONG], + // 128 rows: the walk's fourth batch both meets the window's needs and ends the child. + ["a child whose shown rows end on a batch boundary", CHILD_EDGE, 16, 8, PARENT], +] as const; + +const runsOf = (items: ReadonlyArray) => [ + ...new Set(items.map((item) => Number(String(item.runId).split(":").at(-1)))), +]; + +const seed = Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const threads: ReadonlyArray< + readonly [ThreadId, ReadonlyArray, ReturnType] + > = [ + [BOT, botThread(), threadCreated(BOT)], + [PAGED, pagedThread(), threadCreated(PAGED)], + [REPLIES, repliesThread(), threadCreated(REPLIES)], + [PARENT, parentThread(), threadCreated(PARENT)], + ...CHILDREN.map( + ([, child, runs, sends]) => + [ + child, + childThread(child, runs, sends), + threadCreated(child, { threadId: PARENT, runId: runIdOf(PARENT, FORK_RUN) }), + ] as const, + ), + [NAMES, namesThread(), threadCreated(NAMES)], + [GUARD, guardThread(), threadCreated(GUARD)], + [QUIET, quietThread(), threadCreated(QUIET)], + [HIDDEN, hiddenThread(), threadCreated(HIDDEN)], + [UNANSWERED, unansweredThread(), threadCreated(UNANSWERED)], + [LARGE, largeThread(), threadCreated(LARGE)], + [REWRITES, rewritesThread(), threadCreated(REWRITES)], + ]; + for (const [threadId, items, created] of threads) { + yield* projections.apply(created); + for (const run of runsOf(items)) yield* projections.apply(runEvent(threadId, run)); + for (const item of items) yield* projections.apply(itemEvent(item)); + } + yield* projections.apply(runEvent(PARENT, ROLLED_BACK_RUN, "rolled_back")); + yield* projections.apply(runEvent(GUARD, 3, "failed")); + for (const run of range(1, 3)) yield* projections.apply(runEvent(QUIET, run, "failed")); + yield* projections.apply({ + id: EventId.make("event:shown:question"), + type: "runtime-request.updated", + threadId: GUARD, + nodeId: nodeIdOf(GUARD, 1), + occurredAt: at, + payload: { + id: QUESTION, + nodeId: nodeIdOf(GUARD, 1), + providerTurnId: null, + nativeRequestRef: null, + kind: "user_input", + status: "pending", + responseCapability: { type: "message" }, + createdAt: at, + resolvedAt: null, + }, + }); +}); + +const store = Layer.mergeAll( + ProjectionStore.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)), + SqlitePersistence.layerMemory, +); +const management = Layer.unwrap( + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + return Layer.mock(ThreadManagementService.ThreadManagementService)({ + ensureLegacyTranscript: () => Effect.void, + getThreadSnapshot: (id) => projections.getThreadSnapshot(id).pipe(Effect.orDie), + getThreadSnapshotWindow: (id, options) => + projections.getThreadSnapshotWindow(id, options).pipe(Effect.orDie), + getThreadHistoryPage: (id, cursor, throughEntryId, conversationOnly, workMode) => + projections + .getThreadHistoryPage(id, cursor, throughEntryId, conversationOnly, workMode) + .pipe(Effect.orDie), + streamStoredEventsFrom: () => Stream.empty, + }); + }), +); + +class OrchestrationApi extends HttpApi.make("environment").add( + EnvironmentHttpApi.groups.orchestration, +) {} +const auth = Layer.succeed(EnvironmentAuthenticatedAuth, (effect) => + effect.pipe( + Effect.provideService(EnvironmentAuthenticatedPrincipal, { + sessionId: AuthSessionId.make("shown-history"), + subject: "shown-history", + method: "browser-session-cookie", + scopes: new Set([AuthOrchestrationReadScope]), + }), + ), +); + +const TestLayer = Layer.mergeAll( + management, + Layer.succeed(ServerEnvironment.ServerEnvironmentIdentity, { + getEnvironmentId: Effect.succeed(EnvironmentId.make("shown-history")), + }), + Layer.effectDiscard(seed), + Layer.mock(OrchestrationEventStore.OrchestrationEventStore)({}), + Layer.mock(ProjectStore.ProjectStoreV2)({}), + Layer.mock(ProjectEnrichmentService.ProjectEnrichmentService)({}), +).pipe(Layer.provideMerge(store)); + +const withHttp = (use: (get: (path: string) => Promise) => Promise) => + Effect.gen(function* () { + const context = yield* Effect.context>(); + const appLayer = HttpApiBuilder.layer(OrchestrationApi).pipe( + Layer.provide(OrchestrationHttp.layer), + Layer.provide(auth), + Layer.provide(Layer.succeedContext(context)), + Layer.provide(NodeHttpPlatform.layer), + Layer.provide(Etag.layerWeak), + Layer.provide(NodeServices.layer), + ); + return yield* Effect.acquireUseRelease( + Effect.sync(() => HttpRouter.toWebHandler(appLayer, { disableLogger: true })), + ({ handler }) => + Effect.promise(() => + use((path) => + handler( + new Request(`http://environment.test${path}`, { + headers: { + [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION), + }, + }), + ), + ), + ), + ({ dispose }) => Effect.promise(() => dispose()), + ); + }); + +/** The first snapshot of a subscription, as the client decodes it. */ +const firstSnapshot = (input: OrchestrationV2SubscribeThreadInput) => + subscribeOrchestrationV2Thread(input).pipe( + Effect.flatMap((stream) => Stream.runHead(stream)), + Effect.map((head) => { + if (head._tag !== "Some" || head.value.kind !== "snapshot") throw new Error("No snapshot"); + const decoded = decodeStreamItem(JSON.parse(JSON.stringify(encodeStreamItem(head.value)))); + if (decoded.kind !== "snapshot") throw new Error("No snapshot"); + return decoded; + }), + ); + +const shownSnapshot = (threadId: ThreadId) => + firstSnapshot({ threadId, acceptBoundedSnapshot: true, shownOnly: true }); + +const cursorRow = (cursor: string | null | undefined) => + cursor == null ? null : String(JSON.parse(Buffer.from(cursor, "base64url").toString("utf8")).si); +const cursorThread = (cursor: string | null | undefined) => + cursor == null ? null : String(JSON.parse(Buffer.from(cursor, "base64url").toString("utf8")).st); + +const historyPath = (threadId: ThreadId, cursor: string, throughEntryId?: string) => + `/api/orchestration/threads/${encodeURIComponent(threadId)}/history?cursor=${encodeURIComponent(cursor)}${throughEntryId === undefined ? "" : `&throughEntryId=${encodeURIComponent(throughEntryId)}`}`; + +const fetchPage = (get: (path: string) => Promise, path: string) => + get(path).then(async (response) => { + assert.strictEqual(response.status, 200); + return decodeHistoryPage(await response.text()); + }); + +/** Every older page from `cursor` on, one request each. */ +const allPages = (threadId: ThreadId, cursor: string | null | undefined) => + withHttp(async (get) => { + const pages: Array> = []; + for (let next = cursor ?? null; next !== null;) { + const page = await fetchPage(get, historyPath(threadId, next)); + pages.push(page); + next = page.nextCursor; + } + return pages; + }); + +const visibleIds = (projection: OrchestrationV2ThreadProjection) => + projection.visibleTurnItems.map((row) => String(row.sourceItemId)); + +const isShownItem = (item: OrchestrationV2TurnItem) => { + const event = displayEventFromTurnItem(item); + return event?.type === "user" || event?.type === "sent" || event?.type === "rewrite"; +}; + +/** The ids of `items` that a display mode shows, oldest first. */ +const shownIdsOf = (items: ReadonlyArray) => + items.filter(isShownItem).map((item) => String(item.id)); + +it.layer(TestLayer)("shown-only history", (it) => { + it.effect("a shown subscription opens a bot-heavy thread on its messages", () => + Effect.gen(function* () { + const snapshot = yield* shownSnapshot(BOT); + const projection = boundedSnapshotProjection(snapshot); + const shown = projection.visibleTurnItems.filter((row) => isShownItem(row.item)); + + assert.deepStrictEqual( + shown.map((row) => String(row.sourceItemId)), + shownIdsOf(botThread()), + ); + assert.include(visibleIds(projection), String(TYPED_FIRST)); + assert.notInclude(visibleIds(projection), String(SCHEDULED)); + assert.isFalse(snapshot.hasMoreHistory); + assert.isNull(snapshot.historyCursor); + }), + ); + + it.effect("older shown pages hold only shown rows, in order, and end history", () => + Effect.gen(function* () { + const snapshot = yield* shownSnapshot(PAGED); + const window = boundedSnapshotProjection(snapshot).visibleTurnItems.filter((row) => + isShownItem(row.item), + ); + assert.isTrue(snapshot.hasMoreHistory); + const pages = yield* allPages(PAGED, snapshot.historyCursor); + + for (const page of pages) { + assert.isTrue(page.items.every((row) => isShownItem(row.item))); + } + for (const page of pages.slice(0, -1)) { + assert.isAtLeast(page.items.length, 100, "a page that is not the last holds 100 messages"); + } + assert.isFalse(pages.at(-1)!.hasMoreHistory); + assert.isNull(pages.at(-1)!.nextCursor); + const loaded = [...pages.toReversed().flatMap((page) => page.items), ...window].map((row) => + String(row.sourceItemId), + ); + assert.deepStrictEqual(loaded, shownIdsOf(pagedThread())); + }), + ); + + it.effect("the window brings the first message of every loaded reply", () => + Effect.gen(function* () { + const projection = boundedSnapshotProjection(yield* shownSnapshot(REPLIES)); + const cores = foldCores( + projection.visibleTurnItems.flatMap((row) => { + const event = displayEventFromTurnItem(row.item); + return event === null ? [] : [event]; + }), + ); + const loaded = new Set(cores.map((core) => core.id)); + const replies = cores.filter((core) => core.reply); + + assert.strictEqual(replies.length, 3); + assert.isTrue(replies.every((core) => core.first === FIRST)); + assert.isTrue(loaded.has(FIRST)); + }), + ); + + describe.each(CHILDREN)("%s", (_, child, runs, sends, cursorIn) => { + it.effect("reads its parent up to the fork run, rolled-back runs included", () => + Effect.gen(function* () { + const snapshot = yield* shownSnapshot(child); + assert.strictEqual(cursorThread(snapshot.historyCursor), String(cursorIn)); + const window = boundedSnapshotProjection(snapshot).visibleTurnItems.filter((row) => + isShownItem(row.item), + ); + const pages = yield* allPages(child, snapshot.historyCursor); + const loaded = [...pages.toReversed().flatMap((page) => page.items), ...window]; + + const inherited = parentThread().filter((item) => runOrdinalOf(item) <= FORK_RUN); + assert.deepStrictEqual( + loaded.map((row) => String(row.sourceItemId)), + [...shownIdsOf(inherited), ...shownIdsOf(childThread(child, runs, sends))], + ); + assert.isTrue( + loaded.every( + (row) => (row.sourceThreadId === PARENT) === (row.visibility === "inherited"), + ), + ); + assert.isTrue(loaded.some((row) => runOrdinalOf(row.item) === ROLLED_BACK_RUN)); + assert.isFalse(pages.at(-1)?.hasMoreHistory ?? snapshot.hasMoreHistory); + }), + ); + }); + + it.effect( + "a targeted page reads in one request until it has passed its target, and no further", + () => + Effect.gen(function* () { + const snapshot = yield* shownSnapshot(PAGED); + const cursor = snapshot.historyCursor ?? ""; + const inWindow = new Set(visibleIds(boundedSnapshotProjection(snapshot))); + const sendFarBack = shownIdsOf(pagedThread()).at(-280)!; + const workRow = itemIdOf(PAGED, 6, 3); + const pages = yield* withHttp(async (get) => ({ + plain: await fetchPage(get, historyPath(PAGED, cursor)), + send: await fetchPage(get, historyPath(PAGED, cursor, sendFarBack)), + work: await fetchPage(get, historyPath(PAGED, cursor, workRow)), + missing: await fetchPage(get, historyPath(PAGED, cursor, "item:in-no-thread")), + })); + const idsOf = (page: typeof pages.plain) => + page.items.map((row) => String(row.sourceItemId)); + + assert.notInclude(idsOf(pages.plain), sendFarBack); + assert.include(idsOf(pages.send), sendFarBack); + assert.includeMembers( + idsOf(pages.work), + pagedThread() + .filter( + (item) => isShownItem(item) && item.ordinal > 6_000_003 && !inWindow.has(item.id), + ) + .map((item) => String(item.id)), + "every shown row between the cursor and a work row", + ); + assert.isTrue(pages.work.hasMoreHistory, "the walk stops once it has passed the work row"); + assert.deepStrictEqual(pages.missing, pages.plain, "a row in no thread is no target"); + for (const page of Object.values(pages)) { + assert.isTrue(page.items.every((row) => isShownItem(row.item))); + } + }), + ); + + it.effect("rows that show nothing are neither sent, counted, nor a cursor", () => + Effect.gen(function* () { + const snapshot = yield* shownSnapshot(HIDDEN); + const projection = boundedSnapshotProjection(snapshot); + const items = projection.turnItems.map((item) => String(item.id)); + + assert.deepStrictEqual( + projection.visibleTurnItems + .filter((row) => isShownItem(row.item)) + .map((row) => String(row.sourceItemId)), + shownIdsOf(hiddenThread()), + ); + assert.isFalse(snapshot.hasMoreHistory); + assert.isNull(snapshot.historyCursor); + assert.include(items, String(RUNNING_SEND), "the running send"); + assert.include(items, String(QUEUED), "the queued message"); + }), + ); + + it.effect("an agent reply to no message stops needing one after the window's messages", () => + Effect.gen(function* () { + const snapshot = yield* shownSnapshot(UNANSWERED); + const cores = foldCores( + boundedSnapshotProjection(snapshot).visibleTurnItems.flatMap((row) => { + const event = displayEventFromTurnItem(row.item); + return event === null ? [] : [event]; + }), + ); + + assert.isTrue(cores.at(-1)?.reply, "the reply is in the window"); + assert.isBelow(cores.length, 200, "the window does not walk on for the reply"); + assert.isTrue(snapshot.hasMoreHistory); + }), + ); + + it.effect("older rewrites of a message a newer rewrite sets do not fill the window", () => + Effect.gen(function* () { + const snapshot = yield* shownSnapshot(REWRITES); + const items = boundedSnapshotProjection(snapshot).visibleTurnItems; + const cores = foldCores( + items.flatMap((row) => { + const event = displayEventFromTurnItem(row.item); + return event === null ? [] : [event]; + }), + ); + + assert.deepStrictEqual( + cores.map((core) => core.text), + ["Tell me how the long job goes.", "Progress 201."], + ); + assert.isFalse(snapshot.hasMoreHistory); + }), + ); + + it.effect("a window stopped by its byte cap pages on from the first row it left out", () => + Effect.gen(function* () { + const snapshot = yield* shownSnapshot(LARGE); + const window = boundedSnapshotProjection(snapshot).visibleTurnItems.filter((row) => + isShownItem(row.item), + ); + const shown = shownIdsOf(largeThread()); + const pages = yield* allPages(LARGE, snapshot.historyCursor); + + assert.isBelow(window.length, 120, "the cap, not the message count, ends the window"); + assert.strictEqual(cursorRow(snapshot.historyCursor), shown.at(-window.length - 1)); + assert.isAbove(pages.length, 1); + assert.deepStrictEqual( + [...pages.toReversed().flatMap((page) => page.items), ...window].map((row) => + String(row.sourceItemId), + ), + shown, + ); + }), + ); + + it.effect("HTTP shownOnly returns the socket snapshot's rows and cursor", () => + Effect.gen(function* () { + const flags = { compactTurnItems: true, dedupeBodies: true, compactIds: true }; + const socket = yield* firstSnapshot({ + threadId: PAGED, + acceptBoundedSnapshot: true, + acceptDeferredWork: true, + acceptCompactTurnItems: flags.compactTurnItems, + acceptDedupedBodies: flags.dedupeBodies, + acceptCompactIds: flags.compactIds, + shownOnly: true, + }); + const http = yield* withHttp(async (get) => + decodeBounded( + await ( + await get( + `/api/orchestration/threads/${encodeURIComponent(PAGED)}/bounded?compactTurnItems=1&dedupeBodies=1&compactIds=1&deferWork=1&shownOnly=1`, + ) + ).text(), + ), + ); + const socketRows = boundedSnapshotProjection(socket).visibleTurnItems; + + assert.deepStrictEqual( + visibleIds(boundedSnapshotProjection(http)), + socketRows.map((row) => String(row.sourceItemId)), + ); + assert.strictEqual(http.historyCursor, socket.historyCursor); + assert.strictEqual(http.hasMoreHistory, socket.hasMoreHistory); + assert.strictEqual(http.latestLocalTurnOrdinal, socket.latestLocalTurnOrdinal); + assert.isTrue( + socketRows.slice(0, -1).every((row) => isShownItem(row.item)), + "every row but the newest is a shown row", + ); + }), + ); + + it.effect( + "a send or rewrite under any accepted tool name is shown; untyped messages are not", + () => + Effect.gen(function* () { + const ids = visibleIds(boundedSnapshotProjection(yield* shownSnapshot(NAMES))); + const items = namesThread(); + + for (const [index, toolName] of SEND_NAMES.entries()) { + assert.isNotNull(resolveT3McpToolId(toolName), toolName); + assert.include(ids, String(itemIdOf(NAMES, 2, 2 * index + 1)), toolName); + } + for (const item of items.filter((item) => runOrdinalOf(item) === 3)) { + if (item.type === "user_message") assert.notInclude(ids, String(item.id)); + } + }), + ); + + it.effect("the shown window keeps what the screen needs beside the messages", () => + Effect.gen(function* () { + const snapshot = yield* shownSnapshot(GUARD); + const projection = boundedSnapshotProjection(snapshot); + const ids = projection.turnItems.map((item) => String(item.id)); + + assert.include(ids, String(itemIdOf(GUARD, 1, 1)), "the pending question card"); + assert.include(ids, String(itemIdOf(GUARD, 3, 1)), "the failed error"); + assert.include(ids, String(itemIdOf(GUARD, 3, 2)), "the running command"); + assert.include(ids, String(itemIdOf(GUARD, 3, 5)), "the newest row"); + assert.strictEqual(snapshot.latestLocalTurnOrdinal, 3_000_005); + }), + ); + + it.effect("a window with no shown row of its own keeps only the latest run's errors", () => + Effect.gen(function* () { + const ids = boundedSnapshotProjection(yield* shownSnapshot(QUIET)).turnItems.map((item) => + String(item.id), + ); + + assert.include(ids, String(itemIdOf(QUIET, 3, 1))); + assert.notInclude(ids, String(itemIdOf(QUIET, 2, 1))); + assert.notInclude(ids, String(itemIdOf(QUIET, 1, 1))); + }), + ); + + it.effect("a subscription without shownOnly keeps the Normal window and pages", () => + Effect.gen(function* () { + const snapshot = yield* firstSnapshot({ threadId: BOT, acceptBoundedSnapshot: true }); + const cursor = snapshot.historyCursor ?? ""; + const page = yield* withHttp((get) => fetchPage(get, historyPath(BOT, cursor))); + + assert.isTrue( + boundedSnapshotProjection(snapshot).visibleTurnItems.some((row) => !isShownItem(row.item)), + ); + assert.notProperty(JSON.parse(Buffer.from(cursor, "base64url").toString("utf8")), "fv"); + assert.isTrue(page.items.some((row) => !isShownItem(row.item))); + }), + ); +}); + +function runOrdinalOf(item: Pick): number { + return Math.floor(item.ordinal / 1_000_000); +} diff --git a/packages/client-runtime/src/fork/shownOnlyThread.test.ts b/packages/client-runtime/src/fork/shownOnlyThread.test.ts new file mode 100644 index 0000000000000..d4c3cb7339c26 --- /dev/null +++ b/packages/client-runtime/src/fork/shownOnlyThread.test.ts @@ -0,0 +1,279 @@ +import { + EnvironmentId, + ORCHESTRATION_V2_WS_METHODS, + type OrchestrationV2SubscribeThreadInput, + type OrchestrationV2ThreadHistoryPage, + type OrchestrationV2ThreadProjection, + type OrchestrationV2ThreadStreamItem, + type ThreadId, + TurnItemId, +} from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import { HttpClient, HttpClientResponse } from "effect/http"; +import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; +import * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; +import * as TestClock from "effect/testing/TestClock"; + +import { + AVAILABLE_CONNECTION_STATE, + type PreparedConnection, + PrimaryConnectionTarget, +} from "../connection/model.ts"; +import * as EnvironmentSupervisor from "../connection/supervisor.ts"; +import * as Persistence from "../platform/persistence.ts"; +import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; +import type { RpcSession } from "../rpc/session.ts"; +import { v2Projection, v2ThreadId } from "../state/orchestrationV2TestFixtures.ts"; +import * as ThreadHistoryController from "../state/threadHistoryController.ts"; +import { makeEnvironmentThreadState } from "../state/threads.ts"; +import * as ThreadSnapshotLoader from "../state/threadSnapshotHttp.ts"; + +const TARGET = new PrimaryConnectionTarget({ + environmentId: EnvironmentId.make("environment-1"), + label: "Test environment", + httpBaseUrl: "https://environment.example.test", + wsBaseUrl: "wss://environment.example.test", +}); +const THREAD_ID = v2ThreadId; +const SAVED: OrchestrationV2ThreadProjection = { + ...v2Projection, + thread: { ...v2Projection.thread, title: "Saved Normal view" }, +}; +const SHOWN: OrchestrationV2ThreadProjection = { + ...v2Projection, + thread: { ...v2Projection.thread, title: "Shown view" }, +}; + +// The view is the last argument of makeEnvironmentThreadState, which the base call does not take. +type MakeThreadState = ( + threadId: ThreadId, + resumeCache?: undefined, + shownOnly?: boolean, +) => ReturnType; + +const makeHarness = Effect.fn("TestShownOnlyThread.makeHarness")(function* (shownOnly: boolean) { + const subscriptions = yield* Queue.unbounded<{ + readonly input: OrchestrationV2SubscribeThreadInput; + readonly events: Queue.Queue; + }>(); + const olderLoads = yield* Queue.unbounded<{ + readonly url: URL; + readonly response: Deferred.Deferred; + }>(); + const cache = { loads: 0, saves: 0 }; + const client = { + [ORCHESTRATION_V2_WS_METHODS.subscribeThread]: (input: OrchestrationV2SubscribeThreadInput) => + Stream.unwrap( + Effect.gen(function* () { + const events = yield* Queue.unbounded(); + yield* Queue.offer(subscriptions, { input, events }); + return Stream.fromQueue(events); + }), + ), + } as unknown as WsRpcProtocolClient; + const session: RpcSession = { + client, + initialConfig: Effect.succeed({ threadSnapshotPagination: true } as never), + subscribeServerConfig: (input = {}) => client.subscribeServerConfig(input), + ready: Effect.void, + probe: Effect.void, + inboundFrames: Effect.succeed(0), + closed: Effect.never, + }; + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE), + session: yield* SubscriptionRef.make(Option.some(session)), + prepared: yield* SubscriptionRef.make>( + Option.some({ + environmentId: TARGET.environmentId, + label: TARGET.label, + httpBaseUrl: TARGET.httpBaseUrl, + socketUrl: TARGET.wsBaseUrl, + httpAuthorization: null, + target: TARGET, + }), + ), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + }); + const store = Persistence.EnvironmentCacheStore.of({ + loadShell: () => Effect.succeedNone, + saveShell: () => Effect.void, + loadThread: () => + Effect.sync(() => { + cache.loads += 1; + return Option.some({ snapshotSequence: 7, projection: SAVED }); + }), + saveThread: () => + Effect.sync(() => { + cache.saves += 1; + }), + removeThread: () => Effect.void, + loadServerConfig: () => Effect.succeedNone, + saveServerConfig: () => Effect.void, + loadVcsRefs: () => Effect.succeedNone, + saveVcsRefs: () => Effect.void, + removeVcsRefs: () => Effect.void, + clearVcsRefs: () => Effect.void, + clear: () => Effect.void, + }); + const historyController = yield* Effect.service( + ThreadHistoryController.ThreadHistoryController, + ).pipe(Effect.provide(ThreadHistoryController.layer)); + const snapshotLoader = ThreadSnapshotLoader.ThreadSnapshotLoader.of({ + load: () => Effect.die("Unexpected HTTP snapshot"), + }); + const make: MakeThreadState = makeEnvironmentThreadState; + const scope = yield* Scope.make(); + const threadState = yield* make(THREAD_ID, undefined, shownOnly).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, store), + Effect.provideService(ThreadSnapshotLoader.ThreadSnapshotLoader, snapshotLoader), + Effect.provideService(ThreadHistoryController.ThreadHistoryController, historyController), + Effect.provideService( + HttpClient.HttpClient, + HttpClient.make((request, url) => + Effect.gen(function* () { + const response = yield* Deferred.make(); + yield* Queue.offer(olderLoads, { url, response }); + return HttpClientResponse.fromWeb( + request, + Response.json(yield* Deferred.await(response)), + ); + }), + ), + ), + Scope.provide(scope), + ); + const subscription = yield* Queue.take(subscriptions); + yield* Queue.offer(subscription.events, { + kind: "snapshot", + snapshotSequence: 12, + projection: SHOWN, + historyCursor: "older-1", + hasMoreHistory: true, + latestLocalTurnOrdinal: 4, + }); + yield* Queue.offer(subscription.events, { kind: "synchronized" }); + yield* SubscriptionRef.changes(threadState).pipe( + Stream.filter((state) => state.status === "live"), + Stream.runHead, + ); + return { + input: subscription.input, + cache, + olderLoads, + threadState, + close: Scope.close(scope, Exit.void), + loadEarlier: (throughEntryId?: string) => + historyController.loadEarlier(TARGET.environmentId, THREAD_ID, throughEntryId), + }; +}); + +const sendRow = (key: string) => { + const item = { + id: TurnItemId.make(`older-send-${key}`), + type: "dynamic_tool" as const, + threadId: THREAD_ID, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed" as const, + title: null, + toolName: "mcp__t3-code__send_to_user", + input: { text: `Older ${key}.` }, + startedAt: SHOWN.thread.createdAt, + completedAt: SHOWN.thread.createdAt, + updatedAt: SHOWN.thread.createdAt, + }; + return { + position: 0, + visibility: "local" as const, + sourceThreadId: THREAD_ID, + sourceItemId: item.id, + item, + }; +}; + +/** One request's result, or "another request" when the load asks the server again. */ +const answerOnce = ( + harness: Effect.Success>, + load: Fiber.Fiber, + key: string, +) => + Effect.gen(function* () { + const request = yield* Queue.take(harness.olderLoads); + // A page with no message the user typed and more history after it, which Normal pages past. + yield* Deferred.succeed(request.response, { + snapshotSequence: 12, + items: [sendRow(key)], + nextCursor: `older-after-${key}`, + hasMoreHistory: true, + }); + const outcome = yield* Effect.raceFirst( + Fiber.join(load).pipe(Effect.as("loaded" as const)), + Queue.take(harness.olderLoads).pipe(Effect.as("another request" as const)), + ); + return { url: request.url, outcome }; + }); + +describe("shown-only thread state", () => { + it.effect.each([ + ["Normal", false, { loads: 1, saves: 1, title: "Shown view" }], + ["shown-only", true, { loads: 0, saves: 0, title: "Shown view" }], + ] as const)("a %s state and the saved thread snapshot", ([, shownOnly, expected]) => + Effect.gen(function* () { + const harness = yield* makeHarness(shownOnly); + expect(harness.input.shownOnly).toBe(shownOnly ? true : undefined); + expect(harness.input.afterSequence).toBe(shownOnly ? undefined : 7); + + yield* TestClock.adjust("500 millis"); + yield* harness.close; + const state = yield* SubscriptionRef.get(harness.threadState); + expect({ + ...harness.cache, + title: Option.getOrThrow(state.data).thread.title, + }).toEqual(expected); + }), + ); + + it.effect("a shown-only state loads one page per request, a targeted one too", () => + Effect.gen(function* () { + const harness = yield* makeHarness(true); + + const plain = yield* answerOnce( + harness, + yield* harness.loadEarlier().pipe(Effect.forkScoped), + "1", + ); + expect(plain.outcome).toBe("loaded"); + expect(plain.url.searchParams.get("cursor")).toBe("older-1"); + + const targeted = yield* answerOnce( + harness, + yield* harness.loadEarlier("send-far-back").pipe(Effect.forkScoped), + "2", + ); + expect(targeted.outcome).toBe("loaded"); + expect(targeted.url.searchParams.get("view")).toBeNull(); + expect(targeted.url.searchParams.get("throughEntryId")).toBe("send-far-back"); + expect((yield* SubscriptionRef.get(harness.threadState)).history.historyCursor).toBe( + "older-after-2", + ); + yield* harness.close; + }), + ); +}); diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 0b0f862e4962e..0564cc2caf0c0 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -121,6 +121,8 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ usagePriceOverrides: Schema.optionalKey(Schema.Boolean), usageModelAliases: Schema.optionalKey(Schema.Boolean), threadDisplayModes: Schema.optionalKey(Schema.Boolean), + /** Honors `shownOnly` on thread subscriptions and bounded snapshots. */ + threadShownHistory: Schema.optionalKey(Schema.Boolean), threadPinning: Schema.optionalKey(Schema.Boolean), threadPinReorder: Schema.optionalKey(Schema.Boolean), threadActiveReorder: Schema.optionalKey(Schema.Boolean), diff --git a/packages/contracts/src/environmentHttp.ts b/packages/contracts/src/environmentHttp.ts index db0bd2daae610..a638ae4c1297f 100644 --- a/packages/contracts/src/environmentHttp.ts +++ b/packages/contracts/src/environmentHttp.ts @@ -691,6 +691,7 @@ const EnvironmentOrchestrationThreadBoundedSnapshotQuery = Schema.Struct({ compactTurnItems: Schema.optionalKey(Schema.String), dedupeBodies: Schema.optionalKey(Schema.String), compactIds: Schema.optionalKey(Schema.String), + shownOnly: Schema.optionalKey(Schema.String), }); const EnvironmentOrchestrationThreadHistoryQuery = Schema.Struct({ diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 3fba070ac3206..c6daa251e6a43 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3855,6 +3855,8 @@ export const OrchestrationV2SubscribeThreadInput = Schema.Struct({ /** The client restores `*Omitted` bodies from records it already holds. */ acceptDedupedBodies: Schema.optionalKey(Schema.Boolean), acceptCompactIds: Schema.optionalKey(Schema.Boolean), + /** Fork: a bounded snapshot and its history pages hold only what a display mode shows. */ + shownOnly: Schema.optionalKey(Schema.Boolean), }); export type OrchestrationV2SubscribeThreadInput = typeof OrchestrationV2SubscribeThreadInput.Type; From bea4716b8d3eaee76c3c7902a2e7347d3fe46cde Mon Sep 17 00:00:00 2001 From: Renfei Zhou Date: Fri, 13 Nov 2026 16:49:04 +0000 Subject: [PATCH 2/2] fix(fork): display modes load only the history they show A thread in Messages only, Messages + Dashboard or Chat + Replies could open on an empty pane with only "Load earlier turns": the server picked the window and older pages by Normal rules, and the bot's hidden work filled them. The web now asks for a shown-only view of such a thread when the server has the `threadShownHistory` capability. The server then walks only user-typed messages and `send_to_user` / `replace_send_to_user` calls, newest first across the thread and its fork ancestors, until the window holds 120 messages and the first message of every loaded reply, or a row, byte or run cap. Its cursor carries the view, so each "Load earlier turns" click loads one page of earlier shown messages. The window also keeps the newest row, active items, failed errors and pending request cards, so working state, failures and questions still show. On the client the shown view is a separate keyed thread state that never reads or writes the saved snapshot. The web picks one view per thread from its display mode, with a per-viewer hint for a cold start. The open sub-thread view has no "Load earlier turns", and the Replies list says when replies may be older and can load earlier messages. Normal threads, the mobile app and guest shares keep today's requests. Live events are unchanged. Co-Authored-By: Claude Opus 5.5 --- .../src/environment/ServerEnvironment.ts | 1 + .../fork/displayModeWrap/DisplayModeWrap.ts | 5 +- .../src/fork/shownHistory/ShownHistory.ts | 552 ++++++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 2 + .../src/orchestration-v2/ProjectionStore.ts | 24 + .../src/orchestration-v2/ThreadStream.ts | 2 + apps/server/src/orchestration-v2/http.ts | 5 + .../orchestration-v2/threadHistoryPaging.ts | 17 +- apps/server/src/ws.ts | 4 + apps/web/src/components/ChatView.tsx | 9 +- .../src/fork/displayMode/threadView.test.ts | 74 +++ apps/web/src/fork/displayMode/threadView.ts | 115 ++++ .../fork/displayMode/useThreadDisplayMode.ts | 8 +- .../src/fork/subThreads/SubThreadsPanel.tsx | 53 +- apps/web/src/state/threads.ts | 5 +- .../src/fork/shownOnlyThread.ts | 33 ++ .../src/state/boundedThreadSnapshotHttp.ts | 4 + .../src/state/threadSnapshotHttp.ts | 1 + packages/client-runtime/src/state/threads.ts | 25 +- 19 files changed, 913 insertions(+), 26 deletions(-) create mode 100644 apps/server/src/fork/shownHistory/ShownHistory.ts create mode 100644 apps/web/src/fork/displayMode/threadView.test.ts create mode 100644 apps/web/src/fork/displayMode/threadView.ts create mode 100644 packages/client-runtime/src/fork/shownOnlyThread.ts diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 0b8fe67277b54..20cf1cf9d645a 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -253,6 +253,7 @@ export const make = Effect.gen(function* () { usagePriceOverrides: true, usageModelAliases: true, threadDisplayModes: true, + threadShownHistory: true, threadPinning: true, threadPinReorder: true, threadActiveReorder: true, diff --git a/apps/server/src/fork/displayModeWrap/DisplayModeWrap.ts b/apps/server/src/fork/displayModeWrap/DisplayModeWrap.ts index 85cafb73caee0..3eb6456b0e840 100644 --- a/apps/server/src/fork/displayModeWrap/DisplayModeWrap.ts +++ b/apps/server/src/fork/displayModeWrap/DisplayModeWrap.ts @@ -15,7 +15,6 @@ import { NORMAL_DISPLAY_MODE, resolveDisplayMode, foldCores, - SEND_TO_USER_PAYLOAD_LIKE, SUB_THREAD_WINDOW_CORES, subThreadCores, subThreadOfRecords, @@ -34,6 +33,7 @@ import type * as SqlError from "effect/sql/SqlError"; import * as ProjectionStore from "../../orchestration-v2/ProjectionStore.ts"; import * as ProviderRegistry from "../../provider/ProviderRegistry.ts"; import * as ServerSettings from "../../serverSettings.ts"; +import { sendCallSql } from "../shownHistory/ShownHistory.ts"; import * as SubThreads from "../subThreads/SubThreads.ts"; import { createToldTable } from "./DisplayModeCompactionNote.ts"; @@ -251,8 +251,7 @@ export const layer: Layer.Layer< } // The screen shows only user messages and send_to_user calls, so the query steps over other rows. - const shownOnly = sql`(item.type = 'user_message' OR (item.type = 'dynamic_tool' - AND item.payload_json LIKE ${SEND_TO_USER_PAYLOAD_LIKE}))`; + const shownOnly = sql`(item.type = 'user_message' OR ${sendCallSql(sql)})`; const found: OrchestrationV2TurnItem[] = []; let cores = 0; let passedTyped = before !== null; diff --git a/apps/server/src/fork/shownHistory/ShownHistory.ts b/apps/server/src/fork/shownHistory/ShownHistory.ts new file mode 100644 index 0000000000000..3b6c8f25e6b54 --- /dev/null +++ b/apps/server/src/fork/shownHistory/ShownHistory.ts @@ -0,0 +1,552 @@ +import { + type OrchestrationV2AppThread, + OrchestrationV2AppThreadJson, + type OrchestrationV2ProjectedTurnItem, + type OrchestrationV2ThreadHistoryPage, + type OrchestrationV2ThreadProjection, + type OrchestrationV2TurnItem, + OrchestrationV2TurnItemJson, + type RunId, + ThreadId, + TurnItemId, +} from "@t3tools/contracts"; +import { + type DisplayEvent, + displayEventFromTurnItem, + foldCores, + SEND_TO_USER_TOOL_NAME, + SUB_THREAD_WINDOW_CORES, +} from "@t3tools/shared/displayModes"; +import * as Effect from "effect/Effect"; +import * as Result from "effect/Result"; +import * as Schema from "effect/Schema"; +import type * as SqlClient from "effect/sql/SqlClient"; + +import { + decodeThreadHistoryCursor, + InvalidThreadHistoryCursorError, + projectedRowBoundedSnapshotEncodedBytes, + projectedRowEncodedBytes, + type SelectTimelinePageResult, + THREAD_HISTORY_MAX_RAW_TURNS, + type ThreadHistoryCursorPayload, +} from "../../orchestration-v2/threadHistoryPaging.ts"; +import { projectTurnItemForWire } from "../../orchestration-v2/WireProjection.ts"; + +/** A row's place in its thread's history: rows sort by ordinal, then by item id. */ +export interface ShownKey { + readonly threadId: ThreadId; + readonly ordinal: number; + readonly itemId: TurnItemId; +} + +/** A row that a display mode shows, and what it shows. */ +interface ShownRow extends ShownKey { + readonly rowid: number; + readonly item: OrchestrationV2TurnItem; + readonly event: Exclude; +} + +/** The rows of a window or page, newest first, and the newest shown row older than them. */ +export interface ShownWalk { + readonly rows: ReadonlyArray; + readonly next: ShownKey | null; +} + +/** Where a read starts: rows older than this key, or also the row itself when inclusive. */ +interface ShownBound { + readonly ordinal: number; + readonly itemId: string; + readonly inclusive: boolean; +} + +/** One thread's stretch of a history, read newest first from `from`. */ +interface ShownSegment { + readonly threadId: ThreadId; + readonly from: ShownBound; + /** An inherited stretch keeps the rows at or before the fork run, as isTurnItemAtOrBeforeRun does. */ + readonly inherited: { readonly runOrdinal: number; readonly runlessRows: boolean } | null; +} + +/** The row a targeted page must reach: its segment and its key there. */ +interface ShownTarget { + readonly index: number; + readonly ordinal: number; + readonly itemId: string; +} + +/** A walk stops once it holds `minCores` messages, or at a cap. */ +interface ShownLimits { + readonly minCores: number; + readonly maxRows: number; + readonly maxBytes: number; + readonly maxRuns: number; +} + +const WINDOW_LIMITS: ShownLimits = { + // A send answers only a message among the latest SUB_THREAD_WINDOW_CORES, so a live reply finds it. + minCores: SUB_THREAD_WINDOW_CORES, + maxRows: 600, + maxBytes: 512 * 1024, + // A run's records come along outside the byte cap; Normal stops at as many turns. + maxRuns: THREAD_HISTORY_MAX_RAW_TURNS, +}; +const PAGE_LIMITS: ShownLimits = { + minCores: 100, + maxRows: 1_000, + maxBytes: 1_048_576, + maxRuns: Number.POSITIVE_INFINITY, +}; +const TARGETED_PAGE_LIMITS: ShownLimits = { ...PAGE_LIMITS, maxRows: 2_000, maxBytes: 2_097_152 }; + +const BATCH_ROWS = 32; +// TurnItemPositionStore numbers a run's items from runOrdinal * 1_000_000. +const RUN_ORDINAL_SPAN = 1_000_000; +const NEWEST: ShownBound = { ordinal: Number.MAX_SAFE_INTEGER, itemId: "", inclusive: false }; + +const decodeTurnItem = Schema.decodeUnknownEffect( + Schema.fromJsonString(OrchestrationV2TurnItemJson), +); +const decodeForkFields = Schema.decodeUnknownEffect( + Schema.fromJsonString( + OrchestrationV2AppThreadJson.mapFields(({ forkedFrom, lineage, historyOrigin, deletedAt }) => ({ + forkedFrom, + lineage, + historyOrigin, + deletedAt, + })), + ), +); + +/** ProjectionStore's inheritedRunFork, passed in so this module never imports ProjectionStore. */ +type ForkOf = ( + thread: Pick, +) => { readonly threadId: ThreadId; readonly runId: RunId } | null; + +/** A shown cursor is upstream's history cursor, at the row the next page starts with, plus `fv`. */ +type ShownCursorPayload = ThreadHistoryCursorPayload & { readonly fv: "shown" }; + +function encodeShownCursor(snapshotSequence: number, key: ShownKey): string { + const payload: ShownCursorPayload = { + v: 1, + seq: snapshotSequence, + st: key.threadId, + si: key.itemId, + p: key.ordinal, + inclusive: true, + fv: "shown", + }; + return Buffer.from(JSON.stringify(payload), "utf8").toString("base64url"); +} + +/** The row a shown cursor starts at; null for any other cursor, which pages the Normal way. */ +export function shownCursorOf(cursor: string): ShownKey | null { + const payload = Result.getOrNull(Result.try(() => decodeThreadHistoryCursor(cursor))); + return payload !== null && "fv" in payload && payload.fv === "shown" && payload.inclusive === true + ? { + threadId: ThreadId.make(payload.st), + ordinal: payload.p, + itemId: TurnItemId.make(payload.si), + } + : null; +} + +/** + * A `send_to_user` or `replace_send_to_user` call: a tool name that holds `send_to_user`, a superset + * of the names resolveT3McpToolId takes, completion labels and namespaces included. + */ +export const sendCallSql = (sql: SqlClient.SqlClient) => + sql`(item.type = 'dynamic_tool' + AND json_extract(item.payload_json, '$.toolName') LIKE ${`%${SEND_TO_USER_TOOL_NAME}%`})`; + +/** The rows a display mode can show: messages the user typed (isUserTypedMessage) and send calls. */ +const shownCandidateSql = (sql: SqlClient.SqlClient) => sql`( + (item.type = 'user_message' + AND json_extract(item.payload_json, '$.createdBy') = 'user' + AND json_extract(item.payload_json, '$.creationSource') IN ('web', 'mobile') + AND json_extract(item.payload_json, '$.scheduledTaskId') IS NULL + AND json_extract(item.payload_json, '$.senderThreadId') IS NULL + AND json_extract(item.payload_json, '$.idleCompaction') IS NULL) + OR ${sendCallSql(sql)})`; + +// A thread's own rows follow the eligible rule of ProjectionStore.retainedWindowRows; an inherited +// stretch only the fork cutoff. +const segmentVisible = (sql: SqlClient.SqlClient, segment: ShownSegment) => + segment.inherited === null + ? sql`(run.status IS NULL OR run.status <> 'rolled_back') + AND NOT (item.type = 'user_message' + AND json_extract(item.payload_json, '$.inputIntent') = 'queued_turn' + AND run.status IS 'cancelled')` + : sql`(run.ordinal <= ${segment.inherited.runOrdinal} + OR (item.run_id IS NULL AND ${segment.inherited.runlessRows ? 1 : 0} = 1))`; + +interface CandidateRow { + readonly item_rowid: number; + readonly turn_item_id: string; + readonly ordinal: number; + readonly payload_json: string; +} + +const readCandidates = (sql: SqlClient.SqlClient, segment: ShownSegment, from: ShownBound) => + sql` + SELECT item.rowid AS item_rowid, item.turn_item_id, item.ordinal, item.payload_json + FROM orchestration_v2_projection_turn_items AS item + LEFT JOIN orchestration_v2_projection_runs AS run ON run.run_id = item.run_id + WHERE item.thread_id = ${segment.threadId} + AND item.ordinal <= ${from.ordinal} + AND (item.ordinal < ${from.ordinal} + OR item.turn_item_id ${sql.literal(from.inclusive ? "<=" : "<")} ${from.itemId}) + AND ${shownCandidateSql(sql)} + AND ${segmentVisible(sql, segment)} + ORDER BY item.ordinal DESC, item.turn_item_id DESC + LIMIT ${BATCH_ROWS} + `; + +const isBelow = (key: { readonly ordinal: number; readonly itemId: string }, bound: ShownBound) => + key.ordinal < bound.ordinal || + (key.ordinal === bound.ordinal && + (bound.inclusive ? key.itemId <= bound.itemId : key.itemId < bound.itemId)); + +/** + * The thread's history as stretches, newest first: its own rows, then each fork ancestor's rows up + * to the fork run. Null when the thread is missing or deleted. + */ +const shownSegments = Effect.fn("ShownHistory.segments")(function* ( + sql: SqlClient.SqlClient, + threadId: ThreadId, + forkOf: ForkOf, +) { + const readThread = (id: ThreadId) => + sql<{ readonly payload_json: string }>` + SELECT payload_json FROM orchestration_v2_projection_threads WHERE thread_id = ${id} LIMIT 1 + `.pipe( + Effect.flatMap(([row]) => + row === undefined ? Effect.succeed(null) : decodeForkFields(row.payload_json), + ), + ); + let thread = yield* readThread(threadId); + if (thread === null || thread.deletedAt !== null) return null; + const segments: ShownSegment[] = [{ threadId, from: NEWEST, inherited: null }]; + const seen = new Set([threadId]); + for ( + let fork = forkOf(thread); + fork !== null && !seen.has(fork.threadId); + fork = forkOf(thread) + ) { + seen.add(fork.threadId); + const source = yield* readThread(fork.threadId); + const [run] = yield* sql<{ readonly ordinal: number }>` + SELECT ordinal FROM orchestration_v2_projection_runs + WHERE thread_id = ${fork.threadId} AND run_id = ${fork.runId} LIMIT 1 + `; + if (source === null || run === undefined) break; + segments.push({ + threadId: fork.threadId, + from: { ordinal: (run.ordinal + 1) * RUN_ORDINAL_SPAN, itemId: "", inclusive: false }, + inherited: { runOrdinal: run.ordinal, runlessRows: source.historyOrigin === "v1_import" }, + }); + thread = source; + } + return segments; +}); + +/** The segments from `key` on, the row itself included; null when `key` is in none of them. */ +function segmentsFrom( + segments: ReadonlyArray, + key: ShownKey, +): ReadonlyArray | null { + const index = segments.findIndex((segment) => segment.threadId === key.threadId); + if (index === -1) return null; + const at = { ordinal: key.ordinal, itemId: key.itemId, inclusive: true }; + return [{ ...segments[index]!, from: at }, ...segments.slice(index + 1)]; +} + +/** + * Where the row that `entryId` names (an item id, or a message's id as Normal pages take it) sits in + * `segments`, by index lookups, never a scan; its oldest place when it has several, null when none. + */ +const locateTarget = Effect.fn("ShownHistory.locateTarget")(function* ( + sql: SqlClient.SqlClient, + segments: ReadonlyArray, + entryId: string, +) { + const rows = yield* sql<{ + readonly thread_id: string; + readonly ordinal: number; + readonly turn_item_id: string; + }>` + SELECT thread_id, ordinal, turn_item_id FROM orchestration_v2_projection_turn_items + WHERE turn_item_id = ${entryId} + UNION ALL + SELECT item.thread_id, item.ordinal, item.turn_item_id + FROM orchestration_v2_projection_messages AS message + JOIN orchestration_v2_projection_turn_items AS item + ON item.thread_id = message.thread_id AND item.run_id IS message.run_id + WHERE message.message_id = ${entryId} + AND item.type IN ('user_message', 'assistant_message') + AND json_extract(item.payload_json, '$.messageId') = ${entryId} + `; + let target: ShownTarget | null = null; + for (const row of rows) { + const key = { ordinal: row.ordinal, itemId: row.turn_item_id }; + const index = segments.findIndex( + (segment) => segment.threadId === row.thread_id && isBelow(key, segment.from), + ); + if (index === -1) continue; + if ( + target === null || + index > target.index || + (index === target.index && isBelow(key, { ...target, inclusive: false })) + ) + target = { index, ...key }; + } + return target; +}); + +/** + * Whether `rows`, newest first, hold `minCores` messages and the first message of each reply among + * them. An agent's reply whose answered message is not among the SUB_THREAD_WINDOW_CORES messages + * before it needs none: send_to_user accepts an answer only to those, so that message never existed. + */ +function shownNeedsMet(rows: ReadonlyArray, minCores: number): boolean { + const cores = foldCores(rows.map((row) => row.event).toReversed()); + if (cores.length < minCores) return false; + const loaded = new Set(cores.map((core) => core.id)); + return cores.every( + (core, older) => + !core.reply || + (core.first === null ? older >= SUB_THREAD_WINDOW_CORES : loaded.has(core.first)), + ); +} + +const keyOf = (segment: ShownSegment, row: CandidateRow): ShownKey => ({ + threadId: segment.threadId, + ordinal: row.ordinal, + itemId: TurnItemId.make(row.turn_item_id), +}); + +/** The message a rewrite names; undefined for one that rewrites the latest send before it. */ +const rewriteTarget = (event: Extract) => + event.replyTo !== undefined + ? `reply:${event.replyTo}` + : event.subThread !== undefined + ? `first:${event.subThread}` + : undefined; + +/** A candidate whose display event shows nothing, such as a refused send, is null. */ +const decodeShownRow = (segment: ShownSegment, row: CandidateRow) => + decodeTurnItem(row.payload_json).pipe( + Effect.map((item): ShownRow | null => { + const event = displayEventFromTurnItem(item); + return event === null || event.type === "other" + ? null + : { ...keyOf(segment, row), rowid: row.item_rowid, item, event }; + }), + ); + +/** + * Walks `segments` newest first over the shown rows only, in batches, until it has passed `target` + * and holds the messages `limits` asks for, checked after each batch. A cap ends it at any row. + * `next` is the newest shown row the walk left out, which the next page starts at. + */ +const walkShownRows = Effect.fn("ShownHistory.walk")(function* ( + sql: SqlClient.SqlClient, + segments: ReadonlyArray, + limits: ShownLimits, + rowBytes: (row: ShownRow) => number, + target: ShownTarget | null, +) { + const rows: ShownRow[] = []; + const runs = new Set(); + const rewritten = new Set(); + let bytes = 0; + let met = false; + for (const [index, segment] of segments.entries()) { + let from: ShownBound | null = segment.from; + while (from !== null) { + const batch: ReadonlyArray = yield* readCandidates(sql, segment, from); + for (const candidate of batch) { + const row = yield* decodeShownRow(segment, candidate); + if (row === null) continue; + // A newer rewrite of the same message already sets its text. + const rewrite = row.event.type === "rewrite" ? rewriteTarget(row.event) : undefined; + if (rewrite !== undefined && rewritten.has(rewrite)) continue; + if (rewrite !== undefined) rewritten.add(rewrite); + const cost = rowBytes(row); + const newRun = row.item.runId !== null && !runs.has(row.item.runId); + if ( + met || + (rows.length > 0 && + (rows.length >= limits.maxRows || + bytes + cost > limits.maxBytes || + (newRun && runs.size >= limits.maxRuns))) + ) + return { rows, next: keyOf(segment, candidate) } satisfies ShownWalk; + rows.push(row); + bytes += cost; + if (row.item.runId !== null) runs.add(row.item.runId); + } + const last = batch.at(-1); + from = + last === undefined || batch.length < BATCH_ROWS + ? null + : { ordinal: last.ordinal, itemId: last.turn_item_id, inclusive: false }; + const passed = + target === null || + index > target.index || + (index === target.index && (from === null || !isBelow(target, from))); + met ||= passed && shownNeedsMet(rows, limits.minCores); + } + } + return { rows, next: null } satisfies ShownWalk; +}); + +/** A shown row as a page or a snapshot sends it. */ +const shownWireRow = ( + row: ShownRow, + threadId: ThreadId, + position: number, +): OrchestrationV2ProjectedTurnItem => ({ + position, + visibility: row.threadId === threadId ? "local" : "inherited", + sourceThreadId: row.threadId, + sourceItemId: row.itemId, + item: projectTurnItemForWire(row.item), +}); + +/** The thread's latest shown rows, in the bounded snapshot's byte unit. */ +export const selectShownWindow = Effect.fn("ShownHistory.selectWindow")(function* ( + sql: SqlClient.SqlClient, + threadId: ThreadId, + forkOf: ForkOf, +) { + const segments = yield* shownSegments(sql, threadId, forkOf); + if (segments === null) return { rows: [], next: null } satisfies ShownWalk; + return yield* walkShownRows( + sql, + segments, + WINDOW_LIMITS, + (row) => projectedRowBoundedSnapshotEncodedBytes(shownWireRow(row, threadId, 0), threadId), + null, + ); +}); + +/** The cursor of a shown window, whose walk bounded its rows already; undefined for any other. */ +export function shownWindow(snapshot: { + readonly snapshotSequence: number; + readonly shownHistory?: { readonly next: ShownKey | null } | undefined; +}): Pick | undefined { + const next = snapshot.shownHistory?.next; + return next === undefined + ? undefined + : { + nextCursor: next === null ? null : encodeShownCursor(snapshot.snapshotSequence, next), + hasMoreHistory: next !== null, + }; +} + +/** + * The shown rows older than a shown cursor, as a history page; null when the thread is missing or + * deleted. A page with `throughEntryId` reads on until it has passed that row, and is a plain page + * when the row is not older than the cursor. + */ +export const shownHistoryPage = Effect.fn("ShownHistory.page")(function* ( + sql: SqlClient.SqlClient, + threadId: ThreadId, + cursor: ShownKey, + throughEntryId: string | undefined, + forkOf: ForkOf, +) { + const segments = yield* shownSegments(sql, threadId, forkOf); + if (segments === null) return null; + const from = segmentsFrom(segments, cursor); + if (from === null) return yield* new InvalidThreadHistoryCursorError(); + const target = + throughEntryId === undefined ? null : yield* locateTarget(sql, from, throughEntryId); + const walk = yield* walkShownRows( + sql, + from, + target === null ? PAGE_LIMITS : TARGETED_PAGE_LIMITS, + (row) => projectedRowEncodedBytes(shownWireRow(row, threadId, 0)), + target, + ); + const [revision] = yield* sql<{ readonly sequence: number }>` + SELECT COALESCE(MAX(sequence), 0) AS sequence FROM orchestration_events + WHERE application_event_version = 2 AND aggregate_kind = 'thread' AND stream_id = ${threadId} + `; + const snapshotSequence = revision?.sequence ?? 0; + return { + snapshotSequence, + items: walk.rows.toReversed().map((row, position) => shownWireRow(row, threadId, position)), + nextCursor: walk.next === null ? null : encodeShownCursor(snapshotSequence, walk.next), + hasMoreHistory: walk.next !== null, + } satisfies OrchestrationV2ThreadHistoryPage; +}); + +/** + * Replaces ProjectionStore.retainedWindowRows for a shown window. Next to the walked rows it keeps + * the newest row, active items and failed errors, none older than the oldest walked row so the + * window stays contiguous; with no such row, all active items and the latest run's errors only. + */ +export function shownRetainedRows(sql: SqlClient.SqlClient, threadId: ThreadId, walk: ShownWalk) { + const local = walk.rows.filter((row) => row.threadId === threadId); + const floor = + local.length === 0 || local.length < walk.rows.length ? null : local.at(-1)!.ordinal; + return sql` + WITH retained AS MATERIALIZED ( + SELECT rowid AS item_rowid, ordinal, turn_item_id + FROM orchestration_v2_projection_turn_items + WHERE rowid IN (SELECT value FROM json_each(${JSON.stringify(local.map((row) => row.rowid))})) + UNION + SELECT item_rowid, ordinal, turn_item_id FROM ( + SELECT rowid AS item_rowid, ordinal, turn_item_id + FROM orchestration_v2_projection_turn_items + WHERE thread_id = ${threadId} + ORDER BY ordinal DESC, turn_item_id DESC + LIMIT 1 + ) + UNION + SELECT rowid AS item_rowid, ordinal, turn_item_id + FROM orchestration_v2_projection_turn_items + WHERE thread_id = ${threadId} AND status IN ('pending', 'running', 'waiting') + AND ordinal >= ${floor ?? 0} + UNION + SELECT rowid AS item_rowid, ordinal, turn_item_id + FROM orchestration_v2_projection_turn_items + WHERE thread_id = ${threadId} AND type = 'error' AND status = 'failed' + AND ${ + floor === null + ? sql`run_id = (SELECT run_id FROM orchestration_v2_projection_runs + WHERE thread_id = ${threadId} ORDER BY ordinal DESC LIMIT 1)` + : sql`ordinal >= ${floor}` + } + ORDER BY ordinal ASC, turn_item_id ASC + ), deferred_work AS ( + SELECT NULL AS run_id, NULL AS first_ordinal, NULL AS last_ordinal, NULL AS item_count + WHERE 0 + ) + `; +} + +/** Puts the walked rows of fork ancestors before the thread's own visible rows. */ +export function withInherited( + projection: OrchestrationV2ThreadProjection, + walk: ShownWalk, +): OrchestrationV2ThreadProjection { + const threadId = projection.thread.id; + const inherited = walk.rows.filter((row) => row.threadId !== threadId).reverse(); + if (inherited.length === 0) return projection; + return { + ...projection, + visibleTurnItems: [ + ...inherited.map((row) => ({ + position: 0, + visibility: "inherited" as const, + sourceThreadId: row.threadId, + sourceItemId: row.itemId, + item: row.item, + })), + ...projection.visibleTurnItems, + ].map((row, position) => ({ ...row, position })), + }; +} diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 272d40033415b..9e26cd1237869 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -170,6 +170,7 @@ import { type ShellSnapshotOptions, } from "./ProjectionStore.ts"; import { ProviderAdapterRegistryV2 } from "./ProviderAdapterRegistry.ts"; +import type { ShownKey } from "../fork/shownHistory/ShownHistory.ts"; import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; import * as RunExecutionService from "./RunExecutionService.ts"; @@ -437,6 +438,7 @@ export interface OrchestratorV2Shape { readonly schemaVersion: number; readonly snapshotSequence: number; readonly projection: OrchestrationV2ThreadProjection; + readonly shownHistory?: { readonly next: ShownKey | null }; }, OrchestratorV2Error >; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 333ede4fb6627..264c1b13027c8 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -97,6 +97,7 @@ import type * as Statement from "effect/sql/Statement"; import { MCP_APP_OUTPUT_KEY } from "@t3tools/shared/mcpApp"; import { SEND_TO_USER_PAYLOAD_LIKE } from "@t3tools/shared/displayModes"; import { threadHtmlRenderAttachmentIds } from "../attachmentStore.ts"; +import * as ShownHistory from "../fork/shownHistory/ShownHistory.ts"; import { isThreadHistoryUserTurn, isConversationHistoryItem, @@ -711,12 +712,14 @@ export interface ProjectionStoreV2Shape { readonly anchorItemId?: TurnItemId | undefined; readonly anchorThreadId?: ThreadId | undefined; readonly requiredRunId?: RunId | undefined; + readonly shownOnly?: boolean | undefined; }, ) => Effect.Effect< { readonly schemaVersion: number; readonly snapshotSequence: number; readonly projection: OrchestrationV2ThreadProjection; + readonly shownHistory?: { readonly next: ShownHistory.ShownKey | null }; }, ProjectionStoreV2Error >; @@ -728,6 +731,7 @@ type ProjectionWindow = Omit< > & { readonly suppressLocal?: boolean | undefined; readonly historyAnchor?: { readonly threadId: ThreadId; readonly itemId: TurnItemId } | undefined; + readonly shown?: ShownHistory.ShownWalk | undefined; }; export class ProjectionStoreV2 extends Context.Service()( @@ -2889,6 +2893,8 @@ export const layer: Layer.Layer = threadId: ThreadId, window: ProjectionWindow, ) { + if (window.shown !== undefined) + return ShownHistory.shownRetainedRows(sql, threadId, window.shown); const eligible = sql` WITH eligible AS NOT MATERIALIZED ( SELECT item.rowid AS item_rowid, item.payload_json, item.ordinal, item.turn_item_id, @@ -3819,6 +3825,8 @@ export const layer: Layer.Layer = ? ["runs", "attempts", "turnItems"] : undefined, ); + if (window?.shown !== undefined) + return ShownHistory.withInherited(withLocalVisibleTurnItems(projection), window.shown); const forkedFrom = inheritedRunFork(projection.thread); if (forkedFrom === null || seenThreadIds.has(forkedFrom.threadId)) { return withLocalVisibleTurnItems(projection); @@ -5875,6 +5883,17 @@ export const layer: Layer.Layer = sql .withTransaction( Effect.gen(function* () { + const shown = workMode === "work" ? null : ShownHistory.shownCursorOf(cursor); + if (shown !== null) + return ( + (yield* ShownHistory.shownHistoryPage( + sql, + threadId, + shown, + throughEntryId, + inheritedRunFork, + )) ?? (yield* new ProjectionStoreThreadNotFoundError({ threadId })) + ); const index = yield* readTimelineIndex(threadId, new Set(), workMode !== undefined); if (index.records.thread.deletedAt !== null) return yield* new ProjectionStoreThreadNotFoundError({ threadId }); @@ -6107,6 +6126,9 @@ export const layer: Layer.Layer = `).map((row) => ThreadId.make(row.thread_id))[0] ?? threadId, }; + const shown = options.shownOnly + ? yield* ShownHistory.selectShownWindow(sql, threadId, inheritedRunFork) + : undefined; const projection = yield* readProjection(threadId, new Set(), { rowLimit: options.rowLimit, userTurnLimit: options.userTurnLimit, @@ -6115,6 +6137,7 @@ export const layer: Layer.Layer = ? { anchorItemId: historyAnchor.itemId } : {}), ...(historyAnchor === undefined ? {} : { historyAnchor }), + ...(shown === undefined ? {} : { shown }), }); // Pending requests outlive their turn's window; clients read their prompt from the card. const retainedIds = new Set(projection.turnItems.map((item) => item.id)); @@ -6154,6 +6177,7 @@ export const layer: Layer.Layer = schemaVersion: ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION, snapshotSequence: rows[0]?.snapshot_sequence ?? 0, projection: windowProjection, + ...(shown === undefined ? {} : { shownHistory: { next: shown.next } }), }; }), ) diff --git a/apps/server/src/orchestration-v2/ThreadStream.ts b/apps/server/src/orchestration-v2/ThreadStream.ts index c16a1b9e095aa..fe003e4ef0d95 100644 --- a/apps/server/src/orchestration-v2/ThreadStream.ts +++ b/apps/server/src/orchestration-v2/ThreadStream.ts @@ -71,10 +71,12 @@ export function buildBoundedThreadStreamSnapshot(input: { readonly compactTurnItems?: boolean; readonly dedupeBodies?: boolean; readonly compactIds?: boolean; + readonly window?: Parameters[0]["window"]; }): ThreadSnapshotStreamItem { const bounded = buildBoundedThreadProjection({ snapshotSequence: input.snapshotSequence, projection: projectThreadProjectionForWire(input.projection), + window: input.window, }); return { kind: "snapshot", diff --git a/apps/server/src/orchestration-v2/http.ts b/apps/server/src/orchestration-v2/http.ts index 3bba05f7db63a..6d2457d1136af 100644 --- a/apps/server/src/orchestration-v2/http.ts +++ b/apps/server/src/orchestration-v2/http.ts @@ -34,6 +34,7 @@ import * as ProjectStore from "./ProjectStore.ts"; import { loadSubscriberShellSnapshot } from "./ShellStream.ts"; import { boundedSnapshotResponseFields } from "./ThreadStream.ts"; import { projectThreadProjectionForWire } from "./WireProjection.ts"; +import * as ShownHistory from "../fork/shownHistory/ShownHistory.ts"; const isInvalidThreadHistoryCursorError = Schema.is(InvalidThreadHistoryCursorError); @@ -131,11 +132,13 @@ export const layer = HttpApiBuilder.group( typeof threadManagement.getThreadSnapshotWindow >[1]["anchorThreadId"], deferWork = false, + shownOnly = false, ) { return yield* threadManagement .getThreadSnapshotWindow(threadId, { rowLimit: THREAD_HISTORY_SNAPSHOT_ROW_LIMIT, deferWork, + ...(shownOnly ? { shownOnly } : {}), userTurnLimit: anchorItemId === undefined ? THREAD_HISTORY_PAGE_POLICY.maxUserTurns @@ -227,10 +230,12 @@ export const layer = HttpApiBuilder.group( undefined, undefined, args.query.deferWork === "1", + args.query.shownOnly === "1", ).pipe(traceLocalHandlerWork); const bounded = buildBoundedThreadProjection({ projection: snapshot.projection, snapshotSequence: snapshot.snapshotSequence, + window: ShownHistory.shownWindow(snapshot), }); return { snapshotSequence: snapshot.snapshotSequence, diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.ts index 8bc794ce36c4d..925e8990cf2c6 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.ts @@ -630,6 +630,8 @@ export function buildBoundedThreadProjection(input: { readonly projection: OrchestrationV2ThreadProjection; readonly snapshotSequence: number; readonly policy?: ThreadHistoryPagePolicy | undefined; + /** Set when the caller bounded the visible rows itself: all of them stay, with this cursor. */ + readonly window?: Pick | undefined; }): BoundedProjectionResult { const policy = input.policy ?? THREAD_HISTORY_PAGE_POLICY; const threadId = input.projection.thread.id; @@ -682,12 +684,15 @@ export function buildBoundedThreadProjection(input: { ); const windowPolicy: ThreadHistoryPagePolicy = { ...policy, maxEncodedBytes: windowBudget }; - const window = selectRecentTimelineWindow({ - items: controlProjection.visibleTurnItems, - snapshotSequence: input.snapshotSequence, - policy: windowPolicy, - rowEncodedBytes: (row) => projectedRowBoundedSnapshotEncodedBytes(row, threadId), - }); + const window = + input.window === undefined + ? selectRecentTimelineWindow({ + items: controlProjection.visibleTurnItems, + snapshotSequence: input.snapshotSequence, + policy: windowPolicy, + rowEncodedBytes: (row) => projectedRowBoundedSnapshotEncodedBytes(row, threadId), + }) + : { ...input.window, items: controlProjection.visibleTurnItems }; const visibleTurnItems = renumberPositions(window.items); const windowTurnItems = localTurnItemsForVisibleWindow(controlProjection, visibleTurnItems); const dependencyTurnItems = retainedInterruptRequestTurnItems( diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 45ef1128e13ad..3bd9f57a24049 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -197,6 +197,7 @@ import * as ServerBrowser from "./preview/ServerBrowser.ts"; import * as DeviceService from "./device/DeviceService.ts"; import * as ThreadDashboard from "./fork/threadDashboard/ThreadDashboard.ts"; import * as SubThreads from "./fork/subThreads/SubThreads.ts"; +import { shownWindow } from "./fork/shownHistory/ShownHistory.ts"; import { remoteSshDeviceHosts } from "./device/localSshDeviceHost.ts"; import * as PreviewManager from "./preview/Manager.ts"; import { issueAssetUrl } from "./assets/AssetAccess.ts"; @@ -895,6 +896,7 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs readonly acceptCompactTurnItems?: boolean; readonly acceptDedupedBodies?: boolean; readonly acceptCompactIds?: boolean; + readonly shownOnly?: boolean; }) { const threadManagement = yield* ThreadManagementService.ThreadManagementService; const dedupeBodies = input.acceptDedupedBodies === true; @@ -977,6 +979,7 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs rowLimit: THREAD_HISTORY_SNAPSHOT_ROW_LIMIT, deferWork: input.acceptDeferredWork === true, userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns, + ...(input.shownOnly === true ? { shownOnly: true } : {}), }) : threadManagement.getThreadSnapshot(input.threadId) ).pipe( @@ -993,6 +996,7 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs const snapshotItem = useBoundedSnapshot ? buildBoundedThreadStreamSnapshot({ ...snapshot, + window: shownWindow(snapshot), compactTurnItems: input.acceptCompactTurnItems === true, dedupeBodies, compactIds: input.acceptCompactIds === true, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 0de344951de5c..5c2c26d96f861 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -357,6 +357,7 @@ import { ChatRightPanel } from "./chat/ChatRightPanel"; import { useDashboardSurfaceAdder } from "../fork/dashboard/dashboardSurface"; import { REPLY_BOX_QUEUE_EDIT_BLOCKED, subThreadDraftId } from "../fork/subThreads/subThreadReply"; import { SubThreadViewContext } from "../fork/subThreads/subThreadsState"; +import { useThreadShownView } from "../fork/displayMode/threadView"; import { timelineCitation } from "../fork/dashboard/dashboardCitation"; import { useDraftEnvironmentRouting } from "./chat/useDraftEnvironmentRouting"; import { branchMismatchBannerItem, extensionStatusBannerItem } from "./chat/ThreadActivityBanners"; @@ -535,7 +536,11 @@ export default function ChatView(props: ChatViewProps) { const serverVisibleTurnItems = useQuietThreadVisibleTurnItems(routeThreadDetailRef); const serverThreadHistory = useThreadHistory(routeThreadDetailRef); const threadHistoryControls = (() => { - if (routeThreadDetailRef === null || !shouldShowLoadEarlierControl(serverThreadHistory)) { + if ( + routeThreadDetailRef === null || + subThreadDraft !== null || + !shouldShowLoadEarlierControl(serverThreadHistory) + ) { return undefined; } return { @@ -1588,9 +1593,11 @@ export default function ChatView(props: ChatViewProps) { ); const turnDiffSummaries = useThreadCheckpointSummaries(routeThreadDetailRef); const availableRevertTurnCounts = deriveAvailableRevertTurnCounts(serverProjection); + const threadShownView = useThreadShownView(routeThreadDetailRef); // A paged history window cannot identify the first loaded run as the thread start. const threadStartCheckpointReady = !serverThreadHistory.hasMoreHistory && + !threadShownView && findRevertCheckpoint(serverProjection?.checkpoints ?? [], 0) !== undefined; const workspaceMutationId = deriveWorkspaceMutationId( diff --git a/apps/web/src/fork/displayMode/threadView.test.ts b/apps/web/src/fork/displayMode/threadView.test.ts new file mode 100644 index 0000000000000..1764fee69d43b --- /dev/null +++ b/apps/web/src/fork/displayMode/threadView.test.ts @@ -0,0 +1,74 @@ +import { threadKey } from "@t3tools/client-runtime/state/entities"; +import { EMPTY_ENVIRONMENT_THREAD_STATE } from "@t3tools/client-runtime/state/threads"; +import { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity"; +import { describe, expect, it } from "vite-plus/test"; + +import { createThreadViews, type ShownViewHints, type ThreadViewConfig } from "./threadView"; + +const ref = { environmentId: EnvironmentId.make("environment-1"), threadId: ThreadId.make("t-1") }; + +const configWith = (mode: string | null, capable = true): ThreadViewConfig => ({ + environment: { capabilities: capable ? { threadShownHistory: true } : {} }, + settings: { threadDisplayModes: mode === null ? {} : { [ref.threadId]: mode } }, +}); + +/** A seam over a base whose states record each view they open, as a subscription would. */ +function setup(hinted: boolean) { + const config = Atom.make(null); + const keys = new Set(hinted ? [threadKey(ref)] : []); + const hints: ShownViewHints = { + has: (key) => keys.has(key), + set: (key, shown) => void (shown ? keys.add(key) : keys.delete(key)), + }; + const opened: boolean[] = []; + const baseFamily = Atom.family((shownOnly: boolean) => + Atom.make(() => { + opened.push(shownOnly); + return AsyncResult.success(EMPTY_ENVIRONMENT_THREAD_STATE); + }).pipe(Atom.setIdleTTL(0)), + ); + const views = createThreadViews(() => config, hints); + const states = views.wrap({ stateAtom: (_e, _t, shownOnly = false) => baseFamily(shownOnly) }); + const tasks: Array<() => void> = []; + const registry = AtomRegistry.make({ + scheduleTask: (task) => { + tasks.push(task); + return () => {}; + }, + }); + const unmount = registry.mount(states.stateAtom(ref.environmentId, ref.threadId)); + const setConfig = (value: ThreadViewConfig) => { + registry.set(config, value); + for (let task = tasks.shift(); task !== undefined; task = tasks.shift()) task(); + }; + return { setConfig, keys, opened, unmount }; +} + +describe("display-mode thread views", () => { + it("opens only the shown view on a cold start of a thread last seen in a display mode", () => { + const h = setup(true); + h.setConfig(configWith("messages-only")); + expect(h.opened).toEqual([true]); + h.unmount(); + }); + + it("opens only the Normal view of a thread with no hint, and keeps Normal on an old server", () => { + const h = setup(false); + h.setConfig(configWith("chat-replies", false)); + expect(h.opened).toEqual([false]); + expect(h.keys.size).toBe(0); + h.unmount(); + }); + + it("switches views when the mode changes, and remembers the answer", () => { + const h = setup(false); + h.setConfig(configWith("chat-replies")); + expect(h.keys.has(threadKey(ref))).toBe(true); + h.setConfig(configWith("messages-dashboard")); + h.setConfig(configWith(null)); + expect(h.opened).toEqual([false, true, false]); + expect(h.keys.size).toBe(0); + h.unmount(); + }); +}); diff --git a/apps/web/src/fork/displayMode/threadView.ts b/apps/web/src/fork/displayMode/threadView.ts new file mode 100644 index 0000000000000..d67942e0e2c6d --- /dev/null +++ b/apps/web/src/fork/displayMode/threadView.ts @@ -0,0 +1,115 @@ +import { useAtomValue } from "@effect/atom-react"; +import { parseThreadKey, threadKey } from "@t3tools/client-runtime/state/entities"; +import type { EnvironmentThreadState } from "@t3tools/client-runtime/state/threads"; +import type { EnvironmentId, ScopedThreadRef, ThreadId } from "@t3tools/contracts"; +import { resolveDisplayMode } from "@t3tools/shared/displayModes"; +import * as Schema from "effect/Schema"; +import { type AsyncResult, Atom } from "effect/reactivity"; + +import { getLocalStorageItem, setLocalStorageItem } from "../../hooks/useLocalStorage"; +import { serverEnvironment } from "../../state/server"; + +/** The threads this viewer last saw in a display mode, by thread key. */ +export interface ShownViewHints { + readonly has: (key: string) => boolean; + readonly set: (key: string, shown: boolean) => void; +} + +const HINTS_KEY = "t3code:shown-view-threads:v1"; +const HINTS_KEPT = 200; +const HintKeys = Schema.Array(Schema.String); + +function localStorageHints(): ShownViewHints { + let keys: Set | undefined; + const load = () => { + if (keys !== undefined) return keys; + try { + keys = new Set(getLocalStorageItem(HINTS_KEY, HintKeys) ?? []); + } catch { + keys = new Set(); + } + return keys; + }; + return { + has: (key) => load().has(key), + set: (key, shown) => { + const current = load(); + if (current.has(key) === shown) return; + if (shown) current.add(key); + else current.delete(key); + for (const oldest of current) { + if (current.size <= HINTS_KEPT) break; + current.delete(oldest); + } + try { + setLocalStorageItem(HINTS_KEY, [...current], HintKeys); + } catch { + return; + } + }, + }; +} + +/** What the view of a thread depends on in its server's config. */ +export interface ThreadViewConfig { + readonly environment: { readonly capabilities: { readonly threadShownHistory?: boolean } }; + readonly settings: { readonly threadDisplayModes: Readonly> }; +} + +type ThreadStates = { + readonly stateAtom: ( + environmentId: EnvironmentId, + threadId: ThreadId, + shownOnly?: boolean, + ) => Atom.Atom>; +}; + +/** + * A thread shows its shown-only view when its server serves one and its display mode draws only + * messages. Until the server's config arrives, the viewer's last answer stands in, so a cold start + * opens one view and does not switch. + */ +export function createThreadViews( + configAtom: (environmentId: EnvironmentId) => Atom.Atom, + hints: ShownViewHints, +) { + const shownFamily = Atom.family((key: string) => { + const { environmentId, threadId } = parseThreadKey(key); + return Atom.make((get) => { + const config = get(configAtom(environmentId)); + if (config === null) return hints.has(key); + const shown = + config.environment.capabilities.threadShownHistory === true && + resolveDisplayMode(config.settings.threadDisplayModes[threadId]).visibleCores !== null; + hints.set(key, shown); + return shown; + }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`web-thread-shown-view:${key}`)); + }); + + return { + shownAtom: (ref: ScopedThreadRef) => shownFamily(threadKey(ref)), + /** Each thread mounts exactly one of its views from `base`, and Normal threads the same state as before. */ + wrap: (base: ThreadStates) => { + const stateFamily = Atom.family((key: string) => { + const { environmentId, threadId } = parseThreadKey(key); + return Atom.make((get) => + get(base.stateAtom(environmentId, threadId, get(shownFamily(key)))), + ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`web-thread-view-state:${key}`)); + }); + return { + stateAtom: (environmentId: EnvironmentId, threadId: ThreadId) => + stateFamily(threadKey({ environmentId, threadId })), + }; + }, + }; +} + +const threadViews = createThreadViews(serverEnvironment.configValueAtom, localStorageHints()); + +export const displayModeThreadStates = threadViews.wrap; + +const NOT_SHOWN = Atom.make(false).pipe(Atom.withLabel("web-thread-shown-view:none")); + +/** Whether the thread's state holds only what its display mode shows. */ +export const useThreadShownView = (ref: ScopedThreadRef | null): boolean => + useAtomValue(ref === null ? NOT_SHOWN : threadViews.shownAtom(ref)); diff --git a/apps/web/src/fork/displayMode/useThreadDisplayMode.ts b/apps/web/src/fork/displayMode/useThreadDisplayMode.ts index 378916d062d04..c9e59cb9bb24a 100644 --- a/apps/web/src/fork/displayMode/useThreadDisplayMode.ts +++ b/apps/web/src/fork/displayMode/useThreadDisplayMode.ts @@ -56,7 +56,10 @@ export function useDisplayModeTimeline( ): { readonly entries: ReadonlyArray; readonly repliesChosen: boolean; - /** The key that the thread-switch cache keeps these rows under. A sub-thread's view keeps none. */ + /** + * The key that the thread-switch cache keeps these rows under, one per mode, since modes load + * different views. A sub-thread's view keeps none. + */ readonly rememberKey: string | null; } { const mode = useThreadDisplayMode(threadRef); @@ -64,7 +67,8 @@ export function useDisplayModeTimeline( return { entries: applyDisplayMode(mode, entries, mode.subThreads ? subThread : null), repliesChosen: mode.visibleCores !== null, - rememberKey: subThread === null ? activeThreadKey : null, + rememberKey: + subThread === null && activeThreadKey !== null ? `${activeThreadKey}\u0000${mode.id}` : null, }; } diff --git a/apps/web/src/fork/subThreads/SubThreadsPanel.tsx b/apps/web/src/fork/subThreads/SubThreadsPanel.tsx index 7bd5db430a34b..2a0c744587c6d 100644 --- a/apps/web/src/fork/subThreads/SubThreadsPanel.tsx +++ b/apps/web/src/fork/subThreads/SubThreadsPanel.tsx @@ -1,3 +1,4 @@ +import { useAtomValue } from "@effect/atom-react"; import { scopedThreadKey } from "@t3tools/client-runtime/environment"; import type { ScopedThreadRef, ThreadSubThread } from "@t3tools/contracts"; import { ArrowLeftIcon } from "lucide-react"; @@ -9,6 +10,8 @@ import { RelativeTime } from "../../components/RelativeTime"; import { ComposerHandleContext } from "../../composerHandleContext"; import type { SideChatParentPanel } from "../../lib/sideChatPane"; import { cn } from "../../lib/utils"; +import { useQuietThreadProjection, useThreadHistory } from "../../state/entities"; +import { environmentThreadDetails, threadEnvironment } from "../../state/threads"; import { useAtomCommand } from "../../state/use-atom-command"; import { useThreadDisplayMode } from "../displayMode/useThreadDisplayMode"; import { @@ -44,7 +47,14 @@ export function SubThreadsPanel(props: { // A rewind can remove the first message of the sub-thread on screen; the list takes its place. const shown = chosen !== null && summaries.has(chosen) ? chosen : null; if (shown === null) { - return ; + return ( + + ); } return ( ; readonly states: ReadonlyMap; readonly onOpen: (first: string) => void; }) { const [unreadOnly, setUnreadOnly] = useState(false); + const loaded = useQuietThreadProjection(props.threadRef) !== null; + const syncError = useAtomValue(environmentThreadDetails.errorAtom(props.threadRef)); + const history = useThreadHistory(props.threadRef); + const loadEarlier = useAtomCommand(threadEnvironment.loadEarlierHistory, { + label: "load earlier replies", + reportFailure: false, + }); + const emptyText = unreadOnly + ? "No unread replies." + : !loaded + ? (syncError ?? "Loading replies…") + : history.hasMoreHistory + ? "No replies in the loaded messages." + : "No replies yet. Point at a message in the chat and choose Reply."; const rows = [...props.summaries.values()] .filter((summary) => summary.replies > 0) .map((summary) => { @@ -97,10 +122,8 @@ function SubThreadList(props: {
{rows.length === 0 ? ( -

- {unreadOnly - ? "No unread replies." - : "No replies yet. Point at a message in the chat and choose Reply."} +

+ {emptyText}

) : ( rows.map(({ summary, state, unread }) => ( @@ -132,6 +155,26 @@ function SubThreadList(props: { )) )} + {loaded && (history.hasMoreHistory || history.error !== null) ? ( +
+ {history.error !== null ?

{history.error}

: null} + {history.hasMoreHistory ? ( + + ) : null} +
+ ) : null}
); diff --git a/apps/web/src/state/threads.ts b/apps/web/src/state/threads.ts index 6d2ce2a07c90b..56a9a85be6bea 100644 --- a/apps/web/src/state/threads.ts +++ b/apps/web/src/state/threads.ts @@ -17,6 +17,7 @@ import * as Option from "effect/Option"; import { AsyncResult, Atom } from "effect/reactivity"; import { environmentCatalog } from "../connection/catalog"; +import { displayModeThreadStates } from "../fork/displayMode/threadView"; import { connectionAtomRuntime } from "../connection/runtime"; import { serverEnvironment } from "./server"; import { environmentShell } from "./shell"; @@ -33,7 +34,9 @@ export const threadEnvironment = createThreadEnvironmentAtoms( connectionAtomRuntime, threadFamilies.snapshotAtom, ); -const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime); +const environmentThreads = displayModeThreadStates( + createEnvironmentThreadStateAtoms(connectionAtomRuntime), +); export const environmentThreadDetails = createEnvironmentThreadDetailAtoms( environmentThreads.stateAtom, ); diff --git a/packages/client-runtime/src/fork/shownOnlyThread.ts b/packages/client-runtime/src/fork/shownOnlyThread.ts new file mode 100644 index 0000000000000..678688cd87aa5 --- /dev/null +++ b/packages/client-runtime/src/fork/shownOnlyThread.ts @@ -0,0 +1,33 @@ +import * as Effect from "effect/Effect"; + +import type * as Persistence from "../platform/persistence.ts"; + +const SHOWN_KEY_SUFFIX = "\u0000shown"; + +/** + * The state key of a thread's view: its thread key, or for the view that holds only what a + * display mode shows, the thread key with a suffix. Each view has its own state and resume entry. + */ +export const threadViewKey = (threadKey: string, shownOnly: boolean): string => + shownOnly ? `${threadKey}${SHOWN_KEY_SUFFIX}` : threadKey; + +export function parseThreadViewKey(key: string): { + readonly threadKey: string; + readonly shownOnly: boolean; +} { + return key.endsWith(SHOWN_KEY_SUFFIX) + ? { threadKey: key.slice(0, -SHOWN_KEY_SUFFIX.length), shownOnly: true } + : { threadKey: key, shownOnly: false }; +} + +/** + * The cache a shown-only state uses. The saved thread snapshot is the Normal view's, so a shown + * state neither loads it nor saves over it; removing a deleted thread still reaches the store. + */ +export const withoutThreadSnapshots = ( + store: Persistence.EnvironmentCacheStore["Service"], +): Persistence.EnvironmentCacheStore["Service"] => ({ + ...store, + loadThread: () => Effect.succeedNone, + saveThread: () => Effect.void, +}); diff --git a/packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts b/packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts index e3e9b5bb57fec..4d5c852aa1b73 100644 --- a/packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts +++ b/packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts @@ -27,6 +27,7 @@ export const fetchEnvironmentBoundedThreadSnapshot = Effect.fn( RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization["Service"] >; readonly timeoutMs?: number; + readonly shownOnly?: boolean | undefined; }) { const endpoint = { params: { threadId: input.threadId }, @@ -35,6 +36,7 @@ export const fetchEnvironmentBoundedThreadSnapshot = Effect.fn( dedupeBodies: "1", compactIds: "1", deferWork: "1", + ...(input.shownOnly ? { shownOnly: "1" } : {}), }, }; return yield* executeAuthenticatedEnvironmentHttpRequest({ @@ -76,12 +78,14 @@ export const layer: Layer.Layer< load: Effect.fn("ThreadSnapshotLoader.load")(function ( prepared: PreparedConnection, threadId: ThreadId, + shownOnly?: boolean, ) { return fetchEnvironmentBoundedThreadSnapshot({ prepared, threadId, signer, remoteAuthorization, + shownOnly, }).pipe( Effect.map((bounded): ThreadSnapshotLoader.ThreadSnapshotLoadResult => ({ _tag: "present", diff --git a/packages/client-runtime/src/state/threadSnapshotHttp.ts b/packages/client-runtime/src/state/threadSnapshotHttp.ts index f03ea60bfdf4a..e38ce900194d4 100644 --- a/packages/client-runtime/src/state/threadSnapshotHttp.ts +++ b/packages/client-runtime/src/state/threadSnapshotHttp.ts @@ -25,6 +25,7 @@ export class ThreadSnapshotLoader extends Context.Service< readonly load: ( prepared: PreparedConnection, threadId: ThreadId, + shownOnly?: boolean, ) => Effect.Effect; } >()("@t3tools/client-runtime/state/threadSnapshotHttp/ThreadSnapshotLoader") {} diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 3e8f3709a915e..a1395262849dd 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -31,6 +31,7 @@ import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as ConnectionWakeups from "../connection/wakeups.ts"; import * as Persistence from "../platform/persistence.ts"; import { runCachePersistence } from "./cachePersistence.ts"; +import * as ShownOnly from "../fork/shownOnlyThread.ts"; import * as ManagedRelay from "../relay/managedRelay.ts"; import { isAuthorizationFailure, request, subscribeDynamic } from "../rpc/client.ts"; import { parseThreadKey, threadKey } from "./entities.ts"; @@ -205,9 +206,11 @@ function hasMissingPendingQuestionCard(projection: OrchestrationV2ThreadProjecti export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make")(function* ( threadId: ThreadIdType, resumeCache?: ThreadResumeCache, + shownOnly = false, ) { const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor; - const cache = yield* Persistence.EnvironmentCacheStore; + const store = yield* Persistence.EnvironmentCacheStore; + const cache = shownOnly ? ShownOnly.withoutThreadSnapshots(store) : store; const snapshotLoader = yield* ThreadSnapshotLoader.ThreadSnapshotLoader; const historyController = yield* Effect.serviceOption( ThreadHistoryController.ThreadHistoryController, @@ -753,7 +756,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make const views = workAnchorId !== undefined ? ["work" as const] - : throughEntryId === undefined + : throughEntryId === undefined || shownOnly ? [undefined] : ["conversation" as const, undefined]; let nextCursor: string | null = null; @@ -776,6 +779,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make (workAnchorId !== undefined ? page.nextCursor !== null : throughEntryId === undefined && + !shownOnly && page.hasMoreHistory && !page.items.some( ({ item }) => item.type === "user_message" && item.createdBy === "user", @@ -1012,7 +1016,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ), ); const httpResult: ThreadSnapshotLoader.ThreadSnapshotLoadResult = - yield* snapshotLoader.load(prepared, threadId); + yield* snapshotLoader.load(prepared, threadId, shownOnly); switch (httpResult._tag) { case "present": { if (canLoadHistory && httpResult.history !== undefined) { @@ -1070,6 +1074,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make acceptCompactIds: true as const, } : {}), + ...(shownOnly ? { shownOnly: true as const } : {}), acceptDedupedBodies: true as const, }; }), @@ -1100,11 +1105,14 @@ function threadStateChanges( environmentId: EnvironmentIdType, threadId: ThreadIdType, resumeCache?: ThreadResumeCache, + shownOnly?: boolean, ) { return followStreamInEnvironment( environmentId, Stream.unwrap( - makeEnvironmentThreadState(threadId, resumeCache).pipe(Effect.map(SubscriptionRef.changes)), + makeEnvironmentThreadState(threadId, resumeCache, shownOnly).pipe( + Effect.map(SubscriptionRef.changes), + ), ), ); } @@ -1130,14 +1138,15 @@ export function createEnvironmentThreadStateAtoms( ), ); const family = Atom.family((key: string) => { - const { environmentId, threadId } = parseThreadKey(key); + const view = ShownOnly.parseThreadViewKey(key); + const { environmentId, threadId } = parseThreadKey(view.threadKey); const resumeAtom = resumeFamily(key); return runtime .atom( (get) => { get.mount(resumeAtom); const resume = get.once(resumeAtom); - const live = threadStateChanges(environmentId, threadId, resume); + const live = threadStateChanges(environmentId, threadId, resume, view.shownOnly); return resume.snapshot === undefined ? live : Stream.concat(Stream.succeed(cachedThreadState(resume.snapshot.state)), live); @@ -1150,8 +1159,8 @@ export function createEnvironmentThreadStateAtoms( }); return { - stateAtom: (environmentId: EnvironmentIdType, threadId: ThreadIdType) => - family(threadKey({ environmentId, threadId })), + stateAtom: (environmentId: EnvironmentIdType, threadId: ThreadIdType, shownOnly = false) => + family(ShownOnly.threadViewKey(threadKey({ environmentId, threadId }), shownOnly)), }; }