diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index bc7828dd854..6a87269165b 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -40,6 +40,9 @@ const failSetConfigOption = process.env.T3_ACP_FAIL_SET_CONFIG_OPTION === "1"; const exitOnSetConfigOption = process.env.T3_ACP_EXIT_ON_SET_CONFIG_OPTION === "1"; const promptResponseText = process.env.T3_ACP_PROMPT_RESPONSE_TEXT; const promptDelayMs = Number(process.env.T3_ACP_PROMPT_DELAY_MS ?? "0"); +const promptResponseChunkCount = Number(process.env.T3_ACP_PROMPT_RESPONSE_CHUNK_COUNT ?? "0"); +const promptResponseChunkText = process.env.T3_ACP_PROMPT_RESPONSE_CHUNK_TEXT; +const promptChunkDelayMs = Number(process.env.T3_ACP_PROMPT_CHUNK_DELAY_MS ?? "0"); const permissionOptionIds = { allowOnce: process.env.T3_ACP_ALLOW_ONCE_OPTION_ID ?? "allow-once", allowAlways: process.env.T3_ACP_ALLOW_ALWAYS_OPTION_ID ?? "allow-always", @@ -865,6 +868,28 @@ const program = Effect.gen(function* () { }, }); + if (Number.isFinite(promptResponseChunkCount) && promptResponseChunkCount > 0) { + // Emit the response as many small chunks to reproduce heavy streaming + // histories: with the `enableAssistantStreaming` server setting on, + // each chunk persists as its own `thread.message-sent` delta event. + for (let index = 0; index < promptResponseChunkCount; index += 1) { + yield* agent.client.sessionUpdate({ + sessionId: requestedSessionId, + update: { + sessionUpdate: "agent_message_chunk", + content: { + type: "text", + text: promptResponseChunkText ?? `chunk ${index} of a long streamed response. `, + }, + }, + }); + if (Number.isFinite(promptChunkDelayMs) && promptChunkDelayMs > 0) { + yield* Effect.sleep(`${promptChunkDelayMs} millis`); + } + } + return { stopReason: "end_turn" }; + } + yield* agent.client.sessionUpdate({ sessionId: requestedSessionId, update: { diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index a3b2e3715fd..1969dce0d06 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -5899,6 +5899,51 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect("subscribeThread sends a fresh snapshot instead of replaying a large gap", () => + Effect.gen(function* () { + let readEventsCalls = 0; + const thread = makeDefaultOrchestrationReadModel().threads[0]!; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + // Head is far ahead of the client's afterSequence (gap > 1000). + latestSequence: Effect.succeed(100_000), + readEvents: () => + Stream.sync(() => { + readEventsCalls += 1; + return {} as OrchestrationEvent; + }), + }, + projectionSnapshotQuery: { + getThreadDetailSnapshot: () => + Effect.succeed(Option.some({ snapshotSequence: 100_000, thread })), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: defaultThreadId, + afterSequence: 5, + requestCompletionMarker: true, + }).pipe(Stream.take(2), Stream.runCollect), + ), + ); + + const [first, second] = Array.from(items); + // Large gap => fresh snapshot, and the unbounded replay is never started. + assert.equal(first?.kind, "snapshot"); + if (first?.kind === "snapshot") { + assert.equal(first.snapshot.thread.id, thread.id); + } + assert.deepEqual(second, { kind: "synchronized" }); + assert.equal(readEventsCalls, 0); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("subscribeShell replaces a cursor ahead of the authoritative head", () => Effect.gen(function* () { let readEventsCalls = 0; diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 2a8be25a728..25f52bc64a3 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -298,6 +298,13 @@ const PROVIDER_STATUS_DEBOUNCE_MS = 200; // Matches the event store's default page size (DEFAULT_READ_FROM_SEQUENCE_LIMIT). const SHELL_RESUME_MAX_GAP = 1_000; +// Resuming a thread subscription replays every persisted event after the +// client's cursor. Past this gap a single snapshot is cheaper than streaming +// (and having the client fold) thousands of events — long agentic turns +// persist one thread.activity-appended per tool transition, so a thread left +// overnight can easily accumulate several thousand. +const THREAD_RESUME_MAX_GAP = 1_000; + const RPC_REQUIRED_SCOPE = new Map([ [ORCHESTRATION_WS_METHODS.dispatchCommand, AuthOrchestrationOperateScope], [ORCHESTRATION_WS_METHODS.getTurnDiff, AuthOrchestrationReadScope], @@ -1401,38 +1408,47 @@ const makeWsRpcLayer = ( // catch-up followed by the buffered/ongoing live events. Overlapping // events are deduped by sequence on the client. // - // Read the full range after the cursor (not the store's default - // page-bounded limit): the range is normally tiny (a fresh HTTP - // snapshot sequence) and the per-thread filter runs after reading, - // so a global cap could otherwise omit this thread's events. + // Read the exact range from the cursor through the head captured + // below (not the store's default page-bounded limit): the + // per-thread filter runs after reading, so a global cap could + // otherwise omit this thread's events. + // + // When the cursor is too far behind (or ahead of this engine's + // authoritative state), fall through to a fresh snapshot instead + // of replaying an unbounded backlog — mirroring the shell + // stream's SHELL_RESUME_MAX_GAP fallback. if (input.afterSequence !== undefined) { const afterSequence = input.afterSequence; - const catchUpStream = orchestrationEngine - .readEvents(afterSequence, Number.MAX_SAFE_INTEGER) - .pipe( - Stream.filter(isThisThreadDetailEvent), - Stream.map((event) => ({ - kind: "event" as const, - event: projectActivityEvent(event), - })), - Stream.mapError( - (cause) => - new OrchestrationGetSnapshotError({ - message: `Failed to replay thread ${input.threadId} events`, - cause, - }), - ), - ); - const afterCatchUp = - input.requestCompletionMarker === true - ? Stream.concat( - Stream.fromEffect( - Queue.offer(liveBuffer, { kind: "synchronized" as const }), - ).pipe(Stream.drain), - bufferedLiveStream, - ) - : bufferedLiveStream; - return Stream.concat(catchUpStream, afterCatchUp); + const headSequence = yield* orchestrationEngine.latestSequence; + const replayGap = headSequence - afterSequence; + if (replayGap >= 0 && replayGap <= THREAD_RESUME_MAX_GAP) { + const catchUpStream = orchestrationEngine + .readEvents(afterSequence, replayGap) + .pipe( + Stream.filter(isThisThreadDetailEvent), + Stream.map((event) => ({ + kind: "event" as const, + event: projectActivityEvent(event), + })), + Stream.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: `Failed to replay thread ${input.threadId} events`, + cause, + }), + ), + ); + const afterCatchUp = + input.requestCompletionMarker === true + ? Stream.concat( + Stream.fromEffect( + Queue.offer(liveBuffer, { kind: "synchronized" as const }), + ).pipe(Stream.drain), + bufferedLiveStream, + ) + : bufferedLiveStream; + return Stream.concat(catchUpStream, afterCatchUp); + } } const snapshot = yield* projectionSnapshotQuery diff --git a/apps/web/perf.html b/apps/web/perf.html new file mode 100644 index 00000000000..68c18b81b71 --- /dev/null +++ b/apps/web/perf.html @@ -0,0 +1,12 @@ + + + + + + Thread Replay Playground + + +
+ + + diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index bdde62a3e0d..a6375f90aaa 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -225,6 +225,12 @@ import { DraftHeroHeadline } from "./chat/DraftHeroHeadline"; import { ExpandedImageDialog } from "./chat/ExpandedImageDialog"; import { PullRequestThreadDialog } from "./PullRequestThreadDialog"; import { MessagesTimeline } from "./chat/MessagesTimeline"; +import { + deriveDisplayServerMessages, + deriveRevertTurnCountByUserMessageId, + deriveTimelineMessages, + deriveTurnDiffSummaryByAssistantMessageId, +} from "./chat/threadTimelineModel"; import { ChatHeader } from "./chat/ChatHeader"; import { PanelLayoutControls, RightPanelMaximizeControl } from "./chat/PanelLayoutControls"; import { type ExpandedImagePreview } from "./chat/ExpandedImagePreview"; @@ -2138,21 +2144,10 @@ function ChatViewContent(props: ChatViewProps) { ), [serverAttachmentIds, serverAttachmentUrls], ); - const displayServerMessages = useMemo>(() => { - if (!serverMessages) return []; - return serverMessages.map((message) => { - if (!message.attachments || message.attachments.length === 0) { - return message; - } - return { - ...message, - attachments: message.attachments.map((attachment) => { - const previewUrl = serverAttachmentUrlById.get(attachment.id); - return previewUrl ? { ...attachment, previewUrl } : attachment; - }), - }; - }); - }, [serverAttachmentUrlById, serverMessages]); + const displayServerMessages = useMemo>( + () => deriveDisplayServerMessages(serverMessages, serverAttachmentUrlById), + [serverAttachmentUrlById, serverMessages], + ); useEffect(() => { if (typeof Image === "undefined" || displayServerMessages.length === 0) { return; @@ -2237,58 +2232,15 @@ function ChatViewContent(props: ChatViewProps) { } }; }, [attachmentPreviewHandoffByMessageId, clearAttachmentPreviewHandoff, displayServerMessages]); - const timelineMessages = useMemo(() => { - const messages = displayServerMessages; - const serverMessagesWithPreviewHandoff = - Object.keys(attachmentPreviewHandoffByMessageId).length === 0 - ? messages - : // Spread only fires for the few messages that actually changed; - // unchanged ones early-return their original reference. - // In-place mutation would break React's immutable state contract. - messages.map((message) => { - if ( - message.role !== "user" || - !message.attachments || - message.attachments.length === 0 - ) { - return message; - } - const handoffPreviewUrls = attachmentPreviewHandoffByMessageId[message.id]; - if (!handoffPreviewUrls || handoffPreviewUrls.length === 0) { - return message; - } - - let changed = false; - let imageIndex = 0; - const attachments = message.attachments.map((attachment) => { - if (attachment.type !== "image") { - return attachment; - } - const handoffPreviewUrl = handoffPreviewUrls[imageIndex]; - imageIndex += 1; - if (!handoffPreviewUrl || attachment.previewUrl === handoffPreviewUrl) { - return attachment; - } - changed = true; - return { - ...attachment, - previewUrl: handoffPreviewUrl, - }; - }); - - return changed ? { ...message, attachments } : message; - }); - - if (optimisticUserMessages.length === 0) { - return serverMessagesWithPreviewHandoff; - } - const serverIds = new Set(serverMessagesWithPreviewHandoff.map((message) => message.id)); - const pendingMessages = optimisticUserMessages.filter((message) => !serverIds.has(message.id)); - if (pendingMessages.length === 0) { - return serverMessagesWithPreviewHandoff; - } - return [...serverMessagesWithPreviewHandoff, ...pendingMessages]; - }, [attachmentPreviewHandoffByMessageId, displayServerMessages, optimisticUserMessages]); + const timelineMessages = useMemo( + () => + deriveTimelineMessages({ + displayServerMessages, + attachmentPreviewHandoffByMessageId, + optimisticUserMessages, + }), + [attachmentPreviewHandoffByMessageId, displayServerMessages, optimisticUserMessages], + ); const timelineEntries = useMemo( () => deriveTimelineEntries(timelineMessages, activeThread?.proposedPlans ?? [], workLogEntries), @@ -2306,46 +2258,19 @@ function ChatViewContent(props: ChatViewProps) { ] = useDraftHeroLayoutTransition(isDraftHeroState); const { turnDiffSummaries, inferredCheckpointTurnCountByTurnId } = useTurnDiffSummaries(activeThread); - const turnDiffSummaryByAssistantMessageId = useMemo(() => { - const byMessageId = new Map(); - for (const summary of turnDiffSummaries) { - if (!summary.assistantMessageId) continue; - byMessageId.set(summary.assistantMessageId, summary); - } - return byMessageId; - }, [turnDiffSummaries]); - const revertTurnCountByUserMessageId = useMemo(() => { - const byUserMessageId = new Map(); - for (let index = 0; index < timelineEntries.length; index += 1) { - const entry = timelineEntries[index]; - if (!entry || entry.kind !== "message" || entry.message.role !== "user") { - continue; - } - - for (let nextIndex = index + 1; nextIndex < timelineEntries.length; nextIndex += 1) { - const nextEntry = timelineEntries[nextIndex]; - if (!nextEntry || nextEntry.kind !== "message") { - continue; - } - if (nextEntry.message.role === "user") { - break; - } - const summary = turnDiffSummaryByAssistantMessageId.get(nextEntry.message.id); - if (!summary) { - continue; - } - const turnCount = - summary.checkpointTurnCount ?? inferredCheckpointTurnCountByTurnId[summary.turnId]; - if (typeof turnCount !== "number") { - break; - } - byUserMessageId.set(entry.message.id, Math.max(0, turnCount - 1)); - break; - } - } - - return byUserMessageId; - }, [inferredCheckpointTurnCountByTurnId, timelineEntries, turnDiffSummaryByAssistantMessageId]); + const turnDiffSummaryByAssistantMessageId = useMemo( + () => deriveTurnDiffSummaryByAssistantMessageId(turnDiffSummaries), + [turnDiffSummaries], + ); + const revertTurnCountByUserMessageId = useMemo( + () => + deriveRevertTurnCountByUserMessageId({ + timelineEntries, + turnDiffSummaryByAssistantMessageId, + inferredCheckpointTurnCountByTurnId, + }), + [inferredCheckpointTurnCountByTurnId, timelineEntries, turnDiffSummaryByAssistantMessageId], + ); const gitCwd = activeProject ? projectScriptCwd({ diff --git a/apps/web/src/components/chat/threadTimelineModel.ts b/apps/web/src/components/chat/threadTimelineModel.ts new file mode 100644 index 00000000000..05399fff705 --- /dev/null +++ b/apps/web/src/components/chat/threadTimelineModel.ts @@ -0,0 +1,149 @@ +import type { MessageId, TurnId } from "@t3tools/contracts"; + +import type { TimelineEntry } from "../../session-logic"; +import type { ChatMessage, TurnDiffSummary } from "../../types"; + +/** + * Pure derivation steps that turn thread detail state into the inputs of + * MessagesTimeline. Extracted from ChatView so the pipeline can run outside + * the full app shell (perf harnesses, tests) and so its per-update cost is + * visible in one place: every function here re-runs on every thread state + * publication, including once per replayed streaming delta event. + */ + +/** Swaps message attachment previews to resolved asset URLs when available. */ +export function deriveDisplayServerMessages( + serverMessages: ReadonlyArray | undefined, + attachmentPreviewUrlById: ReadonlyMap, +): ReadonlyArray { + if (!serverMessages) return []; + return serverMessages.map((message) => { + if (!message.attachments || message.attachments.length === 0) { + return message; + } + return { + ...message, + attachments: message.attachments.map((attachment) => { + const previewUrl = attachmentPreviewUrlById.get(attachment.id); + return previewUrl ? { ...attachment, previewUrl } : attachment; + }), + }; + }); +} + +/** + * Overlays local attachment preview handoffs onto server messages and appends + * optimistic user messages the server has not echoed back yet. + */ +export function deriveTimelineMessages({ + displayServerMessages, + attachmentPreviewHandoffByMessageId, + optimisticUserMessages, +}: { + displayServerMessages: ReadonlyArray; + attachmentPreviewHandoffByMessageId: Record; + optimisticUserMessages: ReadonlyArray; +}): ReadonlyArray { + const messages = displayServerMessages; + const serverMessagesWithPreviewHandoff = + Object.keys(attachmentPreviewHandoffByMessageId).length === 0 + ? messages + : // Spread only fires for the few messages that actually changed; + // unchanged ones early-return their original reference. + // In-place mutation would break React's immutable state contract. + messages.map((message) => { + if (message.role !== "user" || !message.attachments || message.attachments.length === 0) { + return message; + } + const handoffPreviewUrls = attachmentPreviewHandoffByMessageId[message.id]; + if (!handoffPreviewUrls || handoffPreviewUrls.length === 0) { + return message; + } + + let changed = false; + let imageIndex = 0; + const attachments = message.attachments.map((attachment) => { + if (attachment.type !== "image") { + return attachment; + } + const handoffPreviewUrl = handoffPreviewUrls[imageIndex]; + imageIndex += 1; + if (!handoffPreviewUrl || attachment.previewUrl === handoffPreviewUrl) { + return attachment; + } + changed = true; + return { + ...attachment, + previewUrl: handoffPreviewUrl, + }; + }); + + return changed ? { ...message, attachments } : message; + }); + + if (optimisticUserMessages.length === 0) { + return serverMessagesWithPreviewHandoff; + } + const serverIds = new Set(serverMessagesWithPreviewHandoff.map((message) => message.id)); + const pendingMessages = optimisticUserMessages.filter((message) => !serverIds.has(message.id)); + if (pendingMessages.length === 0) { + return serverMessagesWithPreviewHandoff; + } + return [...serverMessagesWithPreviewHandoff, ...pendingMessages]; +} + +export function deriveTurnDiffSummaryByAssistantMessageId( + turnDiffSummaries: ReadonlyArray, +): Map { + const byMessageId = new Map(); + for (const summary of turnDiffSummaries) { + if (!summary.assistantMessageId) continue; + byMessageId.set(summary.assistantMessageId, summary); + } + return byMessageId; +} + +/** + * For each user message, the checkpoint turn count of the first following + * assistant message that has a turn diff summary. + */ +export function deriveRevertTurnCountByUserMessageId({ + timelineEntries, + turnDiffSummaryByAssistantMessageId, + inferredCheckpointTurnCountByTurnId, +}: { + timelineEntries: ReadonlyArray; + turnDiffSummaryByAssistantMessageId: ReadonlyMap; + inferredCheckpointTurnCountByTurnId: Record; +}): Map { + const byUserMessageId = new Map(); + for (let index = 0; index < timelineEntries.length; index += 1) { + const entry = timelineEntries[index]; + if (!entry || entry.kind !== "message" || entry.message.role !== "user") { + continue; + } + + for (let nextIndex = index + 1; nextIndex < timelineEntries.length; nextIndex += 1) { + const nextEntry = timelineEntries[nextIndex]; + if (!nextEntry || nextEntry.kind !== "message") { + continue; + } + if (nextEntry.message.role === "user") { + break; + } + const summary = turnDiffSummaryByAssistantMessageId.get(nextEntry.message.id); + if (!summary) { + continue; + } + const turnCount = + summary.checkpointTurnCount ?? inferredCheckpointTurnCountByTurnId[summary.turnId]; + if (typeof turnCount !== "number") { + break; + } + byUserMessageId.set(entry.message.id, Math.max(0, turnCount - 1)); + break; + } + } + + return byUserMessageId; +} diff --git a/apps/web/src/perf/ThreadReplayPlayground.tsx b/apps/web/src/perf/ThreadReplayPlayground.tsx new file mode 100644 index 00000000000..87c7e5b16b9 --- /dev/null +++ b/apps/web/src/perf/ThreadReplayPlayground.tsx @@ -0,0 +1,511 @@ +import { Profiler, useCallback, useEffect, useMemo, useRef, useState } from "react"; +import type { LegendListRef } from "@legendapp/list/react"; + +import { + EnvironmentId, + EventId, + MessageId, + ProjectId, + ProviderInstanceId, + ThreadId, + TurnId, + type OrchestrationEvent, + type OrchestrationMessage, + type OrchestrationThread, +} from "@t3tools/contracts"; +import { applyThreadDetailEvent } from "@t3tools/client-runtime/state/thread-reducer"; + +import { MessagesTimeline } from "../components/chat/MessagesTimeline"; +import { + deriveDisplayServerMessages, + deriveRevertTurnCountByUserMessageId, + deriveTimelineMessages, + deriveTurnDiffSummaryByAssistantMessageId, +} from "../components/chat/threadTimelineModel"; +import { deriveTimelineEntries, deriveWorkLogEntries } from "../session-logic"; + +/** + * Thread replay perf playground. + * + * Reproduces the "big chugging render" when a thread with a large backlog of + * persisted streaming deltas is reopened: the client applies the replayed + * `thread.message-sent` events one at a time, and every application publishes + * a new thread state that re-runs the timeline derivation and re-renders the + * streaming message's markdown. + * + * This page drives the real reducer (`applyThreadDetailEvent`), the real + * derivation pipeline (`threadTimelineModel` + `deriveTimelineEntries`), and + * the real `MessagesTimeline` component with synthetic events — no server, + * no provider, no auth. Served by the web dev server at /perf.html. + * + * Query params: + * deltas — number of streaming delta events to replay (default 2000) + * activities — number of thread.activity-appended (tool) events interleaved + * with the deltas (default 0); real agentic threads are dominated + * by these even with assistant streaming off + * batch — events applied per state publication (default 1 = today's + * behavior; raise to preview what client-side batching buys) + * history — settled message pairs present before the streaming one (default 8) + * code — 1 to include code fences in the streamed markdown (default 1) + * auto — 1 to start the replay on load + */ + +const THREAD_ID = ThreadId.make("thread-perf"); +const STREAMING_MESSAGE_ID = MessageId.make("msg-streaming"); +const STREAMING_TURN_ID = TurnId.make("turn-perf"); +const ENVIRONMENT_ID = EnvironmentId.make("environment-perf"); +const BASE_TIME_MS = Date.parse("2026-04-01T00:00:00.000Z"); + +const isoAt = (offsetSeconds: number) => + new Date(BASE_TIME_MS + offsetSeconds * 1000).toISOString(); + +function chunkText(index: number, withCode: boolean): string { + if (withCode && index % 40 === 12) { + return "\n\n```ts\nconst step" + index + " = replay(" + index + ");\n```\n\n"; + } + if (index % 23 === 0) { + return "\n\nParagraph " + index + " of the streamed reply covers another detail. "; + } + return "token" + index + " lorem ipsum dolor sit amet "; +} + +function historyMessage(pairIndex: number, role: "user" | "assistant"): OrchestrationMessage { + const offset = pairIndex * 60 + (role === "user" ? 0 : 30); + return { + id: MessageId.make(`msg-history-${pairIndex}-${role}`), + role, + text: + role === "user" + ? `History question ${pairIndex}: how does step ${pairIndex} work?` + : `History answer ${pairIndex} with some **markdown**, a list:\n\n- item one\n- item two\n\nand a fence:\n\n\`\`\`ts\nexport const answer${pairIndex} = ${pairIndex};\n\`\`\`\n`, + turnId: role === "assistant" ? TurnId.make(`turn-history-${pairIndex}`) : null, + streaming: false, + createdAt: isoAt(offset), + updatedAt: isoAt(offset), + }; +} + +function makeBaseThread(historyPairs: number): OrchestrationThread { + const messages: OrchestrationMessage[] = []; + for (let index = 0; index < historyPairs; index += 1) { + messages.push(historyMessage(index, "user"), historyMessage(index, "assistant")); + } + const startedAt = isoAt(historyPairs * 60); + messages.push({ + id: MessageId.make("msg-final-user"), + role: "user", + text: "Please stream a very long answer.", + turnId: null, + streaming: false, + createdAt: startedAt, + updatedAt: startedAt, + }); + return { + id: THREAD_ID, + projectId: ProjectId.make("project-perf"), + title: "Replay perf thread", + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + latestTurn: { + turnId: STREAMING_TURN_ID, + state: "running", + requestedAt: startedAt, + startedAt, + completedAt: null, + assistantMessageId: STREAMING_MESSAGE_ID, + }, + createdAt: isoAt(0), + updatedAt: startedAt, + archivedAt: null, + settledOverride: null, + settledAt: null, + deletedAt: null, + messages, + proposedPlans: [], + activities: [], + checkpoints: [], + session: { + threadId: THREAD_ID, + status: "running", + providerName: "codex", + runtimeMode: "full-access", + activeTurnId: STREAMING_TURN_ID, + lastError: null, + updatedAt: startedAt, + }, + }; +} + +function makeDeltaEvents( + count: number, + activityCount: number, + historyPairs: number, + withCode: boolean, +) { + const streamStartSeconds = historyPairs * 60 + 5; + const events: OrchestrationEvent[] = []; + const total = count + activityCount; + let textIndex = 0; + let activityIndex = 0; + // Interleave text deltas and tool activities evenly, mirroring the event + // mix of real agentic threads (which are dominated by + // thread.activity-appended even with assistant streaming off). + for (let index = 0; index < total; index += 1) { + const occurredAt = isoAt(streamStartSeconds + index); + const emitActivity = + activityIndex < activityCount && + (textIndex >= count || activityIndex / activityCount <= textIndex / count); + if (emitActivity) { + const toolCall = Math.floor(activityIndex / 2); + const started = activityIndex % 2 === 0; + events.push({ + eventId: EventId.make(`event-activity-${activityIndex}`), + sequence: index + 1, + occurredAt, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + aggregateKind: "thread", + aggregateId: THREAD_ID, + type: "thread.activity-appended", + payload: { + threadId: THREAD_ID, + activity: { + id: EventId.make(`activity-${activityIndex}`), + tone: "tool", + kind: started ? "tool.started" : "tool.completed", + summary: started ? `Tool call ${toolCall} started` : `Tool call ${toolCall} completed`, + payload: { itemType: "dynamic_tool_call", detail: `Tool ${toolCall}` }, + turnId: STREAMING_TURN_ID, + createdAt: occurredAt, + }, + }, + } as OrchestrationEvent); + activityIndex += 1; + continue; + } + events.push({ + eventId: EventId.make(`event-delta-${textIndex}`), + sequence: index + 1, + occurredAt, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + aggregateKind: "thread", + aggregateId: THREAD_ID, + type: "thread.message-sent", + payload: { + threadId: THREAD_ID, + messageId: STREAMING_MESSAGE_ID, + role: "assistant", + text: chunkText(textIndex, withCode), + turnId: STREAMING_TURN_ID, + streaming: true, + createdAt: isoAt(streamStartSeconds), + updatedAt: occurredAt, + }, + } as OrchestrationEvent); + textIndex += 1; + } + return events; +} + +const macrotask = () => + new Promise((resolve) => { + const channel = new MessageChannel(); + channel.port1.onmessage = () => { + channel.port1.close(); + resolve(); + }; + channel.port2.postMessage(null); + }); + +interface ReplayStats { + running: boolean; + done: boolean; + applied: number; + totalDeltas: number; + wallMs: number; + commits: number; + commitMs: number; + longTasks: number; + longTaskMs: number; + maxStallMs: number; +} + +const INITIAL_STATS: ReplayStats = { + running: false, + done: false, + applied: 0, + totalDeltas: 0, + wallMs: 0, + commits: 0, + commitMs: 0, + longTasks: 0, + longTaskMs: 0, + maxStallMs: 0, +}; + +function readParams() { + const search = new URLSearchParams(window.location.search); + const num = (key: string, fallback: number, minimum = 1) => { + const raw = Number(search.get(key)); + return Number.isFinite(raw) && raw >= minimum ? Math.floor(raw) : fallback; + }; + return { + deltas: num("deltas", 2000, 0), + activities: num("activities", 0, 0), + batch: num("batch", 1), + history: num("history", 8, 0), + code: search.get("code") !== "0", + auto: search.get("auto") === "1", + }; +} + +function StatsPanel({ statsRef }: { statsRef: React.RefObject }) { + const [snapshot, setSnapshot] = useState(INITIAL_STATS); + useEffect(() => { + const interval = window.setInterval(() => { + setSnapshot({ ...statsRef.current }); + }, 250); + return () => window.clearInterval(interval); + }, [statsRef]); + + const s = snapshot; + const status = s.running ? "replaying…" : s.done ? "done" : "idle"; + return ( +
+ status: {status} + + applied: {s.applied}/{s.totalDeltas} + + wall: {s.wallMs.toFixed(0)}ms + react commits: {s.commits} + commit time: {s.commitMs.toFixed(0)}ms + + long tasks: {s.longTasks} ({s.longTaskMs.toFixed(0)}ms) + + worst stall: {s.maxStallMs.toFixed(0)}ms +
+ ); +} + +const noop = () => {}; + +export function ThreadReplayPlayground() { + const params = useMemo(readParams, []); + const baseThread = useMemo(() => makeBaseThread(params.history), [params.history]); + const [thread, setThread] = useState(baseThread); + const listRef = useRef(null); + const statsRef = useRef({ + ...INITIAL_STATS, + totalDeltas: params.deltas + params.activities, + }); + const runningRef = useRef(false); + + useEffect(() => { + if (typeof PerformanceObserver === "undefined") return; + try { + const observer = new PerformanceObserver((list) => { + for (const entry of list.getEntries()) { + statsRef.current.longTasks += 1; + statsRef.current.longTaskMs += entry.duration; + } + }); + observer.observe({ entryTypes: ["longtask"] }); + return () => observer.disconnect(); + } catch { + // Long task timing is unsupported in this browser (e.g. Firefox). + return; + } + }, []); + + const runReplay = useCallback(async () => { + if (runningRef.current) return; + runningRef.current = true; + const stats = statsRef.current; + Object.assign(stats, { + ...INITIAL_STATS, + running: true, + totalDeltas: params.deltas + params.activities, + }); + setThread(baseThread); + await macrotask(); + + const events = makeDeltaEvents(params.deltas, params.activities, params.history, params.code); + const startedAt = performance.now(); + let current = baseThread; + let previousTick = startedAt; + for (let index = 0; index < events.length; index += params.batch) { + for (let offset = 0; offset < params.batch && index + offset < events.length; offset += 1) { + const result = applyThreadDetailEvent( + current, + events[index + offset] as OrchestrationEvent, + ); + if (result.kind === "updated") { + current = result.thread; + } + } + stats.applied = Math.min(index + params.batch, events.length); + setThread(current); + // One macrotask per publication mirrors the one-event-at-a-time stream + // apply in threads.ts and lets React commit between applications. + await macrotask(); + const now = performance.now(); + stats.maxStallMs = Math.max(stats.maxStallMs, now - previousTick); + previousTick = now; + stats.wallMs = now - startedAt; + } + stats.wallMs = performance.now() - startedAt; + stats.running = false; + stats.done = true; + runningRef.current = false; + }, [baseThread, params]); + + useEffect(() => { + if (params.auto) { + void runReplay(); + } + }, [params.auto, runReplay]); + + const onProfilerRender = useCallback((_id: string, _phase: string, actualDuration: number) => { + statsRef.current.commits += 1; + statsRef.current.commitMs += actualDuration; + }, []); + + const environmentThread = useMemo(() => ({ ...thread, environmentId: ENVIRONMENT_ID }), [thread]); + const displayServerMessages = useMemo( + () => deriveDisplayServerMessages(environmentThread.messages, EMPTY_URL_MAP), + [environmentThread.messages], + ); + const timelineMessages = useMemo( + () => + deriveTimelineMessages({ + displayServerMessages, + attachmentPreviewHandoffByMessageId: EMPTY_HANDOFFS, + optimisticUserMessages: EMPTY_MESSAGES, + }), + [displayServerMessages], + ); + const workLogEntries = useMemo( + () => deriveWorkLogEntries(environmentThread.activities), + [environmentThread.activities], + ); + const timelineEntries = useMemo( + () => deriveTimelineEntries(timelineMessages, environmentThread.proposedPlans, workLogEntries), + [environmentThread.proposedPlans, timelineMessages, workLogEntries], + ); + const turnDiffSummaryByAssistantMessageId = useMemo( + () => deriveTurnDiffSummaryByAssistantMessageId(environmentThread.checkpoints), + [environmentThread.checkpoints], + ); + const revertTurnCountByUserMessageId = useMemo( + () => + deriveRevertTurnCountByUserMessageId({ + timelineEntries, + turnDiffSummaryByAssistantMessageId, + inferredCheckpointTurnCountByTurnId: EMPTY_TURN_COUNTS, + }), + [timelineEntries, turnDiffSummaryByAssistantMessageId], + ); + + return ( +
+
+

Thread replay playground

+
+ + + + + + + +
+
+ +
+ + + +
+
+ ); +} + +const EMPTY_URL_MAP: ReadonlyMap = new Map(); +const EMPTY_HANDOFFS: Record = {}; +const EMPTY_MESSAGES: ReadonlyArray = []; +const EMPTY_TURN_COUNTS: Record = {}; diff --git a/apps/web/src/perf/main.tsx b/apps/web/src/perf/main.tsx new file mode 100644 index 00000000000..fffbd13dc05 --- /dev/null +++ b/apps/web/src/perf/main.tsx @@ -0,0 +1,14 @@ +import ReactDOM from "react-dom/client"; + +import "@fontsource-variable/dm-sans/index.css"; +import "@fontsource/jetbrains-mono/400.css"; +import "@fontsource/jetbrains-mono/500.css"; +import "../index.css"; + +import { ThreadReplayPlayground } from "./ThreadReplayPlayground"; + +// No StrictMode: its intentional double-rendering would skew every +// measurement this page exists to take. +ReactDOM.createRoot(document.getElementById("root") as HTMLElement).render( + , +); diff --git a/packages/client-runtime/package.json b/packages/client-runtime/package.json index 4fa05f850e5..19682c4d5f6 100644 --- a/packages/client-runtime/package.json +++ b/packages/client-runtime/package.json @@ -123,6 +123,10 @@ "types": "./src/state/threads.ts", "default": "./src/state/threads.ts" }, + "./state/thread-reducer": { + "types": "./src/state/threadReducer.ts", + "default": "./src/state/threadReducer.ts" + }, "./state/thread-sort": { "types": "./src/state/threadSort.ts", "default": "./src/state/threadSort.ts" diff --git a/packages/client-runtime/src/state/threadReplayPerf.test.ts b/packages/client-runtime/src/state/threadReplayPerf.test.ts new file mode 100644 index 00000000000..ab1f77f542d --- /dev/null +++ b/packages/client-runtime/src/state/threadReplayPerf.test.ts @@ -0,0 +1,208 @@ +/** + * Reproduction / benchmark for the thread-resume replay chug. + * + * When a thread accumulates many persisted streaming `thread.message-sent` + * delta events (assistant streaming enabled) and a client resumes its + * subscription with `afterSequence`, the server replays every event + * individually and the client applies them one at a time. Each apply copies + * the whole message array, re-concatenates the whole accumulated text, and + * publishes a new state to every subscriber (i.e. a React render of the full + * timeline + markdown reparse per delta). Total work is quadratic in the + * backlog. + * + * These tests quantify that on the real code paths so a fix can be verified: + * - "reducer replay cost" times the pure `applyThreadDetailEvent` fold, + * including a per-event full read of the accumulated text to model the + * renderer consuming each intermediate state (V8 cons strings make the + * concat itself cheap until the string is read). + * - "resume replay emissions" drives the real threads state machine through + * a resumed subscription and counts how many state values subscribers + * observe. Today this is one per delta event — the deterministic signature + * of the bug. A batching/coalescing fix should collapse it to a small + * number regardless of backlog size, and cut the wall time accordingly. + * + * Run with: + * pnpm --filter @t3tools/client-runtime test src/state/threadReplayPerf.test.ts + * + * For an end-to-end repro in the real app, see the chunked-response mode of + * apps/server/scripts/acp-mock-agent.ts (T3_ACP_PROMPT_RESPONSE_CHUNK_COUNT) + * with the `enableAssistantStreaming` server setting turned on. + */ +import { + EventId, + MessageId, + ProjectId, + ProviderInstanceId, + ThreadId, + TurnId, + type OrchestrationEvent, + type OrchestrationThread, + type OrchestrationThreadStreamItem, +} from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; + +import { applyThreadDetailEvent } from "./threadReducer.ts"; +import { CACHED_SNAPSHOT_SEQUENCE, makeThreadSyncHarness } from "./threadsSyncHarness.test-util.ts"; + +const THREAD_ID = ThreadId.make("thread-1"); +const STREAMING_MESSAGE_ID = MessageId.make("msg-streaming"); +const TURN_ID = TurnId.make("turn-1"); +const CHUNK_TEXT = "another twenty-four chars. "; + +const baseThread: OrchestrationThread = { + id: THREAD_ID, + projectId: ProjectId.make("project-1"), + title: "Replay perf thread", + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + latestTurn: null, + createdAt: "2026-04-01T00:00:00.000Z", + updatedAt: "2026-04-01T00:00:00.000Z", + archivedAt: null, + settledOverride: null, + settledAt: null, + deletedAt: null, + messages: [], + proposedPlans: [], + activities: [], + checkpoints: [], + session: null, +}; + +const streamingDeltaEvent = (index: number, sequence: number): OrchestrationEvent => + ({ + eventId: EventId.make(`event-delta-${index}`), + sequence, + occurredAt: "2026-04-01T01:00:00.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + aggregateKind: "thread", + aggregateId: THREAD_ID, + type: "thread.message-sent", + payload: { + threadId: THREAD_ID, + messageId: STREAMING_MESSAGE_ID, + role: "assistant", + text: CHUNK_TEXT, + turnId: TURN_ID, + streaming: true, + createdAt: "2026-04-01T01:00:00.000Z", + updatedAt: "2026-04-01T01:00:00.000Z", + }, + }) as OrchestrationEvent; + +/** + * Force the accumulated text to be read the way a renderer would. Without + * this, V8 cons strings defer the O(length) concat cost to the first read, + * which in the app happens on every emission (markdown parse of the full + * message text). + */ +const readFullText = (text: string): number => { + let total = 0; + for (let index = 0; index < text.length; index += 256) { + total += text.charCodeAt(index); + } + return total; +}; + +describe("thread replay performance", () => { + it.effect("reducer replay cost grows superlinearly with delta backlog", () => + Effect.gen(function* () { + const timings: Array<{ deltas: number; ms: number }> = []; + + for (const deltas of [1000, 2000, 4000, 8000]) { + let thread = baseThread; + const events = Array.from({ length: deltas }, (_, index) => + streamingDeltaEvent(index, index + 1), + ); + + const startedAt = performance.now(); + for (const event of events) { + const result = applyThreadDetailEvent(thread, event); + if (result.kind === "updated") { + thread = result.thread; + } + // Model the renderer consuming each intermediate state. + readFullText(thread.messages[0]?.text ?? ""); + } + const ms = performance.now() - startedAt; + + expect(thread.messages).toHaveLength(1); + expect(thread.messages[0]?.text).toHaveLength(deltas * CHUNK_TEXT.length); + timings.push({ deltas, ms }); + } + + // Report the curve; doubling the backlog roughly quadruples total cost + // when the behavior is quadratic. + yield* Effect.log( + `[threadReplayPerf] reducer fold: ${timings + .map(({ deltas, ms }) => `${deltas} deltas -> ${ms.toFixed(1)}ms`) + .join(", ")}`, + ); + }), + ); + + it.effect("resume replay publishes one state per delta event", () => + Effect.gen(function* () { + const deltas = 2000; + const harness = yield* makeThreadSyncHarness({ + cached: baseThread, + completionMarker: true, + }); + + // Wait for the cached thread to be published before starting the clock. + yield* Queue.take(harness.observed).pipe( + Effect.repeat({ + until: (state) => Option.isSome(state.data), + }), + ); + + const startedAt = performance.now(); + const items: Array = Array.from( + { length: deltas }, + (_, index) => ({ + kind: "event", + event: streamingDeltaEvent(index, CACHED_SNAPSHOT_SEQUENCE + 1 + index), + }), + ); + yield* Queue.offerAll(harness.inputs, items); + yield* Queue.offer(harness.inputs, { kind: "synchronized" }); + + // Drain observed states until the replay has fully applied, counting + // how many state publications subscribers had to process. + let emissions = 0; + let caughtUp = false; + while (!caughtUp) { + const state = yield* Queue.take(harness.observed); + emissions += 1; + // Model the renderer consuming each published state. + if (Option.isSome(state.data)) { + readFullText(state.data.value.messages[0]?.text ?? ""); + } + caughtUp = + state.status === "live" && + Option.isSome(state.data) && + (state.data.value.messages[0]?.text.length ?? 0) === deltas * CHUNK_TEXT.length; + } + const ms = performance.now() - startedAt; + + yield* Effect.log( + `[threadReplayPerf] resume replay: ${deltas} deltas -> ${emissions} state emissions in ${ms.toFixed(1)}ms`, + ); + + // Drain-based batching in threads.ts must fold a replay backlog into + // far fewer state publications than events; one publication per event + // is the quadratic-reopen regression this test guards against. + expect(emissions).toBeGreaterThanOrEqual(2); + expect(emissions).toBeLessThan(deltas / 10); + }), + ); +}); diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 196229cc8b1..948ab5db4be 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -11,6 +11,7 @@ import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { Atom } from "effect/unstable/reactivity"; @@ -179,46 +180,76 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ); }); - const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* ( - item: OrchestrationThreadStreamItem, + const applyItems = Effect.fn("EnvironmentThreadState.applyItems")(function* ( + batch: Iterable, ) { - if (item.kind === "synchronized") { + // Fold the whole batch into at most one thread publication. A resume + // replay can deliver thousands of persisted events; publishing state per + // event makes reopening a thread quadratic in the backlog, because every + // publication re-derives and re-renders the full timeline. + let sequence = yield* SubscriptionRef.get(lastSequence); + let data = (yield* SubscriptionRef.get(state)).data; + let dirty = false; + let deleted = false; + let synchronized = false; + + for (const item of batch) { + if (item.kind === "synchronized") { + synchronized = true; + continue; + } + if (item.kind === "snapshot") { + sequence = item.snapshot.snapshotSequence; + data = Option.some(item.snapshot.thread); + dirty = true; + deleted = false; + continue; + } + if (item.event.sequence <= sequence) { + continue; + } + sequence = item.event.sequence; + if (Option.isNone(data)) { + if (item.event.type === "thread.deleted") { + deleted = true; + } + continue; + } + const result = applyThreadDetailEvent(data.value, item.event); + if (result.kind === "updated") { + data = Option.some(result.thread); + dirty = true; + } else if (result.kind === "deleted") { + data = Option.none(); + dirty = false; + deleted = true; + } + } + + yield* SubscriptionRef.set(lastSequence, sequence); + if (deleted && Option.isNone(data)) { + yield* setDeleted(); + } else if (dirty && Option.isSome(data)) { + yield* setThread(data.value); + } + if (synchronized) { yield* Ref.set(awaitingCompletion, false); yield* SubscriptionRef.update(state, (current) => Option.isSome(current.data) && current.status !== "deleted" ? { ...current, status: "live" as const, error: Option.none() } : current, ); - return; - } - - if (item.kind === "snapshot") { - yield* SubscriptionRef.set(lastSequence, item.snapshot.snapshotSequence); - yield* setThread(item.snapshot.thread); - return; - } - - const sequence = yield* SubscriptionRef.get(lastSequence); - if (item.event.sequence <= sequence) { - return; - } - yield* SubscriptionRef.set(lastSequence, item.event.sequence); - - const current = yield* SubscriptionRef.get(state); - if (Option.isNone(current.data)) { - if (item.event.type === "thread.deleted") { - yield* setDeleted(); - } - return; - } - const result = applyThreadDetailEvent(current.data.value, item.event); - if (result.kind === "updated") { - yield* setThread(result.thread); - } else if (result.kind === "deleted") { - yield* setDeleted(); } }); + const applyItem = (item: OrchestrationThreadStreamItem) => applyItems([item]); + const applyLock = yield* Semaphore.make(1); + const subscriptionGeneration = yield* Ref.make(0); + const pendingItems = yield* Queue.unbounded<{ + readonly generation: number; + readonly item: OrchestrationThreadStreamItem; + }>(); + yield* SubscriptionRef.changes(supervisor.state).pipe( Stream.runForEach((connectionState) => { switch (connectionProjectionPhase(connectionState)) { @@ -244,58 +275,106 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make subscribeDynamic( ORCHESTRATION_WS_METHODS.subscribeThread, Effect.fn("EnvironmentThreadState.makeSubscribeInput")(function* (session) { - const supportsCompletionMarker = yield* session.initialConfig.pipe( - Effect.map((config) => config.threadResumeCompletionMarker === true), - Effect.orElseSucceed(() => false), - ); - yield* Ref.set(awaitingCompletion, supportsCompletionMarker); - yield* setSynchronizing; + return yield* applyLock.withPermits(1)( + Effect.gen(function* () { + // A switched-out stream may already have queued items. Give every + // subscription attempt a generation and clear its predecessor's + // backlog before loading a replacement snapshot. + yield* Ref.update(subscriptionGeneration, (generation) => generation + 1); + yield* Queue.clear(pendingItems); - let current = yield* SubscriptionRef.get(state); - if (Option.isNone(current.data) && current.status !== "deleted") { - const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe( - Effect.flatMap( - Option.match({ - onSome: Effect.succeed, - onNone: () => - SubscriptionRef.changes(supervisor.prepared).pipe( - Stream.filter(Option.isSome), - Stream.map((value) => value.value), - Stream.runHead, - Effect.map(Option.getOrThrow), - ), - }), - ), - ); - const httpSnapshot = yield* snapshotLoader.load(prepared, threadId); - if (Option.isSome(httpSnapshot)) { - yield* applyItem({ kind: "snapshot", snapshot: httpSnapshot.value }); - current = yield* SubscriptionRef.get(state); - } - } + const supportsCompletionMarker = yield* session.initialConfig.pipe( + Effect.map((config) => config.threadResumeCompletionMarker === true), + Effect.orElseSucceed(() => false), + ); + yield* Ref.set(awaitingCompletion, supportsCompletionMarker); + yield* setSynchronizing; - const sequence = yield* SubscriptionRef.get(lastSequence); - const canResume = Option.isSome(current.data); - if (!supportsCompletionMarker && canResume) { - yield* SubscriptionRef.update(state, (value) => ({ - ...value, - status: value.status === "deleted" ? value.status : ("live" as const), - error: Option.none(), - })); - } + let current = yield* SubscriptionRef.get(state); + if (Option.isNone(current.data) && current.status !== "deleted") { + const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe( + Effect.flatMap( + Option.match({ + onSome: Effect.succeed, + onNone: () => + SubscriptionRef.changes(supervisor.prepared).pipe( + Stream.filter(Option.isSome), + Stream.map((value) => value.value), + Stream.runHead, + Effect.map(Option.getOrThrow), + ), + }), + ), + ); + const httpSnapshot = yield* snapshotLoader.load(prepared, threadId); + if (Option.isSome(httpSnapshot)) { + yield* applyItem({ kind: "snapshot", snapshot: httpSnapshot.value }); + current = yield* SubscriptionRef.get(state); + } + } - return { - threadId, - ...(canResume ? { afterSequence: sequence } : {}), - ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}), - }; + const sequence = yield* SubscriptionRef.get(lastSequence); + const canResume = Option.isSome(current.data); + if (!supportsCompletionMarker && canResume) { + yield* SubscriptionRef.update(state, (value) => ({ + ...value, + status: value.status === "deleted" ? value.status : ("live" as const), + error: Option.none(), + })); + } + + return { + threadId, + ...(canResume ? { afterSequence: sequence } : {}), + ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}), + }; + }), + ); }), { onExpectedFailure: setStreamError, retryExpectedFailureAfter: "250 millis", resubscribe: foregroundResubscriptions, }, - ).pipe(Stream.runForEach(applyItem)), + ).pipe( + Stream.runForEach((item) => + Ref.get(subscriptionGeneration).pipe( + Effect.flatMap((generation) => Queue.offer(pendingItems, { generation, item })), + ), + ), + ), + ); + + // Drain-based batching: apply one item plus everything that accumulated + // while the previous batch was being applied, giving the producer a few + // scheduler turns to flush a burst it is actively delivering (a resume + // replay arrives as one rapid burst). Live updates arriving one at a time + // publish immediately; a replay backlog folds into large batches with a + // single publication each, keeping thread reopen cost linear instead of + // quadratic. No timers, so batching never delays a quiet stream. + yield* Effect.forkScoped( + Effect.forever( + Effect.gen(function* () { + const batch = [yield* Queue.take(pendingItems)]; + for (let round = 0; round < 32; round += 1) { + yield* Effect.yieldNow; + const more = yield* Queue.clear(pendingItems); + if (more.length === 0) break; + batch.push(...more); + } + yield* applyLock.withPermits(1)( + Effect.gen(function* () { + const generation = yield* Ref.get(subscriptionGeneration); + const currentItems = batch + .filter((pending) => pending.generation === generation) + .map((pending) => pending.item); + if (currentItems.length > 0) { + yield* applyItems(currentItems); + } + }), + ); + }), + ), ); yield* Effect.addFinalizer(() => diff --git a/packages/client-runtime/src/state/threadsSyncHarness.test-util.ts b/packages/client-runtime/src/state/threadsSyncHarness.test-util.ts new file mode 100644 index 00000000000..9ca3dad5da7 --- /dev/null +++ b/packages/client-runtime/src/state/threadsSyncHarness.test-util.ts @@ -0,0 +1,208 @@ +/** + * Test harness for driving the environment thread state machine with a fake + * socket, cache, and snapshot loader. Adapted from the inline harness in + * threads-sync.test.ts so perf/repro tests can reuse it; the filename avoids + * the `.test.ts` suffix so it is not collected as a suite itself. + */ +import { + EnvironmentId, + ORCHESTRATION_WS_METHODS, + ThreadId, + type OrchestrationThread, + type OrchestrationThreadDetailSnapshot, + type OrchestrationThreadStreamItem, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; + +import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; +import { + AVAILABLE_CONNECTION_STATE, + PrimaryConnectionTarget, + type PreparedConnection, + type SupervisorConnectionState, +} from "../connection/model.ts"; +import * as ConnectionWakeups from "../connection/wakeups.ts"; +import * as EnvironmentSupervisor from "../connection/supervisor.ts"; +import * as Persistence from "../platform/persistence.ts"; +import * as RpcSession from "../rpc/session.ts"; +import { + EMPTY_ENVIRONMENT_THREAD_STATE, + makeEnvironmentThreadState, + ThreadSnapshotLoader, + type EnvironmentThreadState, +} from "./threads.ts"; + +const TARGET = new PrimaryConnectionTarget({ + environmentId: EnvironmentId.make("environment-1"), + label: "Test environment", + httpBaseUrl: "https://environment.example.test", + wsBaseUrl: "wss://environment.example.test", +}); +export const HARNESS_THREAD_ID = ThreadId.make("thread-1"); +export const CACHED_SNAPSHOT_SEQUENCE = 7; +const PREPARED: PreparedConnection = { + environmentId: TARGET.environmentId, + label: TARGET.label, + httpBaseUrl: TARGET.httpBaseUrl, + socketUrl: TARGET.wsBaseUrl, + httpAuthorization: null, + target: TARGET, +}; + +export type TestThreadInput = OrchestrationThreadStreamItem | Error; + +function testSession( + client: WsRpcProtocolClient, + options?: { readonly completionMarker?: boolean }, +): RpcSession.RpcSession { + return { + client, + initialConfig: Effect.succeed( + options?.completionMarker === true + ? ({ threadResumeCompletionMarker: true } as never) + : ({} as never), + ), + ready: Effect.void, + probe: Effect.void, + closed: Effect.never, + }; +} + +export const makeThreadSyncHarness = Effect.fn("TestEnvironmentThreads.makeHarness")( + function* (options?: { + readonly cached?: OrchestrationThread; + readonly httpSnapshot?: Option.Option; + readonly completionMarker?: boolean; + }) { + const inputs = yield* Queue.unbounded(); + const observed = yield* Queue.unbounded(); + const latest = yield* Ref.make(EMPTY_ENVIRONMENT_THREAD_STATE); + const retryCount = yield* Ref.make(0); + const subscriptionCount = yield* Ref.make(0); + const loaderCalls = yield* Ref.make(0); + const lastSubscribeAfterSequence = yield* Ref.make(undefined); + const lastRequestCompletionMarker = yield* Ref.make(undefined); + const savedThreads = yield* Ref.make>([]); + const removedThreads = yield* Ref.make>([]); + const wakeups = yield* Queue.unbounded(); + const supervisorState = yield* SubscriptionRef.make( + AVAILABLE_CONNECTION_STATE, + ); + const streamFrom = (queue: Queue.Queue) => + Stream.fromQueue(queue).pipe( + Stream.mapEffect((input) => + input instanceof Error ? Effect.fail(input) : Effect.succeed(input), + ), + ); + const client = { + [ORCHESTRATION_WS_METHODS.subscribeThread]: (input: { + readonly afterSequence?: number; + readonly requestCompletionMarker?: boolean; + }) => + Stream.unwrap( + Ref.updateAndGet(subscriptionCount, (count) => count + 1).pipe( + Effect.andThen(Ref.set(lastSubscribeAfterSequence, input.afterSequence)), + Effect.andThen(Ref.set(lastRequestCompletionMarker, input.requestCompletionMarker)), + Effect.as(streamFrom(inputs)), + ), + ), + } as unknown as WsRpcProtocolClient; + const supervisorSession = yield* SubscriptionRef.make>( + Option.some( + testSession( + client, + options?.completionMarker === true ? { completionMarker: true } : undefined, + ), + ), + ); + const prepared = yield* SubscriptionRef.make>( + Option.some(PREPARED), + ); + const snapshotLoader = ThreadSnapshotLoader.of({ + load: (_prepared, threadId) => + Ref.update(loaderCalls, (count) => count + 1).pipe( + Effect.as( + threadId === HARNESS_THREAD_ID + ? (options?.httpSnapshot ?? Option.none()) + : Option.none(), + ), + ), + }); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: supervisorState, + session: supervisorSession, + prepared, + connect: Effect.void, + disconnect: Effect.void, + retryNow: Ref.update(retryCount, (count) => count + 1), + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const cache = Persistence.EnvironmentCacheStore.of({ + loadShell: () => Effect.succeed(Option.none()), + saveShell: () => Effect.void, + loadThread: (_environmentId, threadId) => + Effect.succeed( + threadId === HARNESS_THREAD_ID && options?.cached !== undefined + ? Option.some({ + snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, + thread: options.cached, + }) + : Option.none(), + ), + saveThread: (_environmentId, thread) => + Ref.update(savedThreads, (current) => [...current, thread]), + removeThread: (_environmentId, threadId) => + Ref.update(removedThreads, (current) => [...current, threadId]), + loadServerConfig: () => Effect.succeed(Option.none()), + saveServerConfig: () => Effect.void, + loadVcsRefs: () => Effect.succeed(Option.none()), + saveVcsRefs: () => Effect.void, + clear: () => Effect.void, + }); + const threadState = yield* makeEnvironmentThreadState(HARNESS_THREAD_ID).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ThreadSnapshotLoader, snapshotLoader), + Effect.provideService( + ConnectionWakeups.ConnectionWakeups, + ConnectionWakeups.ConnectionWakeups.of({ changes: Stream.fromQueue(wakeups) }), + ), + ); + yield* SubscriptionRef.changes(threadState).pipe( + Stream.runForEach((state) => + Ref.set(latest, state).pipe(Effect.andThen(Queue.offer(observed, state))), + ), + Effect.forkScoped, + ); + + return { + inputs, + observed, + latest, + retryCount, + subscriptionCount, + loaderCalls, + lastSubscribeAfterSequence, + lastRequestCompletionMarker, + supervisorState, + supervisorSession, + savedThreads, + removedThreads, + wakeups, + replaceSession: SubscriptionRef.set( + supervisorSession, + Option.some( + testSession( + client, + options?.completionMarker === true ? { completionMarker: true } : undefined, + ), + ), + ), + }; + }, +);