merge: propagate second-pass fixes from test/chat-agent-durability-tri-11166

Author: kathiekiwi

.server-changes/watch-summary-cumulative-count.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+The grouped "watch updates" notification now shows the total number of results waiting, instead of only the most recent batch's count.

apps/webapp/app/components/dashboard-agent/DashboardAgent.tsx

@@ -23,7 +23,7 @@ import {
 } from "./panel-layout";
 import { nextPendingTurnChatId } from "./pending-turn";
 import { nextVisibleChat } from "./unread-counts";
-import { startWakePolling, wakesToToast } from "./wake-poll";
+import { planWakeToasts, startWakePolling, wakesToToast } from "./wake-poll";
 import { shouldPollWakeFeed, subscribeWatchActivity } from "./watch-activity";
 import {
   showWatchWakesSummaryToast,
@@ -99,6 +99,11 @@ export function DashboardAgent({
   // which outlives the render that started it, so it has to be a ref.
   const visibleChat = useRef<string | null>(null);
 
+  // The count the still-visible grouped toast claims. Consecutive polls add to it so a
+  // later batch grows the summary instead of overwriting it with only its own count;
+  // reset when the user opens the panel from that toast.
+  const summaryPending = useRef(0);
+
   // Switching environment re-runs the layout loader but does not remount it, so the seeds
   // above would keep the old environment's counts.
   const seededEnvironment = useRef(environment.id);
@@ -224,11 +229,22 @@ export function DashboardAgent({
         const fresh = wakesToToast(data.wakes, toastedWakes.current);
         for (const wake of fresh) rememberToasted(wake.watchId);
 
-        if (fresh.length > WAKE_TOAST_MAX_INDIVIDUAL) {
-          showWatchWakesSummaryToast(fresh.length, () => setPanelOpen(true));
-        } else {
-          for (const wake of [...fresh].reverse()) {
-            showWatchWakeToast(wake, openChat);
+        if (fresh.length > 0) {
+          const { plan, pending } = planWakeToasts(
+            fresh,
+            summaryPending.current,
+            WAKE_TOAST_MAX_INDIVIDUAL
+          );
+          summaryPending.current = pending;
+          if (plan.mode === "summary") {
+            showWatchWakesSummaryToast(plan.count, () => {
+              summaryPending.current = 0;
+              setPanelOpen(true);
+            });
+          } else {
+            for (const wake of [...plan.wakes].reverse()) {
+              showWatchWakeToast(wake, openChat);
+            }
           }
         }
       } catch {

apps/webapp/app/components/dashboard-agent/settled-transcript.test.ts

@@ -75,6 +75,49 @@ describe("merging a re-read transcript", () => {
   });
 });
 
+describe("replacing a stale running step from the re-read", () => {
+  // Same message id, but the stream EOF'd before `get_report` produced an output.
+  const RUNNING_STEP = {
+    id: "msg_step",
+    role: "assistant",
+    parts: [{ type: "tool-get_report", toolCallId: "call_1", state: "input-available" }],
+  };
+
+  const FINISHED_STEP = {
+    id: "msg_step",
+    role: "assistant",
+    parts: [
+      { type: "tool-get_report", toolCallId: "call_1", state: "output-available", output: {} },
+    ],
+  };
+
+  it("swaps the still-running copy for its finished version from the authoritative read", () => {
+    const merged = mergeSettledMessages([RUNNING_STEP], [FINISHED_STEP]);
+    expect(merged).toEqual([FINISHED_STEP]);
+    // The step no longer reads as running, so nothing keeps the panel on Working…
+    expect(transcriptLooksUnfinished(merged)).toBe(false);
+  });
+
+  it("still appends genuinely-new messages while replacing a stale one", () => {
+    const merged = mergeSettledMessages([RUNNING_STEP], [FINISHED_STEP, SETTLED]);
+    expect(merged.map((message) => message.id)).toEqual([FINISHED_STEP.id, SETTLED.id]);
+    expect(merged[0]).toBe(FINISHED_STEP);
+  });
+
+  it("leaves an in-flight message alone when the re-read is itself still running", () => {
+    const merged = mergeSettledMessages([RUNNING_STEP], [RUNNING_STEP]);
+    // Same reference back, no needless render, and the live turn is untouched.
+    expect(merged).toEqual([RUNNING_STEP]);
+    expect(merged[0]).toBe(RUNNING_STEP);
+  });
+
+  it("does not touch a running message the re-read does not mention", () => {
+    const merged = mergeSettledMessages([RUNNING_STEP], [SETTLED]);
+    expect(merged.map((message) => message.id)).toEqual([RUNNING_STEP.id, SETTLED.id]);
+    expect(merged[0]).toBe(RUNNING_STEP);
+  });
+});
+
 describe("reading the transcript endpoint", () => {
   afterEach(() => {
     vi.unstubAllGlobals();

apps/webapp/app/components/dashboard-agent/settled-transcript.ts

@@ -1,4 +1,4 @@
-import { inFlightToolName, liveInvestigation } from "./progress-line";
+import { IN_FLIGHT_TOOL_STATES, inFlightToolName, liveInvestigation } from "./progress-line";
 
 /**
  * Re-reading the stored transcript once a turn settles.
@@ -11,17 +11,47 @@ import { inFlightToolName, liveInvestigation } from "./progress-line";
 
 type Identified = { id: string };
 
+/** A message whose stream died mid-tool: a `tool-*` part still reads as running. */
+function stillRunning(message: unknown): boolean {
+  const parts = (message as { parts?: ReadonlyArray<{ type?: string; state?: string }> })?.parts;
+  if (!Array.isArray(parts)) return false;
+  return parts.some(
+    (part) =>
+      typeof part?.type === "string" &&
+      part.type.startsWith("tool-") &&
+      IN_FLIGHT_TOOL_STATES.has(part.state ?? "")
+  );
+}
+
 /**
- * Append-only, keyed on the message id. Ids are stable (a settlement card is
- * `investigation-settlement:{id}:{revision}`), so re-reading the same transcript any
- * number of times can never produce a second copy of a card, and nothing already
- * rendered is reordered or replaced.
+ * Merge the authoritative re-read into what the panel holds, keyed on the message id.
+ *
+ * Genuinely-new messages (a settlement card is `investigation-settlement:{id}:{revision}`)
+ * are appended, so re-reading the same transcript any number of times never produces a
+ * second copy. A message whose in-memory copy died mid-tool — a stream that EOF'd before
+ * the part settled — is replaced by its finished version from the re-read; otherwise it
+ * would show that step running forever. We only replace a still-running copy with a copy
+ * that has itself settled, so a live turn streaming under the same id is left alone and
+ * ordering is preserved.
  */
 export function mergeSettledMessages<T extends Identified>(current: T[], fetched: T[]): T[] {
+  const byId = new Map(fetched.map((message) => [message.id, message]));
+
+  let replaced = false;
+  const next = current.map((existing) => {
+    const settled = byId.get(existing.id);
+    if (settled && settled !== existing && stillRunning(existing) && !stillRunning(settled)) {
+      replaced = true;
+      return settled;
+    }
+    return existing;
+  });
+
   const missing = fetched.filter(
     (message) => !current.some((existing) => existing.id === message.id)
   );
-  return missing.length === 0 ? current : [...current, ...missing];
+  if (missing.length === 0) return replaced ? next : current;
+  return [...next, ...missing];
 }
 
 /** Whether the transcript still resolves to a card mid-investigation. */

apps/webapp/app/components/dashboard-agent/wake-poll.test.ts

@@ -1,5 +1,10 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
-import { startWakePolling, UNREAD_POLL_INTERVAL_MS, wakesToToast } from "./wake-poll";
+import {
+  planWakeToasts,
+  startWakePolling,
+  UNREAD_POLL_INTERVAL_MS,
+  wakesToToast,
+} from "./wake-poll";
 
 function harness() {
   let hidden = false;
@@ -124,3 +129,42 @@ describe("wakesToToast", () => {
     expect(wakesToToast(undefined, new Set())).toEqual([]);
   });
 });
+
+describe("planWakeToasts", () => {
+  const MAX = 3;
+  const batch = (n: number) => Array.from({ length: n }, (_, i) => i);
+
+  it("toasts a small batch individually but still counts it toward the running total", () => {
+    const { plan, pending } = planWakeToasts(batch(2), 0, MAX);
+
+    expect(plan).toEqual({ mode: "individual", wakes: [0, 1] });
+    expect(pending).toBe(2);
+  });
+
+  it("summarizes when a single batch is over the max", () => {
+    const { plan, pending } = planWakeToasts(batch(4), 0, MAX);
+
+    expect(plan).toEqual({ mode: "summary", count: 4 });
+    expect(pending).toBe(4);
+  });
+
+  it("accumulates across consecutive polls instead of showing only the latest", () => {
+    // First batch of 2 is below the max: individual toasts, nothing pending yet.
+    const first = planWakeToasts(batch(2), 0, MAX);
+    expect(first.plan.mode).toBe("individual");
+
+    // A second batch of 3 pushes the running total to 5, so the grouped toast claims
+    // the cumulative count, not just this batch's 3.
+    const second = planWakeToasts(batch(3), first.pending, MAX);
+    expect(second.plan).toEqual({ mode: "summary", count: 5 });
+    expect(second.pending).toBe(5);
+  });
+
+  it("grows the visible summary as later batches arrive", () => {
+    const first = planWakeToasts(batch(4), 0, MAX);
+    const second = planWakeToasts(batch(3), first.pending, MAX);
+
+    expect(second.plan).toEqual({ mode: "summary", count: 7 });
+    expect(second.pending).toBe(7);
+  });
+});

apps/webapp/app/components/dashboard-agent/wake-poll.ts

@@ -21,6 +21,30 @@ export function wakesToToast<T extends { watchId: string; unread?: boolean }>(
   return (wakes ?? []).filter((wake) => wake.unread === true && !toasted.has(wake.watchId));
 }
 
+export type WakeToastPlan<T> =
+  | { mode: "summary"; count: number }
+  | { mode: "individual"; wakes: T[] };
+
+/**
+ * Whether this poll's fresh wakes join a grouped summary or each get their own toast.
+ * `pending` is the running count of unacknowledged wakes carried from earlier polls; a
+ * batch that pushes the total past `max` shows the summary with that cumulative count, so
+ * a later batch adds to it rather than replacing it with only its own, smaller number.
+ * The returned `pending` is what the caller carries into the next poll; it resets to zero
+ * once the user acknowledges (opens the panel).
+ */
+export function planWakeToasts<T>(
+  fresh: T[],
+  pending: number,
+  max: number
+): { plan: WakeToastPlan<T>; pending: number } {
+  const total = pending + fresh.length;
+  if (total > max) {
+    return { plan: { mode: "summary", count: total }, pending: total };
+  }
+  return { plan: { mode: "individual", wakes: fresh }, pending: total };
+}
+
 export type WakePollOptions = {
   load: () => Promise<void>;
   isHidden: () => boolean;