test: verify latency harness completion flow

.env.example

@@ -125,6 +125,12 @@ ERROR_SERVICE_API_KEY=your-posthog-error-tracking-api-key
 # E2E_TEST=true
 # IS_TEST=true
 
+# Remote-workspace latency debugging
+# Disable ephemeral partial message persistence to compare against durable-per-update behavior.
+# CLINE_DISABLE_EPHEMERAL_MESSAGE_PERSISTENCE=true
+# Override periodic safety flush cadence in milliseconds (default: 1500)
+# CLINE_EPHEMERAL_MESSAGE_FLUSH_CADENCE_MS=1500
+
 # ============================================================================
 # USAGE INSTRUCTIONS
 # ============================================================================

docs/remote-workspace-latency-technique-01-ephemeral-partial-persistence-plan.md

@@ -0,0 +1,479 @@
+# Technique Plan: Ephemeral Partial Persistence Split
+
+This document is the implementation plan for the **ephemeral partial persistence split** technique identified in `docs/remote-workspace-latency-branch-analysis-report.md` as the highest-impact improvement for remote-workspace user-perceived latency.
+
+The core idea is simple:
+
+> **Stop treating every streamed chunk as durable state that must be written to disk immediately.**
+
+That principle matters everywhere in this project, but it is especially important here. In remote workspace mode, partial updates are expensive not because the bytes are individually large, but because there are many of them and each one can pull in remote filesystem I/O, history recomputation, serialization work, and follow-on transport churn. For the user, this shows up as sluggish or jittery streaming, especially during long answers or large-file operations where the agent emits lots of progress text.
+
+This plan focuses on splitting **ephemeral UI mutations** from **durable persistence boundaries** while preserving crash recovery, history correctness, and the rest of Cline’s product behavior.
+
+## How To Use This Plan
+
+This plan should be executed **on a new development branch**, not by continuing to stack work directly onto `eve_troubleshooting-remote-workspaces`.
+
+That branch, `eve_troubleshooting-remote-workspaces`, should be treated as the **fully developed reference implementation** for this technique. In other words, the work described here is not speculative greenfield design work; it is a structured plan for re-deriving, validating, and extracting the technique from the already-working reference implementation into a smaller, more reviewable changeset.
+
+The most effective way to use this document is:
+
+1. read the goal and mental model for the step,
+2. inspect how the reference implementation branch already solved it,
+3. extract the minimum coherent version of that change into your own branch,
+4. run the tests and validation described here,
+5. compare behavior against the reference implementation when in doubt.
+
+Be smart about this. Do not re-invent behavior that already exists in the reference implementation unless there is a very clear reason to improve or simplify it. The branch already contains the end-state shape we want to learn from. Your job is to extract, verify, and explain the minimal version of that behavior safely.
+
+## Document Type, Audience, and Quality Bar
+
+This is an **extraction implementation plan**, not a greenfield design doc and not a PR description. It is intended for a **Staff+ level distributed systems / infrastructure engineer** who is extracting a production-worthy technique from a known-good integrated implementation.
+
+That means the quality bar is:
+
+- the plan should be executable with minimal ambiguity,
+- the reasoning behind each step should be legible to a strong reviewer,
+- and the extraction should preserve product behavior across the rest of Cline while improving remote-workspace latency.
+
+If a step in this plan would not help a strong engineer quickly answer “what exactly am I changing, why now, what must remain true, and how do I verify it?”, then the step is not detailed enough.
+
+## Artifact Stack and Dependency Position
+
+This doc sits in the broader artifact stack as follows:
+
+1. `docs/remote-workspace-latency-branch-analysis-report.md` explains **which techniques matter most and why**.
+2. This document explains **how to extract and implement one technique in a smaller branch/PR**.
+3. A later PR-slicing / execution phase should use this document to drive real implementation work.
+
+When using this plan, always begin by re-reading the branch analysis report so the extraction stays aligned with the larger prioritization and product intent.
+
+## Developer Operating Posture
+
+This plan is written for a Staff+-level engineer who is expected to use judgment, not just mechanically check boxes.
+
+While implementing each step:
+
+- actively compare your changes to `eve_troubleshooting-remote-workspaces`,
+- preserve the original user experience across non-remote product surfaces,
+- prefer coherent extraction over literal copy-paste,
+- and keep asking: **what hot-path work are we removing, and what correctness boundary are we preserving?**
+
+The most important meta-principle is still:
+
+> **Stop treating every streamed chunk as a durable, full-state, immediately-presented event.**
+
+For this technique specifically, the emphasis is on the **durable** part of that sentence.
+
+## Minimal Coherent Extraction Boundary
+
+The smallest coherent PR for this technique should usually include:
+
+- message-state ephemeral mutation APIs,
+- dirty tracking plus explicit flush support,
+- task callsite conversion for partial updates,
+- periodic safety flush,
+- and the minimum set of tests needed to prove correctness.
+
+What should **not** be split away from this technique if avoidable:
+
+- the dirty-bit / flush contract,
+- the durable-boundary logic for partial → complete transitions,
+- safety-flush lifecycle management,
+- and the tests that prove resume/abort correctness.
+
+If those pieces are separated too aggressively, reviewers will have a much harder time understanding whether the extraction is actually safe.
+
+## Common Failure Modes While Extracting
+
+Watch for these failure modes explicitly:
+
+- extracting the ephemeral APIs without converting the real hot-path callsites,
+- clearing the dirty bit too early,
+- leaving abort/resume paths on stale assumptions about immediate persistence,
+- making durable and ephemeral mutation paths diverge semantically,
+- and validating only happy-path streaming while missing crash-recovery or resume regressions.
+
+If any of those happen, the extraction may appear simpler while actually weakening the product.
+
+---
+
+## Why This Technique Matters
+
+When Cline is writing a large file, updating a long reasoning trace, or streaming many incremental tool/progress updates, the old behavior effectively says:
+
+1. mutate message state,
+2. save messages to disk,
+3. update task history,
+4. possibly trigger more UI/state work,
+5. repeat on the next partial chunk.
+
+That is the wrong clock boundary.
+
+The better mental model is:
+
+- **Streaming partials are animation state.** They exist to help the user perceive progress.
+- **Durable persistence is recovery state.** It exists so the task can survive restart, cancellation, or history resume.
+- **Those are related, but they are not the same thing.**
+
+So the goal here is to keep the UI feeling live while only persisting at meaningful boundaries plus an occasional safety flush.
+
+---
+
+## Success Criteria
+
+- Partial `say(...)` and `ask(...)` updates no longer synchronously persist on every mutation.
+- Durable persistence still occurs at semantic boundaries such as completion, cancel, tool completion, and request completion.
+- Long-running streams periodically safety-flush unsaved partial changes.
+- Resume-from-history and crash-recovery behavior remain correct and predictable.
+- Message-state behavior remains mutex-safe and consistent under concurrent tool / stream / abort activity.
+
+---
+
+## Files Most Likely to Change
+
+- `src/core/task/message-state.ts`
+- `src/core/task/index.ts`
+- `src/core/task/EphemeralMessageFlushScheduler.ts`
+- `src/test/message-state-handler.test.ts`
+- `src/core/task/__tests__/EphemeralMessageFlushScheduler.test.ts`
+- `src/core/task/__tests__/latency.test.ts`
+- possibly targeted resume / abort integration tests
+
+---
+
+## Step-by-Step Implementation Plan
+
+## Step 1 — Define the durability contract for message mutations
+
+### Goal
+
+Write down the architectural contract before changing behavior, so future developers know which updates are ephemeral and which are durable.
+
+### Mental model
+
+If developers cannot quickly answer “does this mutation need to survive a crash immediately?”, they will accidentally route new hot-path updates back through synchronous persistence. This step prevents future regression.
+
+### Work
+
+- [x] Add code comments near `MessageStateHandler` describing the distinction between ephemeral and durable mutations.
+- [x] Add code comments in `Task.say(...)` and `Task.ask(...)` documenting which partial flows are intentionally ephemeral.
+- [x] Define a durable-boundary checklist in comments or docstrings, including at minimum:
+  - [x] partial → complete transition
+  - [x] tool completion / tool result boundary
+  - [x] request completion
+  - [x] cancellation / abort
+  - [x] resume-related state changes
+  - [x] checkpoint-relevant events
+
+### Detailed code changes
+
+- In `src/core/task/message-state.ts`, add a short header comment near the class definition explaining:
+  - durable methods write immediately,
+  - ephemeral methods mutate in memory and emit change notifications,
+  - `flushClineMessagesAndUpdateHistory()` is the bridge between the two.
+- In `src/core/task/index.ts`, add comments above the partial-update branches in `say(...)` and `ask(...)` explaining why partial updates are not always durable.
+
+When doing this work, read the corresponding code in `eve_troubleshooting-remote-workspaces` first and copy the intent, not just the surface syntax. The point of this step is to make the extraction self-explanatory for future reviewers and maintainers.
+
+### Tests
+
+- [x] No behavioral tests required for comments alone.
+- [x] Ensure any snapshot/doc-based linting or type-check flow still passes.
+
+---
+
+## Step 2 — Add explicit ephemeral mutation APIs to `MessageStateHandler`
+
+### Goal
+
+Create first-class APIs for in-memory message changes that emit message-change notifications without synchronously persisting.
+
+### Mental model
+
+The presence of dedicated APIs changes developer behavior. If the only available mutation helper is a durable save path, then everything becomes durable by default. We want the inverse for streaming partials: durable only when intentionally requested.
+
+### Work
+
+- [x] Add `addToClineMessagesEphemeral(message)`.
+- [x] Add `updateClineMessageEphemeral(index, updates)`.
+- [x] Add internal dirty tracking for unsaved ephemeral changes.
+- [x] Ensure ephemeral methods emit the same `clineMessagesChanged` events and task UI deltas as durable methods.
+- [x] Ensure all of the above remain protected by the existing mutex.
+
+### Detailed code changes
+
+- In `src/core/task/message-state.ts`:
+- [x] Add a `hasDirtyEphemeralChanges` flag if it does not already exist.
+  - [x] In `addToClineMessagesEphemeral(...)`:
+    - [x] set `conversationHistoryIndex` and `conversationHistoryDeletedRange` exactly as durable add does,
+    - [x] mutate `clineMessages`,
+    - [x] mark dirty,
+    - [x] emit `clineMessagesChanged`.
+  - [x] In `updateClineMessageEphemeral(...)`:
+    - [x] validate index,
+    - [x] capture previous message,
+    - [x] mutate in place,
+    - [x] mark dirty,
+    - [x] emit `clineMessagesChanged`.
+  - [x] Keep delta emission behavior identical between ephemeral and durable changes so frontend live behavior stays consistent.
+
+The smart way to execute this step is to compare the durable and ephemeral codepaths side by side in the reference implementation branch and preserve the shared invariants exactly. The extraction should reduce write amplification, not create a shadow message-state model with slightly different semantics.
+
+### Tests
+
+- [x] Unit test: ephemeral add mutates in-memory state without calling persistence.
+- [x] Unit test: ephemeral update mutates in-memory state without calling persistence.
+- [x] Unit test: ephemeral mutation emits `clineMessagesChanged` with correct shape.
+- [x] Unit test: ephemeral mutation still emits task UI deltas when delta sync is enabled.
+- [x] Unit test: invalid index still throws in ephemeral update path.
+
+---
+
+## Step 3 — Add explicit flush behavior for previously-ephemeral changes
+
+### Goal
+
+Provide a single durable flush method that persists all dirty ephemeral changes and updates task history once.
+
+### Mental model
+
+We are not removing durability; we are **batching** durability at the right semantic times. The flush method is the “commit” for a burst of ephemeral UI activity.
+
+### Work
+
+- [x] Add `flushClineMessagesAndUpdateHistory()` if not already present.
+- [x] Make it a no-op when no ephemeral changes are dirty.
+- [x] Ensure it reuses the same internal persistence logic as durable mutations.
+
+### Detailed code changes
+
+- In `src/core/task/message-state.ts`:
+  - [x] Add `flushClineMessagesAndUpdateHistory()` guarded by `withStateLock(...)`.
+  - [x] If `hasDirtyEphemeralChanges` is false, return early.
+  - [x] Otherwise call `saveClineMessagesAndUpdateHistoryInternal()`.
+  - [x] Ensure `saveClineMessagesAndUpdateHistoryInternal()` clears the dirty flag only after successful save/update-history flow.
+
+This step is where the implementation starts to feel like a deliberate state machine rather than a collection of helper methods. Be smart about failure ordering here: if the code clears the dirty bit too early or conflates flush success with mutation success, recovery correctness will quietly degrade.
+
+### Tests
+
+- [x] Unit test: flush persists previously-ephemeral changes.
+- [x] Unit test: flush is a cheap no-op when there are no dirty changes.
+- [x] Unit test: task history reflects flushed content after prior ephemeral mutation.
+
+---
+
+## Step 4 — Switch streaming partial update callsites to ephemeral APIs
+
+### Goal
+
+Move the actual hot-path streaming mutations onto the new ephemeral methods.
+
+### Mental model
+
+The APIs only matter if the streaming loop uses them. This is the step that converts theory into latency improvement.
+
+### Work
+
+- [x] Audit all partial `say(...)` paths.
+- [x] Audit all partial `ask(...)` paths.
+- [x] Switch normal streaming partial updates from durable to ephemeral mutation methods.
+- [x] Keep complete/finalized messages on durable paths unless explicitly flushed immediately after ephemeral completion.
+
+### Detailed code changes
+
+- In `src/core/task/index.ts`, update `say(...)`:
+  - [x] partial update of existing partial `say` message should use `updateClineMessageEphemeral(...)`.
+  - [x] new partial `say` message insertion should use `addToClineMessagesEphemeral(...)` where appropriate.
+- In `src/core/task/index.ts`, update `ask(...)`:
+  - [x] partial update of existing partial `ask` message should use `updateClineMessageEphemeral(...)`.
+  - [x] new partial `ask` insertion should use `addToClineMessagesEphemeral(...)` where appropriate.
+- For reasoning/tool-progress-related live updates:
+  - [x] ensure they remain visible to the UI through message-change events / partial-message events,
+  - [x] but no longer synchronously persist each partial mutation.
+
+This is the step where it becomes easy to accidentally under-extract or over-extract. Use the reference implementation branch aggressively here. If a callsite was made ephemeral in `eve_troubleshooting-remote-workspaces`, understand why. If a callsite remained durable, understand why. Preserve that distinction intentionally.
+
+### Tests
+
+- [x] Integration-style test: many partial text updates do not trigger per-update durable saves.
+- [ ] Unit test: partial `say(...)` path still emits live UI updates while skipping persistence.
+- [ ] Unit test: partial `ask(...)` path still emits live UI updates while skipping persistence.
+
+---
+
+## Step 5 — Define and enforce durable flush boundaries
+
+### Goal
+
+Ensure the system persists at the correct semantic points so correctness is preserved.
+
+### Mental model
+
+This is the balancing step. We are intentionally reducing durability frequency, so we must be precise about where durability is still required.
+
+### Work
+
+- [x] Identify all partial → complete transitions.
+- [x] Flush at request completion.
+- [x] Flush on abort / cancellation.
+- [x] Flush when tool execution reaches a stable durable boundary.
+- [x] Flush when history / resume semantics require consistency.
+
+### Detailed code changes
+
+- In `src/core/task/index.ts`:
+  - [x] when `partial: false` finalizes a previously partial message, use durable `updateClineMessage(...)` or explicit flush right after ephemeral completion.
+  - [x] in request-finalization logic, call `flushClineMessagesAndUpdateHistory()` before or alongside final durable save path where needed.
+  - [x] in abort/cancel paths, ensure pending ephemeral changes are flushed before task shutdown is considered complete.
+  - [x] in resume-related flows, ensure history-visible state is not left behind dirty.
+
+This step is the heart of the technique. The way to think about it is: we are removing durability from the hot path only because we are reintroducing durability at the right semantic boundaries. If those boundaries are fuzzy, the technique is incomplete.
+
+### Tests
+
+- [x] Unit test: partial → complete transition results in durable persistence.
+- [ ] Integration test: abort during stream persists a recoverable state.
+- [x] Regression test: resume-from-history still works after deferred partial persistence.
+- [x] Regression test: tool result flows remain properly visible in history after finalization.
+
+---
+
+## Step 6 — Add the periodic safety-flush scheduler
+
+### Goal
+
+Bound the amount of live UI state that could be lost if the extension host crashes during a long-running stream.
+
+### Mental model
+
+The correct behavior is not “persist every partial” and not “never persist until the very end.” The practical middle ground is:
+
+- partials stay ephemeral during normal live streaming,
+- but dirty state gets checkpointed periodically at low frequency.
+
+### Work
+
+- [x] Add or finish `EphemeralMessageFlushScheduler`.
+- [x] Start it when a request begins streaming.
+- [x] Stop it when the request completes or aborts.
+- [x] Make it call `flushClineMessagesAndUpdateHistory()` on cadence only when dirty.
+
+### Detailed code changes
+
+- In `src/core/task/EphemeralMessageFlushScheduler.ts`:
+  - [x] ensure a single timer is active,
+  - [x] prevent overlap between flushes,
+  - [x] support clean start/stop/dispose semantics.
+- In `src/core/task/index.ts`:
+  - [x] start scheduler near request start,
+  - [x] stop scheduler in both success and error/finally paths,
+  - [x] make cadence conservative (for example ~1.5s) to preserve UX win while bounding recovery loss.
+
+Use the reference implementation’s chosen cadence and lifecycle wiring as the default starting point. Only diverge if you can clearly articulate why a different extraction improves isolation or reviewability without changing the core behavior.
+
+### Tests
+
+- [x] Unit test: scheduler flushes pending ephemeral changes on cadence.
+- [x] Unit test: scheduler does nothing when there is no dirty ephemeral state.
+- [x] Unit test: scheduler stops cleanly on request completion/abort.
+
+---
+
+## Step 7 — Validate large-file and long-stream scenarios explicitly
+
+### Goal
+
+Confirm that the technique helps exactly the user scenario we care about: long-running, high-churn operations such as writing or rewriting large files.
+
+### Mental model
+
+Large-file operations are effectively “progress-heavy” workloads. Even if the tool invocation itself is durable, the surrounding reasoning, progress, and partial text can create a huge amount of avoidable mutation churn.
+
+This is the direct answer to the large-file-write scenario: this technique helps because large-file operations tend to produce many small intermediate UI updates, and those should not all be treated like crash-critical persistence boundaries.
+
+### Work
+
+- [x] Add a targeted validation scenario for large streamed output / large-file write behavior.
+- [x] Confirm persistence flush count drops sharply versus baseline.
+- [x] Confirm the user still sees smooth progress and correct final durable state.
+
+### Detailed code changes
+
+- Extend existing latency validation or tests to simulate:
+  - [x] long streaming response,
+  - [x] tool progress and completion,
+  - [ ] large-file write workflow or equivalent sustained partial-update workload.
+- Use telemetry fields already added in latency instrumentation to compare:
+  - [ ] persistence flush count,
+  - [ ] save durations,
+  - [ ] chunk-to-webview timing.
+
+### Tests
+
+- [x] Performance/regression test: long stream causes far fewer persistence flushes than partial-update count.
+- [x] Validation harness comparison: baseline vs ephemeral-persistence-enabled variant.
+- [x] Regression test: final message history and resume state remain correct.
+
+Implementation note: the standalone validation harness at `scripts/validate-latency-scenarios.ts` is now wired to `npm run test:latency:validate` and completes successfully for local/remote plus enabled/disabled variants. In this standalone test environment, task-UI delta streaming is still absent (`taskDeltaCount: 0`), so the harness currently validates state/partial-message timing and completion behavior rather than delta-channel behavior.
+
+---
+
+## Step 8 — Rollout safeguards and developer controls
+
+### Goal
+
+Make the feature easy to disable, validate, and debug during rollout.
+
+### Mental model
+
+Hot-path changes need escape hatches. If behavior regresses in an edge case, the team should be able to isolate the feature quickly.
+
+### Work
+
+- [x] Keep or add env flag gating for ephemeral persistence behavior.
+- [x] Ensure telemetry can compare enabled vs disabled behavior.
+- [x] Add debug logging only if low-noise and useful.
+
+### Detailed code changes
+
+- In `src/core/task/latency.ts` / `.env.example`:
+  - [x] preserve `CLINE_DISABLE_EPHEMERAL_MESSAGE_PERSISTENCE` or equivalent.
+  - [x] document intended use for A/B validation.
+
+Treat the feature flag and validation path as first-class extraction requirements, not afterthoughts. Since the reference implementation already exists, the smoother development process is to keep comparison easy between “extracted technique enabled” and “feature disabled” modes.
+
+### Tests
+
+- [x] Unit test: disable flag routes behavior back to durable-per-update path where applicable.
+- [x] Validation harness variant: feature-disabled mode still behaves correctly.
+
+Implementation note: the validation harness includes an `ephemeral_disabled` variant for A/B runs and now verifies that the feature-disabled mode completes successfully in both local and remote hostbridge scenarios.
+
+---
+
+## Developer Checklist Summary
+
+- [x] Document the durability contract
+- [x] Add explicit ephemeral mutation APIs
+- [x] Add dirty tracking and explicit flush support
+- [x] Convert streaming partial callsites to ephemeral mutations
+- [x] Enforce durable flushes at semantic boundaries
+- [x] Add periodic safety flush scheduler
+- [x] Validate large-file / long-stream scenarios
+- [x] Preserve rollout flags and debugging support
+- [x] Run unit, integration, and validation-harness checks
+
+---
+
+## Final Mental Model Recap
+
+When implementing this technique, keep this in mind:
+
+- **Partial updates are for perception.**
+- **Durable saves are for recovery.**
+- **Doing recovery work on every animation step is what makes remote mode feel bad.**
+- **The fix is not to persist less carelessly; it is to persist more intentionally.**
+
+If this plan is implemented cleanly, large-file writes, long reasoning streams, and other high-churn tasks should feel markedly smoother in remote workspaces without sacrificing correctness.

