diff --git a/notebooks/ai_audit/internal/test_email_e2e.ipynb b/notebooks/ai_audit/internal/test_email_e2e.ipynb
index aeac7343b01..1a64babb754 100644
--- a/notebooks/ai_audit/internal/test_email_e2e.ipynb
+++ b/notebooks/ai_audit/internal/test_email_e2e.ipynb
@@ -459,9 +459,9 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "from syft_bg.approve.config import AutoApproveConfig\n",
+    "from syft_bg.common.syft_bg_config import SyftBgConfig\n",
     "\n",
-    "cfg = AutoApproveConfig.load()\n",
+    "cfg = SyftBgConfig.load().approve\n",
     "auto_obj = cfg.auto_approvals.objects[FIRST_JOB_NAME]\n",
     "\n",
     "content_names = {e.relative_path for e in auto_obj.file_contents}\n",
@@ -608,7 +608,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "cfg = AutoApproveConfig.load()\n",
+    "cfg = SyftBgConfig.load().approve\n",
     "assert API_JOB_NAME in cfg.auto_approvals.objects, \"auto_approve_job did not register a rule under the job's name\"\n",
     "api_obj = cfg.auto_approvals.objects[API_JOB_NAME]\n",
     "assert {e.relative_path for e in api_obj.file_contents} == {\"code.py\"}\n",
diff --git a/packages/syft-bg/src/syft_bg/api/api.py b/packages/syft-bg/src/syft_bg/api/api.py
index 052f4baa84a..181f0152480 100644
--- a/packages/syft-bg/src/syft_bg/api/api.py
+++ b/packages/syft-bg/src/syft_bg/api/api.py
@@ -1,6 +1,8 @@
 """Pythonic API for syft-bg initialization and configuration."""
 
 import shutil
+import tempfile
+from collections.abc import Sequence
 from pathlib import Path
 
 from syft_bg.api.results import AutoApproveResult, InstallationResult, StatusResult
@@ -14,8 +16,8 @@
     setup_orchestrator,
     validate_auto_approve_job_inputs,
 )
-from syft_bg.approve.config import AutoApproveConfig, AutoApprovalObj
-from syft_bg.common.config import get_syftbg_dir, get_default_paths
+from syft_bg.approve.config import AutoApprovalObj, FileEntry
+from syft_bg.common.config import get_default_paths, get_syftbg_dir
 from syft_bg.common.drive import is_colab
 from syft_bg.common.syft_bg_config import SyftBgConfig
 from syft_bg.services import ServiceManager
@@ -31,15 +33,16 @@ def init(
     token_path: str | Path | None = None,
     settings: dict[str, dict] | None = None,
 ) -> None:
+    token_kwargs: dict = {}
     if token_path is not None:
-        token_path = Path(token_path)
-        move_token_to_syftbg_dir(token_path)
+        # A single OAuth token is used for both the Gmail and Drive APIs.
+        token_path = move_token_to_syftbg_dir(Path(token_path))
+        token_kwargs = {"drive_token_path": token_path, "gmail_token_path": token_path}
 
     config = SyftBgConfig(
         do_email=do_email,
         syftbox_root=syftbox_root,
-        token_path=token_path,
-        drive_token_path=token_path,
+        **token_kwargs,
     )
 
     if settings:
@@ -55,10 +58,8 @@ def ensure_running(
     restart: bool = False,
     install: bool = False,
 ) -> None:
-    # store new settings
-    try:
-        config = SyftBgConfig.from_path()
-    except FileNotFoundError:
+    config_path = get_default_paths().config
+    if not config_path.exists():
         print("No config file found, run init first")
         return
 
