diff --git a/compute_endpoint/globus_compute_endpoint/endpoint/config/config.py b/compute_endpoint/globus_compute_endpoint/endpoint/config/config.py
index 22be80565..941d12a51 100644
--- a/compute_endpoint/globus_compute_endpoint/endpoint/config/config.py
+++ b/compute_endpoint/globus_compute_endpoint/endpoint/config/config.py
@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import inspect
+import os
+import pathlib
 import warnings
 
 from globus_compute_endpoint.engines import HighThroughputEngine
@@ -147,6 +149,7 @@ def __init__(
         # Tuning info
         heartbeat_period=30,
         heartbeat_threshold=120,
+        identity_mapping_config_path: os.PathLike | None = None,
         idle_heartbeats_soft=0,
         idle_heartbeats_hard=5760,  # Two days, divided by `heartbeat_period`
         detach_endpoint=True,
@@ -194,6 +197,11 @@ def __init__(
         self.multi_user = multi_user is True
         self.force_mu_allow_same_user = force_mu_allow_same_user is True
         self.mu_child_ep_grace_period_s = mu_child_ep_grace_period_s
+        self.identity_mapping_config_path = identity_mapping_config_path
+        if self.identity_mapping_config_path:
+            _p = pathlib.Path(self.identity_mapping_config_path)
+            if not _p.exists():
+                warnings.warn(f"Identity mapping config path not found ({_p})")
 
         # Auth
         self.allowed_functions = allowed_functions
diff --git a/compute_endpoint/globus_compute_endpoint/endpoint/config/example_identity_mapping_config.json b/compute_endpoint/globus_compute_endpoint/endpoint/config/example_identity_mapping_config.json
new file mode 100644
index 000000000..28e9f53d3
--- /dev/null
+++ b/compute_endpoint/globus_compute_endpoint/endpoint/config/example_identity_mapping_config.json
@@ -0,0 +1,18 @@
+[
+  {
+    "comment": "For more examples, see: https://docs.globus.org/globus-connect-server/v5.4/identity-mapping-guide/",
+    "DATA_TYPE": "external_identity_mapping#1.0.0",
+    "command": ["/bin/false", "--some", "flag", "-a", "-b", "-c"]
+  },
+  {
+    "comment": "For more examples, see: https://docs.globus.org/globus-connect-server/v5.4/identity-mapping-guide/",
+    "DATA_TYPE": "expression_identity_mapping#1.0.0",
+    "mappings": [
+      {
+        "source": "{username}",
+        "match": "(.*)@example.com",
+        "output": "{0}"
+      }
+    ]
+  }
+]
diff --git a/compute_endpoint/globus_compute_endpoint/endpoint/config/model.py b/compute_endpoint/globus_compute_endpoint/endpoint/config/model.py
index cec6b7b5a..3ae44985c 100644
--- a/compute_endpoint/globus_compute_endpoint/endpoint/config/model.py
+++ b/compute_endpoint/globus_compute_endpoint/endpoint/config/model.py
@@ -3,12 +3,13 @@
 import typing as t
 from types import ModuleType
 
+import pydantic
 from globus_compute_endpoint import engines, strategies
 from parsl import addresses as parsl_addresses
 from parsl import channels as parsl_channels
 from parsl import launchers as parsl_launchers
 from parsl import providers as parsl_providers
-from pydantic import BaseModel, validator
+from pydantic import BaseModel, FilePath, validator
 
 
 def _validate_import(field: str, package: ModuleType):
@@ -94,7 +95,7 @@ class Config:
 
 
 class ConfigModel(BaseConfigModel):
-    engine: EngineModel
+    engine: t.Optional[EngineModel]
     display_name: t.Optional[str]
     environment: t.Optional[str]
     funcx_service_address: t.Optional[str]
@@ -102,6 +103,7 @@ class ConfigModel(BaseConfigModel):
     allowed_functions: t.Optional[t.List[str]]
     heartbeat_period: t.Optional[int]
     heartbeat_threshold: t.Optional[int]
+    identity_mapping_config_path: t.Optional[FilePath]
     idle_heartbeats_soft: t.Optional[int]
     idle_heartbeats_hard: t.Optional[int]
     detach_endpoint: t.Optional[bool]
@@ -114,10 +116,26 @@ class ConfigModel(BaseConfigModel):
 
     _validate_engine = _validate_params("engine")
 
+    @pydantic.root_validator
+    @classmethod
+    def _validate(cls, values):
+        is_mt = values.get("multi_user") is True
+
+        if is_mt:
+            msg_engine = "no engine if multi-user"
+            msg_identity = "multi-user requires identity_mapping_config_path"
+        else:
+            msg_engine = "missing engine"
+            msg_identity = "identity_mapping_config_path should not be specified"
+        assert is_mt is not bool(values.get("engine")), msg_engine
+        assert is_mt is bool(values.get("identity_mapping_config_path")), msg_identity
+        return values
+
     def dict(self, *args, **kwargs):
         # Slight modification is needed here since we still
         # store the engine/executor in a list named executors
         ret = super().dict(*args, **kwargs)
-        executor = ret.pop("engine")
-        ret["executors"] = [executor]
+
+        engine = ret.pop("engine", None)
+        ret["executors"] = [engine] if engine else None
         return ret
diff --git a/compute_endpoint/globus_compute_endpoint/endpoint/endpoint.py b/compute_endpoint/globus_compute_endpoint/endpoint/endpoint.py
index e946b4e4b..87428f727 100644
--- a/compute_endpoint/globus_compute_endpoint/endpoint/endpoint.py
+++ b/compute_endpoint/globus_compute_endpoint/endpoint/endpoint.py
@@ -66,6 +66,12 @@ def user_config_schema_path(endpoint_dir: pathlib.Path) -> pathlib.Path:
     def _user_environment_path(endpoint_dir: pathlib.Path) -> pathlib.Path:
         return endpoint_dir / "user_environment.yaml"
 
+    @staticmethod
+    def _example_identity_mapping_configuration_path(
+        endpoint_dir: pathlib.Path,
+    ) -> pathlib.Path:
+        return endpoint_dir / "example_identity_mapping_config.json"
+
     @staticmethod
     def update_config_file(
         original_path: pathlib.Path,
@@ -85,6 +91,12 @@ def update_config_file(
 
         if multi_user:
             config_dict["multi_user"] = multi_user
+            config_dict.pop("engine", None)
+            config_dict["identity_mapping_config_path"] = str(
+                Endpoint._example_identity_mapping_configuration_path(
+                    target_path.parent
+                )
+            )
 
         config_text = yaml.safe_dump(config_dict)
         target_path.write_text(config_text)
@@ -133,6 +145,7 @@ def init_endpoint_dir(
             if multi_user:
                 # template must be readable by user-endpoint processes (see
                 # endpoint_manager.py)
+                owner_only = 0o0600
                 world_readable = 0o0644 & ((0o0777 - user_umask) | 0o0444)
                 world_executable = 0o0711 & ((0o0777 - user_umask) | 0o0111)
                 endpoint_dir.chmod(world_executable)
@@ -140,17 +153,25 @@ def init_endpoint_dir(
                 src_user_tmpl_path = package_dir / "config/user_config_template.yaml"
                 src_user_schem_path = package_dir / "config/user_config_schema.json"
                 src_user_env_path = package_dir / "config/user_environment.yaml"
+                src_example_idmap_path = (
+                    package_dir / "config/example_identity_mapping_config.json"
+                )
                 dst_user_tmpl_path = Endpoint.user_config_template_path(endpoint_dir)
                 dst_user_schem_path = Endpoint.user_config_schema_path(endpoint_dir)
                 dst_user_env_path = Endpoint._user_environment_path(endpoint_dir)
+                dst_idmap_conf_path = (
+                    Endpoint._example_identity_mapping_configuration_path(endpoint_dir)
+                )
 
                 shutil.copy(src_user_tmpl_path, dst_user_tmpl_path)
                 shutil.copy(src_user_schem_path, dst_user_schem_path)
                 shutil.copy(src_user_env_path, dst_user_env_path)
+                shutil.copy(src_example_idmap_path, dst_idmap_conf_path)
 
                 dst_user_tmpl_path.chmod(world_readable)
                 dst_user_schem_path.chmod(world_readable)
                 dst_user_env_path.chmod(world_readable)
+                dst_idmap_conf_path.chmod(owner_only)
 
         finally:
             os.umask(user_umask)
@@ -177,10 +198,14 @@ def configure_endpoint(
         if multi_user:
             user_conf_tmpl_path = Endpoint.user_config_template_path(conf_dir)
             user_env_path = Endpoint._user_environment_path(conf_dir)
+            idmap_ex_conf_path = Endpoint._example_identity_mapping_configuration_path(
+                conf_dir
+            )
 
             print(f"Created multi-user profile for endpoint named <{ep_name}>")
             print(
                 f"\n\tConfiguration file: {config_path}\n"
+                f"\n\tExample identity mapping configuration: {idmap_ex_conf_path}\n"
                 f"\n\tUser endpoint configuration template: {user_conf_tmpl_path}"
                 f"\n\tUser endpoint environment variables: {user_env_path}"
                 "\n\nUse the `start` subcommand to run it:\n"
diff --git a/compute_endpoint/globus_compute_endpoint/endpoint/endpoint_manager.py b/compute_endpoint/globus_compute_endpoint/endpoint/endpoint_manager.py
index a5b61643b..fb1f9caa0 100644
--- a/compute_endpoint/globus_compute_endpoint/endpoint/endpoint_manager.py
+++ b/compute_endpoint/globus_compute_endpoint/endpoint/endpoint_manager.py
@@ -20,6 +20,7 @@
 
 import globus_compute_sdk as GC
 from cachetools import TTLCache
+from globus_compute_endpoint.endpoint.identity_mapper import PosixIdentityMapper
 from pydantic import BaseModel
 
 try:
@@ -114,6 +115,15 @@ def __init__(
 
         endpoint_uuid = Endpoint.get_or_create_endpoint_uuid(conf_dir, endpoint_uuid)
 
+        if not config.identity_mapping_config_path:
+            msg = (
+                "No identity mapping file specified; please specify"
+                " identity_mapping_config_path"
+            )
+            log.error(msg)
+            print(msg, file=sys.stderr)
+            exit(os.EX_OSFILE)
+
         if not reg_info:
             try:
                 client_options = {
@@ -188,11 +198,32 @@ def __init__(
                 " considered a very dangerous override -- please use with care,"
                 " especially if allowing this endpoint to be utilized by multiple"
                 " users."
-                f"\n  Endpoint (UID, GID): ({os.getuid()}, {os.getgid()}) "
+                f"\n  Endpoint (UID, GID): ({os.getuid()}, {os.getgid()})"
             )
         else:
             self._allow_same_user = not is_privileged(self._mu_user)
 
+        try:
+            self.identity_mapper = PosixIdentityMapper(
+                config.identity_mapping_config_path, self._endpoint_uuid_str
+            )
+
+        except PermissionError as e:
+            msg = f"({type(e).__name__}) {e}"
+            log.error(msg)
+            print(msg, file=sys.stderr)
+            exit(os.EX_NOPERM)
+
+        except Exception as e:
+            msg = (
+                f"({type(e).__name__}) {e} -- Unable to read identity mapping"
+                f" configuration from: {config.identity_mapping_config_path}"
+            )
+            log.debug(msg, exc_info=e)
+            log.error(msg)
+            print(msg, file=sys.stderr)
+            exit(os.EX_CONFIG)
+
         # sanitize passwords in logs
         log_reg_info = re.subn(r"://.*?@", r"://***:***@", repr(reg_info))
         log.debug(f"Registration information: {log_reg_info}")
@@ -337,6 +368,9 @@ def start(self):
         self._command_stop_event.set()
         self._kill_event.set()
 
+        if self.identity_mapper:
+            self.identity_mapper.stop_watching()
+
         os.killpg(os.getpgid(0), signal.SIGTERM)
 
         proc_uid, proc_gid = os.getuid(), os.getgid()
@@ -374,19 +408,6 @@ def start(self):
     def _event_loop(self):
         self._command.start()
 
-        local_user_lookup = {}
-        try:
-            with open("local_user_lookup.json") as f:
-                local_user_lookup = json.load(f)
-        except Exception as e:
-            msg = (
-                f"Unable to load local users ({e.__class__.__name__}) {e}\n"
-                "  Will be unable to respond to any commands; update the lookup file"
-                f" and either restart (stop, start) or SIGHUP ({os.getpid()}) this"
-                " endpoint."
-            )
-            log.error(msg)
-
         valid_method_name_re = re.compile(r"^cmd_[A-Za-z][0-9A-Za-z_]{0,99}$")
         max_skew_s = 180  # 3 minutes; ignore commands with out-of-date timestamp
         while not self._time_to_stop:
@@ -444,25 +465,43 @@ def _event_loop(self):
                 continue
 
             try:
-                globus_uuid = msg["globus_uuid"]
-                globus_username = msg["globus_username"]
+                effective_identity = msg["globus_effective_identity"]
+                identity_set = msg["globus_identity_set"]
             except Exception as e:
                 log.error(f"Invalid server command.  ({e.__class__.__name__}) {e}")
                 self._command.ack(d_tag)
                 continue
 
+            identity_for_log = (
+                f"\n  Globus effective identity: {effective_identity}"
+                f"\n  Globus identity set: {identity_set}"
+            )
             try:
-                local_user = local_user_lookup[globus_username]
+                local_username = self.identity_mapper.map_identity(identity_set)
+                if not local_username:
+                    raise LookupError()
+            except LookupError as e:
+                log.error(
+                    "Identity failed to map to a local user name."
+                    f"  ({e.__class__.__name__}) {e}{identity_for_log}"
+                )
+                self._command.ack(d_tag)
+                continue
             except Exception as e:
-                log.warning(f"Invalid or unknown user.  ({e.__class__.__name__}) {e}")
+                msg = "Unhandled error attempting to map user."
+                log.debug(f"{msg}{identity_for_log}", exc_info=e)
+                log.error(f"{msg}  ({e.__class__.__name__}) {e}{identity_for_log}")
                 self._command.ack(d_tag)
                 continue
 
             try:
-                local_user_rec = pwd.getpwnam(local_user)
+                local_user_rec = pwd.getpwnam(local_username)
+
             except Exception as e:
-                log.warning(
-                    f"Invalid or unknown local username.  ({e.__class__.__name__}) {e}"
+                log.error(
+                    f"({type(e).__name__}) {e}\n"
+                    "  Identity mapped to a local user name, but local user does not"
+                    f" exist.\n  Local user name: {local_username}{identity_for_log}"
                 )
                 self._command.ack(d_tag)
                 continue
@@ -477,24 +516,24 @@ def _event_loop(self):
 
                 command_func(local_user_rec, command_args, command_kwargs)
                 log.info(
-                    f"Command process successfully forked for '{globus_username}'"
-                    f" ('{globus_uuid}')."
+                    f"Command process successfully forked for '{local_username}'"
+                    f" (Globus effective identity: {effective_identity})."
                 )
-            except (InvalidCommandError, InvalidUserError) as err:
-                log.error(str(err))
+            except (InvalidCommandError, InvalidUserError) as e:
+                log.error(f"({type(e).__name__}) {e}{identity_for_log}")
 
             except Exception:
                 log.exception(
                     f"Unable to execute command: {command}\n"
                     f"    args: {command_args}\n"
-                    f"  kwargs: {command_kwargs}"
+                    f"  kwargs: {command_kwargs}{identity_for_log}"
                 )
             finally:
                 self._command.ack(d_tag)
 
     def cmd_start_endpoint(
         self,
-        local_user_rec: pwd.struct_passwd,
+        user_record: pwd.struct_passwd,
         args: list[str] | None,
         kwargs: dict | None,
     ):
@@ -513,15 +552,11 @@ def cmd_start_endpoint(
                     f"User endpoint {ep_name} is already running (pid: {p}); "
                     "caching arguments in case it's about to shut down"
                 )
-                self._cached_cmd_start_args[p] = (local_user_rec, args, kwargs)
+                self._cached_cmd_start_args[p] = (user_record, args, kwargs)
                 return
 
-        udir, uid, gid, uname = (
-            local_user_rec.pw_dir,
-            local_user_rec.pw_uid,
-            local_user_rec.pw_gid,
-            local_user_rec.pw_name,
-        )
+        udir, uid, gid = user_record.pw_dir, user_record.pw_uid, user_record.pw_gid
+        uname = user_record.pw_name
 
         if not self._allow_same_user:
             p_uname = self._mu_user.pw_name
@@ -553,7 +588,7 @@ def cmd_start_endpoint(
         if pid > 0:
             proc_args_s = f"({uname}, {ep_name}) {' '.join(proc_args)}"
             self._children[pid] = UserEndpointRecord(
-                ep_name=ep_name, local_user_info=local_user_rec, arguments=proc_args_s
+                ep_name=ep_name, local_user_info=user_record, arguments=proc_args_s
             )
             log.info(f"Creating new user endpoint (pid: {pid}) [{proc_args_s}]")
             return
@@ -693,7 +728,7 @@ def cmd_start_endpoint(
 
             # fcntl.F_GETPIPE_SZ is not available in Python versions less than 3.10
             F_GETPIPE_SZ = 1032
-            # 256 - Allow some head room for multiple kernel-specific factors
+            # 256 - Allow some headroom for multiple kernel-specific factors
             max_buf_size = fcntl.fcntl(write_handle, F_GETPIPE_SZ) - 256
             stdin_data_size = len(stdin_data)
             if stdin_data_size > max_buf_size:
diff --git a/compute_endpoint/globus_compute_endpoint/endpoint/identity_mapper.py b/compute_endpoint/globus_compute_endpoint/endpoint/identity_mapper.py
new file mode 100644
index 000000000..51af77736
--- /dev/null
+++ b/compute_endpoint/globus_compute_endpoint/endpoint/identity_mapper.py
@@ -0,0 +1,215 @@
+from __future__ import annotations
+
+import json
+import logging
+import os
+import pathlib
+import threading
+import typing as t
+
+from globus_identity_mapping.loader import load_mappers
+from globus_identity_mapping.protocol import IdentityMappingProtocol
+
+log = logging.getLogger(__name__)
+
+# From: https://docs.globus.org/globus-connect-server/v5.4/identity-mapping-guide/#command_line_options  # noqa
+POSIX_CONNECTOR_ID = "145812c8-decc-41f1-83cf-bb2a85a2a70b"
+
+
+class PosixIdentityMapper:
+    """
+    IdentityMapper is a wrapper around IdentityMappingProtocol implementations,
+    performing two main functions:
+
+        - auto-loading the correct IdentityMappingProtocol implementation class,
+          based on a configuration file's defined DATA_TYPE.  (See
+          https://docs.globus.org/globus-connect-server/v5.4/identity-mapping-guide/ )
+        - atomically updating the loaded configuration without requiring an
+          application restart
+
+    To construct, specify an identity-mapping configuration in a JSON file and
+    then pass the path to the constructor with a tool or utility identifier.
+    Example:
+
+        >>> import json, pathlib
+        >>> from globus_compute_endpoint.endpoint.identity_mapper import PosixIdentityMapper  # noqa
+
+        >>> identity_mapping_conf_path = "some/identity_mapping_configuration.json"
+        >>> im_path = pathlib.Path(identity_mapping_conf_path)  # for convenience
+        >>> print(im_path.read_text())
+        [
+          {
+            "DATA_TYPE": "expression_identity_mapping#1.0.0",
+            "mappings": [
+              {
+                "source": "{username}",
+                "match": "(.*)@university\\.example\\.edu",
+                "output": "{0}"
+              }
+            ]
+          }
+        ]
+
+        >>> some_utility_identifier = "some identifier"
+        >>> im = PosixIdentityMapper(
+        ...    identity_mapping_conf_path, some_utility_identifier
+        ... )
+
+    Currently, the DATA_TYPE must map to an implementing class in the
+    ``globus-identity-mapping`` Python library.
+
+    Now any valid Globus identity can be mapped to a local user via either the
+    ``.map_identity()`` method:
+
+        >>> globus_identity_set = json.loads(  # usually collected from an API ...
+        ... '''
+        ... [
+        ...     {
+        ...         "id": "1c6dbe57-4a3e-44e5-826e-0280003585ae",
+        ...         "sub": "1c6dbe57-4a3e-44e5-826e-0280003585ae",
+        ...         "organization": "Example Widgets, Co",
+        ...         "name": "Jessie Jess",
+        ...         "username": "jess@example.org",
+        ...         "identity_provider": "62dc25b7-c693-4528-831b-4557a7a0d1e4",
+        ...         "identity_provider_display_name": "Example Identities, Ltd",
+        ...         "email": "jess@example.org",
+        ...         "last_authentication": 1029384756
+        ...     },
+        ...     {
+        ...         "id": "d427502d-722b-4361-ad34-9327010a7b81",
+        ...         "sub": "d427502d-722b-4361-ad34-9327010a7b81",
+        ...         "name": "Jessie Jess",
+        ...         "username": "jessica.jazz.jess@university.example.edu",
+        ...         "identity_provider": "09a7033a-b53b-44e2-8c71-b60f0468df07",
+        ...         "identity_provider_display_name": "University of Example",
+        ...         "email": "jessica.jess@example-2.com",
+        ...         "last_authentication": 1629384750
+        ...     }
+        ... ]
+        ... '''
+        ... )
+
+        >>> im.map_identity(globus_identity_set)
+        'jessica.jazz.jess'
+
+    Meanwhile, if requirements change and the identity mapping configuration
+    must be changed, the changes will be atomically discovered:
+
+        >>> im_path.write_text('''
+        ... [
+        ...   {
+        ...     "DATA_TYPE": "expression_identity_mapping#1.0.0",
+        ...     "mappings": [
+        ...       {
+        ...         "source": "{username}",
+        ...         "match": "(.*)@example\\.org",
+        ...         "output": "{0}"
+        ...       }
+        ...     ]
+        ...   }
+        ... ]
+        ... ''')
+        >>> time.sleep(im.poll_interval_s)  # for example, wait until reloaded
+        >>> im.map_identity(globus_identity_set)
+        'jess'
+
+
+    """
+
+    def __init__(
+        self,
+        identity_mapping_config_path: os.PathLike,
+        endpoint_identifier: str,
+        poll_interval_s: float = 5.0,
+    ):
+        """
+        :param identity_mapping_config_path:
+        :param endpoint_identifier:
+        :param poll_interval_s:
+        """
+        self._time_to_stop = threading.Event()
+        self.config_path = pathlib.Path(identity_mapping_config_path)
+        self._config_stat = self.config_path.stat()
+        self._endpoint_identifier = endpoint_identifier
+        self.poll_interval_s = poll_interval_s
+
+        self._identity_mappings: list[IdentityMappingProtocol] = []
+
+        self.load_configuration()  # Executed on main thread
+
+        threading.Thread(target=self._poll_config, daemon=True).start()
+
+    def __del__(self):
+        self.stop_watching()
+
+    def stop_watching(self):
+        self._time_to_stop.set()
+
+    @property
+    def poll_interval_s(self) -> float:
+        return self._poll_interval_s
+
+    @poll_interval_s.setter
+    def poll_interval_s(self, new_interval_s: float) -> None:
+        self._poll_interval_s = max(0.5, new_interval_s)
+
+    def _poll_config(self):
+        fail_msg_fmt = (
+            "Unable to update identity configuration -- ({}) {}"
+            f"\n  Identity configuration path: {self.config_path}"
+        )
+
+        while not self._time_to_stop.wait(self.poll_interval_s):
+            try:
+                self._update_if_config_changed()
+            except Exception as e:
+                msg = fail_msg_fmt.format(type(e).__name__, e)
+                log.debug(msg, exc_info=e)
+                log.error(msg)
+        log.debug("Polling thread stops")
+
+    def _update_if_config_changed(self):
+        cstat = self._config_stat  # "current stat"
+        nstat = self.config_path.stat()  # "new stat"
+        cur_s = (cstat.st_ino, cstat.st_ctime, cstat.st_mtime, cstat.st_size)
+        new_s = (nstat.st_ino, nstat.st_ctime, nstat.st_mtime, nstat.st_size)
+
+        if cur_s == new_s:
+            # no change; suggests the file has *also* not changed
+            return
+
+        log.info(
+            "Identity mapping configuration change detected; rereading:"
+            f" {self.config_path}"
+        )
+        self.load_configuration()
+
+    def load_configuration(self):
+        self._config_stat = self.config_path.stat()
+        self.identity_mappings = json.loads(self.config_path.read_bytes())
+
+    @property
+    def identity_mappings(self) -> list[IdentityMappingProtocol]:
+        return self._identity_mappings
+
+    @identity_mappings.setter
+    def identity_mappings(self, mappings_list: t.Iterable[dict] | None):
+        self._identity_mappings = load_mappers(
+            mappings_list, POSIX_CONNECTOR_ID, self._endpoint_identifier
+        )
+
+    @identity_mappings.deleter
+    def identity_mappings(self):
+        self.identity_mappings = None
+
+    def map_identity(self, identity_set: t.Iterable[t.Mapping[str, str]]) -> str | None:
+        for mapper in self.identity_mappings:
+            for identity_data in identity_set:
+                try:
+                    identity = mapper.map_identity(identity_data)
+                except Exception as e:
+                    log.warning(f"Identity mapper failed -- ({type(e).__name__}) {e}")
+                    continue
+                if identity:
+                    return identity
+        return None
diff --git a/compute_endpoint/setup.py b/compute_endpoint/setup.py
index 75ad9e384..d0d66bc5f 100644
--- a/compute_endpoint/setup.py
+++ b/compute_endpoint/setup.py
@@ -8,6 +8,7 @@
     "globus-sdk",  # version will be bounded by `globus-compute-sdk`
     "globus-compute-sdk==2.6.0",
     "globus-compute-common==0.3.0a2",
+    "globus-identity-mapping==0.1.0",
     # table printing used in list-endpoints
     "texttable>=1.6.4,<2",
     # although psutil does not declare itself to use semver, it appears to offer
diff --git a/compute_endpoint/tests/integration/endpoint/endpoint/test_endpoint_manager.py b/compute_endpoint/tests/integration/endpoint/endpoint/test_endpoint_manager.py
index c7bb49f09..f85c9ecca 100644
--- a/compute_endpoint/tests/integration/endpoint/endpoint/test_endpoint_manager.py
+++ b/compute_endpoint/tests/integration/endpoint/endpoint/test_endpoint_manager.py
@@ -42,6 +42,7 @@ def test_setup_teardown(self, fs):
         fs.add_real_file(DEF_CONFIG_DIR / "user_config_template.yaml")
         fs.add_real_file(DEF_CONFIG_DIR / "user_config_schema.json")
         fs.add_real_file(DEF_CONFIG_DIR / "user_environment.yaml")
+        fs.add_real_file(DEF_CONFIG_DIR / "example_identity_mapping_config.json")
 
         yield
 
@@ -98,6 +99,9 @@ def test_configure_multi_user(self, mu):
             config_dict = yaml.safe_load(f)
         if mu:
             assert config_dict["multi_user"] == mu is True
+            assert "engine" not in config_dict
+            assert "identity_mapping_config_path" in config_dict
+            assert pathlib.Path(config_dict["identity_mapping_config_path"]).exists()
         else:
             assert "multi_user" not in config_dict
 
diff --git a/compute_endpoint/tests/integration/endpoint/endpoint/test_interchange.py b/compute_endpoint/tests/integration/endpoint/endpoint/test_interchange.py
index e83fea14b..d9794ed38 100644
--- a/compute_endpoint/tests/integration/endpoint/endpoint/test_interchange.py
+++ b/compute_endpoint/tests/integration/endpoint/endpoint/test_interchange.py
@@ -37,6 +37,7 @@ def mock_spt(mocker):
 def test_endpoint_id_conveyed_to_executor(funcx_dir):
     manager = Endpoint()
     config_dir = funcx_dir / "mock_endpoint"
+    expected_ep_id = "mock_endpoint_id"
 
     manager.configure_endpoint(config_dir, None)
 
@@ -46,11 +47,11 @@ def test_endpoint_id_conveyed_to_executor(funcx_dir):
     ic = EndpointInterchange(
         endpoint_config,
         reg_info={"task_queue_info": {}, "result_queue_info": {}},
-        endpoint_id="mock_endpoint_id",
+        endpoint_id=expected_ep_id,
     )
     ic.executor = engines.ThreadPoolEngine()  # test does not need a child process
     ic.start_engine()
-    assert ic.executor.endpoint_id == "mock_endpoint_id"
+    assert ic.executor.endpoint_id == expected_ep_id
     ic.executor.shutdown()
 
 
diff --git a/compute_endpoint/tests/unit/conftest.py b/compute_endpoint/tests/unit/conftest.py
new file mode 100644
index 000000000..d0d4fd797
--- /dev/null
+++ b/compute_endpoint/tests/unit/conftest.py
@@ -0,0 +1,4 @@
+def pytest_configure(config):
+    config.addinivalue_line(
+        "markers", "no_mock_pim: In test_endpointmanager_unit, disable autouse fixture"
+    )
diff --git a/compute_endpoint/tests/unit/test_cli_behavior.py b/compute_endpoint/tests/unit/test_cli_behavior.py
index a8e951b80..5dd12989a 100644
--- a/compute_endpoint/tests/unit/test_cli_behavior.py
+++ b/compute_endpoint/tests/unit/test_cli_behavior.py
@@ -349,7 +349,7 @@ def test_start_ep_incorrect_config_yaml(
     conf.unlink()
     conf.write_text("asdf: asdf")
     res = run_line(f"start {ep_name}", assert_exit_code=1)
-    assert "field required" in res.stderr
+    assert "missing engine" in res.stderr, "Engine: specification required!"
 
 
 def test_start_ep_incorrect_config_py(
@@ -390,6 +390,36 @@ def test_start_ep_config_py_takes_precedence(
     assert not read_config.called, "Key outcome: config.py takes precendence"
 
 
+def test_single_user_requires_engine_configured(mock_command_ensure, ep_name, run_line):
+    ep_dir = mock_command_ensure.endpoint_config_dir / ep_name
+    ep_dir.mkdir(parents=True)
+    data = {"config": ""}
+
+    config = {}
+    data["config"] = yaml.safe_dump(config)
+    rc = run_line(f"start {ep_name}", stdin=json.dumps(data), assert_exit_code=1)
+    assert "validation error" in rc.stderr
+    assert "missing engine" in rc.stderr
+
+    config = {"multi_user": False}
+    data["config"] = yaml.safe_dump(config)
+    rc = run_line(f"start {ep_name}", stdin=json.dumps(data), assert_exit_code=1)
+    assert "validation error" in rc.stderr
+    assert "missing engine" in rc.stderr
+
+
+def test_multi_user_enforces_no_engine(mock_command_ensure, ep_name, run_line):
+    ep_dir = mock_command_ensure.endpoint_config_dir / ep_name
+    ep_dir.mkdir(parents=True)
+    data = {"config": ""}
+
+    config = {"engine": {"type": "ThreadPoolEngine"}, "multi_user": True}
+    data["config"] = yaml.safe_dump(config)
+    rc = run_line(f"start {ep_name}", stdin=json.dumps(data), assert_exit_code=1)
+    assert "validation error" in rc.stderr
+    assert "no engine if multi" in rc.stderr
+
+
 @mock.patch("globus_compute_endpoint.endpoint.config.utils._load_config_py")
 def test_delete_endpoint(read_config, run_line, mock_cli_state, ep_name):
     run_line(f"delete {ep_name} --yes")
@@ -407,10 +437,9 @@ def test_die_with_parent_detached(
 
     if die_with_parent:
         run_line(f"start {ep_name} --die-with-parent")
-        assert not config.detach_endpoint
     else:
         run_line(f"start {ep_name}")
-        assert config.detach_endpoint
+    assert config.detach_endpoint is (not die_with_parent)
 
 
 def test_self_diagnostic(
diff --git a/compute_endpoint/tests/unit/test_endpoint_config.py b/compute_endpoint/tests/unit/test_endpoint_config.py
index 5b139bd4f..55005ffe5 100644
--- a/compute_endpoint/tests/unit/test_endpoint_config.py
+++ b/compute_endpoint/tests/unit/test_endpoint_config.py
@@ -1,7 +1,9 @@
 import typing as t
 
 import pytest
+from globus_compute_endpoint.endpoint.config import Config
 from globus_compute_endpoint.endpoint.config.model import ConfigModel
+from pydantic.error_wrappers import ValidationError
 
 
 @pytest.fixture
@@ -9,6 +11,16 @@ def config_dict():
     return {"engine": {"type": "HighThroughputEngine"}}
 
 
+@pytest.fixture
+def config_dict_mu(tmp_path):
+    idc = tmp_path / "idconf.json"
+    idc.write_text("[]")
+    return {
+        "identity_mapping_config_path": idc,
+        "multi_user": True,
+    }
+
+
 @pytest.mark.parametrize(
     "data",
     [
@@ -31,3 +43,59 @@ def test_config_model_tuple_conversions(config_dict: dict, data: t.Tuple[str, t.
     config_dict["engine"][field] = 50000
     with pytest.raises(ValueError):
         ConfigModel(**config_dict)
+
+
+def test_config_enforces_engine(config_dict):
+    del config_dict["engine"]
+    with pytest.raises(ValidationError) as pyt_exc:
+        ConfigModel(**config_dict)
+
+    assert "missing engine" in str(pyt_exc.value)
+
+
+def test_config_enforces_no_identity_mapping_conf(config_dict, tmp_path):
+    conf_p = tmp_path / "some file"
+    conf_p.write_text("[]")
+    config_dict["identity_mapping_config_path"] = conf_p
+    with pytest.raises(ValidationError) as pyt_exc:
+        ConfigModel(**config_dict)
+
+    assert "identity_mapping_config_path should not be specified" in str(pyt_exc.value)
+
+
+def test_mu_config_enforces_no_engine(config_dict_mu):
+    config_dict_mu["engine"] = {"type": "ThreadPoolEngine"}
+    with pytest.raises(ValidationError) as pyt_exc:
+        ConfigModel(**config_dict_mu)
+
+    assert "no engine if multi-user" in str(pyt_exc), pyt_exc
+
+
+def test_mu_config_requires_identity_mapping(config_dict_mu):
+    del config_dict_mu["identity_mapping_config_path"]
+    with pytest.raises(ValidationError) as pyt_exc:
+        ConfigModel(**config_dict_mu)
+
+    assert "requires identity_mapping_config_path" in str(pyt_exc.value)
+
+
+def test_mu_config_requires_identity_mapping_exists(config_dict_mu, tmp_path):
+    config_dict_mu["identity_mapping_config_path"] = tmp_path / "not exists file"
+    with pytest.raises(ValidationError) as pyt_exc:
+        ConfigModel(**config_dict_mu)
+
+    assert "not exists file" in str(pyt_exc.value)
+    assert "does not exist" in str(pyt_exc.value)
+
+
+def test_config_warns_bad_identity_mapping_path(mocker, config_dict_mu, tmp_path):
+    conf_p = tmp_path / "not exists file"
+    config_dict_mu["identity_mapping_config_path"] = conf_p
+    mock_warn = mocker.patch("globus_compute_endpoint.endpoint.config.config.warnings")
+    Config(**config_dict_mu)
+
+    warn_a = mock_warn.warn.call_args[0][0]
+    assert mock_warn.warn.called
+    assert "Identity mapping config" in warn_a
+    assert "path not found" in warn_a
+    assert str(conf_p) in warn_a, "expect include location of file in warning"
diff --git a/compute_endpoint/tests/unit/test_endpointmanager_unit.py b/compute_endpoint/tests/unit/test_endpointmanager_unit.py
index 88c96e6eb..26d68bcd3 100644
--- a/compute_endpoint/tests/unit/test_endpointmanager_unit.py
+++ b/compute_endpoint/tests/unit/test_endpointmanager_unit.py
@@ -60,8 +60,15 @@ def conf_dir(fs):
 
 
 @pytest.fixture
-def mock_conf():
-    yield Config(executors=[])
+def identity_map_path(conf_dir):
+    im_path = conf_dir / "some_identity_mapping_configuration.json"
+    im_path.write_text("[]")
+    yield im_path
+
+
+@pytest.fixture
+def mock_conf(identity_map_path):
+    yield Config(executors=[], identity_mapping_config_path=identity_map_path)
 
 
 @pytest.fixture
@@ -101,17 +108,31 @@ def mock_client(mocker):
     yield ep_uuid, mock_gcc
 
 
+@pytest.fixture(autouse=True)
+def mock_pim(request):
+    if "no_mock_pim" in request.keywords:
+        yield
+        return
+
+    with mock.patch(f"{_MOCK_BASE}PosixIdentityMapper") as mock_pim:
+        mock_pim.return_value = mock_pim
+        yield mock_pim
+
+
 @pytest.fixture
-def epmanager(mocker, conf_dir, mock_conf, mock_client):
+def epmanager(mocker, conf_dir, mock_conf, mock_client, mock_pim):
     ep_uuid, mock_gcc = mock_client
 
     # Needed to mock the pipe buffer size
     mocker.patch.object(fcntl, "fcntl", return_value=512)
+    mock_pim.map_identity.return_value = "an_account_that_doesnt_exist_abc123"
 
     em = EndpointManager(conf_dir, ep_uuid, mock_conf)
     em._command = mocker.Mock()
 
     yield conf_dir, mock_conf, mock_client, em
+    em.identity_mapper.stop_watching()
+    em.request_shutdown(None, None)
 
 
 @pytest.fixture
@@ -129,11 +150,9 @@ def create_response(status_code: int = 200, err_msg: str = "some error msg"):
 
 
 @pytest.fixture
-def successful_exec(mocker, epmanager, user_conf_template, fs):
-    # fs (pyfakefs) not used directly in this fixture, but is intentionally
-    # utilized in epmanager -> conf_dir.  It is *assumed* in this fixture,
-    # however (e.g., local_user_lookup.json), so make it an explicit detail.
+def successful_exec(mocker, epmanager, user_conf_template):
     mock_os = mocker.patch(f"{_MOCK_BASE}os")
+    mock_pwnam = mocker.patch(f"{_MOCK_BASE}pwd.getpwnam")
     conf_dir, mock_conf, mock_client, em = epmanager
 
     mock_os.fork.return_value = 0
@@ -141,8 +160,7 @@ def successful_exec(mocker, epmanager, user_conf_template, fs):
     mock_os.dup2.side_effect = [0, 1, 2]
     mock_os.open.return_value = 4
 
-    with open("local_user_lookup.json", "w") as f:
-        json.dump({"a": getpass.getuser()}, f)
+    mock_pwnam.return_value = pwd.getpwuid(os.getuid())
 
     props = pika.BasicProperties(
         content_type="application/json",
@@ -152,13 +170,15 @@ def successful_exec(mocker, epmanager, user_conf_template, fs):
     )
 
     pld = {
-        "globus_uuid": "a",
-        "globus_username": "a",
+        "globus_effective_identity": 1,
+        "globus_identity_set": ["a"],
         "command": "cmd_start_endpoint",
         "kwargs": {"name": "some_ep_name", "user_opts": {"heartbeat": 10}},
     }
     queue_item = (1, props, json.dumps(pld).encode())
 
+    em.identity_mapper = mocker.Mock()
+    em.identity_mapper.map_identity.return_value = "typicalglobusname@somehost.org"
     em._command_queue = mocker.Mock()
     em._command_stop_event.set()
     em._command_queue.get.side_effect = [queue_item, queue.Empty()]
@@ -574,19 +594,54 @@ def test_emits_endpoint_id_if_isatty(mocker, epmanager):
     assert not mock_sys.stderr.write.called
 
 
-def test_warns_of_no_local_lookup(mocker, epmanager):
+def test_fails_to_start_if_no_identity_mapper_configuration(mocker, conf_dir):
     mock_log = mocker.patch(f"{_MOCK_BASE}log")
-    conf_dir, mock_conf, mock_client, em = epmanager
+    mock_print = mocker.patch(f"{_MOCK_BASE}print")
 
-    em._time_to_stop = True
-    em._event_loop()
+    ep_uuid = str(uuid.uuid1())
+    with pytest.raises(SystemExit) as pyt_exc:
+        EndpointManager(conf_dir, ep_uuid, Config())
 
+    assert pyt_exc.value.code == os.EX_OSFILE
     assert mock_log.error.called
-    a = mock_log.error.call_args[0][0]
-    assert "FileNotFoundError" in a, "Expected class name in error output -- help dev!"
-    assert " unable to respond " in a
-    assert " restart " in a
-    assert f"{os.getpid()}" in a
+    assert mock_print.called
+    for a in (mock_log.error.call_args[0][0], mock_print.call_args[0][0]):
+        assert "No identity mapping file specified" in a
+        assert "identity_mapping_config_path" in a, "Expected required config item"
+
+
+@pytest.mark.no_mock_pim
+def test_gracefully_handles_unreadable_identity_mapper_conf(mocker, conf_dir):
+    mock_log = mocker.patch(f"{_MOCK_BASE}log")
+    mock_print = mocker.patch(f"{_MOCK_BASE}print")
+
+    ep_uuid = str(uuid.uuid1())
+    reg_info = {
+        "endpoint_id": ep_uuid,
+        "command_queue_info": {"connection_url": 1, "queue": 1},
+    }
+    conf_p = conf_dir / "idmap.json"
+    conf_p.touch(mode=0o000)
+    conf = Config(identity_mapping_config_path=conf_p)
+    with pytest.raises(SystemExit) as pyt_exc:
+        EndpointManager(conf_dir, ep_uuid, conf, reg_info)
+
+    assert pyt_exc.value.code == os.EX_NOPERM
+    assert mock_log.error.called
+    assert mock_print.called
+    for a in (mock_log.error.call_args[0][0], mock_print.call_args[0][0]):
+        assert "PermissionError" in a
+
+    conf_p.chmod(mode=0o644)
+    conf_p.write_text("[{asfg")
+    with pytest.raises(SystemExit) as pyt_exc:
+        EndpointManager(conf_dir, ep_uuid, conf, reg_info)
+
+    assert pyt_exc.value.code == os.EX_CONFIG
+    assert mock_log.error.called
+    assert mock_print.called
+    for a in (mock_log.error.call_args[0][0], mock_print.call_args[0][0]):
+        assert "Unable to read identity mapping" in a
 
 
 def test_iterates_even_if_no_commands(mocker, epmanager):
@@ -701,9 +756,6 @@ def test_ignores_stale_commands(mocker, epmanager):
     em._command_queue = mocker.Mock()
     em._command_stop_event.set()
 
-    with open("local_user_lookup.json", "w") as f:
-        json.dump({"a": "a_user"}, f)
-
     props = pika.BasicProperties(
         content_type="application/json",  # the test
         content_encoding="utf-8",
@@ -746,7 +798,7 @@ def test_handles_invalid_server_msg_gracefully(mocker, epmanager):
     assert em._command.ack.called, "Command always ACKed"
 
 
-def test_handles_unknown_user_gracefully(mocker, epmanager):
+def test_handles_unknown_identity_gracefully(mocker, epmanager):
     mock_log = mocker.patch(f"{_MOCK_BASE}log")
     conf_dir, mock_conf, mock_client, em = epmanager
 
@@ -757,25 +809,34 @@ def test_handles_unknown_user_gracefully(mocker, epmanager):
         expiration="10000",
     )
 
-    pld = {"globus_uuid": 1, "globus_username": 1}
+    pld = {"globus_effective_identity": 1, "globus_identity_set": []}
     queue_item = (1, props, json.dumps(pld).encode())
+    em.identity_mapper.map_identity.return_value = None
 
     em._command_queue = mocker.Mock()
     em._command_stop_event.set()
     em._command_queue.get.side_effect = [queue_item, queue.Empty()]
     em._event_loop()
-    a = mock_log.warning.call_args[0][0]
-    assert "Invalid or unknown user" in a
-    assert "KeyError" in a, "Expected exception name in log line"
+    a = mock_log.error.call_args[0][0]
+    assert "Identity failed to map to a local user name" in a
+    assert "(LookupError)" in a, "Expected exception name in log line"
+    assert "Globus effective identity: " in a
+    assert "Globus identity set: " in a
+    assert str(pld["globus_effective_identity"]) in a
+    assert str(pld["globus_identity_set"]) in a
     assert em._command.ack.called, "Command always ACKed"
 
 
-def test_handles_unknown_local_username_gracefully(mocker, epmanager):
+@pytest.mark.parametrize(
+    "cmd_name", ("", "_private", "9c", "valid_but_do_not_exist", " ", "a" * 101)
+)
+def test_handles_unknown_or_invalid_command_gracefully(mocker, epmanager, cmd_name):
     mock_log = mocker.patch(f"{_MOCK_BASE}log")
     conf_dir, mock_conf, mock_client, em = epmanager
 
-    with open("local_user_lookup.json", "w") as f:
-        json.dump({"a": "a_user"}, f)
+    mocker.patch(f"{_MOCK_BASE}pwd")
+    em.identity_mapper = mocker.Mock()
+    em.identity_mapper.map_identity.return_value = "a"
 
     props = pika.BasicProperties(
         content_type="application/json",
@@ -785,31 +846,36 @@ def test_handles_unknown_local_username_gracefully(mocker, epmanager):
     )
 
     pld = {
-        "globus_uuid": "a",
-        "globus_username": "a",
+        "globus_effective_identity": "a",
+        "globus_identity_set": "a",
+        "command": cmd_name,
+        "user_opts": {"heartbeat": 10},
     }
     queue_item = (1, props, json.dumps(pld).encode())
 
-    mocker.patch(f"{_MOCK_BASE}pwd.getpwnam", side_effect=Exception())
+    mocker.patch(f"{_MOCK_BASE}pwd.getpwnam")
 
     em._command_queue = mocker.Mock()
     em._command_stop_event.set()
     em._command_queue.get.side_effect = [queue_item, queue.Empty()]
     em._event_loop()
-    a = mock_log.warning.call_args[0][0]
-    assert "Invalid or unknown local user" in a
+    a = mock_log.error.call_args[0][0]
+    assert "Unknown or invalid command" in a
+    assert "Globus effective identity: " in a
+    assert "Globus identity set: " in a
+    assert str(pld["globus_effective_identity"]) in a
+    assert str(pld["globus_identity_set"]) in a
+
+    assert str(cmd_name) in a
     assert em._command.ack.called, "Command always ACKed"
 
 
-@pytest.mark.parametrize(
-    "cmd_name", ("", "_private", "9c", "valid_but_do_not_exist", " ", "a" * 101)
-)
-def test_handles_invalid_command_gracefully(mocker, epmanager, cmd_name):
+def test_handles_local_user_not_found_gracefully(mocker, epmanager, randomstring):
+    invalid_user_name = "username_that_is_not_on_localhost6_" + randomstring()
     mock_log = mocker.patch(f"{_MOCK_BASE}log")
     conf_dir, mock_conf, mock_client, em = epmanager
-
-    with open("local_user_lookup.json", "w") as f:
-        json.dump({"a": "a_user"}, f)
+    em.identity_mapper = mocker.Mock()
+    em.identity_mapper.map_identity.return_value = invalid_user_name
 
     props = pika.BasicProperties(
         content_type="application/json",
@@ -818,23 +884,21 @@ def test_handles_invalid_command_gracefully(mocker, epmanager, cmd_name):
         expiration="10000",
     )
 
-    pld = {
-        "globus_uuid": "a",
-        "globus_username": "a",
-        "command": cmd_name,
-        "user_opts": {"heartbeat": 10},
-    }
+    pld = {"globus_effective_identity": "a", "globus_identity_set": "a"}
     queue_item = (1, props, json.dumps(pld).encode())
 
-    mocker.patch(f"{_MOCK_BASE}pwd.getpwnam")
-
     em._command_queue = mocker.Mock()
     em._command_stop_event.set()
     em._command_queue.get.side_effect = [queue_item, queue.Empty()]
     em._event_loop()
     a = mock_log.error.call_args[0][0]
-    assert "Unknown or invalid command" in a
-    assert str(cmd_name) in a
+    assert "Identity mapped to a local user name, but local user does not exist" in a
+    assert f"Local user name: {invalid_user_name}" in a
+    assert "Globus effective identity: " in a
+    assert "Globus identity set: " in a
+    assert str(pld["globus_effective_identity"]) in a
+    assert str(pld["globus_identity_set"]) in a
+
     assert em._command.ack.called, "Command always ACKed"
 
 
@@ -845,8 +909,9 @@ def test_handles_failed_command(mocker, epmanager):
     )
     conf_dir, mock_conf, mock_client, em = epmanager
 
-    with open("local_user_lookup.json", "w") as f:
-        json.dump({"a": "a_user"}, f)
+    mocker.patch(f"{_MOCK_BASE}pwd")
+    em.identity_mapper = mocker.Mock()
+    em.identity_mapper.map_identity.return_value = "a"
 
     props = pika.BasicProperties(
         content_type="application/json",
@@ -856,8 +921,8 @@ def test_handles_failed_command(mocker, epmanager):
     )
 
     pld = {
-        "globus_uuid": "a",
-        "globus_username": "a",
+        "globus_effective_identity": "a",
+        "globus_identity_set": [],
         "command": "cmd_start_endpoint",
         "user_opts": {"heartbeat": 10},
     }
@@ -896,7 +961,7 @@ def test_handles_shutdown_signal(successful_exec, sig, reset_signals):
     assert not em._command_queue.get.called, " ... that we've now confirmed works"
 
 
-def test_environment_default_path(successful_exec, mocker):
+def test_environment_default_path(successful_exec):
     mock_os, *_, em = successful_exec
     with pytest.raises(SystemExit) as pyexc:
         em._event_loop()
@@ -1019,7 +1084,9 @@ def test_start_endpoint_privileges_dropped(successful_exec):
     assert a == (expected_uid, expected_uid, expected_uid)
 
 
-def test_run_as_same_user_disabled_if_admin(mocker, conf_dir, mock_conf, mock_client):
+def test_run_as_same_user_disabled_if_admin(
+    mocker, conf_dir, mock_conf, mock_client, mock_pim
+):
     ep_uuid, mock_gcc = mock_client
 
     mock_pwd = mocker.patch(f"{_MOCK_BASE}pwd")
diff --git a/compute_endpoint/tests/unit/test_identity_mapper.py b/compute_endpoint/tests/unit/test_identity_mapper.py
new file mode 100644
index 000000000..900cdc8c5
--- /dev/null
+++ b/compute_endpoint/tests/unit/test_identity_mapper.py
@@ -0,0 +1,185 @@
+import json
+import pathlib
+import threading
+
+import pytest
+from globus_compute_endpoint.endpoint.identity_mapper import PosixIdentityMapper
+from tests.utils import try_assert
+
+_MOCK_BASE = "globus_compute_endpoint.endpoint.identity_mapper."
+
+
+def _expression_identity_mapper_conf() -> str:
+    return """[{
+        "DATA_TYPE": "expression_identity_mapping#1.0.0",
+        "mappings": [
+            {"source": "{uname}", "match": "(.*)@a.local", "output": "{0}"},
+            {"source": "{uname}", "match": "(.*)@b.local", "output": "{0}"}
+        ]
+    }]"""
+
+
+def _external_identity_mapper_conf() -> str:
+    return """
+    [{
+        "DATA_TYPE": "external_identity_mapping#1.0.0",
+        "command": ["/some/executable.py_or_sh_or_rb_or_exe_or..."]
+    }]""".strip()
+
+
+@pytest.fixture
+def conf_p(tmp_path):
+    yield tmp_path / "some_file"
+
+
+class NoWaitPosixIdentityMapper(PosixIdentityMapper):
+    def __init__(self, *args, poll_interval_s: float = 0.0, **kwargs):
+        self.test_poll_interval_s = poll_interval_s
+        super().__init__(*args, **kwargs)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.stop_watching()
+
+    @property
+    def poll_interval_s(self) -> float:
+        return self.test_poll_interval_s
+
+    @poll_interval_s.setter
+    def poll_interval_s(self, new_interval_s: float) -> None:
+        pass
+
+
+def test_fails_to_handle_configuration_load_failures_at_initialization(fs):
+    # important to give main thread chance to handle failure (e.g., to quit)
+    conf_p = pathlib.Path("/some_file")
+
+    with pytest.raises(FileNotFoundError):
+        PosixIdentityMapper(conf_p, "some_id")
+
+    conf_p.touch(mode=0o000)
+    with pytest.raises(PermissionError):
+        PosixIdentityMapper(conf_p, "some_id")
+
+    conf_p.chmod(mode=0o644)
+    conf_p.write_text('{"unclosed": "quote}')
+    with pytest.raises(json.JSONDecodeError):
+        PosixIdentityMapper(conf_p, "some_id")
+
+
+def test_poll_interval_setter_enforces_minimum(fs):
+    conf_p = pathlib.Path("/some_file")
+    conf_p.write_text("[]")
+    pim = PosixIdentityMapper(conf_p, "some_id", poll_interval_s=-5)
+    assert pim.poll_interval_s >= 0.5
+
+    pim.poll_interval_s = 0.3
+    assert pim.poll_interval_s >= 0.5
+    pim.stop_watching()
+
+
+def test_atomically_loads_new_configurations(mocker, conf_p):
+    def write_configs():
+        assert pim.identity_mappings == []
+
+        mock_log.reset_mock()
+        conf_p.write_text("[adsfasdf")
+        yield False
+
+        assert pim.identity_mappings == [], "bad config; expect not changed"
+
+        info_a = mock_log.info.call_args[0][0]
+        err_a = mock_log.error.call_args[0][0]
+        mock_log.reset_mock()
+        assert "change detected" in info_a
+        assert "rereading" in info_a
+        assert str(conf_p) in info_a
+        assert "Unable to update identity configuration" in err_a
+
+        conf_p.write_text(_external_identity_mapper_conf())
+        yield False
+
+        assert isinstance(pim.identity_mappings, list)
+        assert "py_or_sh_or_rb_or_exe_or" in str(pim.identity_mappings)
+
+        info_a = mock_log.info.call_args[0][0]
+        mock_log.reset_mock()
+
+        assert "change detected" in info_a
+        assert "rereading" in info_a
+        assert str(conf_p) in info_a
+
+        conf_p.write_text("]")
+        yield False
+
+        assert isinstance(pim.identity_mappings, list)
+        assert "py_or_sh_or_rb_or_exe_or" in str(
+            pim.identity_mappings
+        ), "bad config; expect not changed"
+
+        conf_p.write_text("[]")
+        yield False
+        assert pim.identity_mappings == []
+
+        yield True
+
+    mock_log = mocker.patch(f"{_MOCK_BASE}log")
+    mock_evt = mocker.Mock(spec=threading.Event)
+    mock_evt.wait.side_effect = write_configs()
+    mocker.patch(f"{_MOCK_BASE}threading.Event", return_value=mock_evt)
+    mocker.patch(f"{_MOCK_BASE}threading.Thread")
+
+    conf_p.write_text("[]")
+
+    pim = PosixIdentityMapper(conf_p, "some_id")
+    pim._poll_config()
+    pim.stop_watching()
+
+
+def test_automatically_loads_new_configurations(mocker, conf_p):
+    conf_p.write_text("[]")
+    mocker.patch(f"{_MOCK_BASE}log")
+    with NoWaitPosixIdentityMapper(conf_p, "some_id") as pim:
+        assert pim.identity_mappings == []
+
+        conf_p.write_text(_external_identity_mapper_conf())
+        try_assert(lambda: len(pim.identity_mappings) > 0)
+        assert "/some/executable.py_or_s" in str(pim.identity_mappings)
+
+        conf_p.write_text("[]")
+        try_assert(lambda: len(pim.identity_mappings) == 0)
+
+        conf_p.write_text(_expression_identity_mapper_conf())
+        try_assert(lambda: len(pim.identity_mappings) > 0)
+        assert "@a.local" in str(pim.identity_mappings)
+
+
+def test_delete_identity_mappings(conf_p):
+    conf_p.write_text(_expression_identity_mapper_conf())
+    with NoWaitPosixIdentityMapper(conf_p, "some_id") as pim:
+        assert len(pim.identity_mappings) > 0
+        del pim.identity_mappings
+        try_assert(lambda: pim.identity_mappings == [])
+
+
+@pytest.mark.parametrize("idset", ((), ({"some": "id"}, {"other": "id"})))
+def test_map_identity_falls_back_to_none(conf_p, idset):
+    conf_p.write_text(_expression_identity_mapper_conf())
+    with NoWaitPosixIdentityMapper(conf_p, "some_id", poll_interval_s=0.001) as pim:
+        assert pim.map_identity(idset) is None
+
+
+@pytest.mark.parametrize(
+    "expected,idset",
+    (
+        ("a", ({"uname": "a@a.local"}, {"uname": "b@b.local"})),
+        ("b", ({"uname": "b@b.local"}, {"uname": "a@a.local"})),
+        ("b", ({"uname": "a@c.local"}, {"uname": "b@b.local"})),
+    ),
+)
+def test_map_identity_returns_first_found_identity(conf_p, expected, idset):
+    conf_p.write_text(_expression_identity_mapper_conf())
+    with NoWaitPosixIdentityMapper(conf_p, "some_id", poll_interval_s=0.001) as pim:
+        assert pim.map_identity(idset) == expected