package.json

@@ -423,6 +423,7 @@
 		"test:unit": "cross-env TS_NODE_PROJECT=./tsconfig.unit-test.json mocha",
 		"test:coverage": "vscode-test --coverage",
 		"test:sca-server": "npx tsx watch scripts/test-standalone-core-api-server.ts",
+		"test:latency:validate": "npx tsx scripts/validate-latency-scenarios.ts",
 		"test:tp-orchestrator": "npx tsx scripts/testing-platform-orchestrator.ts",
 		"e2e": "playwright test -c playwright.config.ts",
 		"test:e2e:build": "vsce package --allow-package-secrets sendgrid --out dist/e2e.vsix",

scripts/validate-latency-scenarios.ts

@@ -0,0 +1,433 @@
+#!/usr/bin/env npx tsx
+
+import { type ChildProcess, spawn } from "node:child_process"
+import { once } from "node:events"
+import fs from "node:fs/promises"
+import net from "node:net"
+import os from "node:os"
+import path from "node:path"
+import { fileURLToPath } from "node:url"
+import { credentials } from "@grpc/grpc-js"
+import { AccountServiceClient } from "../src/generated/grpc-js/cline/account"
+import { StateServiceClient } from "../src/generated/grpc-js/cline/state"
+import { TaskServiceClient } from "../src/generated/grpc-js/cline/task"
+import { UiServiceClient } from "../src/generated/grpc-js/cline/ui"
+
+type ValidationMode = "local" | "remote"
+
+type ValidationVariant = {
+	name: string
+	env: Record<string, string>
+}
+
+type ScenarioResult = {
+	variant: string
+	mode: ValidationMode
+	newTaskRpcMs: number
+	firstStateMs: number | null
+	firstPartialMessageMs: number | null
+	firstTaskDeltaMs: number | null
+	completionMs: number | null
+	stateUpdateCount: number
+	partialMessageCount: number
+	taskDeltaCount: number
+	statePayloadBytes: number
+	taskDeltaPayloadBytes: number
+	messageCountAtCompletion: number | null
+	completed: boolean
+	error?: string
+}
+
+type StartedServer = {
+	child: ChildProcess
+	grpcPort: number
+	hostbridgePort: number
+	workspaceDir: string
+	getStderr: () => string
+}
+
+const SCRIPT_DIR = path.dirname(fileURLToPath(import.meta.url))
+const PROJECT_ROOT = path.resolve(SCRIPT_DIR, "..")
+
+const variants: ValidationVariant[] = [
+	{ name: "default", env: {} },
+	{
+		name: "presentation_disabled",
+		env: {
+			CLINE_DISABLE_PRESENTATION_SCHEDULER: "true",
+		},
+	},
+	{
+		name: "ephemeral_disabled",
+		env: {
+			CLINE_DISABLE_EPHEMERAL_MESSAGE_PERSISTENCE: "true",
+		},
+	},
+	{
+		name: "delta_disabled",
+		env: {
+			CLINE_DISABLE_TASK_UI_DELTA_SYNC: "true",
+		},
+	},
+]
+
+async function waitForPort(port: number, timeoutMs = 20_000): Promise<void> {
+	const startedAt = Date.now()
+	while (Date.now() - startedAt < timeoutMs) {
+		try {
+			await new Promise<void>((resolve, reject) => {
+				const socket = net.connect(port, "127.0.0.1", () => {
+					socket.destroy()
+					resolve()
+				})
+				socket.on("error", reject)
+			})
+			return
+		} catch {
+			await new Promise((resolve) => setTimeout(resolve, 100))
+		}
+	}
+	throw new Error(`Timed out waiting for port ${port}`)
+}
+
+async function isPortBusy(port: number): Promise<boolean> {
+	try {
+		await new Promise<void>((resolve, reject) => {
+			const socket = net.connect(port, "127.0.0.1", () => {
+				socket.destroy()
+				resolve()
+			})
+			socket.on("error", reject)
+		})
+		return true
+	} catch {
+		return false
+	}
+}
+
+async function getFreePort(): Promise<number> {
+	return await new Promise((resolve, reject) => {
+		const server = net.createServer()
+		server.listen(0, "127.0.0.1", () => {
+			const address = server.address()
+			if (!address || typeof address === "string") {
+				server.close(() => reject(new Error("Unable to allocate port")))
+				return
+			}
+			const { port } = address
+			server.close((error) => {
+				if (error) {
+					reject(error)
+					return
+				}
+				resolve(port)
+			})
+		})
+		server.on("error", reject)
+	})
+}
+
+function unaryCall<T>(fn: (callback: (error: Error | null, response: T) => void) => void): Promise<T> {
+	return new Promise((resolve, reject) => {
+		fn((error, response) => {
+			if (error) {
+				reject(error)
+				return
+			}
+			resolve(response)
+		})
+	})
+}
+
+async function startServer(mode: ValidationMode, envOverrides: Record<string, string>): Promise<StartedServer> {
+	if (await isPortBusy(7777)) {
+		throw new Error(
+			"Cannot start latency validation harness because the mock API port 7777 is already in use. Stop the conflicting process and retry.",
+		)
+	}
+
+	const grpcPort = await getFreePort()
+	const hostbridgePort = await getFreePort()
+	const workspaceDir = await fs.mkdtemp(path.join(os.tmpdir(), "cline-latency-validate-"))
+	await fs.writeFile(path.join(workspaceDir, "test.ts"), 'export const name = "john"\n', "utf8")
+	const env: Record<string, string> = {
+		...process.env,
+		PROTOBUS_PORT: String(grpcPort),
+		HOSTBRIDGE_PORT: String(hostbridgePort),
+		WORKSPACE_DIR: workspaceDir,
+		E2E_TEST: "true",
+		CLINE_ENVIRONMENT: "local",
+		TEST_HOSTBRIDGE_REMOTE_NAME: mode === "remote" ? "ssh-remote" : "",
+		TEST_HOSTBRIDGE_PLATFORM: mode === "remote" ? "VS Code Remote" : "VS Code",
+		GRPC_RECORDER_ENABLED: "false",
+		...envOverrides,
+	}
+
+	const child = spawn("npx", ["tsx", path.join(PROJECT_ROOT, "scripts", "test-standalone-core-api-server.ts")], {
+		cwd: PROJECT_ROOT,
+		env,
+		stdio: ["ignore", "pipe", "pipe"],
+	})
+
+	child.stdout?.on("data", () => {
+		// Drain stdout so the spawned server cannot block on a full pipe buffer.
+	})
+
+	let stderr = ""
+	let exitedEarly = false
+	child.stderr?.on("data", (chunk) => {
+		stderr += chunk.toString()
+	})
+	child.once("exit", () => {
+		exitedEarly = true
+	})
+
+	try {
+		await waitForPort(grpcPort)
+	} catch (error) {
+		if (exitedEarly) {
+			throw new Error(`Latency validation server exited before becoming ready.${stderr ? `\n${stderr.trim()}` : ""}`)
+		}
+		throw error
+	}
+	return { child, grpcPort, hostbridgePort, workspaceDir, getStderr: () => stderr }
+}
+
+async function stopServer(server: StartedServer) {
+	const { child, workspaceDir } = server
+	if (child.killed || child.exitCode !== null) {
+		await fs.rm(workspaceDir, { recursive: true, force: true })
+		return
+	}
+	child.kill("SIGINT")
+	try {
+		await Promise.race([once(child, "exit"), new Promise((resolve) => setTimeout(resolve, 5_000))])
+	} catch {
+		child.kill("SIGKILL")
+	}
+	await fs.rm(workspaceDir, { recursive: true, force: true })
+}
+
+async function runScenario(mode: ValidationMode, variant: ValidationVariant): Promise<ScenarioResult> {
+	const server = await startServer(mode, variant.env)
+	const address = `127.0.0.1:${server.grpcPort}`
+	const accountClient = new AccountServiceClient(address, credentials.createInsecure())
+	const stateClient = new StateServiceClient(address, credentials.createInsecure())
+	const taskClient = new TaskServiceClient(address, credentials.createInsecure())
+	const uiClient = new UiServiceClient(address, credentials.createInsecure())
+	const respondedAskTs = new Set<number>()
+
+	let currentTaskId: string | undefined
+	const startedAt = Date.now()
+	let firstStateMs: number | null = null
+	let firstPartialMessageMs: number | null = null
+	let firstTaskDeltaMs: number | null = null
+	let completionMs: number | null = null
+	let stateUpdateCount = 0
+	let partialMessageCount = 0
+	let taskDeltaCount = 0
+	let statePayloadBytes = 0
+	let taskDeltaPayloadBytes = 0
+	let messageCountAtCompletion: number | null = null
+	let completed = false
+	let lastMessagesSummary = ""
+
+	const stateStream = stateClient.subscribeToState({})
+	const partialStream = uiClient.subscribeToPartialMessage({})
+	type DeltaReadableStream = {
+		on(event: "data", listener: (event: { deltaJson?: string }) => void): DeltaReadableStream
+		on(event: "error", listener: (error: unknown) => void): DeltaReadableStream
+		cancel(): void
+	}
+	const deltaStreamFactory = (
+		uiClient as typeof uiClient & {
+			subscribeToTaskUiDeltas?: () => DeltaReadableStream
+		}
+	).subscribeToTaskUiDeltas
+	const deltaStream = deltaStreamFactory?.call(uiClient)
+	const activeStreams = [stateStream, partialStream, deltaStream].filter(
+		(stream): stream is Exclude<typeof stream, undefined> => stream !== undefined,
+	)
+
+	for (const stream of activeStreams) {
+		stream.on("error", (streamError: any) => {
+			if (streamError?.code === 1 || streamError?.details === "Cancelled on client") {
+				return
+			}
+			console.error("validation stream error", streamError)
+		})
+	}
+
+	stateStream.on("data", (response: { stateJson?: string }) => {
+		stateUpdateCount += 1
+		const stateJson = response.stateJson || "{}"
+		statePayloadBytes += Buffer.byteLength(stateJson, "utf8")
+		if (firstStateMs === null) {
+			firstStateMs = Date.now() - startedAt
+		}
+		try {
+			const state = JSON.parse(stateJson)
+			const activeTaskId = state.currentTaskItem?.id
+			if (activeTaskId) {
+				currentTaskId = activeTaskId
+			}
+			const clineMessages = Array.isArray(state.clineMessages) ? state.clineMessages : []
+			lastMessagesSummary = clineMessages
+				.slice(-5)
+				.map((message: any) => {
+					const kind = message.type === "ask" || message.type === 0 ? "ask" : "say"
+					const subtype = kind === "ask" ? message.ask : message.say
+					return `${kind}:${String(subtype)}:${String(message.text ?? "").slice(0, 60)}`
+				})
+				.join(" | ")
+			const lastMessage = clineMessages.at(-1)
+			const askType = lastMessage?.ask
+			const askTs = typeof lastMessage?.ts === "number" ? lastMessage.ts : undefined
+			const isAskMessage = lastMessage?.type === "ask" || lastMessage?.type === 0 || askType !== undefined
+			if (isAskMessage && askTs !== undefined && !respondedAskTs.has(askTs)) {
+				if (
+					askType === "tool" ||
+					askType === 5 ||
+					askType === "api_req_failed" ||
+					askType === 6 ||
+					askType === "completion_result" ||
+					askType === 4 ||
+					askType === "resume_completed_task" ||
+					askType === 8
+				) {
+					respondedAskTs.add(askTs)
+					void unaryCall((callback) =>
+						taskClient.askResponse(
+							{
+								metadata: undefined,
+								responseType: "yesButtonClicked",
+								text: "",
+								images: [],
+								files: [],
+							},
+							callback as any,
+						),
+					).catch((error) => {
+						console.error("validation askResponse error", error)
+					})
+				}
+			}
+			const hasCompletion = clineMessages.some(
+				(message: any) =>
+					message.ask === "completion_result" ||
+					message.ask === "resume_completed_task" ||
+					message.ask === 4 ||
+					message.ask === 8,
+			)
+			if (hasCompletion && completionMs === null) {
+				completionMs = Date.now() - startedAt
+				messageCountAtCompletion = clineMessages.length
+				completed = true
+			}
+		} catch {
+			// ignore parse errors in validation harness
+		}
+	})
+
+	partialStream.on("data", (message: { say?: string | number; text?: string }) => {
+		partialMessageCount += 1
+		if (
+			firstPartialMessageMs === null &&
+			(message.say === "text" || message.say === "reasoning" || message.say === 4 || message.say === 5)
+		) {
+			firstPartialMessageMs = Date.now() - startedAt
+		}
+	})
+
+	deltaStream?.on("data", (event: { deltaJson?: string }) => {
+		taskDeltaCount += 1
+		const deltaJson = event.deltaJson || ""
+		taskDeltaPayloadBytes += Buffer.byteLength(deltaJson, "utf8")
+		if (firstTaskDeltaMs === null) {
+			try {
+				const delta = JSON.parse(deltaJson)
+				if (delta.type?.startsWith("message_") || delta.type === "task_metadata_updated") {
+					firstTaskDeltaMs = Date.now() - startedAt
+				}
+			} catch {
+				// ignore parse errors
+			}
+		}
+	})
+
+	let newTaskRpcMs = 0
+	let error: string | undefined
+
+	try {
+		await unaryCall<{ value?: string }>((callback) => accountClient.accountLoginClicked({}, callback as any))
+		await unaryCall((callback) => accountClient.getUserOrganizations({}, callback as any))
+
+		const rpcStartedAt = Date.now()
+		const newTaskResponse = await unaryCall<{ value?: string }>((callback) =>
+			taskClient.newTask(
+				{
+					metadata: undefined,
+					text: "latency_validation",
+					images: [],
+					files: [],
+					taskSettings: undefined,
+				},
+				callback as any,
+			),
+		)
+		newTaskRpcMs = Date.now() - rpcStartedAt
+		currentTaskId = newTaskResponse.value || currentTaskId
+
+		const timeoutAt = Date.now() + 20_000
+		while (!completed && Date.now() < timeoutAt) {
+			await new Promise((resolve) => setTimeout(resolve, 100))
+		}
+		if (!completed) {
+			error = `Scenario timed out before completion. Last messages: ${lastMessagesSummary}`
+		}
+	} catch (scenarioError) {
+		error = scenarioError instanceof Error ? scenarioError.message : String(scenarioError)
+	}
+
+	stateStream.cancel()
+	partialStream.cancel()
+	deltaStream?.cancel()
+	accountClient.close()
+	stateClient.close()
+	taskClient.close()
+	uiClient.close()
+	await stopServer(server)
+
+	return {
+		variant: variant.name,
+		mode,
+		newTaskRpcMs,
+		firstStateMs,
+		firstPartialMessageMs,
+		firstTaskDeltaMs,
+		completionMs,
+		stateUpdateCount,
+		partialMessageCount,
+		taskDeltaCount,
+		statePayloadBytes,
+		taskDeltaPayloadBytes,
+		messageCountAtCompletion,
+		completed,
+		error,
+	}
+}
+
+async function main() {
+	const results: ScenarioResult[] = []
+	for (const mode of ["local", "remote"] as const) {
+		for (const variant of variants) {
+			results.push(await runScenario(mode, variant))
+		}
+	}
+	console.log(JSON.stringify({ results }, null, 2))
+}
+
+main().catch((error) => {
+	console.error(error)
+	process.exit(1)
+})

