fix: deduplicate batched agent progress summaries (#5049)

This is only needed temporarily until we have Callum's researcher-first
view.

## Summary
- Batched agents handle K rows per trace, causing the Engine to return K
identical summaries with different `row_index` values
- Added `dedupe_summaries()` in the MCP layer that collapses these into
one entry per unique text with a `row_indices` list
- Applied in both `_fetch_summaries` (tools.py) and
`_fetch_summaries_rest` (routes.py), covering all consumers:
`futuresearch_progress`, `futuresearch_poll`, and the widget HTTP
polling endpoint

## Test plan
- [x] `uv run pytest tests/ -x` — 351 passed
- [ ] Run a rank task with 5+ rows, verify progress summaries show
`[Rows 0, 1, 2]` grouping instead of 3 identical lines

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Sourced from commit f27dd37c9dc9084b0ab51e336de7486b72d7223f

Author: RafaelPo

futuresearch-mcp/src/futuresearch_mcp/routes.py

@@ -18,7 +18,12 @@
 from futuresearch_mcp import redis_store
 from futuresearch_mcp.config import settings
 from futuresearch_mcp.result_store import _sanitize_records
-from futuresearch_mcp.tool_helpers import _UI_EXCLUDE, TaskState, _fetch_task_result
+from futuresearch_mcp.tool_helpers import (
+    _UI_EXCLUDE,
+    TaskState,
+    _fetch_task_result,
+    dedupe_summaries,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -97,7 +102,10 @@ async def _fetch_summaries_rest(
         )
         if resp.status_code == 200:
             data = resp.json()
-            return data.get("summaries") or None, data.get("cursor") or cursor
+            raw = data.get("summaries") or None
+            if raw:
+                raw = dedupe_summaries(raw)
+            return raw, data.get("cursor") or cursor
     except Exception:
         logger.debug("Failed to fetch summaries for task %s via REST", task_id)
     return None, cursor
@@ -123,9 +131,12 @@ async def _fetch_aggregate_rest(
         )
         if resp.status_code == 200:
             data = resp.json()
+            micros = data.get("micro_summaries") or None
+            if micros:
+                micros = dedupe_summaries(micros)
             return (
                 data.get("aggregate") or None,
-                data.get("micro_summaries") or None,
+                micros,
                 data.get("cursor") or cursor,
             )
     except Exception:

futuresearch-mcp/src/futuresearch_mcp/templates.py

@@ -292,7 +292,7 @@
   if(d.aggregate_summary){
     const now=new Date();
     const ts=now.toLocaleTimeString([],{hour:"2-digit",minute:"2-digit",second:"2-digit"});
-    const micros=(d.summaries||[]).map(s=>({text:s.summary||String(s),row_index:s.row_index}));
+    const micros=(d.summaries||[]).map(s=>({text:s.summary||String(s),row_indices:s.row_indices||null,row_index:s.row_index}));
     /* only add if aggregate text is new */
     const lastAgg=aggHistory.length?aggHistory[aggHistory.length-1].aggregate:"";
     if(d.aggregate_summary!==lastAgg){
@@ -303,7 +303,7 @@
     /* fallback: no aggregate, just micro-summaries — create a placeholder entry */
     const now=new Date();
     const ts=now.toLocaleTimeString([],{hour:"2-digit",minute:"2-digit",second:"2-digit"});
-    const micros=d.summaries.map(s=>({text:s.summary||String(s),row_index:s.row_index}));
+    const micros=d.summaries.map(s=>({text:s.summary||String(s),row_indices:s.row_indices||null,row_index:s.row_index}));
     const fallbackAgg=micros[0]?.text||"Agent activity";
     const lastAgg=aggHistory.length?aggHistory[aggHistory.length-1].aggregate:"";
     if(fallbackAgg!==lastAgg){
@@ -365,7 +365,9 @@
       if(hasMicros){
         al+=`<ul class="agg-micros">`;
         for(const m of a.micros){
-          const rowLabel=m.row_index!=null?`<span class="agg-micro-row">Row ${m.row_index+1}</span>`:"";
+          let rowLabel="";
+          if(m.row_indices&&m.row_indices.length>1){rowLabel=`<span class="agg-micro-row">Rows ${m.row_indices.map(r=>r+1).join(", ")}</span>`;}
+          else if(m.row_index!=null){rowLabel=`<span class="agg-micro-row">Row ${m.row_index+1}</span>`;}
           al+=`<li>${rowLabel}${esc(m.text)}</li>`;
         }
         al+=`</ul>`;

futuresearch-mcp/src/futuresearch_mcp/tool_helpers.py

@@ -291,33 +291,56 @@ async def create_tool_response(
 
 
 def _format_summary_lines(summaries: list[dict[str, Any]]) -> str:
-    """Collapse duplicate summaries from batched agents into grouped lines.
+    """Format summaries as text lines with row index prefixes.
 
-    One trace handling multiple rows produces the same text per row.
-    Groups by text and merges row indices: ``[Rows 29, 17] Summarizing...``
+    Accepts both raw (K duplicates from Engine) and already-deduped
+    summaries (with ``row_indices`` lists from ``dedupe_summaries``).
     """
-    grouped: dict[str, list[int]] = {}
-    grouped_order: list[str] = []
-    for s in summaries:
-        text = s["summary"]
-        row_idx = s.get("row_index")
-        if text not in grouped:
-            grouped[text] = []
-            grouped_order.append(text)
-        if row_idx is not None:
-            grouped[text].append(row_idx)
+    # Dedupe only if input lacks row_indices (raw from Engine)
+    if summaries and "row_indices" not in summaries[0]:
+        summaries = dedupe_summaries(summaries)
     lines = ""
-    for text in grouped_order:
-        rows = grouped[text]
+    for entry in summaries:
+        text = entry.get("summary", "")
+        rows = entry.get("row_indices") or []
         if rows:
             label = "Row" if len(rows) == 1 else "Rows"
-            prefix = f"[{label} {', '.join(str(r) for r in sorted(rows))}] "
+            prefix = f"[{label} {', '.join(str(r) for r in rows)}] "
         else:
             prefix = ""
         lines += f"\n- {prefix}{text}"
     return lines
 
 
+def dedupe_summaries(summaries: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Collapse duplicate summaries from batched agents into one per unique text.
+
+    The Engine returns K identical summaries (one per row) when a batched
+    agent handles K rows.  This merges them into a single entry with a
+    ``row_indices`` list, preserving order.
+    """
+    grouped: dict[str, dict[str, Any]] = {}
+    order: list[str] = []
+    for s in summaries:
+        text = s.get("summary", "")
+        if text not in grouped:
+            grouped[text] = {**s, "row_indices": []}
+            order.append(text)
+        row_idx = s.get("row_index")
+        if row_idx is not None:
+            grouped[text]["row_indices"].append(row_idx)
+    result = []
+    for text in order:
+        entry = grouped[text]
+        indices = sorted(entry["row_indices"])
+        entry["row_indices"] = indices or None
+        entry.pop("row_index", None)
+        if indices:
+            entry["row_index"] = indices[0]
+        result.append(entry)
+    return result
+
+
 class TaskState(BaseModel):
     """Parsed progress snapshot from an API status response."""
 

futuresearch-mcp/src/futuresearch_mcp/tools.py

@@ -66,6 +66,7 @@
     _get_client,
     _record_task_ownership,
     create_tool_response,
+    dedupe_summaries,
     log_client_info,
 )
 from futuresearch_mcp.utils import fetch_csv_from_url, is_url, save_result_to_csv
@@ -869,7 +870,10 @@ async def _fetch_partial_rows(
 async def _fetch_summaries(
     httpx_client: Any, task_id: str, cursor: str | None
 ) -> tuple[list[dict[str, Any]] | None, str | None]:
-    """Fetch progress summaries. Returns (summaries, updated_cursor)."""
+    """Fetch progress summaries, deduplicating batched agent copies.
+
+    Returns (summaries, updated_cursor).
+    """
     try:
         query: dict[str, Any] = {}
         if cursor:
@@ -881,7 +885,10 @@ async def _fetch_summaries(
         )
         if resp.status_code == 200:
             data = resp.json()
-            return data.get("summaries") or None, data.get("cursor") or cursor
+            raw = data.get("summaries") or None
+            if raw:
+                raw = dedupe_summaries(raw)
+            return raw, data.get("cursor") or cursor
         logger.warning("summaries returned %s for task %s", resp.status_code, task_id)
     except Exception:
         logger.debug("Failed to fetch summaries for task %s", task_id)