From b832d43928b568966e15da338d7601a2972bf441 Mon Sep 17 00:00:00 2001 From: Carl Kibler Date: Wed, 1 Jul 2026 14:31:11 -0600 Subject: [PATCH] fix(doctor): add --fix pruning for disk-usage, skipping recent sessions check_disk_usage had no fix function, so `cozempic doctor --fix` skipped it and left only a generic "treat " hint. This wires fix_disk_usage into ALL_CHECKS with a conservative, age-gated selection: - `cozempic doctor --fix` now applies the gentle prescription to sessions that are both larger than 5MB and untouched for more than 7 days, writing a .bak per session first. - Recent sessions (modified within 7 days) are skipped so an active or in-flight session is never rewritten; the fix reports treated vs skipped counts. - The check's fix_description now shows the candidate count and threshold instead of the generic placeholder. No change to the check's pass/fail status or to any other check. Co-Authored-By: Claude --- src/cozempic/doctor.py | 61 ++++++++++++++++++++++++++++++++++++++++-- tests/test_doctor.py | 57 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 116 insertions(+), 2 deletions(-) diff --git a/src/cozempic/doctor.py b/src/cozempic/doctor.py index 3cb6468d..303ca827 100644 --- a/src/cozempic/doctor.py +++ b/src/cozempic/doctor.py @@ -205,8 +205,13 @@ def fix_stale_backups() -> str: return f"Deleted {len(bak_files)} backup file(s), freed {total / 1024 / 1024:.1f}MB" +_DISK_BLOAT_MIN_BYTES = 5 * 1024 * 1024 # sessions smaller than this aren't worth touching +_DISK_BLOAT_RECENT_SECS = 7 * 24 * 3600 # skip sessions modified within this window + + def check_disk_usage() -> CheckResult: """Check total Claude session disk usage.""" + import os, time as _time sessions = find_sessions() total = sum(s["size"] for s in sessions) @@ -217,11 +222,63 @@ def check_disk_usage() -> CheckResult: else: status = "issue" + now = _time.time() + candidates = [ + s for s in sessions + if s["size"] >= _DISK_BLOAT_MIN_BYTES + and now - os.path.getmtime(s["path"]) >= _DISK_BLOAT_RECENT_SECS + ] + fix_desc = None + if status != "ok": + fix_desc = ( + f"Run 'cozempic doctor --fix' to apply gentle pruning to " + f"{len(candidates)} session(s) >{_DISK_BLOAT_MIN_BYTES // (1024*1024)}MB " + f"and older than 7 days (skips recent sessions)" + ) + return CheckResult( name="disk-usage", status=status, message=f"{len(sessions)} sessions using {total / 1024 / 1024:.1f}MB total", - fix_description="Run: cozempic treat -rx standard --execute" if status != "ok" else None, + fix_description=fix_desc, + ) + + +def fix_disk_usage() -> str: + """Apply gentle pruning to old, bloated sessions — skips recent sessions.""" + import os, time as _time + from .session import load_messages, save_messages + from .registry import PRESCRIPTIONS + from .executor import run_prescription + + sessions = find_sessions() + now = _time.time() + candidates = [ + s for s in sessions + if s["size"] >= _DISK_BLOAT_MIN_BYTES + and now - os.path.getmtime(s["path"]) >= _DISK_BLOAT_RECENT_SECS + ] + + if not candidates: + return "No bloated stale sessions found." + + treated = 0 + skipped = 0 + for sess in candidates: + try: + messages = load_messages(sess["path"]) + pruned, _ = run_prescription(messages, PRESCRIPTIONS["gentle"], {}) + save_messages(sess["path"], pruned, create_backup=True) + treated += 1 + except Exception as exc: + print(f" Warning: could not treat {sess['session_id'][:8]}: {exc}") + skipped += 1 + + total_skipped = len(sessions) - len(candidates) + return ( + f"Treated {treated} session(s) with gentle prescription. " + f"{total_skipped} session(s) skipped (too small or modified within 7 days). " + f"Backups created." ) @@ -1492,7 +1549,7 @@ def fix_unresumable_session() -> str: ("oversized-sessions", check_oversized_sessions, None), ("stale-backups", check_stale_backups, fix_stale_backups), ("stale-tmp-artifacts", check_stale_tmp_artifacts, fix_stale_tmp_artifacts), - ("disk-usage", check_disk_usage, None), + ("disk-usage", check_disk_usage, fix_disk_usage), ] diff --git a/tests/test_doctor.py b/tests/test_doctor.py index 0f6356f3..fc11b8c3 100644 --- a/tests/test_doctor.py +++ b/tests/test_doctor.py @@ -16,9 +16,11 @@ check_hooks_trust_flag, check_orphaned_tool_results, check_oversized_sessions, + check_disk_usage, check_stale_tmp_artifacts, check_zombie_teams, fix_claude_json_corruption, + fix_disk_usage, fix_hooks_trust_flag, fix_stale_tmp_artifacts, ) @@ -526,5 +528,60 @@ def test_fix_description_sorted_largest_first(self): self.assertLess(idx_large, idx_small) +class TestDiskUsageFix(unittest.TestCase): + + def _make_session(self, session_id: str, size_mb: int, path: str) -> dict: + return {"session_id": session_id, "path": path, "size": size_mb * 1024 * 1024} + + def test_disk_usage_fix_description_counts_only_old_large_sessions(self): + sessions = [ + self._make_session("oldlarge12345678", 600, "/tmp/old-large.jsonl"), + self._make_session("recentlarge1234", 8, "/tmp/recent-large.jsonl"), + self._make_session("oldsmall12345678", 1, "/tmp/old-small.jsonl"), + ] + + def fake_mtime(path): + return 0 if path != "/tmp/recent-large.jsonl" else 10_000_000 + + with ( + patch("cozempic.doctor.find_sessions", return_value=sessions), + patch("os.path.getmtime", side_effect=fake_mtime), + patch("time.time", return_value=10_000_000), + ): + result = check_disk_usage() + + self.assertEqual(result.status, "warning") + self.assertIn("1 session(s) >5MB", result.fix_description) + self.assertIn("skips recent sessions", result.fix_description) + + def test_fix_disk_usage_treats_only_old_large_sessions_with_backups(self): + sessions = [ + self._make_session("oldlarge12345678", 600, "/tmp/old-large.jsonl"), + self._make_session("recentlarge1234", 8, "/tmp/recent-large.jsonl"), + ] + + def fake_mtime(path): + return 0 if path == "/tmp/old-large.jsonl" else 10_000_000 + + with ( + patch("cozempic.doctor.find_sessions", return_value=sessions), + patch("os.path.getmtime", side_effect=fake_mtime), + patch("time.time", return_value=10_000_000), + patch("cozempic.session.load_messages", return_value=[{"type": "user"}]) as load_messages, + patch("cozempic.executor.run_prescription", return_value=([{"type": "user", "pruned": True}], None)) as run_prescription, + patch("cozempic.session.save_messages") as save_messages, + ): + message = fix_disk_usage() + + self.assertIn("Treated 1 session", message) + load_messages.assert_called_once_with("/tmp/old-large.jsonl") + from cozempic.registry import PRESCRIPTIONS + self.assertIs(run_prescription.call_args.args[1], PRESCRIPTIONS["gentle"]) + save_messages.assert_called_once_with( + "/tmp/old-large.jsonl", + [{"type": "user", "pruned": True}], + create_backup=True, + ) + if __name__ == "__main__": unittest.main()