Keep integration imports lightweight

src/integrations.py

@@ -9,8 +9,6 @@ from urllib.parse import urljoin, urlparse, urlunparse
 import httpx
 from fastapi import HTTPException
 
-from core.atomic_io import atomic_write_json
-from core.platform_compat import safe_chmod
 from src.secret_storage import decrypt, encrypt, is_encrypted
 from src.constants import DATA_DIR, INTEGRATIONS_FILE, SETTINGS_FILE
 
@@ -18,6 +16,26 @@ log = logging.getLogger(__name__)
 
 DATA_FILE = INTEGRATIONS_FILE
 
+
+def _atomic_write_json(path: str, data: Any, *, indent: Optional[int] = None) -> None:
+    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
+    tmp = f"{path}.tmp.{os.getpid()}"
+    with open(tmp, "w", encoding="utf-8") as f:
+        json.dump(data, f, indent=indent)
+        f.flush()
+        os.fsync(f.fileno())
+    os.replace(tmp, path)
+
+
+def _safe_chmod(path: str, mode: int) -> bool:
+    if os.name == "nt":
+        return False
+    try:
+        os.chmod(path, mode)
+        return True
+    except OSError:
+        return False
+
 # ---------------------------------------------------------------------------
 # Presets
 # ---------------------------------------------------------------------------
@@ -251,8 +269,8 @@ def load_integrations() -> List[Dict[str, Any]]:
 def save_integrations(integrations: List[Dict[str, Any]]) -> None:
     """Persist integrations list to disk with API keys encrypted at rest."""
     _ensure_data_dir()
-    atomic_write_json(DATA_FILE, _encrypt_integration_secrets(integrations), indent=2)
-    safe_chmod(DATA_FILE, 0o600)
+    _atomic_write_json(DATA_FILE, _encrypt_integration_secrets(integrations), indent=2)
+    _safe_chmod(DATA_FILE, 0o600)
 
 
 def get_integration(integration_id: str) -> Optional[Dict[str, Any]]:

src/secret_storage.py

@@ -24,7 +24,6 @@ from pathlib import Path
 
 from cryptography.fernet import Fernet, InvalidToken
 
-from core.platform_compat import safe_chmod
 from src.constants import APP_KEY_FILE
 
 logger = logging.getLogger(__name__)
@@ -34,6 +33,16 @@ _PREFIX = "enc:"
 _fernet: Fernet | None = None
 
 
+def _safe_chmod(path, mode: int) -> bool:
+    if os.name == "nt":
+        return False
+    try:
+        os.chmod(path, mode)
+        return True
+    except OSError:
+        return False
+
+
 def _load_or_create_key() -> bytes:
     if _KEY_PATH.exists():
         return _KEY_PATH.read_bytes()
@@ -42,7 +51,7 @@ def _load_or_create_key() -> bytes:
     _KEY_PATH.write_bytes(key)
     # POSIX: lock the key to 0o600. Windows: no-op (the user-profile data dir is
     # already ACL-restricted); safe_chmod swallows both cases.
-    safe_chmod(_KEY_PATH, 0o600)
+    _safe_chmod(_KEY_PATH, 0o600)
     logger.info(f"Generated new app key at {_KEY_PATH}")
     return key
 

tests/test_integrations_import_lightweight.py

@@ -0,0 +1,25 @@
+import json
+import os
+import subprocess
+import sys
+
+
+def test_integrations_import_does_not_initialize_core_package():
+    env = dict(os.environ)
+    env["PYTHONPATH"] = os.getcwd()
+    code = (
+        "import json, sys; "
+        "import src.integrations; "
+        "import src.secret_storage; "
+        "print(json.dumps({'core_loaded': 'core' in sys.modules}))"
+    )
+    result = subprocess.run(
+        [sys.executable, "-c", code],
+        cwd=os.getcwd(),
+        env=env,
+        text=True,
+        capture_output=True,
+        timeout=10,
+        check=True,
+    )
+    assert json.loads(result.stdout) == {"core_loaded": False}