Token store Step 4: remove legacy file migration

Remove _LEGACY_BASE_DIR, _migrate_from_files, _legacy_token_dir from
TokenStore. save_token no longer returns a legacy path. has_token and
load_token read SQLite directly (garth-cache fallback remains for Garmin).
_garmin_token_dir in tools.py drops legacy filesystem fallback. 4 tests
added, 1 legacy test removed.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: a-deal

engine/gateway/token_store.py

@@ -21,7 +21,6 @@
 
 logger = logging.getLogger("health-engine.token_store")
 
-_LEGACY_BASE_DIR = Path(os.path.expanduser("~/.config/health-engine/tokens"))
 _KEY_PATH = Path(os.path.expanduser("~/.config/health-engine/token.key"))
 # Garth needs a real directory to load/dump tokens. We use a stable per-user
 # path so garth's token refresh can persist across calls within a process.
@@ -69,13 +68,12 @@ def _now_iso() -> str:
 class TokenStore:
     """Manage wearable auth tokens per service and user.
 
-    Reads/writes tokens in SQLite (wearable_token table).
-    Falls back to legacy file paths for migration.
+    Primary storage: SQLite (wearable_token table).
+    Garmin tokens fall back to garth-cache directory for import.
     """
 
-    def __init__(self, base_dir: str | Path | None = None):
-        # Legacy base_dir kept for backward compat during migration
-        self._legacy_dir = Path(base_dir) if base_dir else _LEGACY_BASE_DIR
+    def __init__(self, base_dir=None):
+        # base_dir is accepted but ignored (legacy compat for callers).
         self._fernet = _get_fernet()
 
     def _encrypt(self, data: bytes) -> bytes:
@@ -139,45 +137,15 @@ def _db_list_tokens(self, user_id: str, service: str) -> list[str]:
         ).fetchall()
         return [r["token_name"] for r in rows]
 
-    # --- Legacy file migration ---
-
-    def _legacy_token_dir(self, service: str, user_id: str) -> Path:
-        return self._legacy_dir / service / user_id
-
-    def _migrate_from_files(self, user_id: str, service: str):
-        """One-time migration: copy file tokens into SQLite, then leave files as backup."""
-        legacy_dir = self._legacy_token_dir(service, user_id)
-        if not legacy_dir.exists() or legacy_dir.is_symlink():
-            return
-        if self._db_has_tokens(user_id, service):
-            return  # Already migrated
-
-        migrated = 0
-        for fpath in legacy_dir.iterdir():
-            if not fpath.is_file():
-                continue
-            raw = fpath.read_bytes()
-            # Decrypt if file was Fernet-encrypted, then re-encrypt for DB
-            if self._fernet and raw.startswith(b"gAAAAA"):
-                raw = self._fernet.decrypt(raw)
-            self._db_save_token(user_id, service, fpath.name, raw)
-            migrated += 1
-
-        if migrated:
-            logger.info("Migrated %d token files for %s/%s from disk to SQLite", migrated, service, user_id)
-
     # --- Public API ---
 
-    def save_token(self, service: str, user_id: str, data: dict) -> Path:
-        """Save token data to SQLite. Returns a (legacy) directory path for backward compat."""
+    def save_token(self, service: str, user_id: str, data: dict):
+        """Save token data to SQLite."""
         raw = json.dumps(data, indent=2).encode()
         self._db_save_token(user_id, service, "token.json", raw)
-        # Return legacy path for any callers that expect it
-        return self._legacy_token_dir(service, user_id)
 
     def load_token(self, service: str, user_id: str) -> dict | None:
-        """Load token data from SQLite. Falls back to files + migrates."""
-        self._migrate_from_files(user_id, service)
+        """Load token data from SQLite."""
         raw = self._db_load_token(user_id, service, "token.json")
         if raw is None:
             return None
@@ -186,11 +154,9 @@ def load_token(self, service: str, user_id: str) -> dict | None:
     def has_token(self, service: str, user_id: str) -> bool:
         """Check if tokens exist for a service/user combo.
 
-        Checks SQLite first (after legacy migration). For Garmin, falls back
-        to garth-cache directory: if tokens exist there but not in SQLite,
-        imports them and returns True.
+        For Garmin, falls back to garth-cache directory: if tokens exist
+        there but not in SQLite, imports them and returns True.
         """
