Compare commits

...

2 Commits

4 changed files with 108 additions and 9 deletions

View File

@@ -84,7 +84,7 @@ This file tracks all major tracks for the project. Each track has its own detail
19. [ ] **Track: Manual UX Validation & Review** 19. [ ] **Track: Manual UX Validation & Review**
*Link: [./tracks/manual_ux_validation_20260302/](./tracks/manual_ux_validation_20260302/)* *Link: [./tracks/manual_ux_validation_20260302/](./tracks/manual_ux_validation_20260302/)*
13. [x] **Track: Enhanced Context Control & Cache Awareness** 20. [x] **Track: Enhanced Context Control & Cache Awareness**
*Link: [./tracks/enhanced_context_control_20260307/](./tracks/enhanced_context_control_20260307/)* *Link: [./tracks/enhanced_context_control_20260307/](./tracks/enhanced_context_control_20260307/)*
--- ---

View File

@@ -1012,6 +1012,8 @@ class App:
if imgui.button("Refresh Registry"): if imgui.button("Refresh Registry"):
self._log_registry = log_registry.LogRegistry(str(paths.get_logs_dir() / "log_registry.toml")) self._log_registry = log_registry.LogRegistry(str(paths.get_logs_dir() / "log_registry.toml"))
imgui.same_line() imgui.same_line()
if imgui.button("Force Prune Logs"):
self.controller.event_queue.put("gui_task", {"action": "click", "item": "btn_prune_logs"})
registry = self._log_registry registry = self._log_registry
sessions = registry.data sessions = registry.data
@@ -1066,8 +1068,6 @@ class App:
) )
imgui.end_table() imgui.end_table()
if imgui.button("Force Prune Logs"):
self.controller.event_queue.put("gui_task", {"action": "click", "item": "btn_prune_logs"})
imgui.end() imgui.end()
def _render_files_panel(self) -> None: def _render_files_panel(self) -> None:

View File

@@ -222,6 +222,7 @@ class LogRegistry:
""" """
Retrieves a list of sessions that are older than a specific cutoff time Retrieves a list of sessions that are older than a specific cutoff time
and are not marked as whitelisted. and are not marked as whitelisted.
Also includes non-whitelisted sessions that are empty (message_count=0 or size_kb=0).
Args: Args:
cutoff_datetime (datetime): The threshold time for identifying old sessions. cutoff_datetime (datetime): The threshold time for identifying old sessions.
@@ -241,11 +242,19 @@ class LogRegistry:
else: else:
start_time = start_time_raw start_time = start_time_raw
is_whitelisted = session_data.get('whitelisted', False) is_whitelisted = session_data.get('whitelisted', False)
if start_time is not None and start_time < cutoff_datetime and not is_whitelisted:
old_sessions.append({ # Heuristic: also include non-whitelisted sessions that have 0 messages or 0 KB size
'session_id': session_id, metadata = session_data.get('metadata', {}) or {}
'path': session_data.get('path'), message_count = metadata.get('message_count', -1)
'start_time': start_time_raw size_kb = metadata.get('size_kb', -1)
}) is_empty = (message_count == 0 or size_kb == 0)
if start_time is not None and not is_whitelisted:
if start_time < cutoff_datetime or is_empty:
old_sessions.append({
'session_id': session_id,
'path': session_data.get('path'),
'start_time': start_time_raw
})
return old_sessions return old_sessions

View File

@@ -0,0 +1,90 @@
import unittest
import tempfile
import os
import shutil
from datetime import datetime, timedelta
from src.log_registry import LogRegistry
from src.log_pruner import LogPruner
class TestLogPruningHeuristic(unittest.TestCase):
def setUp(self) -> None:
self.temp_dir = tempfile.TemporaryDirectory()
self.registry_path = os.path.join(self.temp_dir.name, "registry.toml")
self.logs_dir = os.path.join(self.temp_dir.name, "logs")
os.makedirs(self.logs_dir, exist_ok=True)
self.registry = LogRegistry(self.registry_path)
self.pruner = LogPruner(self.registry, self.logs_dir)
def tearDown(self) -> None:
self.temp_dir.cleanup()
def _create_session(self, session_id, start_time, message_count=None, size_kb=None, whitelisted=False):
path = os.path.join(self.logs_dir, session_id)
os.makedirs(path, exist_ok=True)
# Create a dummy file
with open(os.path.join(path, "comms.log"), "w") as f:
f.write("test content")
self.registry.register_session(session_id, path, start_time)
if message_count is not None or size_kb is not None or whitelisted:
self.registry.update_session_metadata(
session_id,
message_count=message_count if message_count is not None else 10,
errors=0,
size_kb=size_kb if size_kb is not None else 10,
whitelisted=whitelisted,
reason="Test"
)
def test_get_old_non_whitelisted_sessions_includes_empty_sessions(self) -> None:
now = datetime.now()
cutoff_time = now - timedelta(days=7)
# 1. Old, not whitelisted (should be included)
self._create_session("old_nw", now - timedelta(days=10))
# 2. Recent, not whitelisted, but empty (message_count=0) (SHOULD be included based on new heuristic)
self._create_session("recent_empty_msgs", now - timedelta(days=1), message_count=0)
# 3. Recent, not whitelisted, but empty (size_kb=0) (SHOULD be included based on new heuristic)
self._create_session("recent_empty_size", now - timedelta(days=1), size_kb=0)
# 4. Recent, not whitelisted, NOT empty (should NOT be included)
self._create_session("recent_not_empty", now - timedelta(days=1), message_count=5, size_kb=5)
# 5. Old, whitelisted (should NOT be included)
self._create_session("old_w", now - timedelta(days=10), whitelisted=True)
sessions = self.registry.get_old_non_whitelisted_sessions(cutoff_time)
session_ids = {s['session_id'] for s in sessions}
self.assertIn("old_nw", session_ids)
self.assertIn("recent_empty_msgs", session_ids)
self.assertIn("recent_empty_size", session_ids)
self.assertNotIn("recent_not_empty", session_ids)
self.assertNotIn("old_w", session_ids)
def test_prune_removes_empty_sessions_regardless_of_age(self) -> None:
now = datetime.now()
# Create a session that is recent but empty
session_id = "recent_empty"
session_path = os.path.join(self.logs_dir, session_id)
os.makedirs(session_path, exist_ok=True)
# Actual file size 0
with open(os.path.join(session_path, "comms.log"), "w") as f:
pass
self.registry.register_session(session_id, session_path, now - timedelta(hours=1))
self.registry.update_session_metadata(session_id, message_count=0, errors=0, size_kb=0, whitelisted=False, reason="Empty")
self.assertTrue(os.path.exists(session_path))
# Prune with max_age_days=30 (so 1 hour old is NOT "old" by age)
self.pruner.prune(max_age_days=30, min_size_kb=1)
self.assertFalse(os.path.exists(session_path))
self.assertNotIn(session_id, self.registry.data)
if __name__ == '__main__':
unittest.main()