diff --git a/apps/server/src/provider/EventNdjsonLogger.test.ts b/apps/server/src/provider/EventNdjsonLogger.test.ts index eed20381b46cb..0bb55b9bcb39b 100644 --- a/apps/server/src/provider/EventNdjsonLogger.test.ts +++ b/apps/server/src/provider/EventNdjsonLogger.test.ts @@ -255,6 +255,65 @@ it.layer(NodeServices.layer)("EventNdjsonLogger", (it) => { }), ); + it.effect("drops Cursor streaming deltas while keeping its lifecycle updates", () => + Effect.gen(function* () { + const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-")); + const basePath = NodePath.join(tempDir, "events.log"); + const threadId = ThreadId.make("cursor-deltas"); + const cursorUpdate = (update: object) => ({ + provider: "cursor", + kind: "protocol", + event: { + direction: "incoming", + stage: "decoded", + payload: { type: "interaction.update", runId: "run-1", update }, + }, + }); + const toolCall = { type: "shell", args: { command: "ls" } }; + const transient = [ + { type: "text-delta", text: "Hel" }, + { type: "thinking-delta", text: "Hmm" }, + { type: "token-delta", tokens: 3 }, + { type: "shell-output-delta", event: { stdout: "a.ts\n" } }, + { type: "partial-tool-call", callId: "call-1", modelCallId: "m-1", toolCall }, + { + type: "tool-call-delta", + callId: "call-1", + modelCallId: "m-1", + taskUpdate: { type: "text-delta", text: "Sub" }, + }, + ].map(cursorUpdate); + const retained = [ + { type: "tool-call-started", callId: "call-1", modelCallId: "m-1", toolCall }, + { + type: "tool-call-delta", + callId: "call-1", + modelCallId: "m-1", + taskUpdate: { type: "tool-call-completed", callId: "call-2", toolCall }, + }, + { type: "tool-call-completed", callId: "call-1", modelCallId: "m-1", toolCall }, + { type: "turn-ended" }, + ].map(cursorUpdate); + + try { + const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 0 }); + const native = store.logger("native"); + for (const event of [...transient, ...retained]) { + yield* native.write(event, threadId); + } + yield* store.close(); + + const records = NodeFS.readFileSync(ownedLogPath(basePath, "cursor-deltas"), "utf8") + .trim() + .split("\n") + .map((line) => decodeUnknownJson(parseLogLine(line).payload)); + assert.deepEqual(records, retained); + } finally { + NodeFS.rmSync(tempDir, { recursive: true, force: true }); + } + }), + ); + it.effect("bounds oversized records while retaining failure details", () => Effect.gen(function* () { const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-")); diff --git a/apps/server/src/provider/EventNdjsonLogger.ts b/apps/server/src/provider/EventNdjsonLogger.ts index 6a1d3844b2373..5a7b3578578f1 100644 --- a/apps/server/src/provider/EventNdjsonLogger.ts +++ b/apps/server/src/provider/EventNdjsonLogger.ts @@ -63,6 +63,13 @@ const transientNativeMethods = new Set([ "turn/diff/updated", ]); const transientAcpUpdates = new Set(["agent_message_chunk", "agent_thought_chunk"]); +const transientCursorUpdates = new Set([ + "text-delta", + "thinking-delta", + "token-delta", + "shell-output-delta", + "partial-tool-call", +]); export type EventNdjsonStream = "native" | "canonical" | "orchestration"; @@ -219,6 +226,18 @@ export function shouldPersistProviderEvent(stream: EventNdjsonStream, event: unk } } + if (nativeType === "interaction.update") { + let update = Reflect.get(nativeEvent, "update"); + if (typeof update !== "object" || update === null) return true; + // Subagent updates arrive wrapped; keep the wrapper when it carries a lifecycle record. + if (Reflect.get(update, "type") === "tool-call-delta") { + update = Reflect.get(update, "taskUpdate"); + if (typeof update !== "object" || update === null) return true; + } + const updateType = Reflect.get(update, "type"); + return typeof updateType !== "string" || !transientCursorUpdates.has(updateType); + } + const payload = Reflect.get(nativeEvent, "payload"); if (typeof payload !== "object" || payload === null) return true;