fix(photon): serialize auth.json writes with the shared cross-process lock

store_photon_token/store_project_credentials/store_user_numbers read-modify-
wrote auth.json without hermes_cli/auth.py's _auth_store_lock(), the
cross-process flock every other writer of that file (credential_pool
refresh, model_switch, fallback_cmd, nous_portal adapter, etc.) already
holds. A concurrent write from either side during the unlocked
load-mutate-save window silently drops the other side's update.

plugins/platforms/photon/auth.py

@@ -170,11 +170,14 @@ def load_photon_token() -> Optional[str]:
 
 def store_photon_token(token: str) -> None:
     """Persist a dashboard bearer token under ``credential_pool.photon``."""
-    auth = _load_auth()
-    auth.setdefault("credential_pool", {})["photon"] = [
-        {"access_token": token, "issued_at": int(time.time())}
-    ]
-    _save_auth(auth)
+    from hermes_cli.auth import _auth_store_lock
+
+    with _auth_store_lock():
+        auth = _load_auth()
+        auth.setdefault("credential_pool", {})["photon"] = [
+            {"access_token": token, "issued_at": int(time.time())}
+        ]
+        _save_auth(auth)
 
 
 def load_project_credentials() -> Tuple[Optional[str], Optional[str]]:
@@ -239,18 +242,21 @@ def store_project_credentials(
     ``auth.json`` so management commands work even when ``.env`` hasn't been
     loaded into the current process.
     """
-    auth = _load_auth()
-    record: Dict[str, Any] = {
-        "spectrum_project_id": spectrum_project_id,
-        "project_secret": project_secret,
-        "issued_at": int(time.time()),
-    }
-    if dashboard_project_id:
-        record["dashboard_project_id"] = dashboard_project_id
-    if name:
-        record["name"] = name
-    auth.setdefault("credential_pool", {})["photon_project"] = [record]
-    _save_auth(auth)
+    from hermes_cli.auth import _auth_store_lock
+
+    with _auth_store_lock():
+        auth = _load_auth()
+        record: Dict[str, Any] = {
+            "spectrum_project_id": spectrum_project_id,
+            "project_secret": project_secret,
+            "issued_at": int(time.time()),
+        }
+        if dashboard_project_id:
+            record["dashboard_project_id"] = dashboard_project_id
+        if name:
+            record["name"] = name
+        auth.setdefault("credential_pool", {})["photon_project"] = [record]
+        _save_auth(auth)
     _persist_runtime_env(spectrum_project_id, project_secret)
 
 
@@ -264,18 +270,21 @@ def store_user_numbers(
     """Persist non-secret Photon user numbers for offline ``status`` output."""
     if not phone_number and not assigned_phone_number:
         return
-    auth = _load_auth()
-    record: Dict[str, Any] = {"issued_at": int(time.time())}
-    if phone_number:
-        record["phone_number"] = phone_number
-    if assigned_phone_number:
-        record["assigned_phone_number"] = assigned_phone_number
-    if user_id:
-        record["user_id"] = user_id
-    if dashboard_project_id:
-        record["dashboard_project_id"] = dashboard_project_id
-    auth.setdefault("credential_pool", {})["photon_user"] = [record]
-    _save_auth(auth)
+    from hermes_cli.auth import _auth_store_lock
+
+    with _auth_store_lock():
+        auth = _load_auth()
+        record: Dict[str, Any] = {"issued_at": int(time.time())}
+        if phone_number:
+            record["phone_number"] = phone_number
+        if assigned_phone_number:
+            record["assigned_phone_number"] = assigned_phone_number
+        if user_id:
+            record["user_id"] = user_id
+        if dashboard_project_id:
+            record["dashboard_project_id"] = dashboard_project_id
+        auth.setdefault("credential_pool", {})["photon_user"] = [record]
+        _save_auth(auth)
 
 
 def _persist_runtime_env(spectrum_project_id: str, project_secret: str) -> None:

tests/plugins/platforms/photon/test_auth.py

@@ -4,6 +4,8 @@ from __future__ import annotations
 import json
 import os
 import stat
+import threading
+import time
 from base64 import b64encode
 from pathlib import Path
 from typing import Any, Dict
@@ -265,6 +267,110 @@ def test_load_project_credentials_env_override(
     assert secret == "secret-env"
 
 
+# ---------------------------------------------------------------------------
+# Cross-process auth.json lock (issue: photon wrote auth.json without the
+# cross-process lock hermes_cli/auth.py's ~15 other writers all use, so a
+# concurrent refresh from elsewhere could silently lose photon's update or
+# vice versa).
+
+def _hold_auth_lock_then_release(hold_event: threading.Event, release_event: threading.Event) -> None:
+    from hermes_cli.auth import _auth_store_lock
+
+    with _auth_store_lock():
+        hold_event.set()
+        release_event.wait(timeout=5)
+
+
+@pytest.mark.parametrize(
+    "call",
+    [
+        lambda: photon_auth.store_photon_token("blocked-token"),
+        lambda: photon_auth.store_user_numbers(phone_number="+15551234567"),
+    ],
+)
+def test_store_functions_block_on_auth_store_lock(
+    tmp_hermes_home: Path, call,
+) -> None:
+    """store_* writers must serialize against hermes_cli.auth's cross-process lock.
+
+    Before the fix, photon's store_* functions never acquired
+    ``_auth_store_lock`` at all, so a concurrent writer elsewhere in the
+    process (e.g. a Nous OAuth token refresh) could interleave with photon's
+    load-mutate-save cycle and silently drop one side's update. This test
+    proves the lock is actually taken (not just imported) by holding it on
+    a background thread and confirming the photon writer blocks until it is
+    released.
+    """
+    holding = threading.Event()
+    release = threading.Event()
+    holder = threading.Thread(
+        target=_hold_auth_lock_then_release, args=(holding, release), daemon=True,
+    )
+    holder.start()
+    try:
+        assert holding.wait(timeout=5), "background thread never acquired the auth lock"
+
+        finished = threading.Event()
+
+        def _run_call() -> None:
+            call()
+            finished.set()
+
+        caller = threading.Thread(target=_run_call, daemon=True)
+        caller.start()
+        try:
+            # The lock is held by the background thread — the store_* call
+            # must not complete while it waits for the lock.
+            assert not finished.wait(timeout=0.3), (
+                "store_* call completed while a concurrent writer held the "
+                "auth.json lock — it is not actually taking the lock"
+            )
+        finally:
+            release.set()
+            caller.join(timeout=5)
+        assert finished.is_set(), "store_* call never completed after the lock was released"
+    finally:
+        release.set()
+        holder.join(timeout=5)
+
+
+def test_store_project_credentials_blocks_on_auth_store_lock(
+    tmp_hermes_home: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(photon_auth, "_persist_runtime_env", lambda *a, **k: None)
+    holding = threading.Event()
+    release = threading.Event()
+    holder = threading.Thread(
+        target=_hold_auth_lock_then_release, args=(holding, release), daemon=True,
+    )
+    holder.start()
+    try:
+        assert holding.wait(timeout=5), "background thread never acquired the auth lock"
+
+        finished = threading.Event()
+
+        def _run_call() -> None:
+            photon_auth.store_project_credentials(
+                spectrum_project_id="sp-blocked", project_secret="secret-blocked",
+            )
+            finished.set()
+
+        caller = threading.Thread(target=_run_call, daemon=True)
+        caller.start()
+        try:
+            assert not finished.wait(timeout=0.3), (
+                "store_project_credentials completed while a concurrent writer "
+                "held the auth.json lock — it is not actually taking the lock"
+            )
+        finally:
+            release.set()
+            caller.join(timeout=5)
+        assert finished.is_set()
+    finally:
+        release.set()
+        holder.join(timeout=5)
+
+
 # ---------------------------------------------------------------------------
 # Device login flow