fix: pruned skill_view/read_file results reload after a proactive prune

A committed proactive tool-result prune (`prune_tool_results_only`, driven from
`agent/turn_preflight.py::compress_after_tool_results`) demotes old skill_view and
read_file bodies to one-line markers, but only the full-compaction path called
`_reset_read_dedup_caches`. The repeat-view dedup therefore kept answering
"unchanged / content_returned: false" for content that no longer existed in the
transcript, so the `[SKILL_PRUNED: ... reload with skill_view(...)]` marker asked
for a reload the tool then refused (#112763). Treat the committed prune as the same
content-loss boundary compaction already is: reset the task's read/skill dedup right
where the pruned list is committed.

Sibling surface: manual `/compress` on CLI, TUI and the messaging gateway called
`compress_now()` without `task_id`, so the boundary reset hit the "default" bucket
instead of the session's. Pass the session-scoped task id on all three.

Test double in tests/hermes_cli/test_cli_manual_compress.py gains the `task_id`
kwarg the real `_compress_context` facade already accepts.

Supersedes #103268 (@jo0wz), which reached the same path with a process-global
ghost registry (not task-keyed, skill_view only).

agent/turn_preflight.py

@@ -15,7 +15,7 @@ from typing import Any, Dict, List, Optional
 
 from agent.context_engine import automatic_compaction_status_message
 from agent.conversation_compression import (
-    PRE_API_COMPRESSION_STATUS_TEMPLATE, compression_blocked_transiently,
+    PRE_API_COMPRESSION_STATUS_TEMPLATE, _reset_read_dedup_caches, compression_blocked_transiently,
     compression_skipped_due_to_lock, context_compression_timed_out,
     conversation_history_after_compression,
 )
@@ -374,4 +374,8 @@ def compress_after_tool_results(
             # stale in-place flag the helper could seed unpersisted rows.
             if _pruned_n and _pruned_msgs is not messages:
                 messages = _pruned_msgs
+                # A committed prune is a content-loss boundary like compaction: demoted skill_view /
+                # read_file bodies survive only as one-line markers, so the repeat-read dedup must
+                # stop answering "unchanged" for them or the reload the marker asks for is refused.
+                _reset_read_dedup_caches(effective_task_id, session_id=agent.session_id or "")
     return _verdict(False)

gateway/slash_commands_session.py

@@ -550,7 +550,8 @@ class GatewaySessionCommandsMixin:
             # Not a bare run_in_executor: the profile secret scope is a contextvar the default
             # executor hop would drop, failing aux-client credential resolution closed.
             result = await self._run_in_executor_with_context(
-                lambda: compress_now(tmp_agent, msgs, request, system_message="", skip_without_window=True))
+                lambda: compress_now(tmp_agent, msgs, request, system_message="", skip_without_window=True,
+                                     task_id=session_entry.session_id or "default"))
             if result.status == "nothing_to_do":
                 return t("gateway.compress.nothing_to_do")
             if result.status != "compressed":

hermes_cli/cli_session_mixin.py

@@ -968,7 +968,8 @@ class CLISessionMixin:
                     print(f"🗜️  Compressing {original_count} messages, focus: \"{request.focus_topic}\"...")
                 else:
                     print(f"🗜️  Compressing {original_count} messages...")
-                result = compress_now(self.agent, self.conversation_history, request)
+                result = compress_now(self.agent, self.conversation_history, request,
+                                      task_id=self.session_id or "default")
                 if result.status != "compressed":
                     for line in render_compress_result(result):
                         print(f"  {line}")

tests/agent/test_proactive_prune_loop_wiring.py

@@ -117,7 +117,7 @@ def agent():
     return a
 
 
-def _run_tool_loop(agent, n_tool_iterations: int):
+def _run_tool_loop(agent, n_tool_iterations: int, task_id=None):
     responses = [_tool_response(i) for i in range(n_tool_iterations)]
     responses.append(_stop_response())
     agent.client.chat.completions.create.side_effect = responses
@@ -131,7 +131,7 @@ def _run_tool_loop(agent, n_tool_iterations: int):
             lambda name, args, task_id=None, **kwargs: json.dumps({"ok": True}),
         ),
     ):
-        result = agent.run_conversation("do a lot of tool work")
+        result = agent.run_conversation("do a lot of tool work", task_id=task_id)
 
     return result
 
@@ -258,3 +258,61 @@ class TestProactivePruneLoopWiring:
         # tool output may be wrapped in an untrusted_tool_result envelope —
         # assert the original payload survived un-pruned.
         assert all('"ok": true' in m["content"] for m in tool_rows)
