fix: preserve only unseen tool results

s08_context_compact/README.ja.md

@@ -117,12 +117,15 @@ messages = [*messages[:head_end], marker, *messages[tail_start:]]
 
 ## ステップ 3：micro_compact
 
-`micro_compact` は最新の `tool_result` バッチを完全に保持し、さらに以前のバッチから最新 3 件を残します。それより古く 120 文字を超える結果を短くします。保存済みの結果にはファイルパスを残し、それ以外はプレースホルダーに置き換えます。
+`micro_compact` は直近の assistant 応答より後に追加されたすべての `tool_result` を完全に保持し、モデルが各結果を少なくとも 1 回は完全な形で読めるようにします。モデルがすでに読んだ結果については最新 3 件を残し、それより古く 120 文字を超える結果を短くします。保存済みの結果にはファイルパスを残し、それ以外はプレースホルダーに置き換えます。
 
 ![古い結果を置き換える](images/micro-compact.ja.svg)
 
 ```python
-for block in results[:-self.KEEP_RECENT_RESULTS]:
+unseen = self.unseen_tool_result_positions(messages)
+consumed = [entry for entry in results if entry[:2] not in unseen]
+
+for _, _, block in consumed[:-self.KEEP_RECENT_RESULTS]:
     content = str(block.get("content", ""))
     if len(content) <= 120:
         continue
@@ -305,7 +308,7 @@ s01_agent_loop から s05_todo_write までの README.md を読み、
 各ファイルの最上位見出しを比較して、命名の規則をまとめてください。
 ```
 
-このタスクでは少なくとも 5 件のファイル結果が生成されます。最新のバッチと、それ以前の最新 3 件は完全に残り、それより前の長い結果は `[Earlier tool result omitted.]` に変わります。保存済みの結果には保存先のパスが残ります。
+このタスクでは少なくとも 5 件のファイル結果が生成されます。各新規結果はモデルが初めて読むまで完全に保持されます。以降のターンでは、すでに読まれた最新 3 件を残し、それより前の長い結果は `[Earlier tool result omitted.]` に変わります。保存済みの結果には保存先のパスが残ります。
 
 ### 実験 2：大きな結果を保存する
 

s08_context_compact/README.md

@@ -117,12 +117,15 @@ This step controls the number of messages. Tool results inside the retained mess
 
 ## Step 3: micro_compact
 
-`micro_compact` preserves the newest `tool_result` batch in full, then keeps the latest 3 results from earlier batches and shortens older results longer than 120 characters. Persisted results keep their file path; the rest become placeholders:
+`micro_compact` preserves every `tool_result` added after the most recent assistant response, so the model sees each new result in full once. Among results the model has already consumed, it keeps the latest 3 and shortens older results longer than 120 characters. Persisted results keep their file path; the rest become placeholders:
 
 ![Replacing old results](images/micro-compact.en.svg)
 
 ```python
-for block in results[:-self.KEEP_RECENT_RESULTS]:
+unseen = self.unseen_tool_result_positions(messages)
+consumed = [entry for entry in results if entry[:2] not in unseen]
+
+for _, _, block in consumed[:-self.KEEP_RECENT_RESULTS]:
     content = str(block.get("content", ""))
     if len(content) <= 120:
         continue
@@ -305,7 +308,7 @@ Read the README.md files from s01_agent_loop through s05_todo_write.
 Compare their top-level headings and summarize the naming pattern.
 ```
 
-This task produces at least 5 file results. The newest batch and the latest 3 earlier results remain complete, while older long results become `[Earlier tool result omitted.]`. A persisted result retains its saved path.
+This task produces at least 5 file results. Every result remains complete until the model sees it once. On later turns, the latest 3 consumed results remain complete while older long results become `[Earlier tool result omitted.]`. A persisted result retains its saved path.
 
 ### Experiment 2: Persist a Large Result
 

s08_context_compact/README.zh.md

@@ -117,12 +117,15 @@ messages = [*messages[:head_end], marker, *messages[tail_start:]]
 
 ## 第三步：micro_compact
 
