feat(worker): edit admin users on a live worker

bioengine/worker/worker.py

@@ -1,4 +1,5 @@
 import asyncio
+import json
 import logging
 import os
 import time
@@ -262,6 +263,9 @@ def __init__(
             f"Initializing {self.__class__.__name__} v{__version__} with mode '{mode}'"
         )
 
+        self._admin_users_file = self.workspace_dir / "admin_users.json"
+        self._load_persisted_admin_users()
+
         # Hypha server configuration
         self.server_url = server_url
         self.server: Optional[RemoteService] = None
@@ -271,6 +275,7 @@ def __init__(
         self.client_id = client_id
         self.service_id = "bioengine-worker"
         self.worker_name = worker_name
+        self._worker_user_email = None
 
         # Worker state management
         self.start_time = None
@@ -437,6 +442,56 @@ async def _fetch_geo_location(self) -> None:
             except Exception as e:
                 self.logger.warning(f"Failed to fetch geo coordinates: {e}")
 
+    def _load_persisted_admin_users(self) -> None:
+        """Overlay the persisted admin users on the ``admin_users`` startup seed.
+
+        The seed is replayed verbatim on every restart, so without this a user
+        added at runtime disappears on the next pod roll and a removed one comes
+        back — the silent rollback that startup flags plus out-of-band runtime
+        state always produce. The overlay wins and the divergence is logged.
+        """
+        if not self._admin_users_file.exists():
+            return
+
+        try:
+            persisted = json.loads(self._admin_users_file.read_text())
+            if not isinstance(persisted, list) or not all(
+                isinstance(user, str) for user in persisted
+            ):
+                raise ValueError("expected a list of strings")
+        except Exception as e:
+            self.logger.error(
+                f"Ignoring unreadable admin users file '{self._admin_users_file}': {e}. "
+                f"Falling back to the startup admin users: {self.admin_users}"
+            )
+            return
+
+        added = [user for user in persisted if user not in self.admin_users]
+        removed = [user for user in self.admin_users if user not in persisted]
+        if added or removed:
+            self.logger.info(
+                f"Admin users from '{self._admin_users_file}' override the startup list "
+                f"— added at runtime: {added or 'none'}, removed at runtime: {removed or 'none'}."
+            )
+        self.admin_users[:] = persisted
+
+    def _persist_admin_users(self, admin_users: List[str]) -> None:
+        """Write the admin users so the next restart overlays them on the seed.
+
+        Called before the in-memory list changes: a granted permission that only
+        exists in memory would revert on the next restart without ever failing.
+        """
+        try:
+            self._admin_users_file.parent.mkdir(parents=True, exist_ok=True)
+            tmp_file = self._admin_users_file.with_suffix(".json.tmp")
+            tmp_file.write_text(json.dumps(admin_users, indent=2))
+            os.replace(tmp_file, self._admin_users_file)
+        except Exception as e:
+            raise RuntimeError(
+                f"Failed to persist admin users to '{self._admin_users_file}': {e}. "
+                "Admin users are unchanged."
+            )
+
     async def _ping_data_server(self) -> None:
         """Ping the dataset server with retries to verify connectivity."""
         # No data server configured
@@ -606,6 +661,7 @@ async def _connect_to_server(self) -> None:
         if user_email in self.admin_users:
             self.admin_users.remove(user_email)
         self.admin_users.insert(0, user_email)
+        self._worker_user_email = user_email
 
         # Create admin context for internal operations
         self._admin_context = create_context(user_id, user_email)
@@ -638,6 +694,9 @@ async def _register_bioengine_worker_service(self) -> None:
             "stop_worker": self.stop,  # Requires admin permissions
             "check_access": self.check_access,
             "get_logs": self.get_logs,  # Requires admin permissions
+            "list_admin_users": self.list_admin_users,  # Requires admin permissions
+            "add_admin_user": self.add_admin_user,  # Requires admin permissions
+            "remove_admin_user": self.remove_admin_user,  # Requires admin permissions
             # 📦 Dataset management
             "list_datasets": self.list_datasets,
             # 🧮 Code execution
@@ -1168,6 +1227,154 @@ async def get_logs(
         except Exception as e:
             raise RuntimeError(f"Failed to read log file {self.log_file}: {e}")
 
+    @schema_method
+    async def list_admin_users(
+        self,
+        context: Dict[str, Any] = Field(
+            ...,
+            description="Authentication context containing user information, automatically provided by Hypha during service calls.",
+        ),
+    ) -> List[str]:
+        """List the users with admin permissions on this BioEngine worker.
+
+        Requires admin permissions.
+
+        Returns:
+            List[str]: The current admin users, most recently connected first.
+        """
+        check_permissions(
+            context=context,
+            authorized_users=self.admin_users,
+            resource_name="listing BioEngine Worker admin users",
+        )
+        return list(self.admin_users)
+
+    @schema_method
+    async def add_admin_user(
+        self,
+        user: str = Field(
+            ...,
+            description="User ID or email address to grant admin permissions on this worker.",
+        ),
+        context: Dict[str, Any] = Field(
+            ...,
+            description="Authentication context containing user information, automatically provided by Hypha during service calls.",
+        ),
+    ) -> List[str]:
+        """Grant a user admin permissions on this BioEngine worker.
+
+        Takes effect immediately for every worker method — permissions are checked
+        per call — and survives a restart. Already-running applications keep the
+        authorized users they were deployed with; redeploy them to widen access.
+
+        Requires admin permissions.
+
+        Args:
+            user: User ID or email address to grant admin permissions to.
+            context: Authentication context (automatically provided by Hypha)
+
+        Returns:
+            List[str]: The updated admin users.
+
+        Raises:
+            PermissionError: If the caller is not an admin.
+            ValueError: If the user identifier is empty or the wildcard '*'.
+        """
+        check_permissions(
+            context=context,
+            authorized_users=self.admin_users,
+            resource_name="adding a BioEngine Worker admin user",
+        )
+
+        user = user.strip()
+        if not user:
+            raise ValueError("Admin user identifier must not be empty.")
+        if user == "*":
+            # Only the operator starting the worker gets to make it world-writable.
+            raise ValueError(
+                "Refusing to add the wildcard '*' as an admin user. Pass "
+                "--admin-users '*' at worker startup if that is intended."
+            )
+
+        if user not in self.admin_users:
+            self._persist_admin_users(self.admin_users + [user])
+            self.admin_users.append(user)
+            self.logger.info(
+                f"Added admin user '{user}' (requested by "
+                f"'{context['user'].get('email') or context['user'].get('id')}'). "
+                f"Admin users: {', '.join(self.admin_users)}"
+            )
+
+        return list(self.admin_users)
+
+    @schema_method
+    async def remove_admin_user(
+        self,
+        user: str = Field(
+            ...,
+            description="User ID or email address to revoke admin permissions from on this worker.",
+        ),
+        context: Dict[str, Any] = Field(
+            ...,
+            description="Authentication context containing user information, automatically provided by Hypha during service calls.",
+        ),
+    ) -> List[str]:
+        """Revoke a user's admin permissions on this BioEngine worker.
+
+        Takes effect immediately and survives a restart. Removing a user who is
+        not an admin is a no-op.
+
+        Requires admin permissions.
+
+        Args:
+            user: User ID or email address to revoke admin permissions from.
+            context: Authentication context (automatically provided by Hypha)
+
+        Returns:
+            List[str]: The updated admin users.
+
+        Raises:
+            PermissionError: If the caller is not an admin.
+            ValueError: If the removal would lock the caller out, remove the last
+                        admin, or drop the worker's own identity.
+        """
+        check_permissions(
+            context=context,
+            authorized_users=self.admin_users,
+            resource_name="removing a BioEngine Worker admin user",
+        )
+
+        user = user.strip()
+        caller = context["user"]
+        if user and user in (caller.get("id"), caller.get("email")):
+            raise ValueError(
+                f"Refusing to remove '{user}': a caller cannot revoke their own "
+                "admin permissions."
+            )
+        if user == self._worker_user_email:
+            # _connect_to_server re-inserts this on every reconnect, so removing
+            # it would revert silently instead of failing here.
+            raise ValueError(
+                f"Refusing to remove '{user}': it is the identity this worker "
+                "connects to Hypha with and would be restored on the next reconnect."
+            )
+        if user in self.admin_users and len(self.admin_users) == 1:
+            raise ValueError(
+                f"Refusing to remove '{user}': it is the last admin user and the "
+                "worker would be left with none."
+            )
+
+        if user in self.admin_users:
+            self._persist_admin_users([u for u in self.admin_users if u != user])
+            self.admin_users.remove(user)
+            self.logger.info(
+                f"Removed admin user '{user}' (requested by "
+                f"'{caller.get('email') or caller.get('id')}'). "
+                f"Admin users: {', '.join(self.admin_users)}"
+            )
+
+        return list(self.admin_users)
+
     @schema_method
     async def get_status(
         self,

docs/glossary.md

@@ -38,7 +38,7 @@ See **Permission model** for the full access-control picture; workspace membersh
 Access to a BioEngine worker and its apps is granted by **any** of three independent layers — they are not nested, and any single layer alone is sufficient.
 
 1. **Workspace membership.** Holding a Hypha token for the worker's workspace (either a user token where the user is a member, or a workspace-scoped token) grants full access to every operation on the worker and every method of every app within it. This is the broadest layer.
-2. **Worker `admin_users`.** A list of user IDs / email addresses passed at worker startup via `--admin-users`. These users can perform every admin operation on the worker (`deploy_app`, `stop_app`, `upload_app`, `delete_app`, `run_code`, etc.) **even without** a token for the worker's workspace. They are also auto-injected into every app's `authorized_users` on deploy, so admins can always call any method of any app on this worker.
+2. **Worker `admin_users`.** A list of user IDs / email addresses seeded at worker startup via `--admin-users` and editable on a live worker via `list_admin_users` / `add_admin_user` / `remove_admin_user` (admin-only; the runtime list is persisted and overrides the startup seed after a restart). These users can perform every admin operation on the worker (`deploy_app`, `stop_app`, `upload_app`, `delete_app`, `run_code`, etc.) **even without** a token for the worker's workspace. They are also auto-injected into every app's `authorized_users` on deploy — that injection happens once, at deploy time, so an admin added later reaches the worker API immediately but not the apps already running; redeploy those to widen access.
 3. **App `authorized_users`.** Set per app package in `manifest.yaml`, per-method. Each method's `authorized_users` is either `"*"` (public — anyone, including anonymous) or a list of specific user IDs / emails. This grants method-level access to users who have **neither** workspace membership nor worker-admin status.
 
 Identity for layers 2 and 3 is established by the caller's Hypha token (which carries `user_id` and `user_email`) — anonymous callers can only reach methods with `authorized_users: "*"`.