fix(manager): keep short delivery histories inline

loopx/capabilities/native_chat/manager_context.py

@@ -5,6 +5,7 @@
 """
 from __future__ import annotations
 
+import json
 import os
 from pathlib import Path
 import shutil
@@ -27,6 +28,11 @@
 )
 
 
+def _evidence_bytes(value: object) -> int:
+    # Match the frozen evidence serialization in Chat's real manager input.
+    return len(json.dumps(value, ensure_ascii=False).encode("utf-8"))
+
+
 def native_manager_context(context: dict[str, Any], adapter: object, *, runtime_root: Path,
                            capture_root: Path) -> dict[str, Any]:
     """Move only known supporting histories; unknown/current facts stay whole.
@@ -68,6 +74,13 @@ def native_manager_context(context: dict[str, Any], adapter: object, *, runtime_
             setting = registry.resolve("context_delivery").public_descriptor()["configuration_template"]
         if setting.get("antigravity_mode", "on") == "off":
             return context
+        # Even the reader instructions alone must fit inside the removed text.
+        # Avoid private capture IO entirely for common short histories.
+        inline_bytes = {index: _evidence_bytes({"status": history["status"], "deliveries": history["deliveries"]})
+                        for index, history in histories.items()}
+        reader_cost = len(NATIVE_MANAGER_READ_INSTRUCTION.encode("utf-8"))
+        if sum(inline_bytes.values()) <= reader_cost:
+            return context
         private_capture_uid()
         # A new 0700 directory in Chat's existing private store, not the selected
         # project's checkout. Do not change another directory's permissions.
@@ -76,19 +89,28 @@ def native_manager_context(context: dict[str, Any], adapter: object, *, runtime_
         path, digest = Path(recovery["path"]), recovery["sha256"]
         reference = write_context_read_reference(path, digest)
         rows = list(goals)
+        savings = 0
         for index, history in histories.items():
             pointer = f"/goals/{index}/recent_delivery_history"
-            rows[index] = {**goals[index], "recent_delivery_history": {
-                **history, "status": "not_read", "deliveries": [],
-                "read_reference": {"pointer": pointer,
-                    "command": context_capture_read_command(path, digest, reference_file=reference, select=[pointer])},
-            }}
-        return {**context, "goals": rows, "context_read": {
+            if "read_reference" in history:
+                continue  # Preserve a history with an existing binding.
+            read = {"pointer": pointer, "command": context_capture_read_command(
+                path, digest, reference_file=reference, select=[pointer])}
+            deferred = {"status": "not_read", "deliveries": [], "read_reference": read}
+            saved = inline_bytes[index] - _evidence_bytes(deferred)
+            if saved > 0:
+                savings += saved
+                rows[index] = {**goals[index], "recent_delivery_history": {**history, **deferred}}
+        reader = {
             "reference_file": str(reference),
             "recovery_command": context_capture_read_command(path, digest, reference_file=reference),
             "index_command": context_capture_read_command(path, digest, reference_file=reference,
                 list_fields=True, list_depth=2),
-        }}
+        }
+        if savings <= reader_cost + _evidence_bytes({"context_read": reader}):
+            shutil.rmtree(directory)
+            return context
+        return {**context, "goals": rows, "context_read": reader}
     except (OSError, ValueError, TypeError):
         if directory is not None:
             shutil.rmtree(directory, ignore_errors=True)

tests/test_manager_native_context.py

@@ -187,3 +187,27 @@ def test_codex_and_antigravity_context_preferences_operate_independently(
         assert result is not context
         assert result["goals"][0]["recent_delivery_history"]["status"] == "not_read"
         assert _read(result["context_read"]["recovery_command"]) == original
+
+def test_short_history_stays_inline_and_mixed_history_keeps_cheap_rows(tmp_path):
+    from loopx.capabilities.native_chat.manager_input import manager_turn_input
+
+    adapter = _adapter()
+    small = _context()
+    small["goals"][0]["recent_delivery_history"]["deliveries"] = [{"summary": "Export accepted"}]
+    assert native_manager_context(small, adapter, runtime_root=tmp_path, capture_root=tmp_path) is small
+    assert not list(tmp_path.iterdir())
+    prior = tmp_path / "previous-capture"
+    prior.mkdir()
+    medium = deepcopy(small)
+    medium["goals"][0]["recent_delivery_history"]["deliveries"] = [{"summary": "history" * 170}]
+    assert native_manager_context(medium, adapter, runtime_root=tmp_path, capture_root=tmp_path) is medium
+    assert list(tmp_path.iterdir()) == [prior], "discard only this losing candidate"
+    mixed = _context()
+    mixed["goals"].extend(small["goals"])
+    result = native_manager_context(mixed, adapter, runtime_root=tmp_path, capture_root=tmp_path)
+    assert result["goals"][1] == small["goals"][0]
+    assert result["goals"][0]["recent_delivery_history"]["status"] == "not_read"
+    assert _read(result["context_read"]["recovery_command"]) == mixed
+    def wire(context):
+        return manager_turn_input(adapter, "Status?", json.dumps(context, ensure_ascii=False)).encode()
+    assert len(wire(result)) < len(wire(mixed))