-`micro_compact` 会完整保留最新一批 `tool_result`，再保留更早批次中最近 3 条结果；其余超过 120 个字符的旧结果会缩短。已经转存的结果保留文件路径，其他结果只留下占位符：
+`micro_compact` 会完整保留最近一次 assistant 响应之后新增的所有 `tool_result`，确保模型至少完整读取每条新结果一次。对于模型已经读取过的结果，它保留最近 3 条，并缩短其余超过 120 个字符的旧结果。已经转存的结果保留文件路径，其他结果只留下占位符：
 
 ![旧结果替换为占位符](images/micro-compact.svg)
 
 ```python
-for block in results[:-self.KEEP_RECENT_RESULTS]:
+unseen = self.unseen_tool_result_positions(messages)
+consumed = [entry for entry in results if entry[:2] not in unseen]
+
+for _, _, block in consumed[:-self.KEEP_RECENT_RESULTS]:
     content = str(block.get("content", ""))
     if len(content) <= 120:
         continue
@@ -305,7 +308,7 @@ python s08_context_compact/code.py
 比较它们的一级标题，并总结这些标题的命名规律。
 ```
 
-任务会产生至少 5 条文件读取结果。最新一批以及更早批次中最近 3 条结果保持完整，更早且较长的结果会变成 `[Earlier tool result omitted.]`。已经转存的结果会保留保存路径。
+任务会产生至少 5 条文件读取结果。每条新结果在模型首次读取前都会保持完整；后续轮次只保留最近 3 条已读取结果，更早且较长的结果会变成 `[Earlier tool result omitted.]`。已经转存的结果会保留保存路径。
 
 ### 实验二：大结果转存
 

s08_context_compact/code.py

@@ -267,6 +267,23 @@ class ContextCompactor:
                     for block in content)
         )
 
+    @staticmethod
+    def unseen_tool_result_positions(messages: list) -> set[tuple[int, int]]:
+        """Return results added since the model's most recent response."""
+        last_assistant = next(
+            (index for index in range(len(messages) - 1, -1, -1)
+             if messages[index].get("role") == "assistant"),
+            -1,
+        )
+        return {
+            (message_index, block_index)
+            for message_index in range(last_assistant + 1, len(messages))
+            if messages[message_index].get("role") == "user"
+            and isinstance(messages[message_index].get("content"), list)
+            for block_index, block in enumerate(messages[message_index]["content"])
+            if isinstance(block, dict) and block.get("type") == "tool_result"
+        }
+
     def write_transcript(self, messages: list) -> Path:
         self.transcript_dir.mkdir(parents=True, exist_ok=True)
         path = self.transcript_dir / f"transcript_{uuid.uuid4().hex}.jsonl"
@@ -325,26 +342,15 @@ class ContextCompactor:
 
     def micro_compact(self, messages: list) -> list:
         results = [
-            block
+            (message_index, block_index, block)
-            for message in messages
+            for message_index, message in enumerate(messages)
             if message.get("role") == "user" and isinstance(message.get("content"), list)
-            for block in message["content"]
+            for block_index, block in enumerate(message["content"])
             if isinstance(block, dict) and block.get("type") == "tool_result"
         ]
-        latest_batch = []
+        unseen = self.unseen_tool_result_positions(messages)
-        for message in reversed(messages):
+        consumed = [entry for entry in results if entry[:2] not in unseen]
-            content = message.get("content")
+        for _, _, block in consumed[:-self.KEEP_RECENT_RESULTS]:
-            if message.get("role") != "user" or not isinstance(content, list):
-                continue
-            latest_batch = [
-                block for block in content
-                if isinstance(block, dict) and block.get("type") == "tool_result"
-            ]
-            if latest_batch:
-                break
-        latest_batch_ids = {id(block) for block in latest_batch}
-        older_results = [block for block in results if id(block) not in latest_batch_ids]
-        for block in older_results[:-self.KEEP_RECENT_RESULTS]:
             content = str(block.get("content", ""))
             if len(content) <= 120:
                 continue

s15_integrated_harness/code.py

@@ -1903,6 +1903,23 @@ def collect_tool_results(messages: list):
     return found
 
 
+def unseen_tool_result_positions(messages: list) -> set[tuple[int, int]]:
+    """Return results added since the model's most recent response."""
+    last_assistant = next(
+        (index for index in range(len(messages) - 1, -1, -1)
+         if messages[index].get("role") == "assistant"),
+        -1,
+    )
+    return {
+        (message_index, block_index)
+        for message_index in range(last_assistant + 1, len(messages))
+        if messages[message_index].get("role") == "user"
+        and isinstance(messages[message_index].get("content"), list)
+        for block_index, block in enumerate(messages[message_index]["content"])
+        if isinstance(block, dict) and block.get("type") == "tool_result"
+    }
+
+
 def persist_large_output(tool_use_id: str, output: str) -> str:
     if len(output) <= PERSIST_THRESHOLD:
         return output
