refactor(runtime): replace DictRenderer and domains fact with cached lookups

Removes the placeholder interpreter (utils/dict_renderer.py) in favor of
standard {{ lookup('config', ...) }} / {{ lookup('domain', ...) }} calls
inside role configs, and eliminates the `domains` set_fact stage in
tasks/stages/01_constructor.yml by introducing a cached get_merged_domains
helper plus a new `domains` lookup plugin (full map + by-app subtree).

The `domain` lookup no longer depends on a pre-computed `domains` variable;
it delegates to get_merged_domains, which respects inventory-level overrides
from both `applications:` (authoritative canonical source) and `domains:`
(direct override map). Caching mirrors the applications pipeline — keyed on
(roles_dir, variables signature, override fingerprint) with a re-entry guard.

Replaced `{{ domains[app_id] }}` / `{{ domains }}` usages across web-app-matrix,
-bluesky, -mastodon, -peertube, -keycloak, test-e2e-playwright, and
group_vars. Added two performance tests for lookup('domains') mirroring the
applications cache assertions (≥50x warm-to-hot ratio, 120s cold budget).

group_vars/all/14_about.yml

@@ -17,7 +17,7 @@ defaults_service_provider:
     favicon:        "{{ lookup('config', 'web-svc-asset', 'url') ~ '/img/favicon.ico' }}"
   contact:
     web-app-bluesky: >-
-      {{ ('@' ~ lookup('users', 'contact').username ~ '.' ~ domains['web-app-bluesky'].api)
+      {{ ('@' ~ lookup('users', 'contact').username ~ '.' ~ lookup('domains', 'web-app-bluesky').api)
          if 'web-app-bluesky' in group_names else '' }}
     email:          "{{ lookup('users', 'contact').username ~ '@' ~ DOMAIN_PRIMARY if 'web-app-mailu' in group_names else 'contact@infinito.nexus' }}"
     mastodon:       "{{ '@' ~ lookup('users', 'contact').username ~ '@' ~ lookup('domain','web-app-mastodon') if 'web-app-mastodon' in group_names else '@infinito@microblog.infinito.nexus' }}"

plugins/lookup/domain.py

@@ -8,16 +8,18 @@ from typing import Any, Dict, Optional
 from ansible.errors import AnsibleError
 from ansible.plugins.lookup import LookupBase
 
+from utils.runtime_lookup_data import get_merged_domains
+
 
 class LookupModule(LookupBase):
     """
     Usage:
       {{ lookup('domain', application_id) }}
 
-    Reads:
-      - variables['domains'] (required)
-    Returns:
-      - domain as string (resolved via utils.domains.primary_domain.get_domain)
+    Resolves the canonical primary domain for `application_id` via
+    utils.runtime_lookup_data.get_merged_domains (cached). Accepts an optional
+    `variables['domains']` override map recursively merged on top of the
+    computed canonical-domains map.
     """
 
     def run(self, terms, variables: Optional[Dict[str, Any]] = None, **kwargs):
@@ -32,11 +34,14 @@ class LookupModule(LookupBase):
                 f"lookup('domain'): application_id must be a non-empty string, got {application_id!r}"
             )
 
-        variables = variables or {}
-        if "domains" not in variables:
-            raise AnsibleError("lookup('domain'): missing required variable 'domains'")
+        variables = variables or getattr(self._templar, "available_variables", {}) or {}
+
+        domains = get_merged_domains(
+            variables=variables,
+            roles_dir=kwargs.get("roles_dir"),
+            templar=getattr(self, "_templar", None),
+        )
 
-        # Make utils importable (project_root/utils)
         plugin_dir = os.path.dirname(os.path.abspath(__file__))
         project_root = os.path.abspath(os.path.join(plugin_dir, "..", ".."))
         if project_root not in sys.path:
@@ -50,7 +55,7 @@ class LookupModule(LookupBase):
             )
 
         try:
