Let module callers pass in-memory mapping rules to Set

README.md

@@ -147,6 +147,20 @@ for provider in get_result.auth_providers:
 for code_host in get_result.code_hosts:
     ...
 
+# Mapping rules can be passed in memory instead of a maps YAML file —
+# same structure and validation as maps.yaml entries:
+rules: list[src.MappingRule] = [
+    {
+        "name": "Map 1",
+        "users": {"usernameRegexes": [".*"]},
+        "repos": {"codeHostConnection": {"kind": "GITHUB"}},
+    },
+]
+result = src.Set(config, mapping_rules=rules)
+# When files are enabled, the rules actually used are written into the
+# run directory for auditability. Snapshots still gate apply=True unless
+# no_files=True and no_backup=True are both set explicitly.
+
 # Other command wrappers:
 # result = src.Restore(config)
 # result = src.SyncSamlOrgs(config)

dev/PLAN.md

@@ -1,11 +1,11 @@
 # PLAN: file handling + observability redesign
 
-> **Status (2026-06-12): implemented**, except Track A Phase A4
-> (in-memory mapping rules for `Set`), which the plan marks optional —
-> tracked in [TODO.md](./TODO.md). Shipped as src-py-lib v0.3.0 and the
-> src-auth-perms-sync `refactor-logging-and-files` PR. Phases were
-> compressed into one PR per repo (no ContextVar bridge phase was
-> needed); everything else landed as specified.
+> **Status (2026-06-12): fully implemented**, including Track A Phase A4
+> (in-memory mapping rules for `Set`). Shipped as src-py-lib v0.3.0 and
+> src-auth-perms-sync v0.5.0 (`refactor-logging-and-files` PR), with A4
+> following in the `in-memory-mapping-rules` PR. Phases were compressed
+> into one PR per repo (no ContextVar bridge phase was needed);
+> everything landed as specified.
 
 Spans both repos (we own both, both greenfield, no external users):
 

dev/TODO.md

@@ -1,15 +1,5 @@
 # TODO
 
-## Follow-up: in-memory mapping rules for Set (PLAN.md Track A Phase A4)
-
-The rest of [PLAN.md](./PLAN.md) is implemented (src-py-lib v0.3.0 +
-the consumer refactor-logging-and-files PR). The one deliberately
-deferred piece, marked optional in the plan: let module callers pass
-parsed mapping rules to `Set` instead of a maps file, so the full
-get → assemble → dry-run loop never touches disk. Snapshots must stay
-on disk for `--apply` (reversibility invariant); `no_files` + `apply`
-must keep requiring `no_backup`.
-
 ## High priority: Remote trigger on demand
 
 - Sourcegraph webhook for new user coming in v7.4.0

src/src_auth_perms_sync/__init__.py

@@ -16,6 +16,7 @@
 )
 
 from .cli import CommandResult, Config, Get, GetResult, Restore, Set, SyncSamlOrgs