src/core/controller/ui/subscribeToPartialMessage.test.ts

@@ -0,0 +1,97 @@
+import { EmptyRequest } from "@shared/proto/cline/common"
+import { ClineAsk, ClineMessage, ClineMessageType, ClineSay } from "@shared/proto/cline/ui"
+import { expect } from "chai"
+import { afterEach, describe, it } from "mocha"
+import * as sinon from "sinon"
+import { resetTelemetryService } from "@/services/telemetry"
+import { setVscodeHostProviderMock } from "@/test/host-provider-test-utils"
+import { getRequestRegistry, StreamingResponseHandler } from "../grpc-handler"
+import { registerPartialMessageCallback, sendPartialMessageEvent, subscribeToPartialMessage } from "./subscribeToPartialMessage"
+
+describe("subscribeToPartialMessage", () => {
+	afterEach(() => {
+		for (const [requestId] of getRequestRegistry().getAllRequests()) {
+			getRequestRegistry().cancelRequest(requestId)
+		}
+		resetTelemetryService()
+		setVscodeHostProviderMock()
+		sinon.restore()
+	})
+
+	function createPartialSayMessage(text: string): ClineMessage {
+		return ClineMessage.create({
+			ts: Date.now(),
+			type: ClineMessageType.SAY,
+			say: ClineSay.TEXT,
+			ask: ClineAsk.FOLLOWUP,
+			text,
+			partial: true,
+		})
+	}
+
+	it("broadcasts partial messages to gRPC subscribers and unregisters on cancel", async () => {
+		setVscodeHostProviderMock()
+		const responseStream = sinon.stub().resolves() as unknown as StreamingResponseHandler<ClineMessage>
+
+		await subscribeToPartialMessage({} as any, EmptyRequest.create({}), responseStream, "partial-req-1")
+		expect(getRequestRegistry().hasRequest("partial-req-1")).to.equal(true)
+
+		const firstMessage = createPartialSayMessage("stream-1")
+		await sendPartialMessageEvent(firstMessage)
+
+		expect((responseStream as any).calledOnce).to.equal(true)
+		expect((responseStream as any).firstCall.args[0]).to.deep.equal(firstMessage)
+		expect((responseStream as any).firstCall.args[1]).to.equal(false)
+
+		getRequestRegistry().cancelRequest("partial-req-1")
+		expect(getRequestRegistry().hasRequest("partial-req-1")).to.equal(false)
+
+		const secondMessage = createPartialSayMessage("stream-2")
+		await sendPartialMessageEvent(secondMessage)
+
+		expect((responseStream as any).calledOnce).to.equal(true)
+	})
+
+	it("broadcasts partial messages to callback subscribers until unsubscribed", async () => {
+		setVscodeHostProviderMock()
+		const received: ClineMessage[] = []
+		const unsubscribe = registerPartialMessageCallback((message: ClineMessage) => {
+			received.push(message)
+		})
+
+		const firstMessage = createPartialSayMessage("callback-1")
+		await sendPartialMessageEvent(firstMessage)
+
+		expect(received).to.deep.equal([firstMessage])
+
+		unsubscribe()
+
+		const secondMessage = createPartialSayMessage("callback-2")
+		await sendPartialMessageEvent(secondMessage)
+
+		expect(received).to.deep.equal([firstMessage])
+	})
+
+	it("removes failing gRPC subscribers without failing the broadcast", async () => {
+		setVscodeHostProviderMock()
+		const failingStream = sinon
+			.stub()
+			.rejects(new Error("stream failed")) as unknown as StreamingResponseHandler<ClineMessage>
+		const healthyStream = sinon.stub().resolves() as unknown as StreamingResponseHandler<ClineMessage>
+
+		await subscribeToPartialMessage({} as any, EmptyRequest.create({}), failingStream, "partial-req-fail")
+		await subscribeToPartialMessage({} as any, EmptyRequest.create({}), healthyStream, "partial-req-ok")
+
+		const firstMessage = createPartialSayMessage("fanout-1")
+		await sendPartialMessageEvent(firstMessage)
+
+		expect((failingStream as any).calledOnce).to.equal(true)
+		expect((healthyStream as any).calledOnce).to.equal(true)
+
+		const secondMessage = createPartialSayMessage("fanout-2")
+		await sendPartialMessageEvent(secondMessage)
+
+		expect((failingStream as any).calledOnce).to.equal(true)
+		expect((healthyStream as any).calledTwice).to.equal(true)
+	})
+})

