fix: floor TPS time-span to 1s + clear token state on stream-error

Streaming TPS pill could briefly show inflated values at the start of a
new stream then 'drop abruptly' to a real value, looking like cached
stale data. Two root causes:

1. calculateTPS divided by (now - firstDelta.timestamp). For a brand
   new stream's first delta that span is just a few ms — e.g.
   '50 tokens / 0.005s = 10000 t/s'. As more deltas accumulate the
   window broadens and TPS settles, hence the visible drop. Phase 1's
   microtask cadence exposed this where the prior idle-callback batching
   used to mask it. Floor the divisor to a 1s minimum window so the rate
   smoothly ramps up over the first second of a stream instead of
   spiking. Underestimation during the settling window is acceptable;
   order-of-magnitude overestimation isn't.

2. The stream-error event handler in applyWorkspaceChatEventToAggregator
   didn't call clearTokenState, leaving the errored message's
   deltaHistory entry to leak. Match stream-end / stream-abort and clear
   it so a follow-up stream starts with a clean slate.

Adds tests for both behaviors.

Author: ammar-agent

src/browser/utils/messages/StreamingTPSCalculator.test.ts

@@ -81,14 +81,47 @@ describe("StreamingTPSCalculator", () => {
       expect(calculateTPS(deltas, now)).toBe(15);
     });
 
-    test("returns 0 when time span is zero", () => {
+    test("uses minimum 1s window when time span is zero", () => {
       const now = 1000;
       const deltas: DeltaRecord[] = [
         { tokens: 100, timestamp: 1000, type: "text" }, // Same timestamp as now
       ];
+      // 100 tokens divided by the 1s minimum window = 100 t/s.
+      // (Floor avoids the divide-by-tiny-window artifact that otherwise reports
+      // tens of thousands of t/s when only one chunk has arrived.)
+      expect(calculateTPS(deltas, now)).toBe(100);
+    });
+
+    test("returns 0 when time span is negative (clock skew)", () => {
+      const now = 500;
+      const deltas: DeltaRecord[] = [
+        { tokens: 100, timestamp: 1000, type: "text" }, // future-dated, treat as invalid
+      ];
       expect(calculateTPS(deltas, now)).toBe(0);
     });
 
+    test("floors tiny time spans to the 1s minimum to avoid inflated rates", () => {
+      // Realistic scenario: first delta of a fresh stream — only a few ms have
+      // elapsed between the chunk's timestamp and the read.
+      const now = 1005;
+      const deltas: DeltaRecord[] = [{ tokens: 50, timestamp: 1000, type: "text" }];
+      // Without the floor: 50 tokens / 0.005s = 10000 t/s (bogus).
+      // With the 1s floor: 50 tokens / 1s = 50 t/s.
+      expect(calculateTPS(deltas, now)).toBe(50);
+    });
+
+    test("uses raw time span once it exceeds the 1s minimum", () => {
+      // 1.5s after first delta — the floor stops applying and we return the
+      // honest rate.
+      const now = 2500;
+      const deltas: DeltaRecord[] = [
+        { tokens: 50, timestamp: 1000, type: "text" },
+        { tokens: 100, timestamp: 2000, type: "text" },
+      ];
+      // 150 tokens over 1.5s = 100 t/s.
+      expect(calculateTPS(deltas, now)).toBe(100);
+    });
+
     test("uses current time by default", () => {
       const now = Date.now();
       const deltas: DeltaRecord[] = [

src/browser/utils/messages/applyWorkspaceChatEventToAggregator.test.ts

@@ -208,6 +208,24 @@ describe("applyWorkspaceChatEventToAggregator", () => {
     expect(aggregator.calls).toEqual(["clearTokenState:msg-1", "handleStreamAbort:msg-1"]);
   });
 
+  test("stream-error clears token state after calling handleStreamError", () => {
+    const aggregator = new StubAggregator();
+
+    const event: WorkspaceChatMessage = {
+      type: "stream-error",
+      messageId: "msg-1",
+      error: "boom",
+      errorType: "network",
+    };
+
+    const hint = applyWorkspaceChatEventToAggregator(aggregator, event);
+
+    expect(hint).toBe("immediate");
+    // Without the clearTokenState call, the next stream's TPS/token-count
+    // calculation could draw from the errored message's leaked deltaHistory.
+    expect(aggregator.calls).toEqual(["handleStreamError:msg-1", "clearTokenState:msg-1"]);
+  });
+
   test("tool-call-delta routes to handleToolCallDelta and is throttled", () => {
     const aggregator = new StubAggregator();
 

src/browser/utils/messages/applyWorkspaceChatEventToAggregator.ts

@@ -171,6 +171,10 @@ export function applyWorkspaceChatEventToAggregator(
     }
 
     aggregator.handleStreamError(event);
+    // Match stream-end / stream-abort: drop the per-message deltaHistory entry
+    // so a subsequent stream isn't influenced by leaked TPS/token state from
+    // the errored one.
+    aggregator.clearTokenState(event.messageId);
     return "immediate";
   }
 

src/common/utils/tokens/tps.ts

@@ -12,6 +12,25 @@ export interface DeltaRecord {
 
 const DEFAULT_TPS_WINDOW_MS = 60000; // 60 second trailing window
 
+/**
+ * Minimum time span used in the TPS divisor.
+ *
+ * Without this floor, the first delta of a stream produces a wildly inflated
+ * TPS: with one delta `timeSpanMs = now - delta.timestamp` is typically a
+ * single-digit milliseconds value (the gap between when the chunk arrived and
+ * when we read it), so e.g. `50 tokens / 0.005s = 10000 t/s`. The user-visible
+ * effect is "TPS starts very high then drops abruptly" as more deltas arrive
+ * and broaden the window. Flooring to a 1s minimum window makes the rate
+ * smoothly ramp up from ~0 toward the steady-state value over the first
+ * second of a stream — a much more honest measurement.
+ *
+ * Trade-off: in the very first second the reported TPS slightly *under*-counts
+ * (we divide by 1s even when only 200ms have elapsed). That's acceptable —
+ * understatement during a settling window is far better than a misleading
+ * order-of-magnitude overstatement.
+ */
+const MIN_TPS_TIME_SPAN_MS = 1000;
+
 /**
  * Calculate tokens-per-second from a history of delta records.
  */
@@ -23,9 +42,9 @@ export function calculateTPS(deltas: DeltaRecord[], now: number = Date.now()): n
   if (recentDeltas.length === 0) return 0;
 
   const totalTokens = recentDeltas.reduce((sum, d) => sum + (d.tokens || 0), 0);
-  const timeSpanMs = now - recentDeltas[0].timestamp;
-  const timeSpanSec = timeSpanMs / 1000;
-  if (timeSpanSec <= 0) return 0;
+  const rawTimeSpanMs = now - recentDeltas[0].timestamp;
+  if (rawTimeSpanMs < 0) return 0;
+  const timeSpanSec = Math.max(rawTimeSpanMs, MIN_TPS_TIME_SPAN_MS) / 1000;
 
   return Math.round(totalTokens / timeSpanSec);
 }