fix(doctor): add --fix pruning for disk-usage, skipping recent sessions

src/cozempic/doctor.py

@@ -205,8 +205,13 @@
     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 @@
     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 <session> -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 @@
     ("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),
 ]
 
 

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 @@
         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()