From b14ba9aefa5c63884ebeb154a7fff688a66910a3 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sat, 24 Oct 2026 09:05:48 -0500 Subject: [PATCH 1/4] perf(cursor): coalesce streamed text snapshots Each Cursor text or thinking delta emitted a node update plus message and turn-item updates carrying the whole segment text so far, so adapter event volume grew with the square of the reply length. Emit a segment's first text at once, then one full-text snapshot per 50 ms window, and the final text when the segment completes. Emit the text node only when it opens and completes. This matches the ACP adapter (#19626). Co-Authored-By: Claude Opus 5.5 --- .../src/server/adapter.test.ts | 158 ++++++++++++++++ .../provider-cursor/src/server/adapter.ts | 175 ++++++++++++------ 2 files changed, 280 insertions(+), 53 deletions(-) diff --git a/packages/provider-cursor/src/server/adapter.test.ts b/packages/provider-cursor/src/server/adapter.test.ts index 0686b0c831dd2..1b10343825ecb 100644 --- a/packages/provider-cursor/src/server/adapter.test.ts +++ b/packages/provider-cursor/src/server/adapter.test.ts @@ -14,6 +14,7 @@ import { } from "@t3tools/contracts"; import { CursorSettings } from "../settings.ts"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; @@ -21,6 +22,7 @@ import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; @@ -383,6 +385,162 @@ describe("CursorAdapterV2", () => { ), ); + it.effect("coalesces streamed text into bounded snapshots without delaying its start", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const workspace = yield* fileSystem.makeTempDirectoryScoped({ prefix: "cursor-v2-text-" }); + const instanceId = ProviderInstanceId.make("cursor"); + const threadId = ThreadId.make("cursor-text-thread"); + const modelSelection = { instanceId, model: "composer-2.5" }; + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: workspace, + }); + const finished = yield* Deferred.make(); + let sendUpdate: ((update: InteractionUpdate) => Effect.Effect) | undefined; + const adapter = yield* makeCursorAdapterV2({ + instanceId, + settings: yield* decodeCursorSettings({}), + environment: { HOME: workspace }, + }).pipe( + Effect.provideService(CursorAgentSdk.CursorAgentSdkRunner, { + assertComplete: Effect.void, + open: () => + Effect.succeed({ + agentId: "native-cursor-text", + listMessages: Effect.succeed([]), + close: Effect.void, + send: (input) => + Effect.sync(() => { + sendUpdate = (update) => input.onDelta!(update).pipe(Effect.orDie); + return { + agentId: "native-cursor-text", + runId: "native-cursor-text-run", + wait: Deferred.await(finished).pipe( + Effect.as({ + id: "native-cursor-text-run", + requestId: "native-request", + status: "finished" as const, + model: { id: "composer-2.5" }, + durationMs: 1, + }), + ), + cancel: Effect.void, + }; + }), + }), + }), + Effect.provide(TestProviderHost.layer({ cwd: workspace })), + ); + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("cursor-text-session"), + modelSelection, + runtimePolicy, + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + const now = yield* DateTime.now; + yield* runtime.startTurn({ + threadId, + providerThread, + modelSelection, + runtimePolicy, + runId: RunId.make("cursor-text-run"), + runOrdinal: 1, + providerTurnOrdinal: 1, + attemptId: RunAttemptId.make("cursor-text-attempt"), + rootNodeId: NodeId.make("cursor-text-root"), + appThread: { + id: threadId, + projectId: ProjectId.make("cursor-text-project"), + createdBy: "user", + creationSource: "web", + title: "Cursor text", + providerInstanceId: instanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: providerThread.id, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + message: { + messageId: MessageId.make("cursor-text-message"), + createdBy: "user", + creationSource: "web", + text: "hi", + attachments: [], + }, + }); + assert.isDefined(sendUpdate); + const chunk = "tok0 word "; + for (let index = 0; index < 1_000; index += 1) { + yield* sendUpdate!({ type: "text-delta", text: chunk }); + } + yield* TestClock.adjust("50 millis"); + yield* sendUpdate!({ type: "text-delta", text: "end" }); + // Reasoning closes the assistant segment while its next flush is pending. + yield* sendUpdate!({ type: "thinking-delta", text: "next" }); + yield* TestClock.adjust("50 millis"); + yield* Deferred.succeed(finished, undefined); + const streamed = yield* runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + ); + + const reply = chunk.repeat(1_000); + assert.deepEqual( + streamed.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "assistant_message" + ? [[event.turnItem.status, event.turnItem.text]] + : [], + ), + [ + ["running", chunk], + ["running", reply], + ["completed", `${reply}end`], + ], + ); + assert.deepEqual( + streamed.flatMap((event) => + event.type === "message.updated" ? [[event.message.streaming, event.message.text]] : [], + ), + [ + [true, chunk], + [true, reply], + [false, `${reply}end`], + ], + ); + assert.deepEqual( + streamed.flatMap((event) => + event.type === "node.updated" && event.node.kind === "assistant_message" + ? [event.node.status] + : [], + ), + ["running", "completed"], + ); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(NodeServices.layer, IdAllocator.layer, McpProviderSessions.layer), + ), + ), + ); + it.effect("fails standalone SDK transport diagnostics and sends compaction as /compress", () => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; diff --git a/packages/provider-cursor/src/server/adapter.ts b/packages/provider-cursor/src/server/adapter.ts index 825596c04e6da..0cbe5a7f06c5c 100644 --- a/packages/provider-cursor/src/server/adapter.ts +++ b/packages/provider-cursor/src/server/adapter.ts @@ -32,6 +32,7 @@ import { } from "@t3tools/contracts"; import { CursorSettings } from "../settings.ts"; import * as DateTime from "effect/DateTime"; +import type * as Duration from "effect/Duration"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; @@ -40,6 +41,7 @@ import * as Path from "effect/Path"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost"; @@ -825,6 +827,12 @@ interface CursorProjectionTarget { // ever shows a bounded tail, so keep a little more than that. const MAX_CURSOR_STREAMED_OUTPUT_CHARS = 256 * 1024; +// A text segment's first delta and its close emit at once. Later deltas share one +// full-text snapshot this long after them, instead of each re-emitting the whole +// reply, which grew with the square of the reply length. The flush is scheduled, +// not a deadline, so it can land later under host load. +const CURSOR_TEXT_FLUSH_INTERVAL: Duration.Input = "50 millis"; + interface ActiveCursorToolCall { readonly callId: string; toolCall: ToolCall; @@ -850,6 +858,10 @@ interface ActiveCursorTextSegment { readonly nativeItemId: string; readonly startedAt: DateTime.Utc; text: string; + // The running node and first text have been emitted. + emitted: boolean; + // Text changed since the last emission; a delayed flush is scheduled. + dirty: boolean; } interface ActiveCursorTextStream { @@ -964,14 +976,41 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( providerTurnId: context.providerTurnId, }); + const textEmissionPermit = yield* Semaphore.make(1); + /** + * Claims the stream's current segment for one emission. A `flushOf` call + * is a delayed flush: it emits only while that segment is still current + * and has unsent text. + */ + const takeTextSnapshot = ( + stream: ActiveCursorTextStream, + completed: boolean, + flushOf: ActiveCursorTextSegment | undefined, + ) => { + const segment = stream.current; + if (segment === null || segment.text.length === 0) return null; + if (flushOf !== undefined && (segment !== flushOf || !segment.dirty)) return null; + // The node carries no text, so it changes only when the segment opens + // and when it completes. + const emitsNode = completed || !segment.emitted; + segment.emitted = true; + segment.dirty = false; + // Read the text in the same step that clears `dirty`. Deltas appended + // while this emission yields mark the segment dirty again and reach the + // next flush or the close. + return { segment, text: segment.text, emitsNode }; + }; + const emitAssistant = Effect.fnUntraced(function* ( context: ActiveCursorTurn, completed: boolean, + flushOf?: ActiveCursorTextSegment, ) { - const segment = context.assistant.current; - if (segment === null || segment.text.length === 0) { + const snapshot = takeTextSnapshot(context.assistant, completed, flushOf); + if (snapshot === null) { return; } + const { segment, text } = snapshot; const now = yield* DateTime.now; const ordinal = yield* resolveItemOrdinal(context, segment.nativeItemId); const nodeId = idAllocator.derive.nodeFromProviderItem({ @@ -991,27 +1030,29 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( nativeId: segment.nativeItemId, strength: "weak" as const, }; - yield* emitProviderEvent({ - type: "node.updated", - driver: CursorAgentSdk.CURSOR_PROVIDER, - node: { - id: nodeId, - threadId: context.input.threadId, - runId: context.input.runId, - parentNodeId: context.input.rootNodeId, - rootNodeId: context.input.rootNodeId, - kind: "assistant_message", - status: completed ? "completed" : "running", - countsForRun: false, - providerThreadId: context.input.providerThread.id, - providerTurnId: context.providerTurnId, - nativeItemRef, - runtimeRequestId: null, - checkpointScopeId: null, - startedAt: segment.startedAt, - completedAt: completed ? now : null, - }, - }); + if (snapshot.emitsNode) { + yield* emitProviderEvent({ + type: "node.updated", + driver: CursorAgentSdk.CURSOR_PROVIDER, + node: { + id: nodeId, + threadId: context.input.threadId, + runId: context.input.runId, + parentNodeId: context.input.rootNodeId, + rootNodeId: context.input.rootNodeId, + kind: "assistant_message", + status: completed ? "completed" : "running", + countsForRun: false, + providerThreadId: context.input.providerThread.id, + providerTurnId: context.providerTurnId, + nativeItemRef, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: segment.startedAt, + completedAt: completed ? now : null, + }, + }); + } yield* emitProviderEvent({ type: "message.updated", driver: CursorAgentSdk.CURSOR_PROVIDER, @@ -1023,7 +1064,7 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( runId: context.input.runId, nodeId, role: "assistant", - text: segment.text, + text, attachments: [], streaming: !completed, createdAt: segment.startedAt, @@ -1050,20 +1091,22 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( updatedAt: now, type: "assistant_message", messageId, - text: segment.text, + text, streaming: !completed, }, }); - }); + }, textEmissionPermit.withPermit); const emitReasoning = Effect.fnUntraced(function* ( context: ActiveCursorTurn, completed: boolean, + flushOf?: ActiveCursorTextSegment, ) { - const segment = context.reasoning.current; - if (segment === null || segment.text.length === 0) { + const snapshot = takeTextSnapshot(context.reasoning, completed, flushOf); + if (snapshot === null) { return; } + const { segment, text } = snapshot; const now = yield* DateTime.now; const ordinal = yield* resolveItemOrdinal(context, segment.nativeItemId); const nodeId = idAllocator.derive.nodeFromProviderItem({ @@ -1079,27 +1122,29 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( nativeId: segment.nativeItemId, strength: "weak" as const, }; - yield* emitProviderEvent({ - type: "node.updated", - driver: CursorAgentSdk.CURSOR_PROVIDER, - node: { - id: nodeId, - threadId: context.input.threadId, - runId: context.input.runId, - parentNodeId: context.input.rootNodeId, - rootNodeId: context.input.rootNodeId, - kind: "reasoning", - status: completed ? "completed" : "running", - countsForRun: false, - providerThreadId: context.input.providerThread.id, - providerTurnId: context.providerTurnId, - nativeItemRef, - runtimeRequestId: null, - checkpointScopeId: null, - startedAt: segment.startedAt, - completedAt: completed ? now : null, - }, - }); + if (snapshot.emitsNode) { + yield* emitProviderEvent({ + type: "node.updated", + driver: CursorAgentSdk.CURSOR_PROVIDER, + node: { + id: nodeId, + threadId: context.input.threadId, + runId: context.input.runId, + parentNodeId: context.input.rootNodeId, + rootNodeId: context.input.rootNodeId, + kind: "reasoning", + status: completed ? "completed" : "running", + countsForRun: false, + providerThreadId: context.input.providerThread.id, + providerTurnId: context.providerTurnId, + nativeItemRef, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: segment.startedAt, + completedAt: completed ? now : null, + }, + }); + } yield* emitProviderEvent({ type: "turn_item.updated", driver: CursorAgentSdk.CURSOR_PROVIDER, @@ -1119,11 +1164,11 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( completedAt: completed ? now : null, updatedAt: now, type: "reasoning", - text: segment.text, + text, streaming: !completed, }, }); - }); + }, textEmissionPermit.withPermit); const appendTextSegment = Effect.fnUntraced(function* (input: { readonly context: ActiveCursorTurn; @@ -1137,6 +1182,8 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( nativeItemId: `${input.kind}:${input.context.run.runId}:${input.stream.nextSegment}`, startedAt: yield* DateTime.now, text: "", + emitted: false, + dirty: false, }; } input.stream.current.text += input.text; @@ -1158,6 +1205,28 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( context.reasoning.current = null; }); + /** + * Publishes a text change. A segment's first text emits at once so live + * output starts immediately; later changes coalesce into one snapshot per + * CURSOR_TEXT_FLUSH_INTERVAL. Completing the segment emits its final text + * and makes a pending flush a no-op. + */ + const publishTextSegment = Effect.fnUntraced(function* ( + context: ActiveCursorTurn, + kind: "assistant" | "reasoning", + ) { + const emit = kind === "assistant" ? emitAssistant : emitReasoning; + const segment = context[kind].current; + if (segment === null) return; + if (!segment.emitted) return yield* emit(context, false); + if (segment.dirty) return; + segment.dirty = true; + yield* emit(context, false, segment).pipe( + Effect.delay(CURSOR_TEXT_FLUSH_INTERVAL), + Effect.forkIn(sessionScope), + ); + }); + const emitToolArtifacts = Effect.fnUntraced(function* (input: { readonly active: ActiveCursorToolCall; readonly completed: boolean; @@ -1929,7 +1998,7 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( kind: "assistant", text: update.text, }); - yield* emitAssistant(context, false); + yield* publishTextSegment(context, "assistant"); return; case "thinking-delta": yield* completeAssistant(context); @@ -1939,7 +2008,7 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( kind: "reasoning", text: update.text, }); - yield* emitReasoning(context, false); + yield* publishTextSegment(context, "reasoning"); return; case "thinking-completed": yield* completeReasoning(context); From 9c7696feff0d3095af2b4b9883f55e4a43b0eb07 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sat, 24 Oct 2026 16:58:25 -0500 Subject: [PATCH 2/4] perf(cursor): emit streamed text at once after a quiet window Coalescing held every later snapshot for a full 50 ms window, even when the previous one was older than that. On a real Cursor reasoning stream, where deltas arrive about 200 ms apart, that delayed each update by 50 ms and saved little. Emit a delta at once when the segment's last snapshot is at least 50 ms old, and otherwise flush at the end of that window. Snapshots stay at least 50 ms apart. Co-Authored-By: Claude Opus 5.5 --- .../src/server/adapter.test.ts | 11 ++-- .../provider-cursor/src/server/adapter.ts | 52 +++++++++++-------- 2 files changed, 37 insertions(+), 26 deletions(-) diff --git a/packages/provider-cursor/src/server/adapter.test.ts b/packages/provider-cursor/src/server/adapter.test.ts index 1b10343825ecb..1737ed9174381 100644 --- a/packages/provider-cursor/src/server/adapter.test.ts +++ b/packages/provider-cursor/src/server/adapter.test.ts @@ -385,7 +385,7 @@ describe("CursorAdapterV2", () => { ), ); - it.effect("coalesces streamed text into bounded snapshots without delaying its start", () => + it.effect("coalesces streamed text into bounded snapshots without delaying a slow stream", () => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; const workspace = yield* fileSystem.makeTempDirectoryScoped({ prefix: "cursor-v2-text-" }); @@ -492,7 +492,10 @@ describe("CursorAdapterV2", () => { yield* sendUpdate!({ type: "text-delta", text: chunk }); } yield* TestClock.adjust("50 millis"); + // The window has passed since the last snapshot, so this delta emits at once. + yield* TestClock.adjust("50 millis"); yield* sendUpdate!({ type: "text-delta", text: "end" }); + yield* sendUpdate!({ type: "text-delta", text: "!" }); // Reasoning closes the assistant segment while its next flush is pending. yield* sendUpdate!({ type: "thinking-delta", text: "next" }); yield* TestClock.adjust("50 millis"); @@ -512,7 +515,8 @@ describe("CursorAdapterV2", () => { [ ["running", chunk], ["running", reply], - ["completed", `${reply}end`], + ["running", `${reply}end`], + ["completed", `${reply}end!`], ], ); assert.deepEqual( @@ -522,7 +526,8 @@ describe("CursorAdapterV2", () => { [ [true, chunk], [true, reply], - [false, `${reply}end`], + [true, `${reply}end`], + [false, `${reply}end!`], ], ); assert.deepEqual( diff --git a/packages/provider-cursor/src/server/adapter.ts b/packages/provider-cursor/src/server/adapter.ts index 0cbe5a7f06c5c..c7e374f13be92 100644 --- a/packages/provider-cursor/src/server/adapter.ts +++ b/packages/provider-cursor/src/server/adapter.ts @@ -32,7 +32,6 @@ import { } from "@t3tools/contracts"; import { CursorSettings } from "../settings.ts"; import * as DateTime from "effect/DateTime"; -import type * as Duration from "effect/Duration"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; @@ -827,11 +826,11 @@ interface CursorProjectionTarget { // ever shows a bounded tail, so keep a little more than that. const MAX_CURSOR_STREAMED_OUTPUT_CHARS = 256 * 1024; -// A text segment's first delta and its close emit at once. Later deltas share one -// full-text snapshot this long after them, instead of each re-emitting the whole -// reply, which grew with the square of the reply length. The flush is scheduled, -// not a deadline, so it can land later under host load. -const CURSOR_TEXT_FLUSH_INTERVAL: Duration.Input = "50 millis"; +// Full-text snapshots of a streaming segment are at least this far apart, instead +// of one per delta, which grew with the square of the reply length. A delta after +// a quiet interval and the segment's close still emit at once. The flush is +// scheduled, not a deadline, so it can land later under host load. +const CURSOR_TEXT_FLUSH_INTERVAL_MS = 50; interface ActiveCursorToolCall { readonly callId: string; @@ -858,8 +857,9 @@ interface ActiveCursorTextSegment { readonly nativeItemId: string; readonly startedAt: DateTime.Utc; text: string; - // The running node and first text have been emitted. - emitted: boolean; + // When the last snapshot was emitted, or null before the running node and + // first text. + emittedAtMs: number | null; // Text changed since the last emission; a delayed flush is scheduled. dirty: boolean; } @@ -986,14 +986,15 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( stream: ActiveCursorTextStream, completed: boolean, flushOf: ActiveCursorTextSegment | undefined, + now: DateTime.Utc, ) => { const segment = stream.current; if (segment === null || segment.text.length === 0) return null; if (flushOf !== undefined && (segment !== flushOf || !segment.dirty)) return null; // The node carries no text, so it changes only when the segment opens // and when it completes. - const emitsNode = completed || !segment.emitted; - segment.emitted = true; + const emitsNode = completed || segment.emittedAtMs === null; + segment.emittedAtMs = DateTime.toEpochMillis(now); segment.dirty = false; // Read the text in the same step that clears `dirty`. Deltas appended // while this emission yields mark the segment dirty again and reach the @@ -1006,12 +1007,12 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( completed: boolean, flushOf?: ActiveCursorTextSegment, ) { - const snapshot = takeTextSnapshot(context.assistant, completed, flushOf); + const now = yield* DateTime.now; + const snapshot = takeTextSnapshot(context.assistant, completed, flushOf, now); if (snapshot === null) { return; } const { segment, text } = snapshot; - const now = yield* DateTime.now; const ordinal = yield* resolveItemOrdinal(context, segment.nativeItemId); const nodeId = idAllocator.derive.nodeFromProviderItem({ driver: CursorAgentSdk.CURSOR_PROVIDER, @@ -1102,12 +1103,12 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( completed: boolean, flushOf?: ActiveCursorTextSegment, ) { - const snapshot = takeTextSnapshot(context.reasoning, completed, flushOf); + const now = yield* DateTime.now; + const snapshot = takeTextSnapshot(context.reasoning, completed, flushOf, now); if (snapshot === null) { return; } const { segment, text } = snapshot; - const now = yield* DateTime.now; const ordinal = yield* resolveItemOrdinal(context, segment.nativeItemId); const nodeId = idAllocator.derive.nodeFromProviderItem({ driver: CursorAgentSdk.CURSOR_PROVIDER, @@ -1182,7 +1183,7 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( nativeItemId: `${input.kind}:${input.context.run.runId}:${input.stream.nextSegment}`, startedAt: yield* DateTime.now, text: "", - emitted: false, + emittedAtMs: null, dirty: false, }; } @@ -1206,10 +1207,11 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( }); /** - * Publishes a text change. A segment's first text emits at once so live - * output starts immediately; later changes coalesce into one snapshot per - * CURSOR_TEXT_FLUSH_INTERVAL. Completing the segment emits its final text - * and makes a pending flush a no-op. + * Publishes a text change. It emits at once when the segment's last + * snapshot is at least CURSOR_TEXT_FLUSH_INTERVAL_MS old, so live output + * starts immediately and a slow stream is never delayed. Faster changes + * share one flush at the end of that interval. Completing the segment + * emits its final text and makes a pending flush a no-op. */ const publishTextSegment = Effect.fnUntraced(function* ( context: ActiveCursorTurn, @@ -1217,12 +1219,16 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( ) { const emit = kind === "assistant" ? emitAssistant : emitReasoning; const segment = context[kind].current; - if (segment === null) return; - if (!segment.emitted) return yield* emit(context, false); - if (segment.dirty) return; + if (segment === null || segment.dirty) return; + // Clamped so a wall clock that steps back cannot stretch the flush delay. + const sinceLastMs = + segment.emittedAtMs === null + ? CURSOR_TEXT_FLUSH_INTERVAL_MS + : Math.max(0, DateTime.toEpochMillis(yield* DateTime.now) - segment.emittedAtMs); + if (sinceLastMs >= CURSOR_TEXT_FLUSH_INTERVAL_MS) return yield* emit(context, false); segment.dirty = true; yield* emit(context, false, segment).pipe( - Effect.delay(CURSOR_TEXT_FLUSH_INTERVAL), + Effect.delay(CURSOR_TEXT_FLUSH_INTERVAL_MS - sinceLastMs), Effect.forkIn(sessionScope), ); }); From 5c6059909b6cfc09d4b2cae6e3c918a662155cf1 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 1 Nov 2026 00:44:37 +0000 Subject: [PATCH 3/4] fix(cursor): follow main's adapter options and trim streaming comments Main removed the Cursor `settings` adapter option and its test decoder, so the coalescing test no longer compiled after merging main. Drop the dead option from that test, and cut the streaming comments to main's four-line cap. Co-Authored-By: Claude Opus 5.5 --- .../src/server/adapter.test.ts | 1 - .../provider-cursor/src/server/adapter.ts | 32 ++++++------------- 2 files changed, 10 insertions(+), 23 deletions(-) diff --git a/packages/provider-cursor/src/server/adapter.test.ts b/packages/provider-cursor/src/server/adapter.test.ts index ef2082be83e42..fa4ea38b6e389 100644 --- a/packages/provider-cursor/src/server/adapter.test.ts +++ b/packages/provider-cursor/src/server/adapter.test.ts @@ -464,7 +464,6 @@ describe("CursorAdapterV2", () => { let sendUpdate: ((update: InteractionUpdate) => Effect.Effect) | undefined; const adapter = yield* makeCursorAdapterV2({ instanceId, - settings: yield* decodeCursorSettings({}), environment: { HOME: workspace }, }).pipe( Effect.provideService(CursorAgentSdk.CursorAgentSdkRunner, { diff --git a/packages/provider-cursor/src/server/adapter.ts b/packages/provider-cursor/src/server/adapter.ts index 9a4a90eb48273..2bc99706361d2 100644 --- a/packages/provider-cursor/src/server/adapter.ts +++ b/packages/provider-cursor/src/server/adapter.ts @@ -892,9 +892,7 @@ const MAX_CURSOR_STREAMED_OUTPUT_CHARS = 256 * 1024; const CURSOR_RESUME_NO_PROGRESS_MESSAGE = "Agent turn stopped after repeated resume attempts made no progress"; -// Full-text snapshots of a streaming segment are at least this far apart, instead -// of one per delta, which grew with the square of the reply length. A delta after -// a quiet interval and the segment's close still emit at once. The flush is +// Minimum gap between full-text snapshots of a streaming segment. The flush is // scheduled, not a deadline, so it can land later under host load. const CURSOR_TEXT_FLUSH_INTERVAL_MS = 50; @@ -923,8 +921,7 @@ interface ActiveCursorTextSegment { readonly nativeItemId: string; readonly startedAt: DateTime.Utc; text: string; - // When the last snapshot was emitted, or null before the running node and - // first text. + // Null until the first snapshot, which also emits the running node. emittedAtMs: number | null; // Text changed since the last emission; a delayed flush is scheduled. dirty: boolean; @@ -1054,11 +1051,8 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( }); const textEmissionPermit = yield* Semaphore.make(1); - /** - * Claims the stream's current segment for one emission. A `flushOf` call - * is a delayed flush: it emits only while that segment is still current - * and has unsent text. - */ + // Claims the current segment for one emission. With `flushOf` it is a delayed + // flush, emitting only while that segment is still current and dirty. const takeTextSnapshot = ( stream: ActiveCursorTextStream, completed: boolean, @@ -1068,14 +1062,12 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( const segment = stream.current; if (segment === null || segment.text.length === 0) return null; if (flushOf !== undefined && (segment !== flushOf || !segment.dirty)) return null; - // The node carries no text, so it changes only when the segment opens - // and when it completes. + // The node carries no text, so it changes only on open and on completion. const emitsNode = completed || segment.emittedAtMs === null; segment.emittedAtMs = DateTime.toEpochMillis(now); segment.dirty = false; - // Read the text in the same step that clears `dirty`. Deltas appended - // while this emission yields mark the segment dirty again and reach the - // next flush or the close. + // Read the text in the step that clears `dirty`; deltas appended while this + // emission yields mark it dirty again for the next flush or the close. return { segment, text: segment.text, emitsNode }; }; @@ -1283,13 +1275,9 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( context.reasoning.current = null; }); - /** - * Publishes a text change. It emits at once when the segment's last - * snapshot is at least CURSOR_TEXT_FLUSH_INTERVAL_MS old, so live output - * starts immediately and a slow stream is never delayed. Faster changes - * share one flush at the end of that interval. Completing the segment - * emits its final text and makes a pending flush a no-op. - */ + // Runs per delta, so untraced. Emits at once when the last snapshot is at least + // the flush interval old; faster changes share one delayed flush, which + // completing the segment turns into a no-op. const publishTextSegment = Effect.fnUntraced(function* ( context: ActiveCursorTurn, kind: "assistant" | "reasoning", From 3e9985eea5d5ae92f86b626aea5ff682b109607e Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Wed, 11 Nov 2026 22:11:25 +0000 Subject: [PATCH 4/4] test(cursor): the streaming test covers reasoning snapshots --- .../src/server/adapter.test.ts | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/packages/provider-cursor/src/server/adapter.test.ts b/packages/provider-cursor/src/server/adapter.test.ts index d3b9cdd4495f1..47cd0b957fe56 100644 --- a/packages/provider-cursor/src/server/adapter.test.ts +++ b/packages/provider-cursor/src/server/adapter.test.ts @@ -641,6 +641,9 @@ describe("CursorAdapterV2", () => { yield* sendUpdate!({ type: "text-delta", text: "!" }); // Reasoning closes the assistant segment while its next flush is pending. yield* sendUpdate!({ type: "thinking-delta", text: "next" }); + for (let index = 0; index < 100; index += 1) { + yield* sendUpdate!({ type: "thinking-delta", text: " step" }); + } yield* TestClock.adjust("50 millis"); yield* Deferred.succeed(finished, undefined); const streamed = yield* runtime.events.pipe( @@ -681,6 +684,27 @@ describe("CursorAdapterV2", () => { ), ["running", "completed"], ); + const thinking = `next${" step".repeat(100)}`; + assert.deepEqual( + streamed.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "reasoning" + ? [[event.turnItem.status, event.turnItem.text]] + : [], + ), + [ + ["running", "next"], + ["running", thinking], + ["completed", thinking], + ], + ); + assert.deepEqual( + streamed.flatMap((event) => + event.type === "node.updated" && event.node.kind === "reasoning" + ? [event.node.status] + : [], + ), + ["running", "completed"], + ); }).pipe( Effect.scoped, Effect.provide(