-            domain = get_domain(variables["domains"], application_id.strip())
+            domain = get_domain(domains, application_id.strip())
         except Exception as e:
             raise AnsibleError(
                 f"lookup('domain'): failed to resolve domain for '{application_id}': {e}"

plugins/lookup/domains.py

@@ -0,0 +1,69 @@
+# lookup_plugins/domains.py
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
+
+from ansible.errors import AnsibleError
+from ansible.plugins.lookup import LookupBase
+
+from utils.runtime_lookup_data import (
+    _reset_cache_for_tests as _reset_runtime_lookup_cache,
+    get_merged_domains,
+)
+
+
+def _reset_cache_for_tests() -> None:
+    _reset_runtime_lookup_cache()
+
+
+class LookupModule(LookupBase):
+    """
+    Usage:
+        {{ lookup('domains') }}                          -> full canonical-domains map
+        {{ lookup('domains', application_id) }}          -> subtree for one role
+        {{ lookup('domains', application_id, default) }} -> subtree with fallback
+
+    Behavior:
+        Builds (and caches) the canonical-domains map via
+        utils.runtime_lookup_data.get_merged_domains. Accepts an optional
+        `variables['domains']` override map, which is merged recursively on top
+        of the computed base. No upstream set_fact required.
+    """
+
+    def run(
+        self,
+        terms: Optional[list[Any]],
+        variables: Optional[Dict[str, Any]] = None,
+        **kwargs: Any,
+    ) -> list[Any]:
+        terms = terms or []
+        if len(terms) > 2:
+            raise AnsibleError(
+                "lookup('domains'[, application_id[, default]]) expects 0, 1 or 2 terms."
+            )
+
+        variables = variables or getattr(self._templar, "available_variables", {}) or {}
+
+        domains = get_merged_domains(
+            variables=variables,
+            roles_dir=kwargs.get("roles_dir"),
+            templar=getattr(self, "_templar", None),
+        )
+
+        if len(terms) == 0:
+            return [domains]
+
+        application_id = str(terms[0]).strip()
+        default_provided = len(terms) == 2
+        default_value = terms[1] if default_provided else None
+
+        if application_id in domains:
+            return [domains[application_id]]
+
+        if default_provided:
+            return [default_value]
+
+        raise AnsibleError(
+            f"lookup('domains'): application '{application_id}' not found. "
+            f"Known application ids: {sorted(domains)}"
+        )

roles/test-e2e-playwright/tasks/run_one.yml

@@ -161,7 +161,7 @@
     content: "{{ lookup('template', _playwright_env_template) }}"
   vars:
     application_id: "{{ application_id }}"
-    domains: "{{ domains }}"
+    domains: "{{ lookup('domains') }}"
 
 # Optional wait until app is ready (uses domain resolution).
 # Accept any 1xx/2xx/3xx response — anything below 400 means the service is up.

roles/web-app-bluesky/vars/main.yml

@@ -22,20 +22,20 @@ BLUESKY_PDS_BLOBSTORE_LOCATION: "{{ [ BLUESKY_PDS_DATA_DIR, 'blocks' ] | path_jo
 
 ## Web
 BLUESKY_WEB_ENABLED:            "{{ lookup('config', application_id, 'compose.services.web.enabled') }}"
-BLUESKY_WEB_DOMAIN:             "{{ domains[application_id].web }}"
+BLUESKY_WEB_DOMAIN:             "{{ lookup('domains', application_id).web }}"
 BLUESKY_WEB_PORT:               "{{ ports.localhost.http['web-app-bluesky_web'] }}"
 BLUESKY_WEB_IMAGE_CUSTOM:       "bluesky_web_custom"
 
 ## View
 BLUESKY_VIEW_ENABLED:           "{{ lookup('config', application_id, 'compose.services.view.enabled') }}"
-BLUESKY_VIEW_DOMAIN_LOCAL:      "{{ domains[application_id].view }}"
+BLUESKY_VIEW_DOMAIN_LOCAL:      "{{ lookup('domains', application_id).view }}"
 BLUESKY_VIEW_DOMAIN_FINAL:      "{{ BLUESKY_VIEW_DOMAIN_LOCAL if BLUESKY_VIEW_ENABLED else 'api.bsky.app' }}"
 BLUESKY_VIEW_URL:               "{{ lookup('tls', application_id, 'protocols.web') }}://{{ BLUESKY_VIEW_DOMAIN_FINAL }}"
 BLUESKY_VIEW_DID:               "did:web:{{ BLUESKY_VIEW_DOMAIN_FINAL }}"
 BLUESKY_VIEW_PORT:              "{{ ports.localhost.http['web-app-bluesky_view'] }}"
 
 ## Server
-BLUESKY_API_DOMAIN:             "{{ domains[application_id].api }}"
+BLUESKY_API_DOMAIN:             "{{ lookup('domains', application_id).api }}"
 BLUESKY_API_PORT:               "{{ ports.localhost.http['web-app-bluesky_api'] }}"
 
 ## Credentials

roles/web-app-gitea/config/main.yml

@@ -69,7 +69,7 @@ compose:
     oauth2:
       origin:
         host:             "application"
-        port:             "<< applications[web-app-gitea].compose.services.gitea.port >>"
+        port:             "{{ lookup('config', 'web-app-gitea', 'compose.services.gitea.port') }}"
       acl:
         blacklist:
           - "/user/login"

roles/web-app-keycloak/vars/main.yml

@@ -58,7 +58,7 @@ KEYCLOAK_SERVER_HOST:               "127.0.0.1:{{ ports.localhost.http[applicati
 KEYCLOAK_REDIRECT_FEATURES:         ["compose.services.oauth2.enabled","compose.services.oidc.enabled"]
 KEYCLOAK_FRONTCHANNEL_LOGOUT_URL:   "{{ lookup('tls', 'web-svc-logout', 'url.base') }}"
 # This will lead to issues later @todo; Optimize:
-KEYCLOAK_REDIRECT_URIS:             "{{ domains | redirect_uris(lookup('applications'), lookup('tls', application_id, 'protocols.web'), '/*', KEYCLOAK_REDIRECT_FEATURES) }}"
+KEYCLOAK_REDIRECT_URIS:             "{{ lookup('domains') | redirect_uris(lookup('applications'), lookup('tls', application_id, 'protocols.web'), '/*', KEYCLOAK_REDIRECT_FEATURES) }}"
 KEYCLOAK_WEB_ORIGINS: >-
   {{ KEYCLOAK_REDIRECT_URIS
       | map('regex_replace','/\\*$','')

roles/web-app-mastodon/tasks/main.yml

@@ -2,7 +2,7 @@
 - name: "Include setup for domain '{{ domain }}'"
   include_role:
     name: sys-stk-front-proxy
-  loop: "{{ domains['web-app-mastodon'] }}"
+  loop: "{{ lookup('domains', 'web-app-mastodon') }}"
   loop_control:
     loop_var: domain
   vars:

roles/web-app-mastodon/templates/env.j2

@@ -4,7 +4,7 @@
 
 
 LOCAL_DOMAIN={{ lookup('domain',application_id) }}
-ALTERNATE_DOMAINS={{ domains['web-app-mastodon'][1:] | join(',') }}
+ALTERNATE_DOMAINS={{ lookup('domains', 'web-app-mastodon')[1:] | join(',') }}
 SINGLE_USER_MODE={{ lookup('config', application_id, 'single_user_mode') }}
 ALLOWED_PRIVATE_ADDRESSES={{ MASTODON_ALLOWED_PRIVATE_ADDRESSES }}
 

roles/web-app-matrix/templates/synapse/homeserver.yaml.j2

@@ -25,7 +25,7 @@ report_stats:                   true
 macaroon_secret_key:            "{{ lookup('config', application_id, 'credentials.macaroon_secret_key') }}"
 form_secret:                    "{{ lookup('config', application_id, 'credentials.form_secret') }}"
 signing_key_path:               "/data/{{ MATRIX_SYNAPSE_DOMAIN }}.signing.key"
-web_client_location:            "{{ lookup('tls', application_id, 'protocols.web') }}://{{ domains[application_id].element}}"
+web_client_location:            "{{ lookup('tls', application_id, 'protocols.web') }}://{{ lookup('domains', application_id).element }}"
 public_baseurl:                 "{{ MATRIX_SYNAPSE_URL }}"
 trusted_key_servers:
   - server_name: "matrix.org"

roles/web-app-matrix/vars/main.yml

@@ -24,7 +24,7 @@ MATRIX_SYNAPSE_VERSION:               "{{ lookup('config', application_id, 'comp
 MATRIX_SYNAPSE_IMAGE:                 "{{ lookup('config', application_id, 'compose.services.synapse.image') }}"
 MATRIX_SYNAPSE_NAME:                  "{{ lookup('config', application_id, 'compose.services.synapse.name') }}"
 MATRIX_SYNAPSE_VOLUME:                "{{ lookup('config', application_id, 'compose.volumes.synapse') }}"
-MATRIX_SYNAPSE_DOMAIN:                "{{ domains[application_id].synapse }}"
+MATRIX_SYNAPSE_DOMAIN:                "{{ lookup('domains', application_id).synapse }}"
 MATRIX_SYNAPSE_PORT:                  "{{ ports.localhost.http['web-app-matrix_synapse'] }}"
 MATRIX_SYNAPSE_CONFIG_PATH_HOST:      "{{ [ lookup('container', application_id, 'directories.config'), 'homeserver.yaml' ] | path_join }}"
 MATRIX_SYNAPSE_CONFIG_PATH_CONTAINER: "/data/homeserver.yaml"
@@ -37,7 +37,7 @@ MATRIX_SYNAPSE_LISTEN_FLAGS:          "{{ MATRIX_FEDERATION_PORT }} {{ 'ssl ' if
 MATRIX_ELEMENT_VERSION:               "{{ lookup('config', application_id, 'compose.services.element.version') }}"
 MATRIX_ELEMENT_IMAGE:                 "{{ lookup('config', application_id, 'compose.services.element.image') }}"
 MATRIX_ELEMENT_NAME:                  "{{ lookup('config', application_id, 'compose.services.element.name') }}"
-MATRIX_ELEMENT_DOMAIN:                "{{ domains[application_id].element }}"
+MATRIX_ELEMENT_DOMAIN:                "{{ lookup('domains', application_id).element }}"
 MATRIX_ELEMENT_PORT:                  "{{ ports.localhost.http['web-app-matrix_element'] }}"
 MATRIX_ELEMENT_CONFIG_PATH_HOST:      "{{ [ lookup('container', application_id, 'directories.config'), 'element-config.json' ] | path_join }}"
 

roles/web-app-peertube/tasks/main.yml

@@ -1,7 +1,7 @@
 ---
 - name: "include _create-domains.yml for peertube"
   include_tasks: _create-domains.yml
-  loop: "{{ domains[application_id] }}"
+  loop: "{{ lookup('domains', application_id) }}"
   loop_control:
     loop_var: domain
   vars:

roles/web-app-yourls/config/main.yml

@@ -44,7 +44,7 @@ compose:
         - "{{ [RBAC.GROUP.NAME, 'web-app-yourls-administrator'] | path_join }}"
       acl:
         blacklist:
-          - "<< applications[web-app-yourls].server.locations.admin >>" # Protects the admin area
+          - "{{ lookup('config', 'web-app-yourls', 'server.locations.admin') }}" # Protects the admin area
     css:
       enabled: true
       shared: true

roles/web-svc-asset/config/main.yml

@@ -1,5 +1,5 @@
 source_directory: "{{ playbook_dir }}/assets"
-url:              "//<< applications['web-svc-file'].server.domains.canonical[0] >>/assets"
+url:              "//{{ lookup('domain', 'web-svc-file') }}/assets"
 server:
   domains:
     canonical:

tasks/stages/01_constructor.yml

@@ -71,15 +71,6 @@
             )
         }}
 
