Add cache_retention_days config for automatic cleanup of synced segments

Default 7 days. Values: positive int = days to keep, 0 = delete
immediately after confirmed sync, -1 = keep forever. Cleanup runs
at the end of each sync pass with triple-gated safety: synced_days
membership, age check, and per-segment server confirmation.

INSTALL.md

···
    }
    ```
 
+   **optional: cache retention.** by default, synced segments are deleted after 7 days. to change this, add `cache_retention_days` to config.json:
+   - positive number: keep synced segments for that many days (default: `7`)
+   - `0`: delete immediately after confirmed sync
+   - `-1`: keep forever (never auto-delete)
+
+   ```json
+   {
+     "server_url": "http://localhost:5015",
+     "key": "THE_API_KEY_FROM_STEP_3",
+     "stream": "HOSTNAME",
+     "cache_retention_days": 7
+   }
+   ```
+
 5. install and start the systemd user service:
    ```
    solstone-linux install-service

src/solstone_linux/cli.py

···
     else:
         print(f"Cache:  {captures_dir} (not created yet)")
 
+    # Retention policy
+    retention = config.cache_retention_days
+    if retention < 0:
+        print("Retain: forever")
+    elif retention == 0:
+        print("Retain: delete after sync")
+    else:
+        print(f"Retain: {retention} day(s)")
+
     # Synced days
     synced_path = config.state_dir / "synced_days.json"
     if synced_path.exists():

src/solstone_linux/config.py

···
         default_factory=lambda: list(DEFAULT_SYNC_RETRY_DELAYS)
     )
     sync_max_retries: int = DEFAULT_SYNC_MAX_RETRIES
+    cache_retention_days: int = 7
     base_dir: Path = DEFAULT_BASE_DIR
 
     @property
···
         config.sync_retry_delays = data["sync_retry_delays"]
     if "sync_max_retries" in data:
         config.sync_max_retries = data["sync_max_retries"]
+    try:
+        config.cache_retention_days = int(data.get("cache_retention_days", 7))
+    except (TypeError, ValueError):
+        config.cache_retention_days = 7
 
     return config
 
···
         "segment_interval": config.segment_interval,
         "sync_retry_delays": config.sync_retry_delays,
         "sync_max_retries": config.sync_max_retries,
+        "cache_retention_days": config.cache_retention_days,
     }
 
     config_path = config.config_path

src/solstone_linux/sync.py

···
 import json
 import logging
 import os
+import shutil
 import time
 from datetime import datetime, timedelta
 from pathlib import Path
···
             )
             self._save_synced_days()
 
+    async def _cleanup_synced_segments(self) -> None:
+        """Delete synced segments older than cache_retention_days.
+
+        Triple-gated safety:
+        1. Day must be in _synced_days (fully synced locally)
+        2. Segment must be older than retention threshold (unless retention=0)
+        3. Segment must be confirmed present on server (fresh query)
+        """
+        retention = self._config.cache_retention_days
+        if retention < 0:
+            return
+
+        captures_dir = self._config.captures_dir
+        if not captures_dir.exists():
+            return
+
+        today = datetime.now().strftime("%Y%m%d")
+        if retention > 0:
+            cutoff = (datetime.now() - timedelta(days=retention)).strftime("%Y%m%d")
+        else:
+            cutoff = today  # 0 means delete immediately — all days qualify
+
+        deleted_total = 0
+
+        for day_dir in sorted(captures_dir.iterdir()):
+            if not day_dir.is_dir():
+                continue
+
+            day = day_dir.name
+
+            if not self._running:
+                break
+
+            # Gate 1: day must be in synced_days
+            if day not in self._synced_days:
+                continue
+
+            # Gate 2: day must be old enough (unless retention=0)
+            if retention > 0 and day >= cutoff:
+                continue
+
+            # Don't clean today's segments
+            if day == today:
+                continue
+
+            # Gate 3: fresh server confirmation
+            server_segments = await asyncio.to_thread(
+                self._client.get_server_segments, day
+            )
+            if server_segments is None:
+                logger.warning("Cleanup: skipping day %s — server unreachable", day)
+                continue
+
+            server_keys: set[str] = set()
+            for seg in server_segments:
+                server_keys.add(seg.get("key", ""))
+                if "original_key" in seg:
+                    server_keys.add(seg["original_key"])
+
+            deleted_day = 0
+
+            for stream_dir in day_dir.iterdir():
+                if not stream_dir.is_dir():
+                    continue
+
+                for seg_dir in sorted(stream_dir.iterdir()):
+                    if not seg_dir.is_dir():
+                        continue
+
+                    name = seg_dir.name
+                    # Never touch incomplete or failed
+                    if name.endswith(".incomplete") or name.endswith(".failed"):
+                        continue
+
+                    if name not in server_keys:
+                        logger.warning(
+                            "Cleanup: keeping %s/%s — not confirmed on server",
+                            day,
+                            name,
+                        )
+                        continue
+
+                    shutil.rmtree(seg_dir)
+                    logger.info("Cleanup: deleted %s/%s", day, name)
+                    deleted_day += 1
+
+                # Remove empty stream dir
+                if stream_dir.is_dir() and not any(stream_dir.iterdir()):
+                    stream_dir.rmdir()
+
+            # Remove empty day dir
+            if day_dir.is_dir() and not any(day_dir.iterdir()):
+                day_dir.rmdir()
+
+            if deleted_day:
+                deleted_total += deleted_day
+
+        if deleted_total:
+            logger.info("Cleanup: deleted %d segment(s) total", deleted_total)
+
     def _circuit_threshold(self) -> int:
         """Get circuit breaker threshold based on last error type."""
         if self._last_error_type == ErrorType.AUTH:
