fix: avoid re-injecting todo subjects after compression

tests/tools/test_read_loop_detection.py

@@ -285,22 +285,24 @@ def test_read_between_searches_resets_consecutive(self, _mock_ops):
 
 
 class TestTodoInjectionFiltering(unittest.TestCase):
-    """Verify that format_for_injection filters completed/cancelled todos."""
+    """Verify that format_for_injection avoids re-injecting todo subjects."""
 
-    def test_filters_completed_and_cancelled(self):
+    def test_filters_completed_cancelled_and_active_subjects(self):
         from tools.todo_tool import TodoStore
         store = TodoStore()
         store.write([
             {"id": "1", "content": "Read codebase", "status": "completed"},
-            {"id": "2", "content": "Write fix", "status": "in_progress"},
-            {"id": "3", "content": "Run tests", "status": "pending"},
+            {"id": "2", "content": "Write fix for client ACME", "status": "in_progress"},
+            {"id": "3", "content": "Run secret-token rotation tests", "status": "pending"},
             {"id": "4", "content": "Abandoned", "status": "cancelled"},
         ])
         injection = store.format_for_injection()
         self.assertNotIn("Read codebase", injection)
         self.assertNotIn("Abandoned", injection)
-        self.assertIn("Write fix", injection)
-        self.assertIn("Run tests", injection)
+        self.assertNotIn("Write fix for client ACME", injection)
+        self.assertNotIn("Run secret-token rotation tests", injection)
+        self.assertIn("1 in_progress", injection)
+        self.assertIn("1 pending", injection)
 
     def test_all_completed_returns_none(self):
         from tools.todo_tool import TodoStore
@@ -316,16 +318,18 @@ def test_empty_store_returns_none(self):
         store = TodoStore()
         self.assertIsNone(store.format_for_injection())
 
-    def test_all_active_included(self):
+    def test_all_active_injects_counts_only(self):
         from tools.todo_tool import TodoStore
         store = TodoStore()
         store.write([
             {"id": "1", "content": "Task A", "status": "pending"},
             {"id": "2", "content": "Task B", "status": "in_progress"},
         ])
         injection = store.format_for_injection()
-        self.assertIn("Task A", injection)
-        self.assertIn("Task B", injection)
+        self.assertNotIn("Task A", injection)
+        self.assertNotIn("Task B", injection)
+        self.assertIn("1 pending", injection)
+        self.assertIn("1 in_progress", injection)
 
 
 if __name__ == "__main__":

tests/tools/test_todo_tool.py

@@ -53,22 +53,21 @@ def test_empty_returns_none(self):
         store = TodoStore()
         assert store.format_for_injection() is None
 
-    def test_non_empty_has_markers(self):
+    def test_non_empty_injects_counts_without_subjects(self):
         store = TodoStore()
         store.write([
             {"id": "1", "content": "Do thing", "status": "completed"},
-            {"id": "2", "content": "Next", "status": "pending"},
-            {"id": "3", "content": "Working", "status": "in_progress"},
+            {"id": "2", "content": "Client ACME launch", "status": "pending"},
+            {"id": "3", "content": "Rotate secret token", "status": "in_progress"},
         ])
         text = store.format_for_injection()
-        # Completed items are filtered out of injection
-        assert "[x]" not in text
+        # Completed items are filtered out of injection.
         assert "Do thing" not in text
-        # Active items are included
-        assert "[ ]" in text
-        assert "[>]" in text
-        assert "Next" in text
-        assert "Working" in text
+        # Active subjects are not re-injected after compression.
+        assert "Client ACME launch" not in text
+        assert "Rotate secret token" not in text
+        assert "1 pending" in text
+        assert "1 in_progress" in text
         assert "context compression" in text.lower()
 
 

tools/todo_tool.py

@@ -4,8 +4,8 @@
 
 Provides an in-memory task list the agent uses to decompose complex tasks,
 track progress, and maintain focus across long conversations. The state
-lives on the AIAgent instance (one per session) and is re-injected into
-the conversation after context compression events.
+lives on the AIAgent instance (one per session) and a privacy-preserving
+active-task summary is re-injected after context compression events.
 
 Design:
 - Single `todo` tool: provide `todos` param to write, omit to read
@@ -89,37 +89,37 @@ def has_items(self) -> bool:
 
     def format_for_injection(self) -> Optional[str]:
         """
-        Render the todo list for post-compression injection.
+        Render a privacy-preserving task summary for post-compression injection.
 
-        Returns a human-readable string to append to the compressed
-        message history, or None if the list is empty.
+        Full todo subjects are already available to the model in the original
+        todo tool result. Re-injecting them after compression can resurface
+        sensitive project names, client context, credentials, or PII into fresh
+        model context. Preserve continuity by injecting counts only.
         """
         if not self._items:
             return None
 
-        # Status markers for compact display
-        markers = {
-            "completed": "[x]",
-            "in_progress": "[>]",
-            "pending": "[ ]",
-            "cancelled": "[~]",
-        }
-
-        # Only inject pending/in_progress items — completed/cancelled ones
-        # cause the model to re-do finished work after compression.
         active_items = [
             item for item in self._items
             if item["status"] in ("pending", "in_progress")
         ]
         if not active_items:
             return None
 
-        lines = ["[Your active task list was preserved across context compression]"]
-        for item in active_items:
-            marker = markers.get(item["status"], "[?]")
-            lines.append(f"- {marker} {item['id']}. {item['content']} ({item['status']})")
+        pending = sum(1 for item in active_items if item["status"] == "pending")
+        in_progress = sum(1 for item in active_items if item["status"] == "in_progress")
+
+        parts = []
+        if in_progress:
+            parts.append(f"{in_progress} in_progress")
+        if pending:
+            parts.append(f"{pending} pending")
+        summary = ", ".join(parts)
 
-        return "\n".join(lines)
+        return (
+            "[Your active task list was preserved across context compression: "
+            f"{summary}. Call the todo tool with no parameters to inspect details if needed.]"
+        )
 
     @staticmethod
     def _validate(item: Dict[str, Any]) -> Dict[str, str]:
@@ -212,6 +212,12 @@ def check_todo_requirements() -> bool:
         "Manage your task list for the current session. Use for complex tasks "
         "with 3+ steps or when the user provides multiple tasks. "
         "Call with no parameters to read the current list.\n\n"
+        "Privacy:\n"
+        "- Do not put secrets, credentials, raw PII, or highly sensitive "
+        "client/project identifiers in todo content.\n"
+        "- Todo content can be visible in the UI and stored in tool results.\n"
+        "- After context compression, only active todo counts are re-injected; "
+        "call the todo tool to inspect details if needed.\n\n"
         "Writing:\n"
         "- Provide 'todos' array to create/update items\n"
         "- merge=false (default): replace the entire list with a fresh plan\n"