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.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/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.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/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)), }; } 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;