src/core/controller/ui/subscribeToPartialMessage.ts

@@ -1,5 +1,6 @@
 import { EmptyRequest } from "@shared/proto/cline/common"
 import { ClineMessage } from "@shared/proto/cline/ui"
+import { telemetryService } from "@/services/telemetry"
 import { Logger } from "@/shared/services/Logger"
 import { getRequestRegistry, StreamingResponseHandler } from "../grpc-handler"
 import { Controller } from "../index"
@@ -11,6 +12,13 @@ const activePartialMessageSubscriptions = new Set<StreamingResponseHandler<Cline
 export type PartialMessageCallback = (message: ClineMessage) => void
 const callbackSubscriptions = new Set<PartialMessageCallback>()
 
+export type PartialMessageDeliveryStats = {
+	payloadBytes: number
+	broadcastDurationMs: number
+	streamSubscriberCount: number
+	callbackSubscriberCount: number
+}
+
 /**
  * Subscribe to partial message events
  * @param controller The controller instance
@@ -54,7 +62,11 @@ export function registerPartialMessageCallback(callback: PartialMessageCallback)
  * Send a partial message event to all active subscribers
  * @param partialMessage The ClineMessage to send
  */
-export async function sendPartialMessageEvent(partialMessage: ClineMessage): Promise<void> {
+export async function sendPartialMessageEvent(partialMessage: ClineMessage): Promise<PartialMessageDeliveryStats> {
+	const payloadBytes = Buffer.byteLength(JSON.stringify(partialMessage), "utf8")
+	await telemetryService.captureGrpcResponseSize(payloadBytes, "cline.UiService", "subscribeToPartialMessage")
+	const startedAt = performance.now()
+
 	// Send to gRPC stream subscribers
 	const streamPromises = Array.from(activePartialMessageSubscriptions).map(async (responseStream) => {
 		try {
@@ -79,4 +91,11 @@ export async function sendPartialMessageEvent(partialMessage: ClineMessage): Pro
 	}
 
 	await Promise.all(streamPromises)
+
+	return {
+		payloadBytes,
+		broadcastDurationMs: Math.max(0, performance.now() - startedAt),
+		streamSubscriberCount: activePartialMessageSubscriptions.size,
+		callbackSubscriberCount: callbackSubscriptions.size,
+	}
 }

src/core/task/EphemeralMessageFlushScheduler.ts

@@ -0,0 +1,59 @@
+type EphemeralMessageFlushSchedulerOptions = {
+	flush: () => Promise<void>
+	getDelayMs: () => number
+	setIntervalFn?: typeof setInterval
+	clearIntervalFn?: typeof clearInterval
+	onFlushError?: (error: unknown) => void
+}
+
+export class EphemeralMessageFlushScheduler {
+	private intervalHandle: ReturnType<typeof setInterval> | undefined
+	private flushInFlight = false
+
+	private readonly flush: () => Promise<void>
+	private readonly getDelayMs: () => number
+	private readonly setIntervalFn: typeof setInterval
+	private readonly clearIntervalFn: typeof clearInterval
+	private readonly onFlushError?: (error: unknown) => void
+
+	constructor(options: EphemeralMessageFlushSchedulerOptions) {
+		this.flush = options.flush
+		this.getDelayMs = options.getDelayMs
+		this.setIntervalFn = options.setIntervalFn ?? setInterval
+		this.clearIntervalFn = options.clearIntervalFn ?? clearInterval
+		this.onFlushError = options.onFlushError
+	}
+
+	start(): void {
+		if (this.intervalHandle) {
+			return
+		}
+
+		this.intervalHandle = this.setIntervalFn(() => {
+			if (this.flushInFlight) {
+				return
+			}
+
+			this.flushInFlight = true
+			void this.flush()
+				.catch((error) => this.onFlushError?.(error))
+				.finally(() => {
+					this.flushInFlight = false
+				})
+		}, this.getDelayMs())
+	}
+
+	stop(): void {
+		if (!this.intervalHandle) {
+			return
+		}
+
+		this.clearIntervalFn(this.intervalHandle)
+		this.intervalHandle = undefined
+		this.flushInFlight = false
+	}
+
+	dispose(): void {
+		this.stop()
+	}
+}

src/core/task/__tests__/EphemeralMessageFlushScheduler.test.ts

@@ -0,0 +1,155 @@
+import { strict as assert } from "assert"
+import { MessageStateHandler } from "../../task/message-state"
+import { TaskState } from "../../task/TaskState"
+import { EphemeralMessageFlushScheduler } from "../EphemeralMessageFlushScheduler"
+
+class FakeIntervalController {
+	private now = 0
+	private nextId = 1
+	private intervals = new Map<number, { delay: number; nextRunAt: number; callback: () => void }>()
+
+	setInterval = (callback: () => void, delay: number) => {
+		const id = this.nextId++
+		this.intervals.set(id, { delay, nextRunAt: this.now + delay, callback })
+		return id as unknown as ReturnType<typeof setInterval>
+	}
+
+	clearInterval = (handle: ReturnType<typeof setInterval>) => {
+		this.intervals.delete(handle as unknown as number)
+	}
+
+	advance(ms: number) {
+		this.now += ms
+		let ran = true
+		while (ran) {
+			ran = false
+			for (const [id, interval] of [...this.intervals.entries()].sort((a, b) => a[1].nextRunAt - b[1].nextRunAt)) {
+				if (interval.nextRunAt <= this.now) {
+					interval.callback()
+					interval.nextRunAt += interval.delay
+					this.intervals.set(id, interval)
+					ran = true
+				}
+			}
+		}
+	}
+}
+
+describe("EphemeralMessageFlushScheduler", () => {
+	function createHandler(): MessageStateHandler {
+		return new MessageStateHandler({
+			taskId: "test-task-id",
+			ulid: "test-ulid",
+			taskState: new TaskState(),
+			updateTaskHistory: async () => [],
+		})
+	}
+
+	it("periodically flushes pending ephemeral message changes", async () => {
+		const timer = new FakeIntervalController()
+		const handler = createHandler()
+		const scheduler = new EphemeralMessageFlushScheduler({
+			flush: async () => handler.flushClineMessagesAndUpdateHistory(),
+			getDelayMs: () => 1500,
+			setIntervalFn: timer.setInterval as typeof setInterval,
+			clearIntervalFn: timer.clearInterval as typeof clearInterval,
+		})
+
+		await handler.addToClineMessagesEphemeral({ ts: Date.now(), type: "say", say: "text", text: "streaming", partial: true })
+		assert.equal(handler.consumeLatencyMetrics().persistenceFlushCount, 0)
+
+		scheduler.start()
+		timer.advance(1499)
+		await Promise.resolve()
+		assert.equal(handler.consumeLatencyMetrics().persistenceFlushCount, 0)
+
+		timer.advance(1)
+		await Promise.resolve()
+		await Promise.resolve()
+
+		assert.equal(handler.consumeLatencyMetrics().persistenceFlushCount, 1)
+	})
+
+	it("does nothing when there are no dirty ephemeral changes", async () => {
+		const timer = new FakeIntervalController()
+		const handler = createHandler()
+		const scheduler = new EphemeralMessageFlushScheduler({
+			flush: async () => handler.flushClineMessagesAndUpdateHistory(),
+			getDelayMs: () => 1500,
+			setIntervalFn: timer.setInterval as typeof setInterval,
+			clearIntervalFn: timer.clearInterval as typeof clearInterval,
+		})
+
+		scheduler.start()
+		timer.advance(1500)
+		await Promise.resolve()
+		await Promise.resolve()
+
+		assert.equal(handler.consumeLatencyMetrics().persistenceFlushCount, 0)
+	})
+
+	it("stops scheduling future flushes after stop is called", async () => {
+		const timer = new FakeIntervalController()
+		let flushCount = 0
+		const scheduler = new EphemeralMessageFlushScheduler({
+			flush: async () => {
+				flushCount += 1
+			},
+			getDelayMs: () => 100,
+			setIntervalFn: timer.setInterval as typeof setInterval,
+			clearIntervalFn: timer.clearInterval as typeof clearInterval,
+		})
+
+		scheduler.start()
+		timer.advance(100)
+		await Promise.resolve()
+		assert.equal(flushCount, 1)
+
+		scheduler.stop()
+		timer.advance(500)
+		await Promise.resolve()
+		assert.equal(flushCount, 1)
+	})
+
+	it("does not create overlapping timers when start is called repeatedly", async () => {
+		const timer = new FakeIntervalController()
+		let flushCount = 0
+		const scheduler = new EphemeralMessageFlushScheduler({
+			flush: async () => {
+				flushCount += 1
+			},
+			getDelayMs: () => 100,
+			setIntervalFn: timer.setInterval as typeof setInterval,
+			clearIntervalFn: timer.clearInterval as typeof clearInterval,
+		})
+
+		scheduler.start()
+		scheduler.start()
+		scheduler.start()
+
+		timer.advance(100)
+		await Promise.resolve()
+
+		assert.equal(flushCount, 1)
+	})
+
+	it("dispose stops future flushes", async () => {
+		const timer = new FakeIntervalController()
+		let flushCount = 0
+		const scheduler = new EphemeralMessageFlushScheduler({
+			flush: async () => {
+				flushCount += 1
+			},
+			getDelayMs: () => 100,
+			setIntervalFn: timer.setInterval as typeof setInterval,
+			clearIntervalFn: timer.clearInterval as typeof clearInterval,
+		})
+
+		scheduler.start()
+		scheduler.dispose()
+		timer.advance(500)
+		await Promise.resolve()
+
+		assert.equal(flushCount, 0)
+	})
+})

src/core/task/__tests__/latency.test.ts

@@ -0,0 +1,42 @@
+import { strict as assert } from "assert"
+import { afterEach, describe, it } from "mocha"
+import { getEphemeralMessageFlushCadenceMs, isEphemeralMessagePersistenceDisabled } from "../latency"
+
+describe("ephemeral message latency helpers", () => {
+	afterEach(() => {
+		delete process.env.CLINE_DISABLE_EPHEMERAL_MESSAGE_PERSISTENCE
+		delete process.env.CLINE_EPHEMERAL_MESSAGE_FLUSH_CADENCE_MS
+	})
+
+	it("defaults ephemeral persistence to enabled", () => {
+		assert.equal(isEphemeralMessagePersistenceDisabled(), false)
+	})
+
+	it("supports disabling ephemeral message persistence via env flag", () => {
+		process.env.CLINE_DISABLE_EPHEMERAL_MESSAGE_PERSISTENCE = "1"
+		assert.equal(isEphemeralMessagePersistenceDisabled(), true)
+
+		process.env.CLINE_DISABLE_EPHEMERAL_MESSAGE_PERSISTENCE = "true"
+		assert.equal(isEphemeralMessagePersistenceDisabled(), true)
+
+		process.env.CLINE_DISABLE_EPHEMERAL_MESSAGE_PERSISTENCE = "yes"
+		assert.equal(isEphemeralMessagePersistenceDisabled(), true)
+	})
+
+	it("uses a conservative default flush cadence", () => {
+		assert.equal(getEphemeralMessageFlushCadenceMs(), 1500)
+	})
+
+	it("supports overriding the flush cadence via env flag", () => {
+		process.env.CLINE_EPHEMERAL_MESSAGE_FLUSH_CADENCE_MS = "2000"
+		assert.equal(getEphemeralMessageFlushCadenceMs(), 2000)
+	})
+
+	it("falls back to the default cadence when the override is invalid", () => {
+		process.env.CLINE_EPHEMERAL_MESSAGE_FLUSH_CADENCE_MS = "invalid"
+		assert.equal(getEphemeralMessageFlushCadenceMs(), 1500)
+
+		process.env.CLINE_EPHEMERAL_MESSAGE_FLUSH_CADENCE_MS = "-1"
+		assert.equal(getEphemeralMessageFlushCadenceMs(), 1500)
+	})
+})

src/core/task/index.ts

@@ -113,7 +113,9 @@ import { refreshWorkflowToggles } from "../context/instructions/user-instruction
 import { Controller } from "../controller"
 import { executeHook } from "../hooks/hook-executor"
 import { StateManager } from "../storage/StateManager"
+import { EphemeralMessageFlushScheduler } from "./EphemeralMessageFlushScheduler"
 import { FocusChainManager } from "./focus-chain"
+import { getEphemeralMessageFlushCadenceMs, isEphemeralMessagePersistenceDisabled } from "./latency"
 import { MessageStateHandler } from "./message-state"
 import { StreamChunkCoordinator } from "./StreamChunkCoordinator"
 import { StreamResponseHandler } from "./StreamResponseHandler"
@@ -256,6 +258,8 @@ export class Task {
 
 	// Command executor for running shell commands (extracted from executeCommandTool)
 	private commandExecutor!: CommandExecutor
+	private readonly ephemeralMessageFlushScheduler: EphemeralMessageFlushScheduler
+	private readonly ephemeralMessagePersistenceDisabled = isEphemeralMessagePersistenceDisabled()
 
 	constructor(params: TaskParams) {
 		const {
@@ -567,6 +571,12 @@ export class Task {
 			this.getActiveHookExecution.bind(this),
 			this.runUserPromptSubmitHook.bind(this),
 		)
+
+		this.ephemeralMessageFlushScheduler = new EphemeralMessageFlushScheduler({
+			flush: async () => this.messageStateHandler.flushClineMessagesAndUpdateHistory(),
+			getDelayMs: () => getEphemeralMessageFlushCadenceMs(),
+			onFlushError: (error) => Logger.debug(`[Task ${this.taskId}] Failed to flush ephemeral message state: ${error}`),
+		})
 	}
 
 	// Communicate with webview
@@ -583,6 +593,8 @@ export class Task {
 		files?: string[]
 		askTs?: number
 	}> {
+		// Partial ask updates are ephemeral animation state until they complete or another
+		// durable boundary is reached (request completion, abort/cancel, resume, checkpoint).
 		// Allow resume asks even when aborted to enable resume button after cancellation
 		if (this.taskState.abort && type !== "resume_task" && type !== "resume_completed_task") {
 			throw new Error("Cline instance aborted")
@@ -598,10 +610,15 @@ export class Task {
 			if (partial) {
 				if (isUpdatingPreviousPartial) {
 					// existing partial message, so update it
-					await this.messageStateHandler.updateClineMessage(lastMessageIndex, {
-						text,
-						partial,
-					})
+					await (this.ephemeralMessagePersistenceDisabled
+						? this.messageStateHandler.updateClineMessage(lastMessageIndex, {
+								text,
+								partial,
+							})
+						: this.messageStateHandler.updateClineMessageEphemeral(lastMessageIndex, {
+								text,
+								partial,
+							}))
 					// todo be more efficient about saving and posting only new data or one whole message at a time so ignore partial for saves, and only post parts of partial message instead of whole array in new listener
 					// await this.saveClineMessagesAndUpdateHistory()
 					// await this.postStateToWebview()
@@ -615,13 +632,21 @@ export class Task {
 				// this.askResponseImages = undefined
 				askTs = Date.now()
 				this.taskState.lastMessageTs = askTs
-				await this.messageStateHandler.addToClineMessages({
-					ts: askTs,
-					type: "ask",
-					ask: type,
-					text,
-					partial,
-				})
+				await (this.ephemeralMessagePersistenceDisabled
+					? this.messageStateHandler.addToClineMessages({
+							ts: askTs,
+							type: "ask",
+							ask: type,
+							text,
+							partial,
+						})
+					: this.messageStateHandler.addToClineMessagesEphemeral({
+							ts: askTs,
+							type: "ask",
+							ask: type,
+							text,
+							partial,
+						}))
 				await this.postStateToWebview()
 				throw new Error("Current ask promise was ignored 2")
 			}
@@ -759,6 +784,10 @@ export class Task {
 		files?: string[],
 		partial?: boolean,
 	): Promise<number | undefined> {
+		// Partial say updates are intentionally ephemeral: they keep streaming UI live
+		// without forcing disk persistence on every chunk. Durable boundaries include
+		// partial→complete transitions, tool completion, request completion, cancel/abort,
+		// resume-visible state changes, and checkpoint-relevant events.
 		// Allow hook messages even when aborted to enable proper cleanup
 		if (this.taskState.abort && type !== "hook_status" && type !== "hook_output_stream") {
 			throw new Error("Cline instance aborted")
@@ -779,12 +808,19 @@ export class Task {
 				if (isUpdatingPreviousPartial) {
 					// existing partial message, so update it
 					const lastIndex = this.messageStateHandler.getClineMessages().length - 1
-					await this.messageStateHandler.updateClineMessage(lastIndex, {
-						text,
-						images,
-						files,
-						partial,
-					})
+					await (this.ephemeralMessagePersistenceDisabled
+						? this.messageStateHandler.updateClineMessage(lastIndex, {
+								text,
+								images,
+								files,
+								partial,
+							})
+						: this.messageStateHandler.updateClineMessageEphemeral(lastIndex, {
+								text,
+								images,
+								files,
+								partial,
+							}))
 
 					const protoMessage = convertClineMessageToProto(lastMessage)
 					await sendPartialMessageEvent(protoMessage)
@@ -793,16 +829,27 @@ export class Task {
 				// this is a new partial message, so add it with partial state
 				const sayTs = Date.now()
 				this.taskState.lastMessageTs = sayTs
-				await this.messageStateHandler.addToClineMessages({
-					ts: sayTs,
-					type: "say",
-					say: type,
-					text,
-					images,
-					files,
-					partial,
-					modelInfo,
-				})
+				await (this.ephemeralMessagePersistenceDisabled
+					? this.messageStateHandler.addToClineMessages({
+							ts: sayTs,
+							type: "say",
+							say: type,
+							text,
+							images,
+							files,
+							partial,
+							modelInfo,
+						})
+					: this.messageStateHandler.addToClineMessagesEphemeral({
+							ts: sayTs,
+							type: "say",
+							say: type,
+							text,
+							images,
+							files,
+							partial,
+							modelInfo,
+						}))
 				await this.postStateToWebview()
 				return sayTs
 			}
@@ -907,6 +954,7 @@ export class Task {
 		this.taskState.didFinishAbortingStream = true
 
 		// Save conversation state to disk
+		await this.messageStateHandler.flushClineMessagesAndUpdateHistory()
 		await this.messageStateHandler.saveClineMessagesAndUpdateHistory()
 		await this.messageStateHandler.overwriteApiConversationHistory(this.messageStateHandler.getApiConversationHistory())
 
@@ -1462,6 +1510,7 @@ export class Task {
 
 	async abortTask() {
 		try {
+			this.ephemeralMessageFlushScheduler.stop()
 			// PHASE 1: Check if TaskCancel should run BEFORE any cleanup
 			// We must capture this state now because subsequent cleanup will
 			// clear the active work indicators that shouldRunTaskCancelHook checks
@@ -1550,6 +1599,7 @@ export class Task {
 
 			// PHASE 5: Immediately update UI to reflect abort state
 			try {
+				await this.messageStateHandler.flushClineMessagesAndUpdateHistory()
 				await this.messageStateHandler.saveClineMessagesAndUpdateHistory()
 				await this.postStateToWebview()
 			} catch (error) {
@@ -2558,6 +2608,9 @@ export class Task {
 		await this.postStateToWebview()
 
 		try {
+			if (!this.ephemeralMessagePersistenceDisabled) {
+				this.ephemeralMessageFlushScheduler.start()
+			}
 			const taskMetrics: {
 				cacheWriteTokens: number
 				cacheReadTokens: number
@@ -2644,6 +2697,7 @@ export class Task {
 				}
 				// update api_req_started to have cancelled and cost, so that we can display the cost of the partial stream
 				await finalizeApiReqMsg(cancelReason, streamingFailedMessage)
+				await this.messageStateHandler.flushClineMessagesAndUpdateHistory()
 				await this.messageStateHandler.saveClineMessagesAndUpdateHistory()
 
 				// Let assistant know their response was interrupted for when task is resumed
@@ -2977,6 +3031,7 @@ export class Task {
 
 			// Update the api_req_started message with final usage and cost details
 			await finalizeApiReqMsg()
+			await this.messageStateHandler.flushClineMessagesAndUpdateHistory()
 			await this.messageStateHandler.saveClineMessagesAndUpdateHistory()
 			await this.postStateToWebview()
 
@@ -3209,6 +3264,8 @@ export class Task {
 		} catch (_error) {
 			// this should never happen since the only thing that can throw an error is the attemptApiRequest, which is wrapped in a try catch that sends an ask where if noButtonClicked, will clear current task and destroy this instance. However to avoid unhandled promise rejection, we will end this loop which will end execution of this instance (see startTask)
 			return true // needs to be true so parent loop knows to end task
+		} finally {
+			this.ephemeralMessageFlushScheduler.stop()
 		}
 	}
 

src/core/task/latency.ts

@@ -0,0 +1,22 @@
+function readBooleanEnv(envVarName: string): boolean {
+	const rawValue = process.env[envVarName]?.toLowerCase()
+	return rawValue === "1" || rawValue === "true" || rawValue === "yes"
+}
+
+export function isEphemeralMessagePersistenceDisabled(): boolean {
+	return readBooleanEnv("CLINE_DISABLE_EPHEMERAL_MESSAGE_PERSISTENCE")
+}
+
+export function getEphemeralMessageFlushCadenceMs(): number {
+	const rawValue = process.env.CLINE_EPHEMERAL_MESSAGE_FLUSH_CADENCE_MS
+	if (!rawValue) {
+		return 1500
+	}
+
+	const parsed = Number.parseInt(rawValue, 10)
+	if (!Number.isFinite(parsed) || parsed < 0) {
+		return 1500
+	}
+
+	return parsed
+}

src/core/task/message-state.ts

@@ -48,12 +48,25 @@ interface MessageStateHandlerParams {
 export class MessageStateHandler extends EventEmitter<MessageStateHandlerEvents> {
 	private apiConversationHistory: ClineStorageMessage[] = []
 	private clineMessages: ClineMessage[] = []
+	/**
+	 * Partial streaming updates are ephemeral UI state until we cross a durable boundary.
+	 * Durable mutation methods persist immediately; ephemeral mutation methods only update
+	 * in-memory state and emit change notifications. flushClineMessagesAndUpdateHistory()
+	 * is the bridge that commits previously-ephemeral message mutations for recovery/history.
+	 */
+	private hasDirtyEphemeralChanges = false
 	private taskIsFavorited: boolean
 	private checkpointTracker: CheckpointTracker | undefined
 	private updateTaskHistory: (historyItem: HistoryItem) => Promise<HistoryItem[]>
 	private taskId: string
 	private ulid: string
 	private taskState: TaskState
+	private readonly latencyMetrics = {
+		persistenceFlushCount: 0,
+		saveMessagesDurationMs: 0,
+		saveConversationDurationMs: 0,
+		updateHistoryDurationMs: 0,
+	}
 
 	// Mutex to prevent concurrent state modifications (RC-4)
 	// Protects against data loss from race conditions when multiple
@@ -105,6 +118,7 @@ export class MessageStateHandler extends EventEmitter<MessageStateHandlerEvents>
 	setClineMessages(newMessages: ClineMessage[]) {
 		const previousMessages = this.clineMessages
 		this.clineMessages = newMessages
+		this.hasDirtyEphemeralChanges = true
 		this.emitClineMessagesChanged({
 			type: "set",
 			messages: this.clineMessages,
@@ -119,7 +133,10 @@ export class MessageStateHandler extends EventEmitter<MessageStateHandlerEvents>
 	 */
 	private async saveClineMessagesAndUpdateHistoryInternal(): Promise<void> {
 		try {
+			this.latencyMetrics.persistenceFlushCount += 1
+			const saveMessagesStartedAt = performance.now()
 			await saveClineMessages(this.taskId, this.clineMessages)
+			this.latencyMetrics.saveMessagesDurationMs += Math.max(0, performance.now() - saveMessagesStartedAt)
 
 			// combined as they are in ChatView
 			const apiMetrics = getApiMetrics(combineApiRequests(combineCommandSequences(this.clineMessages.slice(1))))
@@ -142,6 +159,7 @@ export class MessageStateHandler extends EventEmitter<MessageStateHandlerEvents>
 				Logger.error("Failed to get task directory size:", taskDir, error)
 			}
 			const cwd = await getCwd(getDesktopDir())
+			const updateHistoryStartedAt = performance.now()
 			await this.updateTaskHistory({
 				id: this.taskId,
 				ulid: this.ulid,
@@ -160,6 +178,8 @@ export class MessageStateHandler extends EventEmitter<MessageStateHandlerEvents>
 				checkpointManagerErrorMessage: this.taskState.checkpointManagerErrorMessage,
 				modelId: lastModelInfo?.modelInfo?.modelId,
 			})
+			this.latencyMetrics.updateHistoryDurationMs += Math.max(0, performance.now() - updateHistoryStartedAt)
+			this.hasDirtyEphemeralChanges = false
 		} catch (error) {
 			Logger.error("Failed to save cline messages:", error)
 		}
@@ -179,7 +199,9 @@ export class MessageStateHandler extends EventEmitter<MessageStateHandlerEvents>
 		// Protect with mutex to prevent concurrent modifications from corrupting data (RC-4)
 		return await this.withStateLock(async () => {
 			this.apiConversationHistory.push(message)
+			const saveConversationStartedAt = performance.now()
 			await saveApiConversationHistory(this.taskId, this.apiConversationHistory)
+			this.latencyMetrics.saveConversationDurationMs += Math.max(0, performance.now() - saveConversationStartedAt)
 		})
 	}
 
@@ -187,7 +209,25 @@ export class MessageStateHandler extends EventEmitter<MessageStateHandlerEvents>
 		// Protect with mutex to prevent concurrent modifications from corrupting data (RC-4)
 		return await this.withStateLock(async () => {
 			this.apiConversationHistory = newHistory
+			const saveConversationStartedAt = performance.now()
 			await saveApiConversationHistory(this.taskId, this.apiConversationHistory)
+			this.latencyMetrics.saveConversationDurationMs += Math.max(0, performance.now() - saveConversationStartedAt)
+		})
+	}
+
+	async addToClineMessagesEphemeral(message: ClineMessage) {
+		return await this.withStateLock(async () => {
+			message.conversationHistoryIndex = this.apiConversationHistory.length - 1
+			message.conversationHistoryDeletedRange = this.taskState.conversationHistoryDeletedRange
+			const index = this.clineMessages.length
+			this.clineMessages.push(message)
+			this.hasDirtyEphemeralChanges = true
+			this.emitClineMessagesChanged({
+				type: "add",
+				messages: this.clineMessages,
+				index,
+				message,
+			})
 		})
 	}
 
@@ -223,6 +263,7 @@ export class MessageStateHandler extends EventEmitter<MessageStateHandlerEvents>
 		return await this.withStateLock(async () => {
 			const previousMessages = this.clineMessages
 			this.clineMessages = newMessages
+			this.hasDirtyEphemeralChanges = true
 			this.emitClineMessagesChanged({
 				type: "set",
 				messages: this.clineMessages,
@@ -261,6 +302,26 @@ export class MessageStateHandler extends EventEmitter<MessageStateHandlerEvents>
 		})
 	}
 
+	async updateClineMessageEphemeral(index: number, updates: Partial<ClineMessage>): Promise<void> {
+		return await this.withStateLock(async () => {
+			if (index < 0 || index >= this.clineMessages.length) {
+				throw new Error(`Invalid message index: ${index}`)
+			}
+
+			const previousMessage = { ...this.clineMessages[index] }
+			Object.assign(this.clineMessages[index], updates)
+			this.hasDirtyEphemeralChanges = true
+
+			this.emitClineMessagesChanged({
+				type: "update",
+				messages: this.clineMessages,
+				index,
+				previousMessage,
+				message: this.clineMessages[index],
+			})
+		})
+	}
+
 	/**
 	 * Delete a specific message from the clineMessages array
 	 * The entire operation (validate, delete, save) is atomic to prevent races (RC-4)
@@ -288,4 +349,22 @@ export class MessageStateHandler extends EventEmitter<MessageStateHandlerEvents>
 			await this.saveClineMessagesAndUpdateHistoryInternal()
 		})
 	}
+
+	async flushClineMessagesAndUpdateHistory(): Promise<void> {
+		return await this.withStateLock(async () => {
+			if (!this.hasDirtyEphemeralChanges) {
+				return
+			}
+			await this.saveClineMessagesAndUpdateHistoryInternal()
+		})
+	}
+
+	consumeLatencyMetrics() {
+		const snapshot = { ...this.latencyMetrics }
+		this.latencyMetrics.persistenceFlushCount = 0
+		this.latencyMetrics.saveMessagesDurationMs = 0
+		this.latencyMetrics.saveConversationDurationMs = 0
+		this.latencyMetrics.updateHistoryDurationMs = 0
+		return snapshot
+	}
 }

src/services/telemetry/TelemetryService.ts

@@ -1643,6 +1643,7 @@ export class TelemetryService {
 				taskId,
 				durationMs,
 				hasCheckpoints,
+				ephemeralMessagePersistenceEnabled: !process.env.CLINE_DISABLE_EPHEMERAL_MESSAGE_PERSISTENCE,
 			},
 		})
 	}

src/services/telemetry/__tests__/TelemetryService.metrics.test.ts

@@ -77,6 +77,10 @@ function createTelemetryService(provider: FakeProvider): TelemetryService {
 }
 
 describe("TelemetryService metrics", () => {
+	afterEach(() => {
+		delete process.env.CLINE_DISABLE_EPHEMERAL_MESSAGE_PERSISTENCE
+	})
+
 	it("captureTokenUsage emits token counters and histograms", () => {
 		const provider = new FakeProvider()
 		const service = createTelemetryService(provider)
@@ -332,6 +336,22 @@ describe("TelemetryService metrics", () => {
 		assert.strictEqual(durationMetric?.attributes.scope, "task")
 	})
 
+	it("captureTaskInitialization includes ephemeral persistence mode for A/B comparison", () => {
+		const provider = new FakeProvider()
+		const service = createTelemetryService(provider)
+
+		delete process.env.CLINE_DISABLE_EPHEMERAL_MESSAGE_PERSISTENCE
+		service.captureTaskInitialization("task-5", "task-id-5", 1234, true)
+
+		process.env.CLINE_DISABLE_EPHEMERAL_MESSAGE_PERSISTENCE = "1"
+		service.captureTaskInitialization("task-6", "task-id-6", 5678, false)
+
+		const initEvents = provider.logs.filter((entry) => entry.event === "task.initialization")
+		assert.strictEqual(initEvents.length, 2)
+		assert.strictEqual(initEvents[0]?.properties?.ephemeralMessagePersistenceEnabled, true)
+		assert.strictEqual(initEvents[1]?.properties?.ephemeralMessagePersistenceEnabled, false)
+	})
+
 	it("captureGrpcResponseSize records histogram with correct name, value, and attributes", () => {
 		const provider = new FakeProvider()
 		const service = createTelemetryService(provider)

src/test/e2e/fixtures/server/api.ts

@@ -53,7 +53,7 @@ The user wants me to replace the name "john" with "cline" in the test.ts file. I
 export const name = "john"
 \`\`\`
 
-I need to change "john" to "cline". This is a simple targeted edit, so I should use the replace_in_file tool rather than write_to_file since I\'m only changing one small part of the file.
+I need to change "john" to "cline". This is a simple targeted edit, so I should use the replace_in_file tool rather than write_to_file since I'm only changing one small part of the file.
 
 I need to:
 1. Use replace_in_file to change "john" to "cline" in the test.ts file
@@ -61,7 +61,7 @@ I need to:
 3. The REPLACE block should be: \`export const name = "cline"\`
 </thinking>
 
-I\'ll replace "john" with "cline" in the test.ts file.
+I'll replace "john" with "cline" in the test.ts file.
 
 <replace_in_file>
 <path>test.ts</path>
@@ -74,8 +74,17 @@ export const name = "cline"
 </diff>
 </replace_in_file>`
 
+const latency_validation = `Streaming validation in progress. This response is intentionally verbose enough to exercise partial message delivery while still completing successfully.
+
+<attempt_completion>
+<result>
+Latency validation scenario completed successfully.
+</result>
+</attempt_completion>`
+
 export const E2E_MOCK_API_RESPONSES = {
 	DEFAULT: "Hello! I'm a mock Cline API response.",
 	REPLACE_REQUEST: replace_in_file,
 	EDIT_REQUEST: edit_request,
+	LATENCY_VALIDATION: latency_validation,
 }

src/test/e2e/fixtures/server/index.ts

@@ -377,6 +377,9 @@ export class ClineApiServerMock {
 						const parsed = JSON.parse(body)
 						const { _messages, model = "claude-3-5-sonnet-20241022", stream = true } = parsed
 						let responseText = E2E_MOCK_API_RESPONSES.DEFAULT
+						if (body.includes("latency_validation")) {
+							responseText = E2E_MOCK_API_RESPONSES.LATENCY_VALIDATION
+						}
 						if (body.includes("[replace_in_file for 'test.ts'] Result:")) {
 							responseText = E2E_MOCK_API_RESPONSES.REPLACE_REQUEST
 						}
@@ -454,31 +457,30 @@ export class ClineApiServerMock {
 
 							sendChunk()
 							return
-						} else {
-							const response = {
-								id: generationId,
-								object: "chat.completion",
-								created: Math.floor(Date.now() / 1000),
-								model,
-								choices: [
-									{
-										index: 0,
-										message: {
-											role: "assistant",
-											content: "Hello! I'm a mock Cline API response.",
-										},
-										finish_reason: "stop",
-									},
-								],
-								usage: {
-									prompt_tokens: 140,
-									completion_tokens: responseText.length,
-									total_tokens: 140 + responseText.length,
-									cost: (140 + responseText.length) * 0.00015,
-								},
-							}
-							return sendJson(response)
 						}
+						const response = {
+							id: generationId,
+							object: "chat.completion",
+							created: Math.floor(Date.now() / 1000),
+							model,
+							choices: [
+								{
+									index: 0,
+									message: {
+										role: "assistant",
+										content: "Hello! I'm a mock Cline API response.",
+									},
+									finish_reason: "stop",
+								},
+							],
+							usage: {
+								prompt_tokens: 140,
+								completion_tokens: responseText.length,
+								total_tokens: 140 + responseText.length,
+								cost: (140 + responseText.length) * 0.00015,
+							},
+						}
+						return sendJson(response)
 					}
 
 					// Generation details endpoint

src/test/message-state-handler.test.ts

@@ -1,9 +1,14 @@
-import { describe, it } from "mocha"
+import fs from "fs/promises"
+import { afterEach, describe, it } from "mocha"
+import os from "os"
+import path from "path"
 import "should"
 import should from "should"
+import { getSavedApiConversationHistory, getSavedClineMessages } from "../core/storage/disk"
 import { MessageStateHandler } from "../core/task/message-state"
 import { TaskState } from "../core/task/TaskState"
 import { ClineMessage } from "../shared/ExtensionMessage"
+import { setVscodeHostProviderMock } from "./host-provider-test-utils"
 
 /**
  * Unit tests for MessageStateHandler's mutex protection (RC-4)
@@ -11,6 +16,16 @@ import { ClineMessage } from "../shared/ExtensionMessage"
  * to prevent race conditions, particularly the TOCTOU bug in addToClineMessages
  */
 describe("MessageStateHandler Mutex Protection", () => {
+	let tempGlobalStorageDir: string | undefined
+
+	afterEach(async () => {
+		if (tempGlobalStorageDir) {
+			await fs.rm(tempGlobalStorageDir, { recursive: true, force: true })
+			tempGlobalStorageDir = undefined
+		}
+		setVscodeHostProviderMock()
+	})
+
 	/**
 	 * Helper to create a minimal MessageStateHandler for testing
 	 */
@@ -24,6 +39,16 @@ describe("MessageStateHandler Mutex Protection", () => {
 		})
 	}
 
+	function createTestHandlerWithHistorySpy(updateTaskHistory: (historyItem: any) => Promise<any[]>): MessageStateHandler {
+		const taskState = new TaskState()
+		return new MessageStateHandler({
+			taskId: "test-task-id",
+			ulid: "test-ulid",
+			taskState,
+			updateTaskHistory,
+		})
+	}
+
 	/**
 	 * Helper to create a test ClineMessage
 	 */
@@ -184,6 +209,20 @@ describe("MessageStateHandler Mutex Protection", () => {
 		}
 	})
 
+	it("should throw error for invalid message index in updateClineMessageEphemeral", async () => {
+		const handler = createTestHandler()
+		handler.setClineMessages([createTestMessage("msg1")])
+
+		try {
+			await handler.updateClineMessageEphemeral(5, { text: "invalid" })
+			throw new Error("Should have thrown")
+		} catch (error) {
+			if (error instanceof Error) {
+				error.message.should.match(/Invalid message index/)
+			}
+		}
+	})
+
 	/**
 	 * Test that invalid indices are rejected in deleteClineMessage
 	 */
@@ -264,4 +303,246 @@ describe("MessageStateHandler Mutex Protection", () => {
 		finalHistory[0].content.should.equal("new1")
 		finalHistory[1].content.should.equal("new2")
 	})
+
+	it("should update messages ephemerally without persisting until flush", async () => {
+		const handler = createTestHandler()
+		const changes: Array<{ type: string; text?: string; previousText?: string }> = []
+
+		handler.on("clineMessagesChanged", (change) => {
+			changes.push({
+				type: change.type,
+				text: change.message?.text,
+				previousText: change.previousMessage?.text,
+			})
+		})
+
+		await handler.addToClineMessagesEphemeral(createTestMessage("ephemeral-start"))
+		await handler.updateClineMessageEphemeral(0, { text: "ephemeral-updated", partial: true })
+
+		const messagesBeforeFlush = handler.getClineMessages()
+		messagesBeforeFlush.length.should.equal(1)
+		should.exist(messagesBeforeFlush[0])
+		const pendingMessage = messagesBeforeFlush[0]!
+		pendingMessage.text?.should.equal("ephemeral-updated")
+		should.exist(pendingMessage.partial)
+		pendingMessage.partial!.should.equal(true)
+
+		changes.length.should.equal(2)
+		changes[0]!.type.should.equal("add")
+		changes[0]!.text!.should.equal("ephemeral-start")
+		changes[1]!.type.should.equal("update")
+		changes[1]!.previousText!.should.equal("ephemeral-start")
+		changes[1]!.text!.should.equal("ephemeral-updated")
+
+		handler.consumeLatencyMetrics().persistenceFlushCount.should.equal(0)
+
+		await handler.flushClineMessagesAndUpdateHistory()
+
+		handler.consumeLatencyMetrics().persistenceFlushCount.should.equal(1)
+	})
+
+	it("should not flush when there are no dirty ephemeral changes", async () => {
+		const handler = createTestHandler()
+
+		await handler.flushClineMessagesAndUpdateHistory()
+
+		const metrics = handler.consumeLatencyMetrics()
+		metrics.persistenceFlushCount.should.equal(0)
+		metrics.saveMessagesDurationMs.should.equal(0)
+		metrics.updateHistoryDurationMs.should.equal(0)
+	})
+
+	it("updates task history when flushing previously-ephemeral changes", async () => {
+		tempGlobalStorageDir = await fs.mkdtemp(path.join(os.tmpdir(), "cline-message-history-spy-"))
+		setVscodeHostProviderMock({ globalStorageFsPath: tempGlobalStorageDir })
+
+		let latestHistoryItem: any
+		const handler = createTestHandlerWithHistorySpy(async (historyItem) => {
+			latestHistoryItem = historyItem
+			return [historyItem]
+		})
+
+		await handler.addToClineMessagesEphemeral({
+			...createTestMessage("history-visible partial"),
+			partial: true,
+		})
+
+		await handler.flushClineMessagesAndUpdateHistory()
+
+		should.exist(latestHistoryItem)
+		latestHistoryItem.id.should.equal("test-task-id")
+		latestHistoryItem.ulid.should.equal("test-ulid")
+		latestHistoryItem.task.should.equal("history-visible partial")
+		latestHistoryItem.ts.should.be.a.Number()
+		handler.consumeLatencyMetrics().persistenceFlushCount.should.equal(1)
+	})
+
+	it("should persist when a partial message transitions to complete", async () => {
+		const handler = createTestHandler()
+
+		await handler.addToClineMessagesEphemeral({
+			...createTestMessage("partial-message"),
+			partial: true,
+		})
+
+		handler.consumeLatencyMetrics().persistenceFlushCount.should.equal(0)
+
+		await handler.updateClineMessage(0, { text: "completed-message", partial: false })
+
+		const completedMessage = handler.getClineMessages()[0]!
+		completedMessage.text?.should.equal("completed-message")
+		completedMessage.partial!.should.equal(false)
+
+		handler.consumeLatencyMetrics().persistenceFlushCount.should.equal(1)
+	})
+
+	it("batches long runs of partial updates into a single durable flush", async () => {
+		const handler = createTestHandler()
+
+		await handler.addToClineMessagesEphemeral({
+			...createTestMessage("chunk-0"),
+			partial: true,
+		})
+
+		for (let i = 1; i <= 25; i++) {
+			await handler.updateClineMessageEphemeral(0, {
+				text: `chunk-${i}`,
+				partial: true,
+			})
+		}
+
+		handler.consumeLatencyMetrics().persistenceFlushCount.should.equal(0)
+
+		await handler.flushClineMessagesAndUpdateHistory()
+
+		const metrics = handler.consumeLatencyMetrics()
+		metrics.persistenceFlushCount.should.equal(1)
+		const finalMessage = handler.getClineMessages()[0]
+		should.exist(finalMessage)
+		finalMessage!.text?.should.equal("chunk-25")
+	})
+
+	it("persists flushed ephemeral messages to disk for task recovery", async () => {
+		tempGlobalStorageDir = await fs.mkdtemp(path.join(os.tmpdir(), "cline-message-state-"))
+		setVscodeHostProviderMock({ globalStorageFsPath: tempGlobalStorageDir })
+
+		const handler = createTestHandler()
+		await handler.addToClineMessagesEphemeral({
+			...createTestMessage("recoverable partial"),
+			partial: true,
+		})
+		await handler.flushClineMessagesAndUpdateHistory()
+
+		const savedMessages = await getSavedClineMessages("test-task-id")
+		savedMessages.length.should.equal(1)
+		savedMessages[0]!.text?.should.equal("recoverable partial")
+		savedMessages[0]!.partial!.should.equal(true)
+	})
+
+	it("persists completed conversation history snapshots that resume can reload", async () => {
+		tempGlobalStorageDir = await fs.mkdtemp(path.join(os.tmpdir(), "cline-message-history-"))
+		setVscodeHostProviderMock({ globalStorageFsPath: tempGlobalStorageDir })
+
+		const handler = createTestHandler()
+		await handler.overwriteApiConversationHistory([
+			{ role: "user", content: "task request", ts: 1 },
+			{ role: "assistant", content: "task response", ts: 2 },
+		])
+
+		const savedHistory = await getSavedApiConversationHistory("test-task-id")
+		savedHistory.length.should.equal(2)
+		savedHistory[0]!.content.should.equal("task request")
+		savedHistory[1]!.content.should.equal("task response")
+	})
+
+	it("persists tool result conversation history after finalization", async () => {
+		tempGlobalStorageDir = await fs.mkdtemp(path.join(os.tmpdir(), "cline-message-tool-result-"))
+		setVscodeHostProviderMock({ globalStorageFsPath: tempGlobalStorageDir })
+
+		const handler = createTestHandler()
+		await handler.overwriteApiConversationHistory([
+			{
+				role: "assistant",
+				content: [
+					{ type: "text", text: "I will inspect the file." },
+					{ type: "tool_use", id: "toolu_123", name: "read_file", input: { path: "src/test.ts" } },
+				],
+				ts: 1,
+			},
+			{
+				role: "user",
+				content: [
+					{
+						type: "tool_result",
+						tool_use_id: "toolu_123",
+						content: [{ type: "text", text: "export const value = 1" }],
+					},
+				],
+				ts: 2,
+			},
+		])
+
+		const savedHistory = await getSavedApiConversationHistory("test-task-id")
+		savedHistory.length.should.equal(2)
+		const savedToolResultMessage = savedHistory[1]
+		should.exist(savedToolResultMessage)
+		Array.isArray(savedToolResultMessage!.content).should.equal(true)
+		const toolResultBlocks = savedToolResultMessage!.content as Array<{
+			type: string
+			tool_use_id?: string
+			content?: Array<{ type: string; text?: string }>
+		}>
+		const firstToolResultBlock = toolResultBlocks[0]
+		should.exist(firstToolResultBlock)
+		if (!firstToolResultBlock) {
+			throw new Error("Expected persisted tool result block")
+		}
+		firstToolResultBlock.type.should.equal("tool_result")
+		const firstToolUseId = firstToolResultBlock.tool_use_id
+		should.exist(firstToolUseId)
+		firstToolUseId!.should.equal("toolu_123")
+		should.exist(firstToolResultBlock.content)
+		const firstToolResultContentBlocks = firstToolResultBlock.content!
+		const firstToolResultContent = firstToolResultContentBlocks[0]
+		should.exist(firstToolResultContent)
+		const firstToolResultText = firstToolResultContent!.text
+		should.exist(firstToolResultText)
+		if (!firstToolResultText) {
+			throw new Error("Expected persisted tool result text")
+		}
+		firstToolResultText.should.equal("export const value = 1")
+	})
+
+	it("persists recoverable message and conversation state for resume after an interrupted stream", async () => {
+		tempGlobalStorageDir = await fs.mkdtemp(path.join(os.tmpdir(), "cline-message-resume-"))
+		setVscodeHostProviderMock({ globalStorageFsPath: tempGlobalStorageDir })
+
+		const handler = createTestHandler()
+		await handler.addToClineMessagesEphemeral({
+			...createTestMessage("partial assistant output"),
+			partial: true,
+		})
+		await handler.flushClineMessagesAndUpdateHistory()
+
+		await handler.overwriteApiConversationHistory([
+			{ role: "user", content: "task request", ts: 1 },
+			{
+				role: "assistant",
+				content: [{ type: "text", text: "partial assistant output\n\n[Response interrupted by user]" }],
+				ts: 2,
+			},
+		])
+
+		const savedMessages = await getSavedClineMessages("test-task-id")
+		const savedHistory = await getSavedApiConversationHistory("test-task-id")
+
+		savedMessages.length.should.equal(1)
+		savedMessages[0]!.text?.should.equal("partial assistant output")
+		savedMessages[0]!.partial!.should.equal(true)
+
+		savedHistory.length.should.equal(2)
+		Array.isArray(savedHistory[1]!.content).should.equal(true)
+		const savedAssistantContent = savedHistory[1]!.content as Array<{ type: string; text?: string }>
+		savedAssistantContent[0]!.text!.should.match(/Response interrupted by user/)
+	})
 })