···
             if day != today and not any_needed_upload:
                 self._synced_days.add(day)
                 self._save_synced_days()
+
+        # Cleanup old synced segments
+        if not self._circuit_open and self._running:
+            try:
+                await self._cleanup_synced_segments()
+            except Exception as e:
+                logger.error(f"Cleanup error: {e}", exc_info=True)
 
     def _collect_segments(self, captures_dir: Path) -> dict[str, list[Path]]:
         """Collect completed segments grouped by day."""

tests/test_config.py

···
         loaded = load_config(tmp_path)
         assert loaded.sync_retry_delays == [10, 60, 300]
         assert loaded.sync_max_retries == 5
+
+    def test_cache_retention_days_roundtrip(self, tmp_path: Path):
+        config = Config(base_dir=tmp_path)
+        config.cache_retention_days = 14
+        save_config(config)
+
+        loaded = load_config(tmp_path)
+        assert loaded.cache_retention_days == 14
+
+    def test_cache_retention_days_default(self, tmp_path: Path):
+        """Existing configs without cache_retention_days default to 7."""
+        config_dir = tmp_path / "config"
+        config_dir.mkdir(parents=True)
+        (config_dir / "config.json").write_text('{"server_url": "http://test"}')
+
+        loaded = load_config(tmp_path)
+        assert loaded.cache_retention_days == 7

tests/test_sync.py

···
         config.sync_max_retries = 1
         client = UploadClient(config)
         assert client._max_retries == 1
