[eric] browser: prune superseded page-state attachments and heavy read results from history

backend/apps/agents/browser/browser_history.py

@@ -94,6 +94,62 @@ def prune_old_screenshots(messages: list[dict], keep_first: bool = True, keep_re
     return collapsed
 
 
+# Sentinel prefixing the auto-attached element list on mutating action results.
+# Lives here so the attacher (browser_agent) and the pruner share one spelling.
+PAGE_STATE_MARKER = "[page state after action]"
+_STATE_STUB = "[stale page state pruned; see the latest action result for current state]"
+_HEAVY_READ_TOOLS = {"BrowserListInteractives", "BrowserGetText"}
+_HEAVY_READ_MIN_CHARS = 600
+
+
+def prune_stale_page_state(messages: list[dict], keep_recent: int = 2) -> int:
+    """Collapse superseded page-state attachments and heavy read results, in place.
+
+    Auto-attached element lists arrive with EVERY mutating action, so without
+    this the model re-reads each stale copy every turn (same failure shape as
+    screenshots, just in text). Keep the `keep_recent` newest of each pool;
+    older attachments lose only the state suffix (the action's own result text
+    stays), older heavy reads keep their first line as a breadcrumb.
+    Returns how many blocks were collapsed.
+    """
+    id_to_name: dict[str, str] = {}
+    for msg in messages:
+        if msg.get("role") == "assistant" and isinstance(msg.get("content"), list):
+            for b in msg["content"]:
+                if isinstance(b, dict) and b.get("type") == "tool_use" and b.get("id"):
+                    id_to_name[b["id"]] = b.get("name", "")
+    attached: list[dict] = []
+    heavy: list[dict] = []
+    for msg in messages:
+        if msg.get("role") != "user" or not isinstance(msg.get("content"), list):
+            continue
+        for b in msg["content"]:
+            if not (isinstance(b, dict) and b.get("type") == "tool_result"):
+                continue
+            inner = b.get("content")
+            if not isinstance(inner, list):
+                continue
+            tool = id_to_name.get(b.get("tool_use_id"), "")
+            for ib in inner:
+                if not (isinstance(ib, dict) and ib.get("type") == "text"):
+                    continue
+                txt = ib.get("text") or ""
+                if PAGE_STATE_MARKER in txt:
+                    attached.append(ib)
+                elif tool in _HEAVY_READ_TOOLS and len(txt) >= _HEAVY_READ_MIN_CHARS:
+                    heavy.append(ib)
+    pruned = 0
+    for ib in attached[:-keep_recent] if keep_recent else attached:
+        txt = ib["text"]
+        ib["text"] = txt[: txt.index(PAGE_STATE_MARKER)] + _STATE_STUB
+        pruned += 1
+    for ib in heavy[:-keep_recent] if keep_recent else heavy:
+        head = (ib["text"] or "").splitlines()[0][:100]
+        ib["text"] = f"{head}\n[stale read output pruned; re-run the tool if you need it again]"
+        pruned += 1
+    return pruned
+
+
 def _validate_message_pairing(messages: list[dict]) -> bool:
     """Verify every tool_result references a tool_use_id from a prior assistant
     message in the same list. Returns False if there's an orphan, which means

backend/tests/test_browser_screenshot_prune.py

@@ -79,3 +79,53 @@ def test_keep_recent_is_tunable():
     prune_old_screenshots(msgs, keep_first=False, keep_recent=1)
     # only the most recent survives
     assert _count_images(msgs) == 1
+
+
+def _tool_use_msg(tu_id, name):
+    return {"role": "assistant", "content": [
+        {"type": "tool_use", "id": tu_id, "name": name, "input": {}},
+    ]}
+
+
+def _tool_result_msg(tu_id, text):
+    return {"role": "user", "content": [
+        {"type": "tool_result", "tool_use_id": tu_id,
+         "content": [{"type": "text", "text": text}]},
+    ]}
+
+
+def test_prune_stale_page_state_keeps_last_two_attachments():
+    from backend.apps.agents.browser.browser_history import (
+        PAGE_STATE_MARKER, prune_stale_page_state,
+    )
+    msgs = []
+    for i in range(4):
+        msgs.append(_tool_use_msg(f"t{i}", "BrowserClickIndex"))
+        msgs.append(_tool_result_msg(
+            f"t{i}", f"Clicked [{i}]\n\n{PAGE_STATE_MARKER}\n[1]<button \"A{i}\">",
+        ))
+    pruned = prune_stale_page_state(msgs)
+    assert pruned == 2
+    texts = [m["content"][0]["content"][0]["text"] for m in msgs if m["role"] == "user"]
+    assert PAGE_STATE_MARKER not in texts[0] and "Clicked [0]" in texts[0]
+    assert PAGE_STATE_MARKER not in texts[1]
+    assert PAGE_STATE_MARKER in texts[2] and PAGE_STATE_MARKER in texts[3]
+    # idempotent: a second pass finds nothing new to prune
+    assert prune_stale_page_state(msgs) == 0
+
+
+def test_prune_stale_page_state_collapses_old_heavy_reads_only():
+    from backend.apps.agents.browser.browser_history import prune_stale_page_state
+    big = "28 interactive elements\n" + "\n".join(f"[{i}]<button \"x\">" for i in range(60))
+    msgs = []
+    for i in range(3):
+        msgs.append(_tool_use_msg(f"r{i}", "BrowserListInteractives"))
+        msgs.append(_tool_result_msg(f"r{i}", big))
+    msgs.append(_tool_use_msg("nav", "BrowserNavigate"))
+    msgs.append(_tool_result_msg("nav", "Navigated to https://example.com"))
+    pruned = prune_stale_page_state(msgs)
+    assert pruned == 1
+    first = msgs[1]["content"][0]["content"][0]["text"]
+    assert first.startswith("28 interactive elements") and "pruned" in first
+    assert msgs[3]["content"][0]["content"][0]["text"] == big
+    assert msgs[7]["content"][0]["content"][0]["text"] == "Navigated to https://example.com"