conductor(checkpoint): Checkpoint end of Phase 1

conductor/tracks/context_management_20260223/plan.md

@@ -1,10 +1,10 @@
 # Implementation Plan
 
 ## Phase 1: Context Memory and Token Visualization
-- [ ] Task: Implement token usage summary widget
+- [x] Task: Implement token usage summary widget e34ff7e
     - [ ] Sub-task: Write Tests
     - [ ] Sub-task: Implement Feature
-- [ ] Task: Expose history truncation controls in the Discussion panel
+- [x] Task: Expose history truncation controls in the Discussion panel 94fe904
     - [ ] Sub-task: Write Tests
     - [ ] Sub-task: Implement Feature
 - [ ] Task: Conductor - User Manual Verification 'Phase 1: Context Memory and Token Visualization' (Protocol in workflow.md)

gui.py

@@ -47,6 +47,30 @@ def hide_tk_root() -> Tk:
     root.wm_attributes("-topmost", True)
     return root
 
+def get_total_token_usage() -> dict:
+    """Returns aggregated token usage across the entire session from comms log."""
+    usage = {
+        "input_tokens": 0,
+        "output_tokens": 0,
+        "cache_read_input_tokens": 0,
+        "cache_creation_input_tokens": 0
+    }
+    for entry in ai_client.get_comms_log():
+        if entry.get("kind") == "response" and "usage" in entry.get("payload", {}):
+            u = entry["payload"]["usage"]
+            for k in usage.keys():
+                usage[k] += u.get(k, 0) or 0
+    return usage
+
+def truncate_entries(entries: list[dict], max_pairs: int) -> list[dict]:
+    """Truncates history to the last N pairs of User/AI messages."""
+    if max_pairs <= 0:
+        return []
+    target_count = max_pairs * 2
+    if len(entries) <= target_count:
+        return entries
+    return entries[-target_count:]
+
 
 # ------------------------------------------------------------------ comms rendering helpers
 
@@ -713,6 +737,15 @@ class App:
         for entry in entries:
             self._comms_entry_count += 1
             self._append_comms_entry(entry, self._comms_entry_count)
+        if entries:
+            self._update_token_usage()
+
+    def _update_token_usage(self):
+        if not dpg.does_item_exist("ai_token_usage"):
+            return
+        usage = get_total_token_usage()
+        total = usage["input_tokens"] + usage["output_tokens"]
+        dpg.set_value("ai_token_usage", f"Tokens: {total} (In: {usage['input_tokens']} Out: {usage['output_tokens']})")
 
     def _append_comms_entry(self, entry: dict, idx: int):
         if not dpg.does_item_exist("comms_scroll"):
@@ -1217,6 +1250,7 @@ class App:
         with self._pending_comms_lock:
             self._pending_comms.clear()
         self._comms_entry_count = 0
+        self._update_token_usage()
         if dpg.does_item_exist("comms_scroll"):
             dpg.delete_item("comms_scroll", children_only=True)
 
@@ -1319,6 +1353,12 @@ class App:
         self.disc_entries.clear()
         self._rebuild_disc_list()
 
+    def cb_disc_truncate(self):
+        pairs = dpg.get_value("disc_truncate_pairs") if dpg.does_item_exist("disc_truncate_pairs") else 2
+        self.disc_entries = truncate_entries(self.disc_entries, pairs)
+        self._rebuild_disc_list()
+        self._update_status(f"history truncated to {pairs} pairs")
+
     def cb_disc_collapse_all(self):
         for i, entry in enumerate(self.disc_entries):
             tag = f"disc_content_{i}"
@@ -1736,6 +1776,9 @@ class App:
                 dpg.add_button(label="+ Entry", callback=self.cb_disc_append_entry)
                 dpg.add_button(label="-All", callback=self.cb_disc_collapse_all)
                 dpg.add_button(label="+All", callback=self.cb_disc_expand_all)
+                dpg.add_text("Keep Pairs:", color=(160, 160, 160))
+                dpg.add_input_int(tag="disc_truncate_pairs", default_value=2, width=120, min_value=1)
+                dpg.add_button(label="Truncate", callback=self.cb_disc_truncate)
                 dpg.add_button(label="Clear All", callback=self.cb_disc_clear)
                 dpg.add_button(label="Save", callback=self.cb_disc_save)
             dpg.add_checkbox(
@@ -1864,6 +1907,8 @@ class App:
             with dpg.group(horizontal=True):
                 dpg.add_text("Status: idle", tag="ai_status", color=(200, 220, 160))
                 dpg.add_spacer(width=16)
+                dpg.add_text("Tokens: 0 (In: 0 Out: 0)", tag="ai_token_usage", color=(180, 255, 180))
+                dpg.add_spacer(width=16)
                 dpg.add_button(label="Clear", callback=self.cb_clear_comms)
             dpg.add_separator()
             with dpg.group(horizontal=True):

pyproject.toml

@@ -10,3 +10,8 @@ dependencies = [
     "anthropic",
     "tomli-w"
 ]
+
+[dependency-groups]
+dev = [
+    "pytest>=9.0.2",
+]

tests/test_history_truncation.py

@@ -0,0 +1,22 @@
+import pytest
+
+def test_history_truncation():
+    # A dummy test to fulfill the Red Phase for the history truncation controls.
+    # The new function in gui.py should be cb_disc_truncate_history or a related utility.
+    from project_manager import str_to_entry, entry_to_str
+    
+    entries = [
+        {"role": "User", "content": "1", "collapsed": False, "ts": "10:00:00"},
+        {"role": "AI", "content": "2", "collapsed": False, "ts": "10:01:00"},
+        {"role": "User", "content": "3", "collapsed": False, "ts": "10:02:00"},
+        {"role": "AI", "content": "4", "collapsed": False, "ts": "10:03:00"}
+    ]
+    
+    # We expect a new function truncate_entries(entries, max_pairs) to exist
+    from gui import truncate_entries
+    
+    truncated = truncate_entries(entries, max_pairs=1)
+    # Keeping the last pair (user + ai)
+    assert len(truncated) == 2
+    assert truncated[0]["content"] == "3"
+    assert truncated[1]["content"] == "4"

tests/test_token_usage.py

@@ -0,0 +1,35 @@
+import pytest
+
+def test_token_usage_aggregation():
+    # A dummy test to fulfill the Red Phase for the new token usage widget.
+    # We will implement a function in gui.py or ai_client.py to aggregate tokens.
+    from ai_client import _comms_log, clear_comms_log, _append_comms
+    
+    clear_comms_log()
+    
+    _append_comms("IN", "response", {
+        "usage": {
+            "input_tokens": 100,
+            "output_tokens": 50,
+            "cache_read_input_tokens": 10,
+            "cache_creation_input_tokens": 5
+        }
+    })
+    
+    _append_comms("IN", "response", {
+        "usage": {
+            "input_tokens": 200,
+            "output_tokens": 100,
+            "cache_read_input_tokens": 20,
+            "cache_creation_input_tokens": 0
+        }
+    })
+    
+    # We expect a new function get_total_token_usage() to exist
+    from gui import get_total_token_usage
+    
+    totals = get_total_token_usage()
+    assert totals["input_tokens"] == 300
+    assert totals["output_tokens"] == 150
+    assert totals["cache_read_input_tokens"] == 30
+    assert totals["cache_creation_input_tokens"] == 5