-        self._migrate_from_files(user_id, service)
         if self._db_has_tokens(user_id, service):
             return True
         # Fallback: garth-cache may have tokens from direct garth usage
@@ -220,14 +186,12 @@ def garmin_token_dir(self, user_id: str = "default") -> Path:
 
         Garmin tokens are stored as garth dumps (oauth1_token.json,
         oauth2_token.json). This method:
-        1. Migrates legacy file tokens to SQLite if needed
-        2. Writes SQLite tokens to a cache directory for garth to read
-        3. Returns the cache directory path
+        1. Writes SQLite tokens to a cache directory for garth to read
+        2. Returns the cache directory path
 
         After garth modifies tokens (e.g. refresh), call sync_garmin_tokens()
         to write changes back to SQLite.
         """
-        self._migrate_from_files(user_id, "garmin")
 
         cache_dir = _GARTH_CACHE_DIR / user_id
         cache_dir.mkdir(parents=True, exist_ok=True)

mcp_server/tools.py

@@ -147,24 +147,13 @@ def _get_token_store():
 def _garmin_token_dir(user_id: str | None = None) -> str | None:
     """Resolve per-user Garmin token directory. Returns None if no tokens exist.
 
-    Uses SQLite-backed TokenStore (with automatic migration from legacy files).
-    NEVER falls back to another user's tokens.
+    Uses SQLite-backed TokenStore. NEVER falls back to another user's tokens.
     """
     ts = _get_token_store()
     uid = user_id if user_id and user_id != "default" else "default"
 
-    if uid != "default":
-        if ts.has_token("garmin", uid):
-            return str(ts.garmin_token_dir(uid))
-        return None
-
-    # Legacy CLI path (no user_id context)
-    if ts.has_token("garmin", "default"):
-        return str(ts.garmin_token_dir("default"))
-    # Check old-style legacy path as absolute fallback for CLI
-    legacy = Path(os.path.expanduser("~/.config/health-engine/garmin-tokens"))
-    if legacy.exists() and any(legacy.iterdir()):
-        return str(legacy)
+    if ts.has_token("garmin", uid):
+        return str(ts.garmin_token_dir(uid))
     return None
 
 

tests/test_garmin.py

@@ -466,13 +466,12 @@ def test_db(self, tmp_path):
     def store(self, tmp_path, test_db, monkeypatch):
         garth_cache = tmp_path / "garth-cache"
         monkeypatch.setattr("engine.gateway.token_store._GARTH_CACHE_DIR", garth_cache)
-        monkeypatch.setattr("engine.gateway.token_store._LEGACY_BASE_DIR", tmp_path / "legacy")
         monkeypatch.setattr(
             "engine.gateway.token_store._get_db",
             lambda: __import__("engine.gateway.db", fromlist=["get_db"]).get_db(test_db),
         )
         from engine.gateway.token_store import TokenStore
-        return TokenStore(base_dir=tmp_path / "legacy")
+        return TokenStore()
 
     def test_init_accepts_token_store(self, store):
         """GarminClient.__init__ accepts token_store and user_id params."""

tests/test_token_store.py

@@ -35,22 +35,19 @@ def test_db(tmp_path, monkeypatch):
 @pytest.fixture
 def store_encrypted(token_dir, key_path, test_db, monkeypatch):
     """TokenStore with Fernet encryption enabled."""
-    monkeypatch.setattr("engine.gateway.token_store._LEGACY_BASE_DIR", token_dir)
     monkeypatch.setattr("engine.gateway.token_store._KEY_PATH", key_path)
     monkeypatch.delenv("HE_TOKEN_KEY", raising=False)
     from engine.gateway.token_store import TokenStore
-    return TokenStore(base_dir=token_dir)
+    return TokenStore()
 
 
 @pytest.fixture
 def store_no_crypto(token_dir, test_db, monkeypatch):
     """TokenStore with cryptography unavailable."""
-    monkeypatch.setattr("engine.gateway.token_store._LEGACY_BASE_DIR", token_dir)
-
     import engine.gateway.token_store as mod
     monkeypatch.setattr(mod, "_get_fernet", lambda: None)
     from engine.gateway.token_store import TokenStore
-    store = TokenStore(base_dir=token_dir)
+    store = TokenStore()
     store._fernet = None
     return store
 
@@ -84,17 +81,6 @@ def test_encrypted_blob_is_not_plaintext(store_encrypted, test_db):
     assert raw.startswith(b"gAAAAA")  # Fernet prefix
 
 
-def test_legacy_file_migration(store_encrypted, token_dir):
-    """Plaintext JSON files from before SQLite migration can be loaded via migration."""
-    td = token_dir / "google-calendar" / "legacy"
-    td.mkdir(parents=True)
-    data = {"access_token": "old_token", "refresh_token": "old_refresh"}
-    (td / "token.json").write_text(json.dumps(data))
-
-    loaded = store_encrypted.load_token("google-calendar", "legacy")
-    assert loaded == data
-
-
 def test_no_crypto_roundtrip(store_no_crypto):
     """Without cryptography, tokens still save and load via SQLite."""
     data = {"access_token": "plain", "refresh_token": "text"}
@@ -114,9 +100,8 @@ def test_load_missing_returns_none(store_encrypted):
     assert store_encrypted.load_token("nonexistent", "nobody") is None
 
 
-def test_key_auto_generated(key_path, token_dir, monkeypatch):
+def test_key_auto_generated(key_path, monkeypatch):
     """Key file is auto-generated on first use."""
-    monkeypatch.setattr("engine.gateway.token_store._LEGACY_BASE_DIR", token_dir)
     monkeypatch.setattr("engine.gateway.token_store._KEY_PATH", key_path)
     monkeypatch.delenv("HE_TOKEN_KEY", raising=False)
 
@@ -129,12 +114,11 @@ def test_key_auto_generated(key_path, token_dir, monkeypatch):
     assert mode == "600"
 
 
-def test_env_var_key(token_dir, monkeypatch):
+def test_env_var_key(monkeypatch):
     """HE_TOKEN_KEY env var is used when set."""
     from cryptography.fernet import Fernet
     key = Fernet.generate_key()
     monkeypatch.setenv("HE_TOKEN_KEY", key.decode())
-    monkeypatch.setattr("engine.gateway.token_store._LEGACY_BASE_DIR", token_dir)
 
     from engine.gateway.token_store import _get_fernet
     f = _get_fernet()
@@ -191,3 +175,34 @@ def test_has_token_garth_cache_non_garmin(store_encrypted, garth_cache_dir):
 
     # oura should NOT use garth-cache fallback
     assert not store_encrypted.has_token("oura", "andrew")
+
+
+# --- Step 4: Legacy removal verification ---
+
+
+def test_no_legacy_migration_attribute(store_encrypted):
+    """TokenStore should not have _migrate_from_files."""
+    assert not hasattr(store_encrypted, "_migrate_from_files")
+
+
+def test_no_legacy_base_dir_constant():
+    """_LEGACY_BASE_DIR should not exist in token_store module."""
+    import engine.gateway.token_store as mod
+    assert not hasattr(mod, "_LEGACY_BASE_DIR")
+
+
+def test_has_token_only_checks_sqlite_and_garth_cache(store_encrypted, token_dir):
+    """has_token should NOT check legacy file paths."""
+    # Create a token file at the old legacy location
+    legacy = token_dir / "oura" / "andrew"
+    legacy.mkdir(parents=True)
+    (legacy / "token.json").write_text('{"token": "legacy"}')
+
+    # Should not find it since legacy migration is removed
+    assert not store_encrypted.has_token("oura", "andrew")
+
+
+def test_save_token_returns_none(store_encrypted):
+    """save_token should not return a legacy directory path."""
+    result = store_encrypted.save_token("svc", "u1", {"token": "val"})
+    assert result is None