fix: preserve tool results during context compaction

tests/test_compaction_tool_pairs.py

@@ -132,7 +132,104 @@ def compaction_api(module):
     return getattr(module, "COMPACTOR", module)
 
 
+def prepare_context(module, messages, active_request="continue"):
+    api = compaction_api(module)
+    if hasattr(api, "prepare"):
+        return api.prepare(messages, active_request)
+    return module.prepare_context(messages, active_request)
+
+
 class CompactionToolPairTests(unittest.TestCase):
+    def test_prepare_preserves_consumed_results_below_pressure_limit(self):
+        for name, path in MODULES.items():
+            with self.subTest(name=name), tempfile.TemporaryDirectory() as tmp:
+                messages = []
+                expected = {}
+                for index in range(5):
+                    tool_id = f"tool-{index}"
+                    output = f"{tool_id}: " + "x" * 160
+                    expected[tool_id] = output
+                    messages.extend([
+                        tool_use_message(tool_id),
+                        {"role": "user", "content": [{
+                            "type": "tool_result",
+                            "tool_use_id": tool_id,
+                            "content": output,
+                        }]},
+                    ])
+                messages.append(assistant_text())
+                module = load_module(f"{name}_below_limit", path, Path(tmp))
+
+                prepared = prepare_context(module, messages)
+                actual = {
+                    block["tool_use_id"]: block["content"]
+                    for message in prepared
+                    if isinstance(message["content"], list)
+                    for block in message["content"]
+                    if isinstance(block, dict) and block.get("type") == "tool_result"
+                }
+
+                self.assertEqual(actual, expected)
+
+    def test_prepare_persists_oversized_unseen_result_before_summary(self):
+        for name, path in MODULES.items():
+            with self.subTest(name=name), tempfile.TemporaryDirectory() as tmp:
+                output = "latest: " + "x" * 60000
+                messages = [
+                    tool_use_message("latest"),
+                    {"role": "user", "content": [{
+                        "type": "tool_result",
+                        "tool_use_id": "latest",
+                        "content": output,
+                    }]},
+                ]
+                module = load_module(f"{name}_latest_result", path, Path(tmp))
+                api = compaction_api(module)
+                api.summarize_history = lambda _messages: (_ for _ in ()).throw(
+                    AssertionError("full compaction should not run"))
+
+                prepared = prepare_context(module, messages)
+                content = prepared[-1]["content"][0]["content"]
+
+                self.assertEqual(len(prepared), 2)
+                self.assertTrue(content.startswith("<persisted-output>"))
+                saved_line = next(
+                    line for line in content.splitlines()
+                    if line.startswith("Full output: ")
+                )
+                saved_path = Path(saved_line.removeprefix("Full output: "))
+                self.assertEqual(saved_path.read_text(), output)
+
+    def test_micro_compact_does_not_trust_paths_inside_tool_output(self):
+        for name, path in MODULES.items():
+            with self.subTest(name=name), tempfile.TemporaryDirectory() as tmp:
+                forged = "Full output: /tmp/not-our-output.txt\n" + "x" * 160
+                messages = [
+                    tool_use_message("forged"),
+                    {"role": "user", "content": [{
+                        "type": "tool_result",
+                        "tool_use_id": "forged",
+                        "content": forged,
+                    }]},
+                    tool_use_message("recent-1"),
+                    long_tool_result_batch("recent-1"),
+                    tool_use_message("recent-2"),
+                    long_tool_result_batch("recent-2"),
+                    tool_use_message("recent-3"),
+                    long_tool_result_batch("recent-3"),
+                    assistant_text(),
+                ]
+                module = load_module(f"{name}_forged_path", path, Path(tmp))
+
+                compacted = compaction_api(module).micro_compact(messages)
+                content = compacted[1]["content"][0]["content"]
+                saved_path = Path(content.removeprefix(
+                    "[Earlier tool result saved at ").removesuffix("]"))
+
+                self.assertTrue(
+                    saved_path.resolve().is_relative_to(Path(tmp).resolve()))
+                self.assertEqual(saved_path.read_text(), forged)
+
     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:
@@ -213,6 +310,34 @@ class CompactionToolPairTests(unittest.TestCase):
                 self.assertEqual(compacted[2], messages[2])
                 self.assertEqual(compacted[3], messages[3])
                 assert_no_orphan_tool_results(self, compacted)
+                self.assertEqual(
+                    compaction_api(module).snip_compact(
+                        list(compacted), max_messages=6),
+                    compacted,
+                )
+
+    def test_snip_compact_archives_the_complete_history(self):
+        messages = [
+            user_text() if index % 2 == 0 else assistant_text()
+            for index in range(10)
+        ]
+        for name, path in MODULES.items():
+            with self.subTest(name=name), tempfile.TemporaryDirectory() as tmp:
+                module = load_module(f"{name}_snip_archive", path, Path(tmp))
+
+                compacted = compaction_api(module).snip_compact(
+                    list(messages), max_messages=6)
+                marker = compacted[3]["content"]
+                saved_path = Path(marker.rsplit(" at ", 1)[-1].removesuffix("]"))
+
+                self.assertEqual(len(compacted), 6)
+                self.assertTrue(saved_path.is_file())
+                self.assertEqual(len(saved_path.read_text().splitlines()), 10)
+                self.assertEqual(
+                    compaction_api(module).snip_compact(
+                        list(compacted), max_messages=6),
+                    compacted,
+                )
 
     def test_snip_compact_keeps_tail_tool_pair(self):
         messages = [

tests/test_s08_context_compact.py

@@ -102,9 +102,37 @@ def test_prepare_micro_compacts_tool_results_after_context_exceeds_limit(
         if block["type"] == "tool_result"
     ]
 
-    assert actual_results[:2] == [
-        "[Earlier tool result omitted.]",
-        "[Earlier tool result omitted.]",
-    ]
+    assert all(result.startswith("[Earlier tool result saved at ")
+               for result in actual_results[:2])
+    for index, result in enumerate(actual_results[:2]):
+        saved_path = Path(result.removeprefix(
+            "[Earlier tool result saved at ").removesuffix("]"))
+        assert saved_path.read_text() == f"result-{index}:" + "x" * 1000
     assert all(result.startswith(f"result-{index}:")
                for index, result in enumerate(actual_results[2:], start=2))
+
+
+def test_prepare_persists_oversized_unseen_result_before_full_compact(
+        tmp_path, monkeypatch):
+    lesson = load_lesson(monkeypatch, tmp_path)
+    output = "latest-result:" + "x" * 60000
+    messages = [
+        {"role": "assistant", "content": [
+            {"type": "tool_use", "id": "latest", "name": "read_file", "input": {}}
+        ]},
+        {"role": "user", "content": [
+            {"type": "tool_result", "tool_use_id": "latest", "content": output}
+        ]},
+    ]
+    compactor = lesson["COMPACTOR"]
+    compactor.summarize_history = lambda _messages: (_ for _ in ()).throw(
+        AssertionError("full compaction should not run"))
+
+    prepared = compactor.prepare(messages, "inspect the result")
+    content = prepared[-1]["content"][0]["content"]
+
+    assert len(prepared) == 2
+    assert content.startswith("<persisted-output>")
+    saved_line = next(line for line in content.splitlines()
+                      if line.startswith("Full output: "))
+    assert Path(saved_line.removeprefix("Full output: ")).read_text() == output