Keep service auth PostgreSQL-owned in ConfigMap mode (#1267)

* Preserve service auth identity in ConfigMap mode

* Make PostgreSQL authoritative for service auth

* Hydrate service auth in ConfigMap integration tests

Author: jiaenren

src/service/core/config/configmap_loader.py

@@ -131,10 +131,9 @@ def __init__(
         backend_queue_updater: Callable[..., bool] | None = None,
         backend_test_updater: Callable[..., bool] | None = None,
     ):
-        # Kept for existing service/test wiring. ConfigMap loads must not
-        # hydrate ConfigMap-managed values from Postgres.
-        del postgres
         self._config_file_path = config_file_path
+        self._postgres = postgres
+        self._db_service_auth: Dict[str, Any] | None = None
         self._event_recorder = event_recorder
         self._enable_reconciliation = enable_reconciliation
         self._backend_queue_updater = backend_queue_updater
@@ -239,11 +238,19 @@ def _load_and_apply(self) -> LoadResult:
         # Dataset config is deprecated; tolerate stale ConfigMap blocks without loading them.
         managed_configs.pop('dataset', None)
 
+        # JWT signing identity is PostgreSQL-owned. Discard ConfigMap input before
+        # resolving secret references so it is never interpreted as runtime auth.
+        service_config = managed_configs.get('service')
+        if isinstance(service_config, dict):
+            service_config.pop('service_auth', None)
+
         # Resolve secret file references (reads mounted K8s Secret files)
         for section in managed_configs.values():
             if isinstance(section, dict):
                 _resolve_secret_file_references(section)
 
+        self._hydrate_service_auth(managed_configs)
+
         validation_errors = _validate_configmap_runtime_contract(managed_configs)
         validation_errors.extend(_validate_configs(managed_configs))
         if validation_errors:
@@ -282,6 +289,29 @@ def _load_and_apply(self) -> LoadResult:
 
         return LoadResult.SUCCESS
 
+    def _hydrate_service_auth(
+        self, managed_configs: Dict[str, Any],
+    ) -> None:
+        """Hydrate the stable JWT signing identity from Postgres."""
+        service_config = managed_configs.setdefault('service', {})
+        if not isinstance(service_config, dict):
+            return
+
+        if self._db_service_auth is None and self._postgres is not None:
+            persisted_service_config = self._postgres.get_service_configs()
+            persisted_service_config_dict = (
+                persisted_service_config.plaintext_dict(
+                    by_alias=True, exclude_unset=True))
+            persisted_service_auth = persisted_service_config_dict.get(
+                'service_auth')
+            if isinstance(persisted_service_auth, dict):
+                self._db_service_auth = copy.deepcopy(persisted_service_auth)
+
+        if self._db_service_auth is not None:
+            service_config['service_auth'] = copy.deepcopy(
+                self._db_service_auth)
+
+
 def start_config_watcher(
     config_file: str | None,
     postgres: connectors.PostgresConnector | None = None,
@@ -304,8 +334,10 @@ def start_config_watcher(
     are handled defensively by configmap_events; this gate just avoids
     cross-service duplication.)
 
-    ConfigMap mode is authoritative for service config. The watcher does
-    not read config rows from Postgres.
+    ConfigMap mode is authoritative for managed config except service auth.
+    ConfigMap-supplied service auth is ignored. On the first load, the watcher
+    recovers the stable signing identity from Postgres; reloads preserve only
+    that watcher-local, DB-derived identity.
     """
     if not config_file:
         return None
@@ -723,12 +755,17 @@ def _validate_configmap_runtime_contract(
 ) -> List[str]:
     """Validate runtime fields that ConfigMap mode must own.
 
-    ConfigMap mode does not hydrate config from DB. Fields that used to be
-    DB/runtime-derived but are required for backend side effects must therefore
-    be present in the ConfigMap.
+    Runtime fields must be present after explicit secret resolution and
+    PostgreSQL auth hydration.
     """
     errors: List[str] = []
 
+    service_config = managed_configs.get('service')
+    if (not isinstance(service_config, dict)
+            or 'service_auth' not in service_config):
+        errors.append(
+            'service.service_auth: required from PostgreSQL in ConfigMap mode')
+
     backends = managed_configs.get('backends', {})
     if isinstance(backends, dict):
         for backend_name, backend_config in backends.items():

src/service/core/config/tests/test_configmap_loader_integration.py

@@ -327,7 +327,8 @@ def test_watcher_load_populates_snapshot(self):
                 mode='w', suffix='.yaml', delete=False) as temp_file:
             yaml.dump(_with_service_auth(config), temp_file)
         try:
-            watcher = configmap_loader.ConfigMapWatcher(temp_file.name)
+            watcher = configmap_loader.ConfigMapWatcher(
+                temp_file.name, self._get_postgres())
             result = watcher._load_and_apply()
             self.assertTrue(result)
 
@@ -381,7 +382,8 @@ def test_watcher_resolves_pool_parsed_fields(self):
                 mode='w', suffix='.yaml', delete=False) as temp_file:
             yaml.dump(_with_service_auth(config), temp_file)
         try:
-            watcher = configmap_loader.ConfigMapWatcher(temp_file.name)
+            watcher = configmap_loader.ConfigMapWatcher(
+                temp_file.name, self._get_postgres())
             result = watcher._load_and_apply()
             self.assertTrue(result)