@@ -1959,20 +1976,9 @@ def snip_compact(messages: list, max_messages: int = 50) -> list:
 
 def micro_compact(messages: list) -> list:
     tool_results = collect_tool_results(messages)
-    latest_batch = []
+    unseen = unseen_tool_result_positions(messages)
-    for message in reversed(messages):
+    consumed = [entry for entry in tool_results if entry[:2] not in unseen]
-        content = message.get("content")
+    for _, _, block in consumed[:-KEEP_RECENT_TOOL_RESULTS]:
-        if message.get("role") != "user" or not isinstance(content, list):
-            continue
-        latest_batch = [
-            block for block in content
-            if isinstance(block, dict) and block.get("type") == "tool_result"
-        ]
-        if latest_batch:
-            break
-    latest_batch_ids = {id(block) for block in latest_batch}
-    older_results = [entry for entry in tool_results if id(entry[2]) not in latest_batch_ids]
-    for _, _, block in older_results[:-KEEP_RECENT_TOOL_RESULTS]:
         if len(str(block.get("content", ""))) > 120:
             block["content"] = "[Earlier tool result compacted. Re-run if needed.]"
     return messages

tests/test_compaction_tool_pairs.py

@@ -79,6 +79,16 @@ def tool_use_message(tool_id="tool-1"):
     }
 
 
+def tool_use_batch(*tool_ids):
+    return {
+        "role": "assistant",
+        "content": [
+            types.SimpleNamespace(type="tool_use", id=tool_id, name="bash")
+            for tool_id in tool_ids
+        ],
+    }
+
+
 def tool_result_message(tool_id="tool-1"):
     return {
         "role": "user",
@@ -123,18 +133,26 @@ def compaction_api(module):
 
 
 class CompactionToolPairTests(unittest.TestCase):
-    def test_micro_compact_keeps_latest_tool_result_batch(self):
+    def test_micro_compact_keeps_unseen_tool_result_batch(self):
         for name, path in MODULES.items():
             with self.subTest(name=name), tempfile.TemporaryDirectory() as tmp:
                 messages = [
+                    tool_use_message("old-1"),
                     long_tool_result_batch("old-1"),
+                    tool_use_message("old-2"),
                     long_tool_result_batch("old-2"),
+                    tool_use_message("old-3"),
                     long_tool_result_batch("old-3"),
+                    tool_use_message("old-4"),
                     long_tool_result_batch("old-4"),
-                    user_text(),
+                    tool_use_batch("latest-1", "latest-2", "latest-3", "latest-4"),
                     long_tool_result_batch(
                         "latest-1", "latest-2", "latest-3", "latest-4"
                     ),
+                    {"role": "user", "content": [
+                        {"type": "text", "text": "<task_notification>done</task_notification>"}
+                    ]},
+                    {"role": "user", "content": "<reminder>Update your todos.</reminder>"},
                 ]
                 module = load_module(f"{name}_micro_batch_under_test", path, Path(tmp))
                 compacted = compaction_api(module).micro_compact(messages)
@@ -150,6 +168,28 @@ class CompactionToolPairTests(unittest.TestCase):
                                 "latest-1", "latest-2", "latest-3", "latest-4"):
                     self.assertIn(f"{tool_id}: ", results[tool_id])
 
+    def test_micro_compact_releases_batch_after_model_consumes_it(self):
+        for name, path in MODULES.items():
+            with self.subTest(name=name), tempfile.TemporaryDirectory() as tmp:
+                messages = [
+                    tool_use_batch("seen-1", "seen-2", "seen-3", "seen-4"),
+                    long_tool_result_batch("seen-1", "seen-2", "seen-3", "seen-4"),
+                    assistant_text(),
+                    user_text(),
+                ]
+                module = load_module(f"{name}_consumed_batch_under_test", path, Path(tmp))
+                compacted = compaction_api(module).micro_compact(messages)
+                results = {
+                    block["tool_use_id"]: block["content"]
+                    for message in compacted
+                    if isinstance(message["content"], list)
+                    for block in message["content"]
+                    if isinstance(block, dict) and block.get("type") == "tool_result"
+                }
+                self.assertNotIn("seen-1: ", results["seen-1"])
+                for tool_id in ("seen-2", "seen-3", "seen-4"):
+                    self.assertIn(f"{tool_id}: ", results[tool_id])
+
     def test_snip_compact_keeps_head_tool_pair(self):
         messages = [
             user_text(),