@@ -67,23 +68,24 @@ def ensure_running(
     manager = ServiceManager()
 
     # store new settings
-    for name, service_config in services.items():
-        service = manager.get_service(name)
-        if not service:
-            raise ValueError(f"Unknown service: {name}")
-        config.set_service_config(name, service_config)
-
-    config.save()
+    with SyftBgConfig.edit(config_path) as config:
+        for name, service_config in services.items():
+            service = manager.get_service(name)
+            if not service:
+                raise ValueError(f"Unknown service: {name}")
+            config.set_service_config(name, service_config)
 
     # make sure services are running
     for name, service_config in services.items():
         service = manager.get_service(name)
-        if service.is_running() and not restart:
+
+        # service was already confirmed to exist above, so ignore type checker here
+        if service.is_running() and not restart:  # ty:ignore[unresolved-attribute]
             print(
                 f"{name} is already running, skipping. If you want to restart it, set restart=True."
             )
             continue
-        elif service.is_running() and restart:
+        elif service.is_running() and restart:  # ty:ignore[unresolved-attribute]
             manager.restart_service(name)
         else:
             manager.start_service(name)
@@ -233,7 +235,7 @@ def uninstall(service: str | None = None) -> list[InstallationResult]:
     return results
 
 
-def logs(service: str, n: int = 50, as_list: bool = False) -> list[str]:
+def logs(service: str, n: int = 50, as_list: bool = False) -> list[str] | None:
     """Get recent log lines for a service.
 
     Args:
@@ -306,8 +308,17 @@ def status() -> StatusResult:
 # ---------------------------------------------------------------------------
 
 
+class _AutoApprovalNotFound(Exception):
+    """Raised inside SyftBgConfig.edit() to skip its save()."""
+
+
+class _AutoApproveDirectoryConflict(Exception):
+    """Raised inside SyftBgConfig.edit() to skip its save() when the staging
+    directory can't be moved into place."""
+
+
 def auto_approve(
-    contents: list[str | Path],
+    contents: Sequence[str | Path],
     file_paths: list[str] | None = None,
     peers: list[str] | None = None,
     name: str | None = None,
@@ -344,18 +355,56 @@ def auto_approve(
     if not content_files and not file_paths:
         return AutoApproveResult(success=False, error="No files to process")
 
-    config = AutoApproveConfig.load()
-    name = generate_unique_name(name, content_files, config)
-
-    file_entries = copy_and_hash_files(content_files, name)
-
-    obj = AutoApprovalObj(
-        file_contents=file_entries,
-        file_paths=file_paths,
-        peers=peers,
+    # Copy/hash into a private staging directory (unique per call) unlocked,
+    # so the config lock in SyftBgConfig.edit() is held for the minimum time
+    # possible. Concurrent callers never share a directory before the lock
+    # resolves the final name, unlike copying straight into a name-derived
+    # directory. The staging dir lives inside auto_approvals_dir so the
+    # later rename into place is an atomic same-filesystem move.
+    #
+    auto_approvals_dir = get_default_paths().auto_approvals_dir
+    auto_approvals_dir.mkdir(parents=True, exist_ok=True)
+    current_dir = Path(
+        tempfile.mkdtemp(prefix=".auto_approve_staging_", dir=auto_approvals_dir)
     )
-    config.auto_approvals.objects[name] = obj
-    config.save()
+    succeeded = False
+    try:
+        file_entries = copy_and_hash_files(content_files, current_dir.name)
+
+        with SyftBgConfig.edit() as syft_bg_config:
+            config = syft_bg_config.approve
+            name = generate_unique_name(name, content_files, config)
+            final_dir = auto_approvals_dir / name
+            try:
+                current_dir.rename(final_dir)
+            except OSError as e:
+                raise _AutoApproveDirectoryConflict(str(e)) from e
+            current_dir = final_dir
+
+            file_entries = [
+                FileEntry(
+                    relative_path=entry.relative_path,
+                    path=str(final_dir / entry.relative_path),
+                    hash=entry.hash,
+                )
+                for entry in file_entries
+            ]
+
+            obj = AutoApprovalObj(
+                file_contents=file_entries,
+                file_paths=file_paths,
+                peers=peers,
+            )
+            config.auto_approvals.objects[name] = obj
+        succeeded = True
+    except _AutoApproveDirectoryConflict as e:
+        return AutoApproveResult(
+            success=False,
+            error=f"Could not finalize auto-approval directory '{name}': {e}",
+        )
+    finally:
+        if not succeeded:
+            shutil.rmtree(current_dir, ignore_errors=True)
 
     return AutoApproveResult(
         success=True,
@@ -421,7 +470,7 @@ def list_auto_approvals() -> dict[str, AutoApprovalObj]:
     Returns:
         Mapping of name → AutoApprovalObj.
     """
-    return AutoApproveConfig.load().auto_approvals.objects
+    return SyftBgConfig.load().approve.auto_approvals.objects
 
 
 def remove_auto_approve(name: str) -> AutoApproveResult:
@@ -437,15 +486,20 @@ def remove_auto_approve(name: str) -> AutoApproveResult:
     Returns:
         AutoApproveResult with success/error status.
     """
-    config = AutoApproveConfig.load()
-    if name not in config.auto_approvals.objects:
+    # `return` inside edit() is normal control flow and wouldn't skip its
+    # save() — raise instead so it's actually skipped.
+    try:
+        with SyftBgConfig.edit() as syft_bg_config:
+            config = syft_bg_config.approve
+            if name not in config.auto_approvals.objects:
+                raise _AutoApprovalNotFound(name)
+
+            del config.auto_approvals.objects[name]
+    except _AutoApprovalNotFound:
         return AutoApproveResult(
             success=False, error=f"Auto-approval object '{name}' not found"
         )
 
-    del config.auto_approvals.objects[name]
-    config.save()
-
     obj_dir = get_default_paths().auto_approvals_dir / name
     if obj_dir.exists():
         shutil.rmtree(obj_dir, ignore_errors=True)
diff --git a/packages/syft-bg/src/syft_bg/api/utils.py b/packages/syft-bg/src/syft_bg/api/utils.py
index 21b205ded35..7a313846e7b 100644
--- a/packages/syft-bg/src/syft_bg/api/utils.py
+++ b/packages/syft-bg/src/syft_bg/api/utils.py
@@ -2,6 +2,7 @@
 
 import hashlib
 import shutil
+from collections.abc import Sequence
 from pathlib import Path
 
 from syft_bg.approve.config import AutoApproveConfig, FileEntry
@@ -104,6 +105,8 @@ def move_token_to_syftbg_dir(token_path: Path) -> Path:
         else:
             print(f"Warning: Provided token_path ({token_path}) does not exist.")
 
+    return Path(token_path)
+
 
 def credentials_setup_steps(creds_path: Path, colab: bool) -> str:
     """Return step-by-step instructions for setting up credentials.json."""
@@ -189,9 +192,10 @@ def save_gcp_project_id(credentials_path: Path) -> None:
     """Extract project_id from credentials.json and save to config.yaml."""
     try:
         project_id = get_project_id_from_credentials(credentials_path)
-        config = SyftBgConfig.from_path()
-        config.email_approve.gcp_project_id = project_id
-        config.save()
+        if not get_default_paths().config.exists():
+            return
+        with SyftBgConfig.edit() as config:
+            config.email_approve.gcp_project_id = project_id
     except Exception:
         pass
 
@@ -220,7 +224,7 @@ def generate_unique_name(
 
 
 def resolve_content_files(
-    contents: list[str | Path], base_dir: Path | None
+    contents: Sequence[str | Path], base_dir: Path | None
 ) -> tuple[list[tuple[str, Path]], str | None]:
     """Resolve content paths to (relative_path, absolute_path) pairs.
 
diff --git a/packages/syft-bg/src/syft_bg/approve/config.py b/packages/syft-bg/src/syft_bg/approve/config.py
index f53a28742fe..73b1875a236 100644
--- a/packages/syft-bg/src/syft_bg/approve/config.py
+++ b/packages/syft-bg/src/syft_bg/approve/config.py
@@ -3,7 +3,6 @@
 from pathlib import Path
 from typing import Optional
 
-import yaml
 from pydantic import BaseModel, Field
 
 from syft_bg.common.config import get_default_paths
@@ -78,82 +77,6 @@ class AutoApproveConfig(BaseModel):
     )
     force_ignore_peer_version: bool = False
 
-    @classmethod
-    def load(cls, config_path: Optional[Path] = None) -> "AutoApproveConfig":
-        """Load configuration from YAML file."""
-        if config_path is None:
-            config_path = get_default_paths().config
-        else:
-            config_path = Path(config_path).expanduser()
-
-        if not config_path.exists():
-            return cls()
-
-        with open(config_path) as f:
-            data = yaml.safe_load(f) or {}
-
-        common = {k: v for k, v in data.items() if not isinstance(v, dict)}
-        approve_section = data.get("approve", {})
-
-        auto_approvals_data = approve_section.get("auto_approvals", {})
-        peers_data = approve_section.get("peers", {})
-
-        kwargs: dict = {
-            "interval": approve_section.get("interval", 5),
-            "auto_approvals": AutoApprovalsConfig(**auto_approvals_data)
-            if auto_approvals_data
-            else AutoApprovalsConfig(),
-            "peers": PeerApprovalConfig(**peers_data)
-            if peers_data
-            else PeerApprovalConfig(),
-            "skip_peer_on_patch_version_diff": approve_section.get(
-                "skip_peer_on_patch_version_diff"
-            ),
-            "force_ignore_peer_version": approve_section.get(
-                "force_ignore_peer_version", False
-            ),
-        }
-        if common.get("do_email"):
-            kwargs["do_email"] = common["do_email"]
-        if common.get("syftbox_root"):
-            kwargs["syftbox_root"] = Path(common["syftbox_root"]).expanduser()
-        if common.get("drive_token_path"):
-            kwargs["drive_token_path"] = Path(common["drive_token_path"]).expanduser()
-
-        return cls(**kwargs)
-
-    def save(self, config_path: Optional[Path] = None) -> None:
-        """Save configuration to YAML file."""
-        if config_path is None:
-            config_path = get_default_paths().config
-        else:
-            config_path = Path(config_path).expanduser()
-
-        config_path.parent.mkdir(parents=True, exist_ok=True)
-
-        data = {}
-        if config_path.exists():
-            with open(config_path) as f:
-                data = yaml.safe_load(f) or {}
-
-        if self.do_email:
-            data["do_email"] = self.do_email
-        if self.syftbox_root:
-            data["syftbox_root"] = str(self.syftbox_root)
-        if self.drive_token_path:
-            data["drive_token_path"] = str(self.drive_token_path)
-
-        data["approve"] = {
-            "interval": self.interval,
-            "auto_approvals": self.auto_approvals.model_dump(),
-            "peers": self.peers.model_dump(),
-            "skip_peer_on_patch_version_diff": self.skip_peer_on_patch_version_diff,
-            "force_ignore_peer_version": self.force_ignore_peer_version,
-        }
-
-        with open(config_path, "w") as f:
-            yaml.dump(data, f, default_flow_style=False, sort_keys=False)
-
 
 # --- Backwards-compatible aliases (deprecated, will be removed) ---
 
diff --git a/packages/syft-bg/src/syft_bg/approve/handlers/job.py b/packages/syft-bg/src/syft_bg/approve/handlers/job.py
index fb661ad4bf1..1b6d17ef210 100644
--- a/packages/syft-bg/src/syft_bg/approve/handlers/job.py
+++ b/packages/syft-bg/src/syft_bg/approve/handlers/job.py
@@ -7,15 +7,12 @@
 
 from syft_job.job import JobInfo
 
-from syft_bg.approve.config import (
-    AutoApprovalObj,
-    AutoApprovalsConfig,
-    AutoApproveConfig,
-)
+from syft_bg.approve.config import AutoApprovalObj, AutoApprovalsConfig
 from syft_bg.approve.criteria import (
     AutoApprovalValidationResult,
     _validate_job_against_object,
 )
+from syft_bg.common.syft_bg_config import SyftBgConfig
 
 if TYPE_CHECKING:
     from syft_client.sync.syftbox_manager import SyftboxManager
@@ -49,7 +46,7 @@ def __init__(
     @property
     def config(self) -> AutoApprovalsConfig:
         """Always-fresh auto-approvals config, re-read from disk on every access."""
-        return AutoApproveConfig.load(self._config_path).auto_approvals
+        return SyftBgConfig.load(self._config_path).approve.auto_approvals
 
     def _get_approved_peers(self) -> list[str]:
         """Get list of approved peer emails."""
diff --git a/packages/syft-bg/src/syft_bg/approve/orchestrator.py b/packages/syft-bg/src/syft_bg/approve/orchestrator.py
index 20db094644d..f67d3cab5aa 100644
--- a/packages/syft-bg/src/syft_bg/approve/orchestrator.py
+++ b/packages/syft-bg/src/syft_bg/approve/orchestrator.py
@@ -10,6 +10,7 @@
 from syft_bg.approve.monitors.peer import PeerMonitor
 from syft_bg.common.orchestrator import BaseOrchestrator
 from syft_bg.common.state import JsonStateManager
+from syft_bg.common.syft_bg_config import SyftBgConfig
 
 if TYPE_CHECKING:
     from syft_client.sync.syftbox_manager import SyftboxManager
@@ -49,7 +50,7 @@ def from_client(
                 "ApprovalOrchestrator should only run on Data Owner (DO) side."
             )
 
-        config = AutoApproveConfig.load()
+        config = SyftBgConfig.load().approve
         config.do_email = client.email
         config.syftbox_root = client.syftbox_folder
         config.interval = interval
diff --git a/packages/syft-bg/src/syft_bg/cli/commands.py b/packages/syft-bg/src/syft_bg/cli/commands.py
index 70c06e4d9aa..2eed251b3be 100644
--- a/packages/syft-bg/src/syft_bg/cli/commands.py
+++ b/packages/syft-bg/src/syft_bg/cli/commands.py
@@ -378,20 +378,22 @@ def remove_auto_approval(files: tuple[str, ...], name: str):
 
       syft-bg remove-auto-approval main.py utils.py -n my_analysis
     """
-    from syft_bg.approve.config import AutoApproveConfig
+    from syft_bg.common.syft_bg_config import SyftBgConfig
 
-    config = AutoApproveConfig.load()
+    with SyftBgConfig.edit() as syft_bg_config:
+        config = syft_bg_config.approve
 
-    if name not in config.auto_approvals.objects:
-        click.echo(f"Auto-approval object '{name}' not found in config.", err=True)
-        raise SystemExit(1)
+        if name not in config.auto_approvals.objects:
+            click.echo(f"Auto-approval object '{name}' not found in config.", err=True)
+            raise SystemExit(1)
 
-    obj = config.auto_approvals.objects[name]
-    before = len(obj.file_contents)
-    obj.file_contents = [s for s in obj.file_contents if s.relative_path not in files]
-    removed = before - len(obj.file_contents)
+        obj = config.auto_approvals.objects[name]
+        before = len(obj.file_contents)
+        obj.file_contents = [
+            s for s in obj.file_contents if s.relative_path not in files
+        ]
+        removed = before - len(obj.file_contents)
 
-    config.save()
     click.echo(f"Removed {removed} script(s) from '{name}'.")
 
 
@@ -412,30 +414,30 @@ def remove_peer(peer: str, name: str | None):
 
       syft-bg remove-peer alice@uni.edu -n my_analysis
     """
-    from syft_bg.approve.config import AutoApproveConfig
+    from syft_bg.common.syft_bg_config import SyftBgConfig
 
-    config = AutoApproveConfig.load()
-    removed_from = 0
+    with SyftBgConfig.edit() as syft_bg_config:
+        config = syft_bg_config.approve
+        removed_from = 0
 
-    if name:
-        if name not in config.auto_approvals.objects:
-            click.echo(f"Auto-approval object '{name}' not found.", err=True)
-            raise SystemExit(1)
-        obj = config.auto_approvals.objects[name]
-        if peer in obj.peers:
-            obj.peers.remove(peer)
-            removed_from = 1
-    else:
-        for obj in config.auto_approvals.objects.values():
+        if name:
+            if name not in config.auto_approvals.objects:
+                click.echo(f"Auto-approval object '{name}' not found.", err=True)
+                raise SystemExit(1)
+            obj = config.auto_approvals.objects[name]
             if peer in obj.peers:
                 obj.peers.remove(peer)
-                removed_from += 1
+                removed_from = 1
+        else:
+            for obj in config.auto_approvals.objects.values():
+                if peer in obj.peers:
+                    obj.peers.remove(peer)
+                    removed_from += 1
 
-    if removed_from == 0:
-        click.echo(f"Peer {peer} not found in any auto-approval object.", err=True)
-        raise SystemExit(1)
+        if removed_from == 0:
+            click.echo(f"Peer {peer} not found in any auto-approval object.", err=True)
+            raise SystemExit(1)
 
-    config.save()
     click.echo(f"Removed peer {peer} from {removed_from} object(s).")
 
 
@@ -455,21 +457,21 @@ def list_auto_approvals(name: str | None):
 
       syft-bg list-auto-approvals -n my_analysis
     """
-    from syft_bg.approve.config import AutoApproveConfig
+    from syft_bg.api.api import list_auto_approvals as api_list_auto_approvals
 
-    config = AutoApproveConfig.load()
+    objects = api_list_auto_approvals()
 
-    if not config.auto_approvals.objects:
+    if not objects:
         click.echo("No auto-approval objects configured.")
         return
 
     if name:
-        if name not in config.auto_approvals.objects:
+        if name not in objects:
             click.echo(f"Auto-approval object '{name}' not found.", err=True)
             raise SystemExit(1)
-        objects_to_show = {name: config.auto_approvals.objects[name]}
+        objects_to_show = {name: objects[name]}
     else:
-        objects_to_show = config.auto_approvals.objects
+        objects_to_show = objects
 
     for obj_name, obj in objects_to_show.items():
         click.echo(f"\n[{obj_name}]")
diff --git a/packages/syft-bg/src/syft_bg/common/locking.py b/packages/syft-bg/src/syft_bg/common/locking.py
new file mode 100644
index 00000000000..00b8ebfefeb
--- /dev/null
+++ b/packages/syft-bg/src/syft_bg/common/locking.py
@@ -0,0 +1,19 @@
+"""Shared exclusive file-locking helper for read-modify-write cycles."""
+
+import fcntl
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Iterator
+
+
+@contextmanager
+def file_lock(lock_path: Path) -> Iterator[None]:
+    """Hold an exclusive advisory lock on lock_path for the block's duration."""
+    lock_path.parent.mkdir(parents=True, exist_ok=True)
+    lock_path.touch(exist_ok=True)
+    with open(lock_path) as lock_handle:
+        try:
+            fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX)
+            yield
+        finally:
+            fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN)
diff --git a/packages/syft-bg/src/syft_bg/common/state.py b/packages/syft-bg/src/syft_bg/common/state.py
index 48f5941aa7e..6a36c6e5929 100644
--- a/packages/syft-bg/src/syft_bg/common/state.py
+++ b/packages/syft-bg/src/syft_bg/common/state.py
@@ -1,6 +1,5 @@
 """State management for tracking notified/approved entities."""
 
-import fcntl
 import json
 from contextlib import contextmanager
 from datetime import datetime, timezone
@@ -9,6 +8,8 @@
 
 from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, ValidationError
 
+from syft_bg.common.locking import file_lock
+
 
 class BgState(BaseModel):
     """Typed top-level state for the background services.
@@ -43,13 +44,8 @@ def model_post_init(self, __context: Any) -> None:
     @contextmanager
     def _file_lock(self):
         """Acquire exclusive file lock for thread-safe writes."""
-        self._lock_file.touch(exist_ok=True)
-        with open(self._lock_file, "r") as lock_handle:
-            try:
-                fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX)
-                yield
-            finally:
-                fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN)
+        with file_lock(self._lock_file):
+            yield
 
     def _load(self) -> BgState:
         if not self.state_file.exists():
diff --git a/packages/syft-bg/src/syft_bg/common/syft_bg_config.py b/packages/syft-bg/src/syft_bg/common/syft_bg_config.py
index 6647bf67a91..7316b1b8741 100644
--- a/packages/syft-bg/src/syft_bg/common/syft_bg_config.py
+++ b/packages/syft-bg/src/syft_bg/common/syft_bg_config.py
@@ -1,12 +1,15 @@
 """Top-level SyftBg configuration model (mirrors config.yaml)."""
 
+from contextlib import contextmanager
 from pathlib import Path
+from typing import Iterator
 
 import yaml
-from pydantic import BaseModel, Field, model_validator
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 from syft_bg.approve.config import AutoApproveConfig
 from syft_bg.common.config import get_default_paths
+from syft_bg.common.locking import file_lock
 from syft_bg.email_approve.config import EmailApproveConfig
 from syft_bg.notify.config import NotifyConfig
 from syft_bg.sync.config import SyncConfig
@@ -15,15 +18,19 @@
 class SyftBgConfig(BaseModel):
     """Top-level syft-bg configuration, matching the config.yaml structure."""
 
+    # Preserve top-level keys we don't model (e.g. written by another
+    # syft-bg version) instead of silently dropping them on save().
+    model_config = ConfigDict(extra="allow")
+
     do_email: str | None = None
-    syftbox_root: str | None = None
+    syftbox_root: str | Path | None = None
     credentials_path: Path = Field(
         default_factory=lambda: get_default_paths().credentials
     )
     gmail_token_path: Path = Field(
         default_factory=lambda: get_default_paths().gmail_token
     )
-    drive_token_path: Path = Field(
+    drive_token_path: Path | None = Field(
         default_factory=lambda: get_default_paths().drive_token
     )
     notify: NotifyConfig = Field(default_factory=NotifyConfig)
@@ -64,10 +71,18 @@ def _merge_common_into_services(self) -> None:
             ):
                 if self.syftbox_root is not None:
                     service_config.syftbox_root = Path(self.syftbox_root)
-            for path_field in ("drive_token_path", "gmail_token_path"):
+            # These fields default to a concrete path, not None, so use the
+            # default itself (not "is None") to detect an unset field.
+            default_paths = get_default_paths()
+            path_field_defaults = {
+                "drive_token_path": default_paths.drive_token,
+                "gmail_token_path": default_paths.gmail_token,
+            }
+            for path_field, own_default in path_field_defaults.items():
                 if hasattr(service_config, path_field):
                     parent_val = getattr(self, path_field, None)
-                    if parent_val is not None:
+                    current_val = getattr(service_config, path_field, None)
+                    if parent_val is not None and current_val == own_default:
                         setattr(service_config, path_field, parent_val)
 
     def set_service_config(self, name: str, config: dict) -> None:
@@ -98,8 +113,11 @@ def _repr_html_(self) -> str:
 <pre>{yaml_str}</pre>"""
 
     @classmethod
-    def from_path(cls, config_path: Path | None = None) -> "SyftBgConfig":
-        """Load from a YAML config file, merging common fields into services."""
+    def from_path(
+        cls, config_path: Path | None = None, *, merge: bool = True
+    ) -> "SyftBgConfig":
+        """Load from a YAML config file. Pass merge=False before saving back
+        (e.g. edit()) to avoid baking inherited fields into sub-configs."""
         if config_path is None:
             config_path = get_default_paths().config
 
@@ -107,7 +125,8 @@ def from_path(cls, config_path: Path | None = None) -> "SyftBgConfig":
             data = yaml.safe_load(f) or {}
 
         config = cls.model_validate(data)
-        config._merge_common_into_services()
+        if merge:
+            config._merge_common_into_services()
         return config
 
     def save(self, config_path: Path | None = None) -> None:
@@ -124,3 +143,26 @@ def save(self, config_path: Path | None = None) -> None:
                 default_flow_style=False,
                 sort_keys=False,
             )
+
+    @classmethod
+    def load(
+        cls, config_path: Path | None = None, *, merge: bool = True
+    ) -> "SyftBgConfig":
+        """Load config from path, returning defaults if the file doesn't exist."""
+        try:
+            return cls.from_path(config_path, merge=merge)
+        except FileNotFoundError:
+            return cls()
+
+    @classmethod
+    @contextmanager
+    def edit(cls, config_path: Path | None = None) -> Iterator["SyftBgConfig"]:
+        """Locked load-mutate-save cycle. Skips save() if the block raises."""
+        if config_path is None:
+            config_path = get_default_paths().config
+
+        lock_path = config_path.with_suffix(".lock")
+        with file_lock(lock_path):
+            config = cls.load(config_path, merge=False)
+            yield config
+            config.save(config_path)
diff --git a/tests/integration/syft_bg/test_cli.py b/tests/integration/syft_bg/test_cli.py
index ee39fe7aa82..31bf546fba7 100644
--- a/tests/integration/syft_bg/test_cli.py
+++ b/tests/integration/syft_bg/test_cli.py
@@ -319,8 +319,8 @@ def test_auto_approve_error(self, mock_api, sample_script):
 class TestRemoveAutoApprovalCommand:
     """Tests for the remove-auto-approval command."""
 
-    @patch("syft_bg.approve.config.AutoApproveConfig.load")
-    def test_remove_auto_approval(self, mock_load):
+    @patch("syft_bg.common.syft_bg_config.SyftBgConfig.edit")
+    def test_remove_auto_approval(self, mock_edit):
         """Should remove scripts by filename from an auto-approval object."""
         from syft_bg.approve.config import AutoApprovalObj, FileEntry
 
@@ -340,8 +340,8 @@ def test_remove_auto_approval(self, mock_load):
             peers=["alice@test.com"],
         )
         mock_config = MagicMock()
-        mock_config.auto_approvals.objects = {"my_analysis": obj}
-        mock_load.return_value = mock_config
+        mock_config.approve.auto_approvals.objects = {"my_analysis": obj}
+        mock_edit.return_value.__enter__.return_value = mock_config
 
         runner = CliRunner()
         result = runner.invoke(remove_auto_approval, ["utils.py", "-n", "my_analysis"])
@@ -351,12 +351,12 @@ def test_remove_auto_approval(self, mock_load):
         assert len(obj.file_contents) == 1
         assert obj.file_contents[0].relative_path == "main.py"
 
-    @patch("syft_bg.approve.config.AutoApproveConfig.load")
-    def test_remove_auto_approval_unknown_object(self, mock_load):
+    @patch("syft_bg.common.syft_bg_config.SyftBgConfig.edit")
+    def test_remove_auto_approval_unknown_object(self, mock_edit):
         """Should error when object name not found."""
         mock_config = MagicMock()
-        mock_config.auto_approvals.objects = {}
-        mock_load.return_value = mock_config
+        mock_config.approve.auto_approvals.objects = {}
+        mock_edit.return_value.__enter__.return_value = mock_config
 
         runner = CliRunner()
         result = runner.invoke(remove_auto_approval, ["main.py", "-n", "nonexistent"])
@@ -367,16 +367,16 @@ def test_remove_auto_approval_unknown_object(self, mock_load):
 class TestRemovePeerCommand:
     """Tests for the remove-peer command."""
 
-    @patch("syft_bg.approve.config.AutoApproveConfig.load")
-    def test_remove_peer(self, mock_load):
+    @patch("syft_bg.common.syft_bg_config.SyftBgConfig.edit")
+    def test_remove_peer(self, mock_edit):
         """Should remove peer from auto-approval objects."""
         from syft_bg.approve.config import AutoApprovalObj
 
         mock_config = MagicMock()
-        mock_config.auto_approvals.objects = {
+        mock_config.approve.auto_approvals.objects = {
             "my_analysis": AutoApprovalObj(peers=["alice@test.com"])
         }
-        mock_load.return_value = mock_config
+        mock_edit.return_value.__enter__.return_value = mock_config
 
         runner = CliRunner()
         result = runner.invoke(remove_peer, ["alice@test.com"])
@@ -385,15 +385,15 @@ def test_remove_peer(self, mock_load):
         assert "Removed peer" in result.output
         assert (
             "alice@test.com"
-            not in mock_config.auto_approvals.objects["my_analysis"].peers
+            not in mock_config.approve.auto_approvals.objects["my_analysis"].peers
         )
 
-    @patch("syft_bg.approve.config.AutoApproveConfig.load")
-    def test_remove_peer_not_found(self, mock_load):
+    @patch("syft_bg.common.syft_bg_config.SyftBgConfig.edit")
+    def test_remove_peer_not_found(self, mock_edit):
         """Should error if peer not found."""
         mock_config = MagicMock()
-        mock_config.auto_approvals.objects = {}
-        mock_load.return_value = mock_config
+        mock_config.approve.auto_approvals.objects = {}
+        mock_edit.return_value.__enter__.return_value = mock_config
 
         runner = CliRunner()
         result = runner.invoke(remove_peer, ["unknown@test.com"])
@@ -404,13 +404,12 @@ def test_remove_peer_not_found(self, mock_load):
 class TestListAutoApprovalsCommand:
     """Tests for the list-auto-approvals command."""
 
-    @patch("syft_bg.approve.config.AutoApproveConfig.load")
-    def test_list_auto_approvals(self, mock_load):
+    @patch("syft_bg.api.api.list_auto_approvals")
+    def test_list_auto_approvals(self, mock_list):
         """Should list all auto-approval objects and their scripts."""
         from syft_bg.approve.config import AutoApprovalObj, FileEntry
 
-        mock_config = MagicMock()
-        mock_config.auto_approvals.objects = {
+        mock_list.return_value = {
             "analysis_a": AutoApprovalObj(
                 file_contents=[
                     FileEntry(
@@ -432,7 +431,6 @@ def test_list_auto_approvals(self, mock_load):
                 peers=["bob@test.com"],
             ),
         }
-        mock_load.return_value = mock_config
 
         runner = CliRunner()
         result = runner.invoke(list_auto_approvals)
@@ -443,13 +441,12 @@ def test_list_auto_approvals(self, mock_load):
         assert "bob@test.com" in result.output
         assert "train.py" in result.output
 
-    @patch("syft_bg.approve.config.AutoApproveConfig.load")
-    def test_list_auto_approvals_single_object(self, mock_load):
+    @patch("syft_bg.api.api.list_auto_approvals")
+    def test_list_auto_approvals_single_object(self, mock_list):
         """Should filter to a specific auto-approval object."""
         from syft_bg.approve.config import AutoApprovalObj, FileEntry
 
-        mock_config = MagicMock()
-        mock_config.auto_approvals.objects = {
+        mock_list.return_value = {
             "analysis_a": AutoApprovalObj(
                 file_contents=[
                     FileEntry(
@@ -471,7 +468,6 @@ def test_list_auto_approvals_single_object(self, mock_load):
                 peers=["bob@test.com"],
             ),
         }
-        mock_load.return_value = mock_config
 
         runner = CliRunner()
         result = runner.invoke(list_auto_approvals, ["-n", "analysis_a"])
@@ -480,12 +476,10 @@ def test_list_auto_approvals_single_object(self, mock_load):
         assert "analysis_a" in result.output
         assert "analysis_b" not in result.output
 
-    @patch("syft_bg.approve.config.AutoApproveConfig.load")
-    def test_list_auto_approvals_empty(self, mock_load):
+    @patch("syft_bg.api.api.list_auto_approvals")
+    def test_list_auto_approvals_empty(self, mock_list):
         """Should show message when no auto-approval objects configured."""
-        mock_config = MagicMock()
-        mock_config.auto_approvals.objects = {}
-        mock_load.return_value = mock_config
+        mock_list.return_value = {}
 
         runner = CliRunner()
         result = runner.invoke(list_auto_approvals)
diff --git a/tests/unit/syft_bg/test_auto_approve_api.py b/tests/unit/syft_bg/test_auto_approve_api.py
index 7e2de766a6b..4304ab9730f 100644
--- a/tests/unit/syft_bg/test_auto_approve_api.py
+++ b/tests/unit/syft_bg/test_auto_approve_api.py
@@ -1,12 +1,15 @@
 """Tests for the list/remove auto-approve Python API and config-reload behavior."""
 
+import fcntl
 from contextlib import contextmanager
 from dataclasses import replace
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
+import pytest
 
-from syft_bg.api.api import list_auto_approvals, remove_auto_approve
+from syft_bg.api.api import auto_approve, list_auto_approvals, remove_auto_approve
+from syft_bg.api.utils import copy_and_hash_files
 from syft_bg.approve.config import (
     AutoApprovalObj,
     AutoApprovalsConfig,
@@ -15,6 +18,7 @@
 )
 from syft_bg.approve.handlers.job import JobApprovalHandler
 from syft_bg.common.config import get_default_paths
+from syft_bg.common.syft_bg_config import SyftBgConfig
 
 
 @contextmanager
@@ -28,7 +32,9 @@ def _patched_paths(tmp: Path):
     )
     with (
         patch("syft_bg.api.api.get_default_paths", return_value=patched),
+        patch("syft_bg.api.utils.get_default_paths", return_value=patched),
         patch("syft_bg.approve.config.get_default_paths", return_value=patched),
+        patch("syft_bg.common.syft_bg_config.get_default_paths", return_value=patched),
     ):
         yield patched
 
@@ -36,9 +42,10 @@ def _patched_paths(tmp: Path):
 def _seed_config(tmp: Path, objects: dict[str, AutoApprovalObj]) -> Path:
     """Write a config YAML with the given auto-approval objects to tmp/config.yaml."""
     config_path = tmp / "config.yaml"
-    AutoApproveConfig(auto_approvals=AutoApprovalsConfig(objects=objects)).save(
-        config_path
+    approve_config = AutoApproveConfig(
+        auto_approvals=AutoApprovalsConfig(objects=objects)
     )
+    SyftBgConfig(approve=approve_config).save(config_path)
     return config_path
 
 
@@ -98,6 +105,141 @@ def test_no_files_dir_still_succeeds(self, temp_dir):
             assert result.success is True
             assert list_auto_approvals() == {}
 
+    def test_unknown_does_not_create_config_file(self, temp_dir):
+        """A no-op failure (nothing to remove) must not write config.yaml
+        into existence — the not-found path must not trigger edit()'s save."""
+        with _patched_paths(temp_dir) as patched:
+            assert not patched.config.exists()
+
+            result = remove_auto_approve("does_not_exist")
+
+            assert result.success is False
+            assert not patched.config.exists()
+
+
+class TestAutoApproveLockScope:
+    """auto_approve() must not hold the config lock during file I/O."""
+
+    def test_lock_not_held_during_file_io(self, temp_dir):
+        with _patched_paths(temp_dir) as patched:
+            content_dir = temp_dir / "project"
+            content_dir.mkdir()
+            (content_dir / "script.py").write_text("print('lock test')\n")
+
+            lock_path = patched.config.with_suffix(".lock")
+            observed = {}
+
+            def _check_lock_then_copy(content_files, name):
+                lock_path.parent.mkdir(parents=True, exist_ok=True)
+                lock_path.touch(exist_ok=True)
+                with open(lock_path) as lock_handle:
+                    try:
+                        fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+                        observed["lock_was_free"] = True
+                        fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN)
+                    except BlockingIOError:
+                        observed["lock_was_free"] = False
+                return copy_and_hash_files(content_files, name)
+
+            with patch(
+                "syft_bg.api.api.copy_and_hash_files",
+                side_effect=_check_lock_then_copy,
+            ):
+                result = auto_approve(contents=[str(content_dir / "script.py")])
+
+            assert result.success is True
+            assert observed["lock_was_free"] is True
+
+    def test_name_collision_picks_next_available_name(self, temp_dir):
+        """If the target name is already taken by the time the lock is
+        acquired, auto_approve() must pick the next available name instead
+        of clobbering the existing entry. Naming is now resolved entirely
+        under the lock (after the file I/O lands in a private staging
+        directory), so a plain pre-seeded collision exercises the same
+        code path a genuine race would."""
+        with _patched_paths(temp_dir):
+            content_dir = temp_dir / "project"
+            content_dir.mkdir()
+            (content_dir / "main.py").write_text("print('hi')\n")
+
+            _seed_config(temp_dir, {"main": _make_obj("racer@test.com")})
+
+            result = auto_approve(contents=[str(content_dir / "main.py")])
+
+            assert result.success is True
+            assert result.name == "main_1"
+
+            objects = list_auto_approvals()
+            assert set(objects.keys()) == {"main", "main_1"}
+            assert objects["main"].peers == ["racer@test.com"]
+
+            entry = objects["main_1"].file_contents[0]
+            assert "main_1" in entry.path
+            assert Path(entry.path).read_text() == "print('hi')\n"
+
+    def test_orphaned_directory_at_target_name_returns_failure(self, temp_dir):
+        """A stale, non-empty directory already occupying the resolved name
+        (e.g. left behind by a prior remove_auto_approve's best-effort
+        rmtree) must produce a clean failure, not an unhandled OSError —
+        and the staging directory must be cleaned up rather than leaked."""
+        with _patched_paths(temp_dir) as patched:
+            content_dir = temp_dir / "project"
+            content_dir.mkdir()
+            (content_dir / "main.py").write_text("print('hi')\n")
+
+            orphan_dir = patched.auto_approvals_dir / "main"
+            orphan_dir.mkdir(parents=True)
+            (orphan_dir / "leftover.txt").write_text("stale")
+
+            result = auto_approve(contents=[str(content_dir / "main.py")])
+
+            assert result.success is False
+            assert "main" in (result.error or "")
+            assert list_auto_approvals() == {}
+
+            leftover_staging = [
+                p
+                for p in patched.auto_approvals_dir.iterdir()
+                if p.name.startswith(".auto_approve_staging_")
+            ]
+            assert leftover_staging == []
+
+    def test_staging_dir_cleaned_up_when_copy_and_hash_fails(self, temp_dir):
+        """A failure during the unlocked copy/hash step (e.g. a bad file
+        read) must not leak the staging directory it already created."""
+        with _patched_paths(temp_dir) as patched:
+            content_dir = temp_dir / "project"
+            content_dir.mkdir()
+            (content_dir / "main.py").write_text("print('hi')\n")
+
+            with patch(
+                "syft_bg.api.api.copy_and_hash_files",
+                side_effect=UnicodeDecodeError("utf-8", b"\xff", 0, 1, "bad byte"),
+            ):
+                with pytest.raises(UnicodeDecodeError):
+                    auto_approve(contents=[str(content_dir / "main.py")])
+
+            assert list(patched.auto_approvals_dir.iterdir()) == []
+
+    def test_final_dir_cleaned_up_when_save_fails_after_rename(self, temp_dir):
+        """A failure after the staging->final rename already succeeded
+        (e.g. config.save() itself fails) must not leave an orphaned,
+        unregistered directory behind — not just the rename-collision case."""
+        with _patched_paths(temp_dir) as patched:
+            content_dir = temp_dir / "project"
+            content_dir.mkdir()
+            (content_dir / "main.py").write_text("print('hi')\n")
+
+            with patch(
+                "syft_bg.common.syft_bg_config.SyftBgConfig.save",
+                side_effect=OSError("disk full"),
+            ):
+                with pytest.raises(OSError):
+                    auto_approve(contents=[str(content_dir / "main.py")])
+
+            assert list(patched.auto_approvals_dir.iterdir()) == []
+            assert not patched.config.exists()
+
 
 class TestHandlerReloadsConfig:
     """The approve service must pick up YAML changes without a restart."""
@@ -135,13 +277,15 @@ def test_picks_up_added_object(self, temp_dir):
         assert first.match is False
 
         # Add a matching object directly to the YAML on disk (no restart).
-        AutoApproveConfig(
-            auto_approvals=AutoApprovalsConfig(
-                objects={
-                    "r1": self._create_matching_autoapprove_obj_from_dir(
-                        code_dir, "alice@test.com"
-                    )
-                }
+        SyftBgConfig(
+            approve=AutoApproveConfig(
+                auto_approvals=AutoApprovalsConfig(
+                    objects={
+                        "r1": self._create_matching_autoapprove_obj_from_dir(
+                            code_dir, "alice@test.com"
+                        )
+                    }
+                )
             )
         ).save(config_path)
 
@@ -170,9 +314,9 @@ def test_picks_up_removed_object(self, temp_dir):
         assert first.match is True
 
         # Wipe the object from the YAML.
-        AutoApproveConfig(auto_approvals=AutoApprovalsConfig(objects={})).save(
-            config_path
-        )
+        SyftBgConfig(
+            approve=AutoApproveConfig(auto_approvals=AutoApprovalsConfig(objects={}))
+        ).save(config_path)
 
         second = handler.evaluate_auto_approval(job)
         assert second.match is False
diff --git a/tests/unit/syft_bg/test_config.py b/tests/unit/syft_bg/test_config.py
index 838f069099d..2670ad761d8 100644
--- a/tests/unit/syft_bg/test_config.py
+++ b/tests/unit/syft_bg/test_config.py
@@ -1,7 +1,11 @@
 """Tests for configuration loading."""
 
+import fcntl
 from pathlib import Path
 
+import pytest
+import yaml
+
 from syft_bg.approve.config import (
     AutoApproveConfig,
     AutoApprovalObj,
@@ -9,6 +13,7 @@
     PeerApprovalConfig,
     FileEntry,
 )
+from syft_bg.common.syft_bg_config import SyftBgConfig
 from syft_bg.notify.config import NotifyConfig
 
 
@@ -92,7 +97,7 @@ def test_default_config(self):
         assert config.peers.enabled is False
 
     def test_load_from_file(self, sample_config):
-        config = AutoApproveConfig.load(sample_config)
+        config = SyftBgConfig.from_path(sample_config).approve
         assert config.do_email == "test@example.com"
         assert config.syftbox_root == Path("/tmp/syftbox")
         assert config.interval == 5
@@ -106,19 +111,19 @@ def test_load_from_file(self, sample_config):
         assert "bob@co.com" in obj.peers
 
     def test_load_nonexistent_returns_defaults(self, temp_dir):
-        config = AutoApproveConfig.load(temp_dir / "nonexistent.yaml")
+        config = SyftBgConfig.load(temp_dir / "nonexistent.yaml").approve
         assert config.do_email is None
         assert config.auto_approvals.enabled is True
 
     def test_save_config(self, temp_dir):
         config_path = temp_dir / "config.yaml"
-        config = AutoApproveConfig(
+        approve_config = AutoApproveConfig(
             do_email="save@example.com",
             syftbox_root=Path("/tmp/saved"),
             interval=10,
         )
-        config.auto_approvals.enabled = False
-        config.auto_approvals.objects["test_obj"] = AutoApprovalObj(
+        approve_config.auto_approvals.enabled = False
+        approve_config.auto_approvals.objects["test_obj"] = AutoApprovalObj(
             file_contents=[
                 FileEntry(
                     relative_path="main.py", path="/tmp/main.py", hash="sha256:xyz"
@@ -126,9 +131,9 @@ def test_save_config(self, temp_dir):
             ],
             peers=["alice@test.com"],
         )
-        config.save(config_path)
+        SyftBgConfig(approve=approve_config).save(config_path)
 
-        loaded = AutoApproveConfig.load(config_path)
+        loaded = SyftBgConfig.from_path(config_path).approve
         assert loaded.do_email == "save@example.com"
         assert loaded.interval == 10
         assert loaded.auto_approvals.enabled is False
@@ -139,8 +144,8 @@ def test_save_config(self, temp_dir):
 
     def test_save_reload_multi_script_roundtrip(self, temp_dir):
         config_path = temp_dir / "config.yaml"
-        config = AutoApproveConfig(do_email="rt@test.com")
-        config.auto_approvals.objects["multi"] = AutoApprovalObj(
+        approve_config = AutoApproveConfig(do_email="rt@test.com")
+        approve_config.auto_approvals.objects["multi"] = AutoApprovalObj(
             file_contents=[
                 FileEntry(
                     relative_path="main.py", path="/tmp/main.py", hash="sha256:aaa"
@@ -151,9 +156,9 @@ def test_save_reload_multi_script_roundtrip(self, temp_dir):
             ],
             peers=["ds@test.com"],
         )
-        config.save(config_path)
+        SyftBgConfig(approve=approve_config).save(config_path)
 
-        loaded = AutoApproveConfig.load(config_path)
+        loaded = SyftBgConfig.from_path(config_path).approve
         obj = loaded.auto_approvals.objects["multi"]
         assert len(obj.file_contents) == 2
         assert obj.file_contents[0].relative_path == "main.py"
@@ -168,9 +173,106 @@ def test_load_empty_objects(self, temp_dir):
     enabled: true
     objects: {}
 """)
-        config = AutoApproveConfig.load(config_path)
+        config = SyftBgConfig.from_path(config_path).approve
         assert config.auto_approvals.objects == {}
 
+    def test_gmail_token_path_propagates_from_top_level(self, temp_dir):
+        """Regression test: AutoApproveConfig.load() used to never read
+        gmail_token_path from disk at all, always falling back to the
+        default-factory value. Going through SyftBgConfig fixes this."""
+        config_path = temp_dir / "config.yaml"
+        custom_token_path = Path("/tmp/custom_gmail_token.json")
+        SyftBgConfig(gmail_token_path=custom_token_path).save(config_path)
+
+        loaded = SyftBgConfig.from_path(config_path)
+        assert loaded.approve.gmail_token_path == custom_token_path
+
+
+class TestSyftBgConfigEdit:
+    """Tests for SyftBgConfig.edit(), the locked read-modify-write helper."""
+
+    def test_edit_saves_on_clean_exit(self, temp_dir):
+        config_path = temp_dir / "config.yaml"
+        with SyftBgConfig.edit(config_path) as config:
+            config.approve.auto_approvals.enabled = False
+
+        reloaded = SyftBgConfig.from_path(config_path)
+        assert reloaded.approve.auto_approvals.enabled is False
+
+    def test_edit_does_not_save_on_exception(self, temp_dir):
+        config_path = temp_dir / "config.yaml"
+
+        with pytest.raises(ValueError):
+            with SyftBgConfig.edit(config_path) as config:
+                config.approve.auto_approvals.enabled = False
+                raise ValueError("boom")
+
+        assert not config_path.exists()
+
+    def test_edit_holds_exclusive_lock(self, temp_dir):
+        config_path = temp_dir / "config.yaml"
+
+        with SyftBgConfig.edit(config_path) as config:
+            config.approve.auto_approvals.enabled = False
+
+            lock_path = config_path.with_suffix(".lock")
+            with open(lock_path) as lock_handle:
+                with pytest.raises(BlockingIOError):
+                    fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+
+
+class TestSyftBgConfigMergeAndSave:
+    """Regression tests for the review-flagged merge/save correctness bugs."""
+
+    def test_save_preserves_unknown_top_level_keys(self, temp_dir):
+        """save() must not silently drop YAML keys it doesn't model."""
+        config_path = temp_dir / "config.yaml"
+        config_path.write_text("""
+do_email: test@example.com
+some_future_field: keep_me
+approve:
+  interval: 5
+""")
+        with SyftBgConfig.edit(config_path) as config:
+            config.approve.interval = 10
+
+        raw = yaml.safe_load(config_path.read_text())
+        assert raw["some_future_field"] == "keep_me"
+        assert raw["approve"]["interval"] == 10
+
+    def test_merge_does_not_clobber_custom_service_token_path(self, temp_dir):
+        """A per-service drive_token_path override must survive a reload."""
+        config_path = temp_dir / "config.yaml"
+        custom_path = Path("/custom/approve_token.json")
+
+        syft_bg_config = SyftBgConfig()
+        syft_bg_config.set_service_config("approve", {"drive_token_path": custom_path})
+        syft_bg_config.save(config_path)
+
+        reloaded = SyftBgConfig.from_path(config_path)
+        assert reloaded.approve.drive_token_path == custom_path
+
+    def test_edit_does_not_bake_common_fields_into_approve(self, temp_dir):
+        """edit() must not permanently write do_email into approve's own
+        fields, so a later top-level change is still inherited on reload."""
+        config_path = temp_dir / "config.yaml"
+        SyftBgConfig(do_email="alice@test.com").save(config_path)
+
+        with SyftBgConfig.edit(config_path) as config:
+            config.approve.auto_approvals.enabled = False
+
+        raw = yaml.safe_load(config_path.read_text())
+        assert raw["approve"]["do_email"] is None
+
+        # Top-level do_email changes later; approve should still inherit it.
+        raw["do_email"] = "bob@test.com"
+        config_path.write_text(
+            yaml.safe_dump(raw, sort_keys=False, default_flow_style=False)
+        )
+
+        reloaded = SyftBgConfig.from_path(config_path)
+        assert reloaded.approve.do_email == "bob@test.com"
+
 
 class TestAutoApprovalsConfig:
     """Tests for AutoApprovalsConfig."""
diff --git a/tests/unit/syft_bg/test_criteria.py b/tests/unit/syft_bg/test_criteria.py
index d9eb95b980b..2477406bbe4 100644
--- a/tests/unit/syft_bg/test_criteria.py
+++ b/tests/unit/syft_bg/test_criteria.py
@@ -17,6 +17,7 @@
     _validate_job_against_object,
 )
 from syft_bg.approve.handlers.job import JobApprovalHandler
+from syft_bg.common.syft_bg_config import SyftBgConfig
 
 
 def _write_files(base_dir: Path, files: dict[str, str]) -> Path:
@@ -215,7 +216,7 @@ def _make_handler(config: AutoApprovalsConfig, tmp_dir: Path) -> JobApprovalHand
     it via its config_path-backed property.
     """
     config_path = tmp_dir / "config.yaml"
-    AutoApproveConfig(auto_approvals=config).save(config_path)
+    SyftBgConfig(approve=AutoApproveConfig(auto_approvals=config)).save(config_path)
     return JobApprovalHandler(client=MagicMock(), config_path=config_path)
 
 
diff --git a/tests/unit/syft_bg/test_email_auto_approve_flow.py b/tests/unit/syft_bg/test_email_auto_approve_flow.py
index 0fb0fed921c..7c122723102 100644
--- a/tests/unit/syft_bg/test_email_auto_approve_flow.py
+++ b/tests/unit/syft_bg/test_email_auto_approve_flow.py
@@ -8,10 +8,10 @@
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
-from syft_bg.approve.config import AutoApproveConfig
 from syft_bg.approve.orchestrator import ApprovalOrchestrator
 from syft_bg.common.config import get_default_paths
 from syft_bg.common.state import JsonStateManager
+from syft_bg.common.syft_bg_config import SyftBgConfig
 from syft_bg.email_approve.gmail_message import GmailMessage
 from syft_bg.email_approve.handler import EmailApproveHandler
 from syft_bg.email_approve.monitor import EmailApproveMonitor
@@ -38,7 +38,12 @@ def _temp_config_paths():
         )
         with (
             patch("syft_bg.common.config.get_default_paths", return_value=patched),
+            patch("syft_bg.api.api.get_default_paths", return_value=patched),
+            patch("syft_bg.api.utils.get_default_paths", return_value=patched),
             patch("syft_bg.approve.config.get_default_paths", return_value=patched),
+            patch(
+                "syft_bg.common.syft_bg_config.get_default_paths", return_value=patched
+            ),
         ):
             yield patched
 
@@ -202,7 +207,7 @@ def test_email_auto_approve_creates_object_and_approves_future_jobs():
         assert result["params"]["run"] == 1
 
         # -- Step 5: Verify auto-approve object was created correctly --
-        config = AutoApproveConfig.load()
+        config = SyftBgConfig.load().approve
         obj = config.auto_approvals.objects[job_name]
         content_names = {e.relative_path for e in obj.file_contents}
         assert content_names == {"main.py"}
@@ -225,7 +230,7 @@ def test_email_auto_approve_creates_object_and_approves_future_jobs():
         assert second_job.status == "pending"
 
         # -- Step 7: Run approval orchestrator — should auto-approve --
-        approve_config = AutoApproveConfig.load()
+        approve_config = SyftBgConfig.load().approve
         approve_config.do_email = do_manager.email
         approve_config.syftbox_root = do_manager.syftbox_folder
         approve_config.approve_state_path = tmp / "approve_state.json"
diff --git a/tests/unit/syft_bg/test_ensure_running.py b/tests/unit/syft_bg/test_ensure_running.py
new file mode 100644
index 00000000000..9cbc115fa26
--- /dev/null
+++ b/tests/unit/syft_bg/test_ensure_running.py
@@ -0,0 +1,100 @@
+"""Regression tests for the review-flagged unlocked read-modify-write bugs
+in ensure_running() and save_gcp_project_id() (both now route through the
+locked SyftBgConfig.edit() instead of a bare load()/save() pair)."""
+
+from dataclasses import replace
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from syft_bg.api.api import ensure_running
+from syft_bg.api.utils import save_gcp_project_id
+from syft_bg.common.config import get_default_paths
+from syft_bg.common.syft_bg_config import SyftBgConfig
+
+
+class TestEnsureRunning:
+    def test_no_config_file_does_not_create_one(self, temp_dir):
+        config_path = temp_dir / "config.yaml"
+        patched = replace(get_default_paths(), config=config_path)
+
+        with patch("syft_bg.api.api.get_default_paths", return_value=patched):
+            ensure_running(["approve"])
+
+        assert not config_path.exists()
+
+    def test_persists_settings_through_locked_edit(self, temp_dir):
+        config_path = temp_dir / "config.yaml"
+        SyftBgConfig().save(config_path)
+        patched = replace(get_default_paths(), config=config_path)
+
+        mock_service = MagicMock()
+        mock_service.is_running.return_value = True
+        mock_manager = MagicMock()
+        mock_manager.get_service.return_value = mock_service
+
+        with (
+            patch("syft_bg.api.api.get_default_paths", return_value=patched),
+            patch("syft_bg.api.api.ServiceManager", return_value=mock_manager),
+        ):
+            ensure_running({"approve": {"interval": 42}})
+
+        reloaded = SyftBgConfig.from_path(config_path)
+        assert reloaded.approve.interval == 42
+
+    def test_unknown_service_does_not_save(self, temp_dir):
+        config_path = temp_dir / "config.yaml"
+        SyftBgConfig(do_email="a@test.com").save(config_path)
+        patched = replace(get_default_paths(), config=config_path)
+
+        mock_manager = MagicMock()
+        mock_manager.get_service.return_value = None
+
+        with (
+            patch("syft_bg.api.api.get_default_paths", return_value=patched),
+            patch("syft_bg.api.api.ServiceManager", return_value=mock_manager),
+        ):
+            with pytest.raises(ValueError):
+                ensure_running({"bogus": {}})
+
+        reloaded = SyftBgConfig.from_path(config_path)
+        assert reloaded.do_email == "a@test.com"
+
+
+class TestSaveGcpProjectId:
+    def test_noop_when_config_missing(self, temp_dir):
+        config_path = temp_dir / "config.yaml"
+        patched = replace(get_default_paths(), config=config_path)
+
+        with (
+            patch("syft_bg.api.utils.get_default_paths", return_value=patched),
+            patch(
+                "syft_bg.api.utils.get_project_id_from_credentials",
+                return_value="proj-123",
+            ),
+        ):
+            save_gcp_project_id(Path("/fake/credentials.json"))
+
+        assert not config_path.exists()
+
+    def test_sets_project_id_when_config_exists(self, temp_dir):
+        config_path = temp_dir / "config.yaml"
+        SyftBgConfig().save(config_path)
+        patched = replace(get_default_paths(), config=config_path)
+
+        with (
+            patch("syft_bg.api.utils.get_default_paths", return_value=patched),
+            patch(
+                "syft_bg.common.syft_bg_config.get_default_paths",
+                return_value=patched,
+            ),
+            patch(
+                "syft_bg.api.utils.get_project_id_from_credentials",
+                return_value="proj-123",
+            ),
+        ):
+            save_gcp_project_id(Path("/fake/credentials.json"))
+
+        reloaded = SyftBgConfig.from_path(config_path)
+        assert reloaded.email_approve.gcp_project_id == "proj-123"
diff --git a/tests/unit/syft_bg/test_init.py b/tests/unit/syft_bg/test_init.py
new file mode 100644
index 00000000000..996742f1dfc
--- /dev/null
+++ b/tests/unit/syft_bg/test_init.py
@@ -0,0 +1,69 @@
+"""Tests for syft_bg.api.api.init() and its OAuth token handling."""
+
+from contextlib import contextmanager
+from dataclasses import replace
+from pathlib import Path
+from unittest.mock import patch
+
+from syft_bg.api.api import init
+from syft_bg.api.utils import move_token_to_syftbg_dir
+from syft_bg.common.config import get_default_paths
+from syft_bg.common.syft_bg_config import SyftBgConfig
+
+
+class TestMoveTokenToSyftbgDir:
+    """move_token_to_syftbg_dir must return the canonical (post-copy) path."""
+
+    def test_copies_and_returns_canonical_path(self, temp_dir):
+        source = temp_dir / "external_token.json"
+        source.write_text("{}")
+        syftbg_dir = temp_dir / "syftbg"
+
+        with patch("syft_bg.api.utils.get_syftbg_dir", return_value=syftbg_dir):
+            result = move_token_to_syftbg_dir(source)
+
+        canonical = syftbg_dir / "token.json"
+        assert result == canonical
+        assert canonical.exists()
+
+    def test_returns_same_path_when_already_canonical(self, temp_dir):
+        syftbg_dir = temp_dir / "syftbg"
+        syftbg_dir.mkdir()
+        canonical = syftbg_dir / "token.json"
+        canonical.write_text("{}")
+
+        with patch("syft_bg.api.utils.get_syftbg_dir", return_value=syftbg_dir):
+            result = move_token_to_syftbg_dir(canonical)
+
+        assert result == canonical
+
+
+@contextmanager
+def _patched_paths(tmp: Path):
+    """Redirect config, syft-bg dir, and gmail token default to a temp directory."""
+    original = get_default_paths()
+    syftbg_dir = tmp / "syftbg"
+    patched = replace(original, config=syftbg_dir / "config.yaml")
+    with (
+        patch("syft_bg.api.api.get_default_paths", return_value=patched),
+        patch("syft_bg.api.utils.get_syftbg_dir", return_value=syftbg_dir),
+        patch("syft_bg.common.syft_bg_config.get_default_paths", return_value=patched),
+    ):
+        yield patched
+
+
+class TestInitTokenPath:
+    """api.init(token_path=...) must set both gmail and drive token paths."""
+
+    def test_sets_both_gmail_and_drive_token_path(self, temp_dir):
+        token_source = temp_dir / "external_token.json"
+        token_source.write_text("{}")
+
+        with _patched_paths(temp_dir) as patched:
+            init(do_email="alice@test.com", token_path=token_source)
+
+            config = SyftBgConfig.from_path(patched.config)
+
+        canonical_token_path = temp_dir / "syftbg" / "token.json"
+        assert config.drive_token_path == canonical_token_path
+        assert config.gmail_token_path == canonical_token_path
diff --git a/tests/unit/test_job_auto_approval.py b/tests/unit/test_job_auto_approval.py
index 342cd8ee754..08fc29a542a 100644
--- a/tests/unit/test_job_auto_approval.py
+++ b/tests/unit/test_job_auto_approval.py
@@ -8,8 +8,8 @@
 from unittest.mock import patch
 
 from syft_bg.api import auto_approve_job
-from syft_bg.approve.config import AutoApproveConfig
 from syft_bg.common.config import get_default_paths
+from syft_bg.common.syft_bg_config import SyftBgConfig
 from syft_client.job_auto_approval import auto_approve_and_run_jobs
 from syft_client.sync.syftbox_manager import SyftboxManager
 
@@ -27,7 +27,12 @@ def _temp_config_paths():
         )
         with (
             patch("syft_bg.common.config.get_default_paths", return_value=patched),
+            patch("syft_bg.api.api.get_default_paths", return_value=patched),
+            patch("syft_bg.api.utils.get_default_paths", return_value=patched),
             patch("syft_bg.approve.config.get_default_paths", return_value=patched),
+            patch(
+                "syft_bg.common.syft_bg_config.get_default_paths", return_value=patched
+            ),
         ):
             yield patched
 
@@ -145,7 +150,7 @@ def test_auto_approve_job_default_all_content_matched():
         result = auto_approve_job(job)
         assert result.success is True
 
-        config = AutoApproveConfig.load()
+        config = SyftBgConfig.load().approve
         obj = config.auto_approvals.objects[job.name]
         content_names = {e.relative_path for e in obj.file_contents}
         assert content_names == {"main.py", "data.json"}
@@ -167,7 +172,7 @@ def test_auto_approve_job_file_paths_only():
         result = auto_approve_job(job, file_paths=["data.json"])
         assert result.success is True
 
-        config = AutoApproveConfig.load()
+        config = SyftBgConfig.load().approve
         obj = config.auto_approvals.objects[job.name]
         assert [e.relative_path for e in obj.file_contents] == ["main.py"]
         assert obj.file_paths == ["data.json"]
@@ -186,7 +191,7 @@ def test_auto_approve_job_contents_only():
         result = auto_approve_job(job, contents=["main.py"])
         assert result.success is True
 
-        config = AutoApproveConfig.load()
+        config = SyftBgConfig.load().approve
         obj = config.auto_approvals.objects[job.name]
         assert [e.relative_path for e in obj.file_contents] == ["main.py"]
         assert obj.file_paths == []
@@ -205,7 +210,7 @@ def test_auto_approve_job_both_contents_and_file_paths():
         result = auto_approve_job(job, contents=["main.py"], file_paths=["data.json"])
         assert result.success is True
 
-        config = AutoApproveConfig.load()
+        config = SyftBgConfig.load().approve
         obj = config.auto_approvals.objects[job.name]
         assert [e.relative_path for e in obj.file_contents] == ["main.py"]
         assert obj.file_paths == ["data.json"]
@@ -258,7 +263,7 @@ def test_auto_approve_job_nested_directory():
         result = auto_approve_job(job)
         assert result.success is True
 
-        config = AutoApproveConfig.load()
+        config = SyftBgConfig.load().approve
         obj = config.auto_approvals.objects[job.name]
         entries = {e.relative_path: e for e in obj.file_contents}
         assert set(entries.keys()) == {"main.py", "subdir/helper.py", "data.json"}
@@ -285,7 +290,7 @@ def test_auto_approve_job_default_no_special_treatment_for_non_params_json():
         result = auto_approve_job(job)
         assert result.success is True
 
-        config = AutoApproveConfig.load()
+        config = SyftBgConfig.load().approve
         obj = config.auto_approvals.objects[job.name]
         content_names = {e.relative_path for e in obj.file_contents}
         assert content_names == {"main.py", "somefile.json"}
@@ -307,7 +312,7 @@ def test_auto_approve_job_default_params_json_is_name_only():
         result = auto_approve_job(job)
         assert result.success is True
 
-        config = AutoApproveConfig.load()
+        config = SyftBgConfig.load().approve
         obj = config.auto_approvals.objects[job.name]
         content_names = {e.relative_path for e in obj.file_contents}
         assert content_names == {"main.py"}