+from .permissions.types import MappingRule
 from .shared.backups import RunPaths
 
 __all__ = [
@@ -28,6 +29,7 @@
     "GetResult",
     "InMemoryEventSink",
     "JSONLEventSink",
+    "MappingRule",
     "NullEventSink",
     "Restore",
     "RunPaths",

src/src_auth_perms_sync/cli.py

@@ -26,6 +26,7 @@
 
 from .orgs import command as organizations_command
 from .permissions import command as permissions_command
+from .permissions import mapping as permissions_mapping
 from .permissions import maps as permissions_maps
 from .permissions import types as permission_types
 from .shared import backups, run_context, site_config
@@ -741,6 +742,7 @@ def run_with_client(
     endpoint: str,
     run_paths: backups.RunPaths,
     worker_pool: ThreadPoolExecutor,
+    mapping_rules: list[permission_types.MappingRule] | None = None,
 ) -> run_context.CommandData:
     """Create a client, run the selected command, and always close HTTP resources."""
     http = src.HTTPClient(
@@ -756,7 +758,7 @@ def run_with_client(
         fetch_sg_traces=config.fetch_sg_traces,
     )
     try:
-        return run_command(config, command, client, run_paths, worker_pool)
+        return run_command(config, command, client, run_paths, worker_pool, mapping_rules)
     finally:
         client.http.close()
 
@@ -767,6 +769,7 @@ def run_command(
     client: src.SourcegraphClient,
     run_paths: backups.RunPaths,
     worker_pool: ThreadPoolExecutor,
+    mapping_rules: list[permission_types.MappingRule] | None = None,
 ) -> run_context.CommandData:
     """Dispatch the selected command."""
     sourcegraph_site_config = site_config.validate_site_config(client)
@@ -775,7 +778,13 @@ def run_command(
         command_data = run_get(config, client, sourcegraph_site_config, run_paths, worker_pool)
     elif command.name == "set":
         command_data = run_set(
-            config, command, client, sourcegraph_site_config, run_paths, worker_pool
+            config,
+            command,
+            client,
+            sourcegraph_site_config,
+            run_paths,
+            worker_pool,
+            mapping_rules=mapping_rules,
         )
     elif command.name == "restore":
         run_restore(config, client, sourcegraph_site_config, run_paths, worker_pool)
@@ -810,10 +819,22 @@ def run_set(
     sourcegraph_site_config: site_config.SiteConfig,
     run_paths: backups.RunPaths,
     worker_pool: ThreadPoolExecutor,
+    mapping_rules: list[permission_types.MappingRule] | None = None,
 ) -> run_context.CommandData:
-    """Run the selected repo-permission sync command."""
+    """Run the selected repo-permission sync command.
+
+    With in-memory `mapping_rules` (module callers), no maps file is read;
+    when files are enabled, the rules actually used are written into the run
+    directory so the audit trail stays faithful.
+    """
     assert command.set_options is not None
-    require_set_input_file(run_paths.maps_path)
+    if mapping_rules is None:
+        require_set_input_file(run_paths.maps_path)
+    elif run_paths.write_files:
+        materialized_maps_path = run_paths.input_copy_path(backups.DEFAULT_MAPS_FILE_NAME)
+        permissions_maps.dump_mapping_rules_yaml(materialized_maps_path, mapping_rules)
+        run_paths = dataclasses.replace(run_paths, maps_path=materialized_maps_path)
+        log.info("Wrote in-memory mapping rules for audit: %s", materialized_maps_path)
     return permissions_command.cmd_set(
         client,
         run_paths,
@@ -828,6 +849,7 @@ def run_set(
         do_backup=not config.no_backup,
         retain_saml_group_users=command.sync_saml_organizations,
         worker_pool=worker_pool,
+        mapping_rules=mapping_rules,
     )
 
 
@@ -970,9 +992,22 @@ def Get(config: Config, *, event_sink: src.EventSink | None = None) -> GetResult
     )
 
 
-def Set(config: Config, *, event_sink: src.EventSink | None = None) -> CommandResult:
-    """Run repository permission reconciliation."""
-    succeeded, _, run_paths = _run("set", config, event_sink)
+def Set(
+    config: Config,
+    *,
+    mapping_rules: list[permission_types.MappingRule] | None = None,
+    event_sink: src.EventSink | None = None,
+) -> CommandResult:
+    """Run repository permission reconciliation.
+
+    `mapping_rules` lets module callers pass parsed rules directly (e.g.
+    assembled from `GetResult` data) instead of a maps YAML file. The rules
+    go through the same structural validation as file-loaded rules. When
+    files are enabled, the rules actually used are written into the run
+    directory for auditability; snapshots still gate `apply` unless
+    `no_files` and `no_backup` are both set explicitly.
+    """
+    succeeded, _, run_paths = _run("set", config, event_sink, mapping_rules=mapping_rules)
     return CommandResult(succeeded=succeeded, paths=run_paths)
 
 
@@ -992,11 +1027,21 @@ def _run(
     command_name: CommandName,
     config: Config,
     event_sink: src.EventSink | None,
+    mapping_rules: list[permission_types.MappingRule] | None = None,
 ) -> tuple[bool, run_context.CommandData | None, backups.RunPaths | None]:
     """Run a module-mode command, reporting success instead of raising."""
     try:
-        command_data, run_paths = _run_or_raise(command_name, config, event_sink=event_sink)
+        command_data, run_paths = _run_or_raise(
+            command_name,
+            config,
+            event_sink=event_sink,
+            mapping_rules=mapping_rules,
+        )
     except SystemExit as exception:
+        if isinstance(exception.code, str):
+            # Module mode swallows the exit; surface the diagnostic through
+            # the package logger so host applications can see why it failed.
+            log.error("%s", exception.code)
         return exception.code in (None, 0), None, None
     except Exception:
         log.exception("src-auth-perms-sync run failed.")
@@ -1035,6 +1080,7 @@ def _run_or_raise(
     *,
     cli_mode: bool = False,
     event_sink: src.EventSink | None = None,
+    mapping_rules: list[permission_types.MappingRule] | None = None,
 ) -> tuple[run_context.CommandData, backups.RunPaths]:
     """Run src-auth-perms-sync, preserving CLI-style exceptions.
 
@@ -1043,6 +1089,10 @@ def _run_or_raise(
     application's logging configuration stays in charge.
     """
     validate_config(command_name, config)
+    if mapping_rules is not None:
+        if command_name != "set":
+            config_error("mapping_rules can only be passed to Set")
+        permissions_mapping.validate_mapping_rules(mapping_rules)
     command = resolve_command(command_name, config)
     try:
         endpoint = src.normalize_sourcegraph_endpoint(config.src_endpoint)
@@ -1102,7 +1152,9 @@ def _run_or_raise(
             )
         worker_pool = stack.enter_context(run_context.thread_pool(config.parallelism))
         try:
-            command_data = run_with_client(config, command, endpoint, run_paths, worker_pool)
+            command_data = run_with_client(
+                config, command, endpoint, run_paths, worker_pool, mapping_rules
+            )
         except SystemExit as exception:
             reraise_system_exit_with_logged_error(exception)
         return command_data, run_paths

src/src_auth_perms_sync/permissions/command.py

@@ -26,11 +26,11 @@
 from .workflow import (
     load_discovery,
     load_mapping_context_discovery,
-    load_mapping_rules,
     load_repos_for_mapping_context,
     load_repository_candidates_by_names,
     load_repository_candidates_created_on_or_after,
     parse_cli_date,
+    resolve_mapping_rules,
     sourcegraph_datetime_filter,
     user_ids_created_on_or_after,
     write_maps_backup,
@@ -594,8 +594,13 @@ def cmd_set(
     do_backup: bool,
     retain_saml_group_users: bool = False,
     worker_pool: ThreadPoolExecutor | None = None,
+    mapping_rules: list[permission_types.MappingRule] | None = None,
 ) -> run_context.CommandData:
-    """Dispatch the selected set mode."""
+    """Dispatch the selected set mode.
+
+    `mapping_rules` carries in-memory rules from module callers; when None,
+    rules are loaded from `run_paths.maps_path`.
+    """
     options = set_options
     if options.mode == "full":
         return permissions_full_set.cmd_set_full(
@@ -613,6 +618,7 @@ def cmd_set(
             do_backup=do_backup,
             retain_saml_group_users=retain_saml_group_users,
             worker_pool=worker_pool,
+            mapping_rules=mapping_rules,
         )
     if options.mode == "repos":
         assert options.repository_names
@@ -631,6 +637,7 @@ def cmd_set(
             do_backup=do_backup,
             retain_saml_group_users=retain_saml_group_users,
             worker_pool=worker_pool,
+            mapping_rules=mapping_rules,
         )
     if options.mode == "repos_without_explicit_perms":
         return permissions_full_set.cmd_set_full(
@@ -648,6 +655,7 @@ def cmd_set(
             do_backup=do_backup,
             retain_saml_group_users=retain_saml_group_users,
             worker_pool=worker_pool,
+            mapping_rules=mapping_rules,
         )
     if options.mode == "repos_created_after":
         assert options.repository_created_after is not None
@@ -666,6 +674,7 @@ def cmd_set(
             do_backup=do_backup,
             retain_saml_group_users=retain_saml_group_users,
             worker_pool=worker_pool,
+            mapping_rules=mapping_rules,
         )
     if options.mode == "users":
         assert options.user_identifiers
@@ -680,6 +689,7 @@ def cmd_set(
             saml_groups_attribute_name_by_config_id,
             do_backup,
             worker_pool,
+            mapping_rules=mapping_rules,
         )
     if options.mode == "users_without_explicit_perms":
         return cmd_set_additive_users_without_explicit_perms(
@@ -693,6 +703,7 @@ def cmd_set(
             saml_groups_attribute_name_by_config_id,
             do_backup,
             worker_pool,
+            mapping_rules=mapping_rules,
         )
     if options.mode == "created_after":
         assert options.user_created_after is not None
@@ -706,6 +717,7 @@ def cmd_set(
             saml_groups_attribute_name_by_config_id,
             do_backup,
             worker_pool,
+            mapping_rules=mapping_rules,
         )
     return run_context.CommandData()
 
@@ -721,6 +733,7 @@ def cmd_set_additive_users(
     saml_groups_attribute_name_by_config_id: dict[str, str],
     do_backup: bool,
     worker_pool: ThreadPoolExecutor | None = None,
+    mapping_rules: list[permission_types.MappingRule] | None = None,
 ) -> run_context.CommandData:
     """Add missing mapped permissions for resolved users."""
     with src.span(
@@ -732,7 +745,7 @@ def cmd_set_additive_users(
         parallelism=parallelism,
         do_backup=do_backup,
     ):
-        mapping_rules = load_mapping_rules(run_paths.maps_path)
+        mapping_rules = resolve_mapping_rules(mapping_rules, run_paths.maps_path)
         if not mapping_rules:
             log.warning("No maps defined in %s — nothing to do.", run_paths.maps_path)
             return run_context.CommandData()
@@ -844,6 +857,7 @@ def cmd_set_additive_users_without_explicit_perms(
     saml_groups_attribute_name_by_config_id: dict[str, str],
     do_backup: bool,
     worker_pool: ThreadPoolExecutor | None = None,
+    mapping_rules: list[permission_types.MappingRule] | None = None,
 ) -> run_context.CommandData:
     """Add mapped permissions for users with no explicit API grants."""
     created_after_filter: str | None = None
@@ -859,7 +873,7 @@ def cmd_set_additive_users_without_explicit_perms(
         parallelism=parallelism,
         do_backup=do_backup,
     ):
-        mapping_rules = load_mapping_rules(run_paths.maps_path)
+        mapping_rules = resolve_mapping_rules(mapping_rules, run_paths.maps_path)
         if not mapping_rules:
             log.warning("No maps defined in %s — nothing to do.", run_paths.maps_path)
             return run_context.CommandData()
@@ -1019,6 +1033,7 @@ def cmd_set_additive_created_after(
     saml_groups_attribute_name_by_config_id: dict[str, str],
     do_backup: bool,
     worker_pool: ThreadPoolExecutor | None = None,
+    mapping_rules: list[permission_types.MappingRule] | None = None,
 ) -> run_context.CommandData:
     """Add missing mapped permissions for users created on or after a date."""
     created_after_filter = sourcegraph_datetime_filter(
@@ -1032,7 +1047,7 @@ def cmd_set_additive_created_after(
         parallelism=parallelism,
         do_backup=do_backup,
     ):
-        mapping_rules = load_mapping_rules(run_paths.maps_path)
+        mapping_rules = resolve_mapping_rules(mapping_rules, run_paths.maps_path)
         if not mapping_rules:
             log.warning("No maps defined in %s — nothing to do.", run_paths.maps_path)
             return run_context.CommandData()

src/src_auth_perms_sync/permissions/full_set.py

@@ -21,12 +21,12 @@
 from .workflow import (
     load_mapping_context_discovery,
     load_mapping_context_for_rules,
-    load_mapping_rules,
     load_repository_candidates_by_names,
     load_repository_candidates_created_on_or_after,
     mapping_context_with_repository_candidates,
     projected_snapshot_shell,
     render_projected_snapshot_diff,
+    resolve_mapping_rules,
     user_ids_created_on_or_after,
     validate_post_apply,
     write_maps_backup,
@@ -968,6 +968,7 @@ def cmd_set_full(
     do_backup: bool,
     retain_saml_group_users: bool,
     worker_pool: ThreadPoolExecutor | None = None,
+    mapping_rules: list[permission_types.MappingRule] | None = None,
 ) -> run_context.CommandData:
     """Overwrite each mapped repo with the union of users from all rules."""
     with src.span(
@@ -981,7 +982,7 @@ def cmd_set_full(
         parallelism=parallelism,
         do_backup=do_backup,
     ) as command_event:
-        mapping_rules = load_mapping_rules(run_paths.maps_path)
+        mapping_rules = resolve_mapping_rules(mapping_rules, run_paths.maps_path)
         if not mapping_rules:
             _finish_empty_full_set_mapping_rules(
                 client,