+
+
+class TestCommittedPruneIsDedupBoundary:
+    """A committed proactive prune demotes old skill_view / read_file bodies to one-line markers
+    without a compaction boundary; the repeat-read dedup must stop answering "unchanged" for them
+    or the reload the marker asks for is refused (#112763)."""
+
+    @staticmethod
+    def _seed_dedup(tmp_path, task_id):
+        from tools.file_tools_read_tracking import _read_tracker, _read_tracker_lock, _task_data
+        from tools.skills_tool_dedup import _record_skill_view, reset_skill_view_dedup
+
+        skill_md = tmp_path / "SKILL.md"
+        skill_md.write_text("# s\n", encoding="utf-8")
+        reset_skill_view_dedup(task_id)
+        _record_skill_view(task_id, "bigskill", None, {"name": "bigskill", "_source_path": str(skill_md)})
+        with _read_tracker_lock:
+            _read_tracker.pop(task_id, None)
+            td = _task_data(task_id)
+            td["dedup"][("/x/big.txt", 1, 2000)] = 1.0
+            td["dedup_generation_reads"].add(("/x/big.txt", 1, 2000))
+        return skill_md
+
+    @staticmethod
+    def _dedup_state(task_id):
+        from tools.file_tools_read_tracking import _read_tracker
+        from tools.skills_tool_dedup import _check_skill_view_dedup
+
+        skill_stubbed = _check_skill_view_dedup(task_id, "bigskill", None) is not None
+        file_in_generation = ("/x/big.txt", 1, 2000) in _read_tracker[task_id]["dedup_generation_reads"]
+        return skill_stubbed, file_in_generation
+
+    def test_committed_prune_releases_skill_and_file_dedup(self, agent, tmp_path):
+        task_id = "prune-boundary-task"
+        self._seed_dedup(tmp_path, task_id)
+        assert self._dedup_state(task_id) == (True, True)
+
+        def _prune(messages, current_tokens=None):
+            pruned = [dict(m) for m in messages]
+            changed = 0
+            for m in pruned:
+                if m.get("role") == "tool" and m.get("content") != "[pruned]":
+                    m["content"] = "[pruned]"
+                    changed += 1
+            return (pruned, changed) if changed else (messages, 0)
+
+        agent.context_compressor.prune_tool_results_only = _prune
+        assert _run_tool_loop(agent, n_tool_iterations=1, task_id=task_id)["completed"] is True
+        # Skill: next view serves full content again. File: the generation-read set is cleared so the
+        # first unchanged re-read serves content; the mtime map itself is preserved (later reads stub).
+        assert self._dedup_state(task_id) == (False, False)
+
+    def test_noop_prune_keeps_dedup(self, agent, tmp_path):
+        task_id = "prune-noop-task"
+        self._seed_dedup(tmp_path, task_id)
+        agent.context_compressor.prune_tool_results_only = lambda messages, current_tokens=None: (messages, 0)
+        assert _run_tool_loop(agent, n_tool_iterations=1, task_id=task_id)["completed"] is True
+        assert self._dedup_state(task_id) == (True, True)

tests/hermes_cli/test_cli_manual_compress.py

@@ -38,6 +38,7 @@ class DummyAgent:
         focus_topic=None,
         force=False,
         defer_context_engine_notification=False,
+        task_id="default",
     ):
         self.calls.append(
             {

tools/skills_tool_dedup.py

@@ -1,7 +1,8 @@
 """skill_view repeat-view dedup registry: per-task cache of (skill name, file_path) ->
 (skill file mtime+size). A repeat view of an UNCHANGED file returns a short stub — the earlier
-tool result already carries the content verbatim. Cleared on context compression via
-``reset_skill_view_dedup()`` because the original content is summarized away.
+tool result already carries the content verbatim. Cleared via ``reset_skill_view_dedup()`` on
+context compression AND on a committed proactive tool-result prune, because both replace the
+original content with a one-line marker.
 """
 
 import json

tui_gateway/session_compression.py

@@ -219,7 +219,7 @@ def _compress_session_history(
     request = parse_compress_args(focus_topic or "")
     if request.aggressive:
         raise ValueError(AGGRESSIVE_UNSUPPORTED)
-    result = compress_now(agent, before_messages, request)
+    result = compress_now(agent, before_messages, request, task_id=session.get("session_key") or "default")
     if result.status == "preview":
         return 0, _get_usage(agent)
     # Lock-skipped: raise so callers surface a clear message instead of "No changes from compression".