diff --git a/bioengine/worker/worker.py b/bioengine/worker/worker.py
index 10e8148e..79ebb069 100644
--- a/bioengine/worker/worker.py
+++ b/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,
diff --git a/docs/glossary.md b/docs/glossary.md
index 7681ff85..3bedcb81 100644
--- a/docs/glossary.md
+++ b/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: "*"`.
diff --git a/tests/worker/test_admin_users.py b/tests/worker/test_admin_users.py
new file mode 100644
index 00000000..d3f3c19b
--- /dev/null
+++ b/tests/worker/test_admin_users.py
@@ -0,0 +1,279 @@
+"""Editing the worker's admin users on a live worker.
+
+Permissions are checked per call against ``self.admin_users``, so the list is
+editable at runtime without re-registering the service — the only thing that
+was missing is an API. Two properties have to hold for that API to be safe:
+nobody can lock the worker (or themselves) out, and a runtime change must not
+silently revert on the next restart, when the ``--admin-users`` startup flag is
+replayed verbatim.
+"""
+
+import json
+
+import pytest
+
+from bioengine.utils import create_context
+from bioengine.worker.worker import BioEngineWorker
+
+ADMIN = create_context("admin-id", "admin@example.org")
+NEWCOMER = create_context("new-id", "new@example.org")
+OUTSIDER = create_context("outsider-id", "outsider@example.org")
+
+
+def _bare_worker(tmp_path, admin_users, **attrs):
+    """A worker with only the state the admin-user methods touch."""
+    worker = BioEngineWorker.__new__(BioEngineWorker)
+    worker.start_time = None  # silences __del__ on a half-built instance
+    worker.logger = _Logger()
+    worker.admin_users = list(admin_users)
+    worker._admin_users_file = tmp_path / "admin_users.json"
+    worker._worker_user_email = "worker@service.internal"
+    for key, value in attrs.items():
+        setattr(worker, key, value)
+    return worker
+
+
+class _Logger:
+    def __init__(self):
+        self.records = []
+
+    def _record(self, message):
+        self.records.append(str(message))
+
+    info = warning = error = debug = _record
+
+
+async def test_a_granted_admin_can_immediately_call_admin_methods(tmp_path):
+    worker = _bare_worker(tmp_path, ["admin@example.org"])
+
+    with pytest.raises(PermissionError):
+        await worker.list_admin_users(context=OUTSIDER)
+
+    await worker.add_admin_user(user="outsider@example.org", context=ADMIN)
+
+    assert await worker.list_admin_users(context=OUTSIDER) == [
+        "admin@example.org",
+        "outsider@example.org",
+    ]
+
+
+async def test_the_change_reaches_the_component_managers(tmp_path):
+    """The managers hold the same list object, so mutation must stay in place."""
+
+    class _Holder:
+        pass
+
+    worker = _bare_worker(tmp_path, ["admin@example.org"])
+    apps_manager = _Holder()
+    code_executor = _Holder()
+    apps_manager.admin_users = worker.admin_users
+    code_executor.admin_users = worker.admin_users
+
+    await worker.add_admin_user(user="new@example.org", context=ADMIN)
+    await worker.remove_admin_user(user="admin@example.org", context=NEWCOMER)
+
+    assert apps_manager.admin_users == ["new@example.org"]
+    assert code_executor.admin_users == ["new@example.org"]
+
+
+async def test_a_caller_cannot_revoke_their_own_admin_permissions(tmp_path):
+    worker = _bare_worker(tmp_path, ["admin@example.org", "other@example.org"])
+
+    with pytest.raises(ValueError, match="cannot revoke their own"):
+        await worker.remove_admin_user(user="admin@example.org", context=ADMIN)
+
+    assert worker.admin_users == ["admin@example.org", "other@example.org"]
+
+
+async def test_a_caller_cannot_revoke_themselves_by_user_id(tmp_path):
+    worker = _bare_worker(tmp_path, ["admin-id", "other@example.org"])
+
+    with pytest.raises(ValueError, match="cannot revoke their own"):
+        await worker.remove_admin_user(user="admin-id", context=ADMIN)
+
+
+async def test_the_last_admin_cannot_be_removed(tmp_path):
+    """Reachable on a worker started with ``--admin-users '*'``: every caller is
+    an admin, so nothing else stops one of them from revoking the only entry."""
+    worker = _bare_worker(tmp_path, ["*"])
+
+    with pytest.raises(ValueError, match="last admin user"):
+        await worker.remove_admin_user(user="*", context=OUTSIDER)
+
+    assert worker.admin_users == ["*"]
+
+
+async def test_the_workers_own_identity_cannot_be_removed(tmp_path):
+    """_connect_to_server re-inserts it, so removal would revert silently."""
+    worker = _bare_worker(
+        tmp_path, ["worker@service.internal", "admin@example.org"]
+    )
+
+    with pytest.raises(ValueError, match="restored on the next reconnect"):
+        await worker.remove_admin_user(user="worker@service.internal", context=ADMIN)
+
+    assert "worker@service.internal" in worker.admin_users
+
+
+async def test_the_wildcard_cannot_be_granted_over_rpc(tmp_path):
+    worker = _bare_worker(tmp_path, ["admin@example.org"])
+
+    with pytest.raises(ValueError, match="wildcard"):
+        await worker.add_admin_user(user="*", context=ADMIN)
+
+    assert worker.admin_users == ["admin@example.org"]
+
+
+async def test_an_empty_identifier_is_refused(tmp_path):
+    worker = _bare_worker(tmp_path, ["admin@example.org"])
+
+    with pytest.raises(ValueError, match="must not be empty"):
+        await worker.add_admin_user(user="   ", context=ADMIN)
+
+
+async def test_repeated_grants_and_revocations_are_idempotent(tmp_path):
+    worker = _bare_worker(tmp_path, ["admin@example.org"])
+
+    await worker.add_admin_user(user="new@example.org", context=ADMIN)
+    await worker.add_admin_user(user="new@example.org", context=ADMIN)
+    assert worker.admin_users == ["admin@example.org", "new@example.org"]
+
+    await worker.remove_admin_user(user="new@example.org", context=ADMIN)
+    result = await worker.remove_admin_user(user="new@example.org", context=ADMIN)
+    assert result == ["admin@example.org"]
+
+
+async def test_a_non_admin_cannot_read_or_change_the_admin_list(tmp_path):
+    worker = _bare_worker(tmp_path, ["admin@example.org"])
+
+    for call in (
+        worker.list_admin_users(context=OUTSIDER),
+        worker.add_admin_user(user="outsider@example.org", context=OUTSIDER),
+        worker.remove_admin_user(user="admin@example.org", context=OUTSIDER),
+    ):
+        with pytest.raises(PermissionError):
+            await call
+
+    assert worker.admin_users == ["admin@example.org"]
+
+
+async def test_a_grant_is_written_before_it_takes_effect(tmp_path):
+    worker = _bare_worker(tmp_path, ["admin@example.org"])
+
+    await worker.add_admin_user(user="new@example.org", context=ADMIN)
+
+    assert json.loads((tmp_path / "admin_users.json").read_text()) == [
+        "admin@example.org",
+        "new@example.org",
+    ]
+
+
+async def test_an_unwritable_store_leaves_the_admin_list_unchanged(tmp_path):
+    """A grant that only exists in memory would revert on restart without ever failing."""
+    worker = _bare_worker(tmp_path, ["admin@example.org"])
+    worker._admin_users_file = tmp_path / "not-a-dir" / "admin_users.json"
+    (tmp_path / "not-a-dir").write_text("this is a file, not a directory")
+
+    with pytest.raises(RuntimeError, match="Failed to persist admin users"):
+        await worker.add_admin_user(user="new@example.org", context=ADMIN)
+
+    assert worker.admin_users == ["admin@example.org"]
+
+
+def test_the_persisted_list_overrides_the_startup_seed(tmp_path):
+    """The seed is replayed on every restart; the runtime overlay has to win."""
+    worker = _bare_worker(tmp_path, ["seeded@example.org"])
+    (tmp_path / "admin_users.json").write_text(
+        json.dumps(["seeded@example.org", "added-at-runtime@example.org"])
+    )
+
+    worker._load_persisted_admin_users()
+
+    assert worker.admin_users == [
+        "seeded@example.org",
+        "added-at-runtime@example.org",
+    ]
+
+
+def test_a_runtime_revocation_is_not_undone_by_the_seed(tmp_path):
+    worker = _bare_worker(tmp_path, ["seeded@example.org", "revoked@example.org"])
+    (tmp_path / "admin_users.json").write_text(json.dumps(["seeded@example.org"]))
+
+    worker._load_persisted_admin_users()
+
+    assert worker.admin_users == ["seeded@example.org"]
+
+
+def test_the_divergence_from_the_seed_is_reported(tmp_path):
+    worker = _bare_worker(tmp_path, ["seeded@example.org", "revoked@example.org"])
+    (tmp_path / "admin_users.json").write_text(
+        json.dumps(["seeded@example.org", "added@example.org"])
+    )
+
+    worker._load_persisted_admin_users()
+
+    reported = "\n".join(worker.logger.records)
+    assert "added@example.org" in reported
+    assert "revoked@example.org" in reported
+
+
+def test_a_corrupt_store_falls_back_to_the_seed(tmp_path):
+    worker = _bare_worker(tmp_path, ["seeded@example.org"])
+    (tmp_path / "admin_users.json").write_text("{not json")
+
+    worker._load_persisted_admin_users()
+
+    assert worker.admin_users == ["seeded@example.org"]
+    assert any("Ignoring unreadable" in r for r in worker.logger.records)
+
+
+def test_a_store_of_the_wrong_shape_falls_back_to_the_seed(tmp_path):
+    worker = _bare_worker(tmp_path, ["seeded@example.org"])
+    (tmp_path / "admin_users.json").write_text(json.dumps({"admins": ["a@b.c"]}))
+
+    worker._load_persisted_admin_users()
+
+    assert worker.admin_users == ["seeded@example.org"]
+
+
+def test_no_store_leaves_the_seed_alone(tmp_path):
+    worker = _bare_worker(tmp_path, ["seeded@example.org"])
+
+    worker._load_persisted_admin_users()
+
+    assert worker.admin_users == ["seeded@example.org"]
+    assert worker.logger.records == []
+
+
+async def test_the_admin_methods_are_exposed_on_the_worker_service(tmp_path):
+    registered = {}
+
+    class _Server:
+        async def register_service(self, service):
+            registered.update(service)
+            return type("Info", (), {"id": "ws/client:bioengine-worker"})()
+
+    class _Cluster:
+        mode = "single-machine"
+
+    class _Component:
+        def __getattr__(self, name):
+            return lambda *args, **kwargs: None
+
+    worker = _bare_worker(
+        tmp_path,
+        ["admin@example.org"],
+        server=_Server(),
+        ray_cluster=_Cluster(),
+        apps_manager=_Component(),
+        code_executor=_Component(),
+        service_id="bioengine-worker",
+        worker_name="test-worker",
+        full_service_id="ws/client:bioengine-worker",
+    )
+
+    await worker._register_bioengine_worker_service()
+
+    assert registered["list_admin_users"] == worker.list_admin_users
+    assert registered["add_admin_user"] == worker.add_admin_user
+    assert registered["remove_admin_user"] == worker.remove_admin_user