+
+
+class TestCleanupSyncedSegments:
+    """Test cache retention cleanup of synced segments."""
+
+    def _make_sync(self, tmp_path: Path, retention: int = 7) -> SyncService:
+        config = Config(base_dir=tmp_path)
+        config.cache_retention_days = retention
+        config.ensure_dirs()
+        client = UploadClient(config)
+        return SyncService(config, client)
+
+    def _create_segment(
+        self, captures_dir: Path, day: str, stream: str, name: str
+    ) -> Path:
+        seg_dir = captures_dir / day / stream / name
+        seg_dir.mkdir(parents=True, exist_ok=True)
+        (seg_dir / "screen.webm").write_bytes(b"\x00" * 100)
+        return seg_dir
+
+    @pytest.mark.asyncio
+    async def test_deletes_old_synced_confirmed(self, tmp_path: Path):
+        """Segments in synced_days + confirmed on server + old enough -> deleted."""
+        sync = self._make_sync(tmp_path, retention=7)
+        captures = sync._config.captures_dir
+
+        self._create_segment(captures, "20260101", "archon", "120000_300")
+        sync._synced_days.add("20260101")
+
+        server_response = [{"key": "120000_300"}]
+        with patch(
+            "asyncio.to_thread", new_callable=AsyncMock, return_value=server_response
+        ):
+            await sync._cleanup_synced_segments()
+
+        assert not (captures / "20260101" / "archon" / "120000_300").exists()
+
+    @pytest.mark.asyncio
+    async def test_keeps_unconfirmed_on_server(self, tmp_path: Path):
+        """Segments in synced_days + NOT on server -> not deleted."""
+        sync = self._make_sync(tmp_path, retention=7)
+        captures = sync._config.captures_dir
+
+        self._create_segment(captures, "20260101", "archon", "120000_300")
+        sync._synced_days.add("20260101")
+
+        server_response = [{"key": "999999_300"}]
+        with patch(
+            "asyncio.to_thread", new_callable=AsyncMock, return_value=server_response
+        ):
+            await sync._cleanup_synced_segments()
+
+        assert (captures / "20260101" / "archon" / "120000_300").exists()
+
+    @pytest.mark.asyncio
+    async def test_keeps_segments_not_in_synced_days(self, tmp_path: Path):
+        """Segments NOT in synced_days -> not deleted."""
+        sync = self._make_sync(tmp_path, retention=7)
+        captures = sync._config.captures_dir
+
+        self._create_segment(captures, "20260101", "archon", "120000_300")
+
+        with patch("asyncio.to_thread", new_callable=AsyncMock) as mock_thread:
+            await sync._cleanup_synced_segments()
+
+        assert (captures / "20260101" / "archon" / "120000_300").exists()
+        mock_thread.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_keeps_when_server_unreachable(self, tmp_path: Path):
+        """Server unreachable (returns None) -> nothing deleted."""
+        sync = self._make_sync(tmp_path, retention=7)
+        captures = sync._config.captures_dir
+
+        self._create_segment(captures, "20260101", "archon", "120000_300")
+        sync._synced_days.add("20260101")
+
+        with patch("asyncio.to_thread", new_callable=AsyncMock, return_value=None):
+            await sync._cleanup_synced_segments()
+
+        assert (captures / "20260101" / "archon" / "120000_300").exists()
+
+    @pytest.mark.asyncio
+    async def test_never_touches_incomplete_or_failed(self, tmp_path: Path):
+        """.incomplete and .failed segments are never deleted."""
+        sync = self._make_sync(tmp_path, retention=7)
+        captures = sync._config.captures_dir
+
+        self._create_segment(captures, "20260101", "archon", "120000.incomplete")
+        self._create_segment(captures, "20260101", "archon", "130000.failed")
+        self._create_segment(captures, "20260101", "archon", "140000_300")
+        sync._synced_days.add("20260101")
+
+        server_response = [{"key": "140000_300"}]
+        with patch(
+            "asyncio.to_thread", new_callable=AsyncMock, return_value=server_response
+        ):
+            await sync._cleanup_synced_segments()
+
+        assert (captures / "20260101" / "archon" / "120000.incomplete").exists()
+        assert (captures / "20260101" / "archon" / "130000.failed").exists()
+        assert not (captures / "20260101" / "archon" / "140000_300").exists()
+
+    @pytest.mark.asyncio
+    async def test_retention_negative_one_keeps_forever(self, tmp_path: Path):
+        """cache_retention_days = -1 -> nothing deleted."""
+        sync = self._make_sync(tmp_path, retention=-1)
+        captures = sync._config.captures_dir
+
+        self._create_segment(captures, "20260101", "archon", "120000_300")
+        sync._synced_days.add("20260101")
+
+        with patch("asyncio.to_thread", new_callable=AsyncMock) as mock_thread:
+            await sync._cleanup_synced_segments()
+
+        assert (captures / "20260101" / "archon" / "120000_300").exists()
+        mock_thread.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_retention_zero_deletes_immediately(self, tmp_path: Path):
+        """cache_retention_days = 0 -> deletes immediately (no age check)."""
+        sync = self._make_sync(tmp_path, retention=0)
+        captures = sync._config.captures_dir
+
+        from datetime import datetime, timedelta
+
+        yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
+
+        self._create_segment(captures, yesterday, "archon", "120000_300")
+        sync._synced_days.add(yesterday)
+
+        server_response = [{"key": "120000_300"}]
+        with patch(
+            "asyncio.to_thread", new_callable=AsyncMock, return_value=server_response
+        ):
+            await sync._cleanup_synced_segments()
+
+        assert not (captures / yesterday / "archon" / "120000_300").exists()
+
+    @pytest.mark.asyncio
+    async def test_never_cleans_today(self, tmp_path: Path):
+        """Today's segments are never cleaned, even with retention=0."""
+        sync = self._make_sync(tmp_path, retention=0)
+        captures = sync._config.captures_dir
+
+        from datetime import datetime
+
+        today = datetime.now().strftime("%Y%m%d")
+
+        self._create_segment(captures, today, "archon", "120000_300")
+        sync._synced_days.add(today)
+
+        with patch("asyncio.to_thread", new_callable=AsyncMock) as mock_thread:
+            await sync._cleanup_synced_segments()
+
+        assert (captures / today / "archon" / "120000_300").exists()
+        mock_thread.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_cleans_empty_dirs(self, tmp_path: Path):
+        """Empty stream and day dirs are removed after segment deletion."""
+        sync = self._make_sync(tmp_path, retention=7)
+        captures = sync._config.captures_dir
+
+        self._create_segment(captures, "20260101", "archon", "120000_300")
+        sync._synced_days.add("20260101")
+
+        server_response = [{"key": "120000_300"}]
+        with patch(
+            "asyncio.to_thread", new_callable=AsyncMock, return_value=server_response
+        ):
+            await sync._cleanup_synced_segments()
+
+        assert not (captures / "20260101" / "archon").exists()
+        assert not (captures / "20260101").exists()
+
+    @pytest.mark.asyncio
+    async def test_original_key_lookup(self, tmp_path: Path):
+        """Server segment with original_key should match local segment."""
+        sync = self._make_sync(tmp_path, retention=7)
+        captures = sync._config.captures_dir
+
+        self._create_segment(captures, "20260101", "archon", "120000_300")
+        sync._synced_days.add("20260101")
+
+        server_response = [{"key": "renamed_key", "original_key": "120000_300"}]
+        with patch(
+            "asyncio.to_thread", new_callable=AsyncMock, return_value=server_response
+        ):
+            await sync._cleanup_synced_segments()
+
+        assert not (captures / "20260101" / "archon" / "120000_300").exists()