-  - name: Merge domain definitions for all domains
-    ansible.builtin.set_fact:
-      domains: >-
-        {{
-          lookup('applications')
-          | canonical_domains_map(DOMAIN_PRIMARY)
-          | combine(CURRENT_PLAY_DOMAINS, recursive=True)
-        }}
-
   - name: Merge CURRENT_PLAY_REDIRECT_DOMAINS
     ansible.builtin.set_fact:
       # The following mapping is necessary to define the exceptions for domains which are created, but which aren't used

templates/roles/web-app/config/README.md.j2

@@ -10,14 +10,15 @@ applications:
     variable_c: []             # Replaces the default value (use caution with domains)
 ```
 
-## Placeholder Logic with `<< >>`
+## Cross-Role References via Lookup Plugins
 
-You can reference values from the aggregated `applications` dictionary at build time by embedding `<< ... >>` placeholders inside your template. For example:
+You can reference values from another role's aggregated configuration by using the `config` and `domain` lookup plugins directly inside your template. For example:
 
 ```yaml
-url: "{{ lookup('tls', application_id, 'protocols.web') }}://<< applications.web-svc-file.server.domains.canonical[0] >>/assets"
+url: "{{ lookup('tls', application_id, 'protocols.web') }}://{{ lookup('domain', 'web-svc-file') }}/assets"
+port: "{{ lookup('config', 'web-app-gitea', 'compose.services.gitea.port') }}"
 ```
 
-- The `<< ... >>` placeholders are resolved by the [`DictRenderer`](../../../utils/dict_renderer.py) helper class.
-- The applications lookup aggregates role configurations in memory and then calls the renderer to substitute each `<< ... >>` occurrence.
-- Use the `--verbose` flag on the CLI script to log every replacement step, and rely on the built‑in timeout (default: 10 seconds) to prevent infinite loops.
+- `lookup('config', ROLE_ID, DOTTED_PATH, DEFAULT)` resolves any nested value from another role's `config/main.yml` (the `DEFAULT` argument is optional).
+- `lookup('domain', ROLE_ID)` returns the canonical domain of another role.
+- Both lookups operate on the merged `applications` mapping and Jinja expressions in the referenced values are rendered at lookup time.

tests/integration/lookups/test_runtime_performance.py

@@ -24,6 +24,10 @@ from plugins.lookup.applications_current_play import (
     LookupModule as ApplicationsCurrentPlayLookup,
     _reset_cache_for_tests as _reset_current_play_cache,
 )
+from plugins.lookup.domains import (
+    LookupModule as DomainsLookup,
+    _reset_cache_for_tests as _reset_domains_cache,
+)
 
 
 REPO_ROOT = Path(__file__).resolve().parents[3]
@@ -56,12 +60,14 @@ class TestRuntimeLookupPerformance(unittest.TestCase):
     def setUp(self) -> None:
         _reset_cache_for_tests()
         _reset_current_play_cache()
+        _reset_domains_cache()
         self.shared_applications: dict = {}
         self.shared_users: dict = {}
 
     def tearDown(self) -> None:
         _reset_cache_for_tests()
         _reset_current_play_cache()
+        _reset_domains_cache()
 
     def _call_applications(self, variables: dict) -> None:
         lookup = ApplicationsLookup()
@@ -73,6 +79,11 @@ class TestRuntimeLookupPerformance(unittest.TestCase):
         lookup._templar = Templar(loader=DataLoader())
         lookup.run([], variables=variables, roles_dir=str(ROLES_DIR))
 
+    def _call_domains(self, variables: dict) -> None:
+        lookup = DomainsLookup()
+        lookup._templar = Templar(loader=DataLoader())
+        lookup.run([], variables=variables, roles_dir=str(ROLES_DIR))
+
     def test_applications_lookup_caches_across_fresh_variables_dicts(self) -> None:
         """Warm call populates the cache; subsequent calls with freshly-built
         `variables` mappings AND freshly-copied applications/users dicts
@@ -175,6 +186,65 @@ class TestRuntimeLookupPerformance(unittest.TestCase):
             f"(ratio {warm_over_hot_avg:.1f}x).",
         )
 
+    def test_domains_lookup_caches_across_fresh_variables_dicts(self) -> None:
+        """Warm call populates the domains cache; subsequent calls with freshly-
+        built `variables` mappings AND freshly-copied applications/users dicts
+        must hit the cache and stay under budget.
+
+        Same guarantees as test_applications_lookup_caches_across_fresh_variables_dicts,
+        but for lookup('domains'). Without caching, each call re-runs
+        canonical_domains_map over the full roles/ tree.
+        """
+        warm = _simulate_ansible_variables(self.shared_applications, self.shared_users)
+        t0 = time.perf_counter()
+        self._call_domains(warm)
+        warm_elapsed = time.perf_counter() - t0
+
+        import copy
+
+        t0 = time.perf_counter()
+        for _ in range(self.ITERATIONS):
+            variables = _simulate_ansible_variables(
+                copy.deepcopy(self.shared_applications),
+                copy.deepcopy(self.shared_users),
+            )
+            self._call_domains(variables)
+        hot_elapsed = time.perf_counter() - t0
+
+        avg_ms = hot_elapsed / self.ITERATIONS * 1000
+        warm_over_hot_avg = (warm_elapsed * 1000) / max(avg_ms, 0.001)
+
+        self.assertLess(
+            hot_elapsed,
+            self.WARM_CALLS_BUDGET_SECONDS,
+            f"{self.ITERATIONS} cached domains lookups took {hot_elapsed:.2f}s "
+            f"(avg {avg_ms:.2f}ms). Budget: {self.WARM_CALLS_BUDGET_SECONDS}s. "
+            f"Warm-up: {warm_elapsed:.2f}s ({warm_over_hot_avg:.0f}x avg hot).",
+        )
+        self.assertGreater(
+            warm_over_hot_avg,
+            50.0,
+            f"Domains cache not providing meaningful speed-up: warm={warm_elapsed:.2f}s "
+            f"vs avg hot={avg_ms:.2f}ms (ratio {warm_over_hot_avg:.1f}x). "
+            "Expected cache hits to be >=50x faster than cold render.",
+        )
+
+    def test_domains_lookup_warmup_budget(self) -> None:
+        """First render of lookup('domains') on full roles/ dir must complete
+        in reasonable time."""
+        variables = _simulate_ansible_variables(
+            self.shared_applications, self.shared_users
+        )
+        t0 = time.perf_counter()
+        self._call_domains(variables)
+        warm_elapsed = time.perf_counter() - t0
+
+        self.assertLess(
+            warm_elapsed,
+            120.0,
+            f"Cold domains render took {warm_elapsed:.2f}s (budget 120s).",
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/plugins/lookup/test_domain_lookup.py

@@ -2,6 +2,7 @@
 import os
 import sys
 import unittest
+from unittest.mock import patch
 
 from ansible.errors import AnsibleError
 
@@ -20,40 +21,58 @@ from plugins.lookup.domain import LookupModule  # noqa: E402
 
 
 class TestDomainLookup(unittest.TestCase):
+    """Unit tests for the `domain` lookup.
+
+    The plugin delegates the canonical-domains map build to
+    utils.runtime_lookup_data.get_merged_domains. These tests patch that helper
+    so we assert only the plugin's dispatch + primary-domain extraction logic,
+    not the merge pipeline (which has its own integration tests).
+    """
+
     def setUp(self):
         self.lookup = LookupModule()
 
-    def run_lookup(self, application_id, domains):
-        return self.lookup.run(
-            terms=[application_id],
-            variables={"domains": domains},
-        )
+    def run_lookup(self, application_id, domains_map):
+        with patch(
+            "plugins.lookup.domain.get_merged_domains",
+            return_value=domains_map,
+        ):
+            return self.lookup.run(
+                terms=[application_id],
+                variables={},
+            )
 
     def test_string_domain(self):
-        domains = {"app": "example.com"}
-        self.assertEqual(self.run_lookup("app", domains), ["example.com"])
+        self.assertEqual(
+            self.run_lookup("app", {"app": "example.com"}), ["example.com"]
+        )
 
     def test_list_domain(self):
-        domains = {"app": ["example.com", "alt.example.com"]}
-        self.assertEqual(self.run_lookup("app", domains), ["example.com"])
+        self.assertEqual(
+            self.run_lookup("app", {"app": ["example.com", "alt.example.com"]}),
+            ["example.com"],
+        )
 
     def test_dict_domain(self):
-        domains = {"app": {"primary": "example.com", "secondary": "alt.example.com"}}
-        self.assertEqual(self.run_lookup("app", domains), ["example.com"])
-
-    def test_missing_domains_variable(self):
-        with self.assertRaises(AnsibleError) as ctx:
-            self.lookup.run(terms=["app"], variables={})
-        self.assertIn("missing required variable 'domains'", str(ctx.exception))
+        self.assertEqual(
+            self.run_lookup(
+                "app",
+                {"app": {"primary": "example.com", "secondary": "alt.example.com"}},
+            ),
+            ["example.com"],
+        )
 
     def test_missing_application_id(self):
         with self.assertRaises(AnsibleError):
-            self.lookup.run(terms=[], variables={"domains": {"app": "example.com"}})
+            with patch(
+                "plugins.lookup.domain.get_merged_domains",
+                return_value={"app": "example.com"},
+            ):
+                self.lookup.run(terms=[], variables={})
 
     def test_unknown_application_id(self):
-        with self.assertRaises(AnsibleError) as ctx:
+        with self.assertRaises(AnsibleError):
             self.run_lookup("unknown", {"app": "example.com"})
-        self.assertIn("not found in domains mapping", str(ctx.exception))
 
     def test_empty_string_domain(self):
         with self.assertRaises(AnsibleError):
@@ -69,7 +88,11 @@ class TestDomainLookup(unittest.TestCase):
 
     def test_invalid_application_id_type(self):
         with self.assertRaises(AnsibleError):
-            self.lookup.run(terms=[123], variables={"domains": {"app": "example.com"}})
+            with patch(
+                "plugins.lookup.domain.get_merged_domains",
+                return_value={"app": "example.com"},
+            ):
+                self.lookup.run(terms=[123], variables={})
 
 
 if __name__ == "__main__":

tests/unit/utils/test_dict_renderer.py

@@ -1,77 +0,0 @@
-import unittest
-from utils.dict_renderer import DictRenderer
-
-
-class TestDictRenderer(unittest.TestCase):
-    def setUp(self):
-        # Timeout is small for tests, verbose off
-        self.renderer = DictRenderer(verbose=False, timeout=1.0)
-
-    def test_simple_replacement(self):
-        data = {"foo": "bar", "val": "<<foo>>"}
-        rendered = self.renderer.render(data)
-        self.assertEqual(rendered["val"], "bar")
-
-    def test_nested_replacement(self):
-        data = {"parent": {"child": "value"}, "ref": "<<parent.child>>"}
-        rendered = self.renderer.render(data)
-        self.assertEqual(rendered["ref"], "value")
-
-    def test_list_index(self):
-        data = {"lst": [10, 20, 30], "second": "<<lst[1]>>"}
-        rendered = self.renderer.render(data)
-        self.assertEqual(rendered["second"], "20")
-
-    def test_multi_pass(self):
-        data = {"a": "<<b>>", "b": "<<c>>", "c": "final"}
-        rendered = self.renderer.render(data)
-        self.assertEqual(rendered["a"], "final")
-
-    def test_unresolved_raises(self):
-        data = {"a": "<<missing>>"}
-        with self.assertRaises(ValueError) as cm:
-            self.renderer.render(data)
-        self.assertIn("missing", str(cm.exception))
-
-    def test_leave_curly(self):
-        data = {"tmpl": "{{ not touched }}"}
-        rendered = self.renderer.render(data)
-        self.assertEqual(rendered["tmpl"], "{{ not touched }}")
-
-    def test_mixed_braces(self):
-        data = {"foo": "bar", "tmpl": "{{ <<foo>> }}"}
-        rendered = self.renderer.render(data)
-        self.assertEqual(rendered["tmpl"], "{{ bar }}")
-
-    def test_single_quoted_key(self):
-        # ['foo-bar'] should resolve the key 'foo-bar'
-        data = {"foo-bar": {"val": "xyz"}, "result": "<<['foo-bar'].val>>"}
-        rendered = self.renderer.render(data)
-        self.assertEqual(rendered["result"], "xyz")
-
-    def test_double_quoted_key(self):
-        # ["foo-bar"] should also resolve the key 'foo-bar'
-        data = {"foo-bar": {"val": 123}, "result": '<<["foo-bar"].val>>'}
-        rendered = self.renderer.render(data)
-        self.assertEqual(rendered["result"], "123")
-
-    def test_mixed_bracket_and_dot_with_index(self):
-        # Combine quoted key, dot access and numeric index
-        data = {
-            "web-svc-file": {
-                "server": {"domains": {"canonical": ["file.example.com"]}}
-            },
-            "url": "<<['web-svc-file'].server.domains.canonical[0]>>",
-        }
-        rendered = self.renderer.render(data)
-        self.assertEqual(rendered["url"], "file.example.com")
-
-    def test_double_quoted_key_with_list_index(self):
-        # Double-quoted key and list index together
-        data = {"my-list": ["a", "b", "c"], "pick": '<<["my-list"][2]>>'}
-        rendered = self.renderer.render(data)
-        self.assertEqual(rendered["pick"], "c")
-
-
-if __name__ == "__main__":
-    unittest.main()

utils/dict_renderer.py

@@ -1,129 +0,0 @@
-import re
-import time
-from typing import Any, Dict, Union, List, Set
-
-
-class DictRenderer:
-    """
-    Resolves placeholders in the form << path >> within nested dictionaries,
-    supporting hyphens, numeric list indexing, and quoted keys via ['key'] or ["key"].
-    """
-
-    # Match << path >> where path contains no whitespace or closing >
-    PATTERN = re.compile(r"<<\s*(?P<path>[^\s>]+)\s*>>")
-    # Tokenizes a path into unquoted keys, single-quoted, double-quoted keys, or numeric indices
-    TOKEN_REGEX = re.compile(
-        r"(?P<key>[\w\-]+)"
-        r"|\['(?P<qkey>[^']+)'\]"
-        r"|\[\"(?P<dkey>[^\"]+)\"\]"
-        r"|\[(?P<idx>\d+)\]"
-    )
-
-    def __init__(self, verbose: bool = False, timeout: float = 10.0):
-        self.verbose = verbose
-        self.timeout = timeout
-
-    def render(
-        self, data: Union[Dict[str, Any], List[Any]]
-    ) -> Union[Dict[str, Any], List[Any]]:
-        start = time.monotonic()
-        self.root = data
-        rendered = data
-        pass_num = 0
-
-        while True:
-            pass_num += 1
-            if self.verbose:
-                print(f"[DictRenderer] Pass {pass_num} starting...")
-            rendered, changed = self._render_pass(rendered)
-            if not changed:
-                if self.verbose:
-                    print(f"[DictRenderer] No more placeholders after pass {pass_num}.")
-                break
-            if time.monotonic() - start > self.timeout:
-                raise TimeoutError(
-                    f"Rendering exceeded timeout of {self.timeout} seconds"
-                )
-
-        # After all passes, raise error on unresolved placeholders
-        unresolved = self.find_unresolved(rendered)
-        if unresolved:
-            raise ValueError(
-                f"Unresolved placeholders: {', '.join(sorted(unresolved))}"
-            )
-
-        return rendered
-
-    def _render_pass(self, obj: Any) -> (Any, bool):
-        if isinstance(obj, dict):
-            new = {}
-            changed = False
-            for k, v in obj.items():
-                nv, ch = self._render_pass(v)
-                new[k] = nv
-                changed = changed or ch
-            return new, changed
-        if isinstance(obj, list):
-            new_list = []
-            changed = False
-            for item in obj:
-                ni, ch = self._render_pass(item)
-                new_list.append(ni)
-                changed = changed or ch
-            return new_list, changed
-        if isinstance(obj, str):
-
-            def repl(m):
-                path = m.group("path")
-                val = self._lookup(path)
-                if val is not None:
-                    if self.verbose:
-                        print(f"[DictRenderer] Resolving <<{path}>> -> {val}")
-                    return str(val)
-                return m.group(0)
-
-            new_str = self.PATTERN.sub(repl, obj)
-            return new_str, new_str != obj
-        return obj, False
-
-    def _lookup(self, path: str) -> Any:
-        current = self.root
-        for m in self.TOKEN_REGEX.finditer(path):
-            if m.group("key") is not None:
-                if isinstance(current, dict):
-                    current = current.get(m.group("key"))
-                else:
-                    return None
-            elif m.group("qkey") is not None:
-                if isinstance(current, dict):
-                    current = current.get(m.group("qkey"))
-                else:
-                    return None
-            elif m.group("dkey") is not None:
-                if isinstance(current, dict):
-                    current = current.get(m.group("dkey"))
-                else:
-                    return None
-            elif m.group("idx") is not None:
-                idx = int(m.group("idx"))
-                if isinstance(current, list) and 0 <= idx < len(current):
-                    current = current[idx]
-                else:
-                    return None
-            if current is None:
-                return None
-        return current
-
-    def find_unresolved(self, data: Any) -> Set[str]:
-        """Return all paths of unresolved << placeholders in data."""
-        unresolved: Set[str] = set()
-        if isinstance(data, dict):
-            for v in data.values():
-                unresolved |= self.find_unresolved(v)
-        elif isinstance(data, list):
-            for item in data:
-                unresolved |= self.find_unresolved(item)
-        elif isinstance(data, str):
-            for m in self.PATTERN.finditer(data):
-                unresolved.add(m.group("path"))
-        return unresolved

utils/runtime_lookup_data.py

@@ -13,7 +13,6 @@ import yaml
 
 from plugins.filter.merge_with_defaults import merge_with_defaults
 from plugins.lookup.application_gid import LookupModule as ApplicationGidLookup
-from utils.dict_renderer import DictRenderer
 from utils.templating import _templar_render_best_effort
 
 try:
@@ -62,6 +61,7 @@ _USERS_DEFAULTS_CACHE: dict[str, dict[str, Any]] = {}
 # input+variables; the templar instance is churned per-task by Ansible.
 _MERGED_APPLICATIONS_CACHE: dict[tuple, dict[str, Any]] = {}
 _MERGED_USERS_CACHE: dict[tuple, dict[str, Any]] = {}
+_MERGED_DOMAINS_CACHE: dict[tuple, dict[str, Any]] = {}
 
 # Re-entry guards. Cross-lookups ({{ lookup('users', ...) }} inside applications
 # and vice versa) can otherwise drive unbounded recursion once strings are
@@ -595,11 +595,7 @@ def _build_application_defaults(roles_dir: Path) -> dict[str, Any]:
 
         applications[application_id] = config_data
 
-    rendered = DictRenderer().render({"applications": applications})
-    apps = rendered.get("applications", {})
-    if not isinstance(apps, dict):
-        raise ValueError("Rendered applications payload must be a mapping")
-    return {key: apps[key] for key in sorted(apps)}
+    return {key: applications[key] for key in sorted(applications)}
 
 
 def get_application_defaults(
@@ -738,9 +734,75 @@ def get_merged_users(
     return rendered
 
 
+def get_merged_domains(
+    *,
+    variables: Optional[dict[str, Any]] = None,
+    roles_dir: Optional[str | os.PathLike[str]] = None,
+    templar: Any = None,
+) -> dict[str, Any]:
+    """Build the canonical-domain map lazily, mirroring the previous
+    `domains` set_fact at [tasks/stages/01_constructor.yml].
+
+    The result is:
+        canonical_domains_map(applications, DOMAIN_PRIMARY)
+        | combine(variables['domains'], recursive=True)  # user/play overrides
+
+    Cached keyed on (roles_dir, variables_signature, domains_override_fingerprint).
+    """
+    from plugins.filter.canonical_domains_map import (
+        FilterModule as _CanonicalDomainsFilter,
+    )
+
+    variables = variables or {}
+    resolved_roles_dir = _resolve_roles_dir(roles_dir=roles_dir)
+
+    override = variables.get("domains")
+    if not isinstance(override, Mapping):
+        override = {}
+
+    cache_key = (
+        _cache_key(resolved_roles_dir),
+        _stable_variables_signature(variables),
+        _fingerprint_mapping(override),
+    )
+    cached = _MERGED_DOMAINS_CACHE.get(cache_key)
+    if cached is not None:
+        return cached
+
+    primary_domain = (
+        variables.get("DOMAIN_PRIMARY") or variables.get("SYSTEM_EMAIL_DOMAIN") or ""
+    )
+    if not primary_domain:
+        raise ValueError(
+            "get_merged_domains: DOMAIN_PRIMARY (or SYSTEM_EMAIL_DOMAIN fallback) "
+            "must be set in variables."
+        )
+
+    apps = get_merged_applications(
+        variables=variables,
+        roles_dir=roles_dir,
+        templar=templar,
+    )
+
+    filter_instance = _CanonicalDomainsFilter()
+    base_map = filter_instance.canonical_domains_map(apps, primary_domain)
+
+    merged: dict[str, Any] = copy.deepcopy(base_map)
+    if override:
+        merged = _deep_merge(merged, dict(override))
+        if not isinstance(merged, dict):
+            raise ValueError(
+                "get_merged_domains: merged domain map must be a mapping after override."
+            )
+
+    _MERGED_DOMAINS_CACHE[cache_key] = merged
+    return merged
+
+
 def _reset_cache_for_tests() -> None:
     _APPLICATIONS_DEFAULTS_CACHE.clear()
     _USERS_DEFAULTS_CACHE.clear()
     _MERGED_APPLICATIONS_CACHE.clear()
     _MERGED_USERS_CACHE.clear()
+    _MERGED_DOMAINS_CACHE.clear()
     _FINGERPRINT_BY_ID.clear()