From 6af7ef09ef56c47cb5bd42272e41378b3b7cd2d0 Mon Sep 17 00:00:00 2001 From: tarik02 Date: Sat, 15 Aug 2026 12:31:49 +0300 Subject: [PATCH] feat(thread): page transcripts with upstream cursors --- .changeset/quick-threads-page.md | 5 ++ README.md | 6 ++- skills/t3code-cli/reference/commands.md | 9 +++- src/application/error.ts | 4 +- src/application/service.ts | 11 +++- src/application/threads.ts | 5 +- src/cli/format/thread.ts | 32 ++++++++--- src/cli/threads/messages.ts | 36 +++++++++++-- src/connection/prepared.ts | 14 ++++- src/orchestration/error.ts | 10 ++++ src/orchestration/index.ts | 1 + src/orchestration/layer.ts | 72 +++++++++++++++++++++++++ src/orchestration/service.ts | 14 +++-- src/rpc/ws-group.ts | 1 + src/runtime/layer.ts | 13 +++-- upstream-t3code | 2 +- 16 files changed, 208 insertions(+), 27 deletions(-) create mode 100644 .changeset/quick-threads-page.md create mode 100644 src/orchestration/error.ts diff --git a/.changeset/quick-threads-page.md b/.changeset/quick-threads-page.md new file mode 100644 index 0000000..3a1bfcf --- /dev/null +++ b/.changeset/quick-threads-page.md @@ -0,0 +1,5 @@ +--- +"t3code-cli": minor +--- + +page thread transcripts with upstream user-turn cursors diff --git a/README.md b/README.md index d96dfca..a33e8d4 100644 --- a/README.md +++ b/README.md @@ -194,10 +194,14 @@ t3cli list [--project ] [--archived | --all] t3cli search [--limit <1-50>] # Search conversation content t3cli show [--thread ] # Show thread details t3cli send [--thread ] [message] # Send message to thread -t3cli transcript [--thread ] [--limit] # View messages +t3cli transcript [--thread ] [--turn-limit N] [--before-cursor ] [--all] # View messages t3cli wait [--thread ] # Wait for completion ``` +`transcript` loads the latest 10 user turns by default and includes pagination metadata in JSON +output. Pass the returned `page.beforeCursor` to `--before-cursor` for the next older page, use +`--turn-limit` to set the page size, or use `--all` to load the complete transcript. + ### Advanced Thread Commands ```sh diff --git a/skills/t3code-cli/reference/commands.md b/skills/t3code-cli/reference/commands.md index 3802ba7..6567220 100644 --- a/skills/t3code-cli/reference/commands.md +++ b/skills/t3code-cli/reference/commands.md @@ -109,10 +109,17 @@ t3cli send [--thread ] [--force|-f] [message] [--stdin] [--wait] [--format auto|human|json|ndjson] t3cli show [--thread ] [--format auto|human|json] -t3cli transcript [--thread ] [--limit N] [--full] [--format json] +t3cli transcript [--thread ] [--limit N] + [--turn-limit N] [--before-cursor ] [--all] + [--full] [--format auto|human|json] t3cli wait [--thread ] [--format auto|human|ndjson] ``` +`transcript` loads the latest 10 user turns by default. Older-page requests default to 20 turns. +JSON output includes `page.beforeCursor` and `page.hasMore`; pass the cursor to `--before-cursor` to +load the next older page. `--turn-limit` sets either page size. `--all` loads the full thread and +cannot be combined with the paging flags. `--limit` only caps messages rendered in human output. + ## terminal Thread scope uses `--thread` or `T3CODE_THREAD_ID`. Terminal ids remain positional arguments. diff --git a/src/application/error.ts b/src/application/error.ts index be695ae..6347c24 100644 --- a/src/application/error.ts +++ b/src/application/error.ts @@ -1,4 +1,4 @@ import type { DomainError } from "../domain/error.ts"; -import type { RpcError } from "../rpc/error.ts"; +import type { OrchestrationError } from "../orchestration/service.ts"; -export type ApplicationError = DomainError | RpcError; +export type ApplicationError = DomainError | OrchestrationError; diff --git a/src/application/service.ts b/src/application/service.ts index c686f77..3b9e072 100644 --- a/src/application/service.ts +++ b/src/application/service.ts @@ -9,6 +9,8 @@ import type { OrchestrationSearchThreadsInput, OrchestrationShellSnapshot, OrchestrationThread, + OrchestrationThreadDetailSnapshot, + OrchestrationThreadDetailWindow, OrchestrationThreadShell, ProjectScript, ProjectScriptIcon, @@ -69,6 +71,11 @@ export interface SnoozeThreadInput { export type ListThreadsInclude = "active" | "archived" | "all"; +export interface GetThreadMessagesInput { + readonly threadId: string; + readonly window?: OrchestrationThreadDetailWindow; +} + export type UpdateThreadInput = { readonly threadId: string; readonly title?: string; @@ -218,8 +225,8 @@ export type T3ThreadApplicationService = { ApplicationError >; readonly getThreadMessages: ( - threadId: string, - ) => Effect.Effect; + input: GetThreadMessagesInput, + ) => Effect.Effect; readonly showThread: (threadId: string) => Effect.Effect; readonly approveThread: (input: { readonly threadId: string; diff --git a/src/application/threads.ts b/src/application/threads.ts index 5b0dec1..a182fd8 100644 --- a/src/application/threads.ts +++ b/src/application/threads.ts @@ -7,6 +7,7 @@ import { T3Orchestration } from "../orchestration/service.ts"; import { ProjectLookupError, ThreadLookupError, ThreadSessionError } from "../domain/error.ts"; import { resolveProjectScope } from "../domain/helpers.ts"; import { + type GetThreadMessagesInput, type ListThreadsInclude, type SnoozeThreadInput, type StartThreadInput, @@ -112,9 +113,9 @@ export const makeThreadApplication = Effect.fn("makeThreadApplication")(function }); }); const getThreadMessages = Effect.fn("T3ApplicationLive.getThreadMessages")(function* ( - threadId: string, + input: GetThreadMessagesInput, ) { - return yield* orchestration.getThreadSnapshot(threadId); + return yield* orchestration.getThreadDetailSnapshot(input); }); const showThread = Effect.fn("T3ApplicationLive.showThread")(function* (threadId: string) { const thread = yield* orchestration.getThreadSnapshot(threadId); diff --git a/src/cli/format/thread.ts b/src/cli/format/thread.ts index 95b6e93..541ee23 100644 --- a/src/cli/format/thread.ts +++ b/src/cli/format/thread.ts @@ -1,6 +1,10 @@ import type { ThreadSearchResult, ThreadShow } from "../../application/threads.ts"; import type { WaitEvent } from "../../application/service.ts"; -import type { OrchestrationThread, OrchestrationThreadShell } from "@t3tools/contracts"; +import type { + OrchestrationThread, + OrchestrationThreadDetailSnapshot, + OrchestrationThreadShell, +} from "@t3tools/contracts"; import { latestAssistantMessage, threadStatus } from "../../domain/thread-lifecycle.ts"; import { formatChatTranscript, formatRecord, formatTable } from "./human.ts"; @@ -139,9 +143,15 @@ export function formatThreadStartedHuman(input: { ])}`; } -export function formatThreadMessagesHuman(thread: OrchestrationThread, limit: number) { - const messages = limit === 0 ? thread.messages : thread.messages.slice(-limit); - return formatChatTranscript(messages); +export function formatThreadMessagesHuman( + snapshot: OrchestrationThreadDetailSnapshot, + limit: number, +) { + const messages = limit === 0 ? snapshot.thread.messages : snapshot.thread.messages.slice(-limit); + const transcript = formatChatTranscript(messages); + return snapshot.page?.hasMore === true && snapshot.page.beforeCursor !== null + ? `${transcript}\nearlier turns available\nbefore cursor: ${snapshot.page.beforeCursor}\n` + : transcript; } export function formatWaitDoneHuman(thread: OrchestrationThread) { @@ -151,8 +161,18 @@ export function formatWaitDoneHuman(thread: OrchestrationThread) { }`; } -export function formatThreadMessagesJson(thread: OrchestrationThread, full: boolean) { - return full ? thread : { thread: stripThreadMessages(thread), messages: thread.messages }; +export function formatThreadMessagesJson( + snapshot: OrchestrationThreadDetailSnapshot, + full: boolean, +) { + return full + ? { ...snapshot, page: snapshot.page ?? null } + : { + snapshotSequence: snapshot.snapshotSequence, + thread: stripThreadMessages(snapshot.thread), + messages: snapshot.thread.messages, + page: snapshot.page ?? null, + }; } export function formatWaitEventNdjson(event: WaitEvent) { diff --git a/src/cli/threads/messages.ts b/src/cli/threads/messages.ts index 32b2e82..11f2d40 100644 --- a/src/cli/threads/messages.ts +++ b/src/cli/threads/messages.ts @@ -4,7 +4,7 @@ import { Command, Flag } from "effect/unstable/cli"; import { extraArgsConfig } from "../extra-args.ts"; import { formatFlag, threadFlag } from "../flags.ts"; -import { InvalidLimitError } from "../error.ts"; +import { InvalidFlagCombinationError, InvalidLimitError } from "../error.ts"; import { MissingThreadError } from "../error.ts"; import { resolveThreadId } from "../scope/index.ts"; import { formatThreadMessagesHuman, formatThreadMessagesJson } from "../format/thread.ts"; @@ -19,17 +19,37 @@ export const getThreadTranscriptCommand = Command.make( { thread: threadFlag, limit: Flag.integer("limit").pipe(Flag.withDefault(20)), + turnLimit: Flag.integer("turn-limit").pipe(Flag.optional), + beforeCursor: Flag.string("before-cursor").pipe(Flag.optional), + all: Flag.boolean("all"), full: Flag.boolean("full"), format: formatFlag, ...extraArgsConfig, }, - ({ thread, limit, full, format }) => + ({ thread, limit, turnLimit, beforeCursor, all, full, format }) => Effect.gen(function* () { if (limit < 0) { return yield* Effect.fail( new InvalidLimitError({ message: `invalid limit: ${limit}`, value: String(limit) }), ); } + const turnLimitValue = Option.getOrUndefined(turnLimit); + const beforeCursorValue = Option.getOrUndefined(beforeCursor); + if (turnLimitValue !== undefined && turnLimitValue <= 0) { + return yield* Effect.fail( + new InvalidLimitError({ + message: `invalid turn limit: ${turnLimitValue}`, + value: String(turnLimitValue), + }), + ); + } + if (all && (turnLimitValue !== undefined || beforeCursorValue !== undefined)) { + return yield* Effect.fail( + new InvalidFlagCombinationError({ + message: "--all cannot be combined with --turn-limit or --before-cursor", + }), + ); + } const application = yield* T3Application; const cliRuntime = yield* CliRuntime; const t3CliEnv = yield* loadT3CliEnv; @@ -46,7 +66,17 @@ export const getThreadTranscriptCommand = Command.make( ); } const resolvedFormat = resolveOutputFormat(format, cliRuntime, t3CliEnv, "json"); - const detail = yield* application.getThreadMessages(threadId); + const detail = yield* application.getThreadMessages({ + threadId, + ...(!all + ? { + window: { + turnLimit: turnLimitValue ?? (beforeCursorValue === undefined ? 10 : 20), + ...(beforeCursorValue !== undefined ? { beforeCursor: beforeCursorValue } : {}), + }, + } + : {}), + }); if (resolvedFormat === "json") { return yield* output.printJson(formatThreadMessagesJson(detail, full)); } diff --git a/src/connection/prepared.ts b/src/connection/prepared.ts index 0ab185a..3dfac61 100644 --- a/src/connection/prepared.ts +++ b/src/connection/prepared.ts @@ -16,10 +16,20 @@ import { T3CodeConnectionError } from "./error.ts"; import { T3CodeConnectionProvider } from "./service.ts"; import type { T3CodeConnection } from "./type.ts"; +export interface T3PreparedConnection extends PreparedConnection { + readonly httpAuthorization: { + readonly _tag: "Bearer"; + readonly token: string; + }; +} + export class T3PreparedConnectionProvider extends Context.Service< T3PreparedConnectionProvider, { - readonly get: Effect.Effect; + readonly get: Effect.Effect< + T3PreparedConnection, + ConnectionAttemptError | T3CodeConnectionError + >; } >()("t3cli/T3PreparedConnectionProvider") {} @@ -59,7 +69,7 @@ const makePreparedConnection = Effect.fn("makePreparedConnection")(function* ( label: descriptor.label, connectionId: descriptor.environmentId, }), - } satisfies PreparedConnection; + } satisfies T3PreparedConnection; }); const makeT3PreparedConnectionProvider = Effect.fn("makeT3PreparedConnectionProvider")( diff --git a/src/orchestration/error.ts b/src/orchestration/error.ts new file mode 100644 index 0000000..d034a41 --- /dev/null +++ b/src/orchestration/error.ts @@ -0,0 +1,10 @@ +import * as Schema from "effect/Schema"; + +export class ThreadSnapshotRequestError extends Schema.TaggedErrorClass()( + "ThreadSnapshotRequestError", + { + message: Schema.String, + threadId: Schema.String, + cause: Schema.Defect(), + }, +) {} diff --git a/src/orchestration/index.ts b/src/orchestration/index.ts index 7363bca..867f8e5 100644 --- a/src/orchestration/index.ts +++ b/src/orchestration/index.ts @@ -4,4 +4,5 @@ export { type Orchestration, type OrchestrationError, } from "./service.ts"; +export { ThreadSnapshotRequestError } from "./error.ts"; export { makeT3Orchestration, T3OrchestrationLive } from "./layer.ts"; diff --git a/src/orchestration/layer.ts b/src/orchestration/layer.ts index dd3653e..ca52883 100644 --- a/src/orchestration/layer.ts +++ b/src/orchestration/layer.ts @@ -3,6 +3,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Sink from "effect/Sink"; import * as Stream from "effect/Stream"; +import { HttpClient } from "effect/unstable/http"; import { ORCHESTRATION_WS_METHODS, ThreadId, @@ -12,14 +13,25 @@ import { type OrchestrationShellStreamItem, type OrchestrationThreadStreamItem, } from "@t3tools/contracts"; +import { environmentEndpointUrl } from "@t3tools/client-runtime/environment"; +import { + executeEnvironmentHttpRequest, + makeEnvironmentHttpApiClient, +} from "@t3tools/client-runtime/rpc"; import { applyShellStreamEvent } from "@t3tools/client-runtime/state/shell"; +import { T3PreparedConnectionProvider } from "../connection/prepared.ts"; import { RpcError } from "../rpc/error.ts"; import { T3RpcOperations } from "../rpc/operation.ts"; +import { ThreadSnapshotRequestError } from "./error.ts"; import { T3Orchestration, type OpenThread, type Orchestration } from "./service.ts"; +const THREAD_SNAPSHOT_TIMEOUT_MS = 30_000; + export const makeT3Orchestration = Effect.fn("makeT3Orchestration")(function* () { const rpc = yield* T3RpcOperations; + const preparedConnectionProvider = yield* T3PreparedConnectionProvider; + const httpClient = yield* HttpClient.HttpClient; const watchShellSnapshots: Orchestration["watchShellSnapshots"] = () => rpc @@ -133,6 +145,65 @@ export const makeT3Orchestration = Effect.fn("makeT3Orchestration")(function* () } return value.snapshot.thread; }); + const getThreadDetailSnapshot: Orchestration["getThreadDetailSnapshot"] = Effect.fn( + "T3OrchestrationLive.getThreadDetailSnapshot", + )(function* (input) { + const paginationSupported = + input.window === undefined + ? true + : (yield* getServerConfig()).threadSnapshotPagination === true; + const window = paginationSupported ? input.window : undefined; + const prepared = yield* preparedConnectionProvider.get.pipe( + Effect.mapError( + (cause) => + new ThreadSnapshotRequestError({ + message: "failed to prepare the thread snapshot request", + threadId: input.threadId, + cause, + }), + ), + ); + const threadId = ThreadId.make(input.threadId); + const requestUrl = environmentEndpointUrl( + prepared.httpBaseUrl, + `/api/orchestration/threads/${threadId}`, + ); + const client = yield* makeEnvironmentHttpApiClient(prepared.httpBaseUrl).pipe( + Effect.provideService(HttpClient.HttpClient, httpClient), + Effect.mapError( + (cause) => + new ThreadSnapshotRequestError({ + message: "failed to create the thread snapshot client", + threadId: input.threadId, + cause, + }), + ), + ); + return yield* executeEnvironmentHttpRequest( + requestUrl, + THREAD_SNAPSHOT_TIMEOUT_MS, + client.orchestration.threadSnapshot({ + params: { threadId }, + payload: { + ...(window !== undefined ? { turnLimit: window.turnLimit } : {}), + ...(window?.beforeCursor !== undefined ? { beforeCursor: window.beforeCursor } : {}), + }, + headers: { + authorization: `Bearer ${prepared.httpAuthorization.token}`, + }, + }), + ).pipe( + Effect.provideService(HttpClient.HttpClient, httpClient), + Effect.mapError( + (cause) => + new ThreadSnapshotRequestError({ + message: "failed to load the thread snapshot", + threadId: input.threadId, + cause, + }), + ), + ); + }); const openThread = Effect.fn("T3OrchestrationLive.openThread")(function* (threadId: string) { return yield* watchThreadItems(threadId).pipe( Stream.peel(Sink.head()), @@ -167,6 +238,7 @@ export const makeT3Orchestration = Effect.fn("makeT3Orchestration")(function* () getArchivedShellSnapshot, searchThreads, getThreadSnapshot, + getThreadDetailSnapshot, watchShellSnapshots, watchShellSequence, watchThreadItems, diff --git a/src/orchestration/service.ts b/src/orchestration/service.ts index 874a217..b6661c8 100644 --- a/src/orchestration/service.ts +++ b/src/orchestration/service.ts @@ -10,22 +10,26 @@ import type { OrchestrationSearchThreadsInput, OrchestrationSearchThreadsResult, OrchestrationThread, + OrchestrationThreadDetailSnapshot, + OrchestrationThreadDetailWindow, OrchestrationThreadStreamItem, ServerProviders, } from "@t3tools/contracts"; +import type { ThreadSnapshotRequestError } from "./error.ts"; import type { RpcError } from "../rpc/error.ts"; -export type OrchestrationError = RpcError; +export type OrchestrationError = RpcError | ThreadSnapshotRequestError; export type OpenThread = { readonly snapshot: OrchestrationThread; readonly events: Stream.Stream; }; -export type ServerConfigForCli = { +export interface ServerConfigForCli { readonly providers: ServerProviders; -}; + readonly threadSnapshotPagination?: boolean; +} export type Orchestration = { readonly dispatch: ( @@ -43,6 +47,10 @@ export type Orchestration = { readonly getThreadSnapshot: ( threadId: string, ) => Effect.Effect; + readonly getThreadDetailSnapshot: (input: { + readonly threadId: string; + readonly window?: OrchestrationThreadDetailWindow; + }) => Effect.Effect; readonly watchShellSnapshots: () => Stream.Stream< OrchestrationShellSnapshot, OrchestrationError, diff --git a/src/rpc/ws-group.ts b/src/rpc/ws-group.ts index a3761c7..e18279b 100644 --- a/src/rpc/ws-group.ts +++ b/src/rpc/ws-group.ts @@ -31,6 +31,7 @@ export const CliServerConfig = Schema.Struct({ }), }), providers: ServerProviders, + threadSnapshotPagination: Schema.optionalKey(Schema.Boolean), }); export type CliServerConfig = typeof CliServerConfig.Type; diff --git a/src/runtime/layer.ts b/src/runtime/layer.ts index 931134c..53d482d 100644 --- a/src/runtime/layer.ts +++ b/src/runtime/layer.ts @@ -63,18 +63,23 @@ const T3ConfigConnectionProviderLayer = Layer.effect( }), ).pipe(Layer.provide(T3ConfigLayer)); +const T3PreparedConnectionLayer = T3PreparedConnectionProviderLive.pipe( + Layer.provide(Layer.mergeAll(T3ConfigConnectionProviderLayer, NodeHttpClient.layerUndici)), +); const T3RpcLayer = T3RpcLive.pipe( Layer.provide( Layer.mergeAll( - T3PreparedConnectionProviderLive.pipe( - Layer.provide(Layer.mergeAll(T3ConfigConnectionProviderLayer, NodeHttpClient.layerUndici)), - ), + T3PreparedConnectionLayer, T3RpcSessionFactoryLive.pipe(Layer.provide(NodeSocket.layerWebSocketConstructor)), ), ), ); const T3RpcOperationsLayer = T3RpcOperationsLive.pipe(Layer.provide(T3RpcLayer)); -export const T3OrchestrationLayer = T3OrchestrationLive.pipe(Layer.provide(T3RpcOperationsLayer)); +export const T3OrchestrationLayer = T3OrchestrationLive.pipe( + Layer.provide( + Layer.mergeAll(T3RpcOperationsLayer, T3PreparedConnectionLayer, NodeHttpClient.layerUndici), + ), +); export const T3PreviewAutomationLayer = T3PreviewAutomationLive.pipe( Layer.provide(T3RpcOperationsLayer), ); diff --git a/upstream-t3code b/upstream-t3code index 3b72d17..f0ebc62 160000 --- a/upstream-t3code +++ b/upstream-t3code @@ -1 +1 @@ -Subproject commit 3b72d17cbca691f0b64e6d4a10c9e349f42873a5 +Subproject commit f0ebc628c6dd83fd0c7963078ad7778ce6028d0c