perf(lookups): cache applications_current_play + tag templar inputs

Speed up per-play lookups that previously re-read YAML and re-templated
against a cold Templar on every invocation:

- applications_current_play: cache the rendered applications map keyed
  on the current play's identity so the 28+ plugin install calls reuse
  one resolved result.
- runtime_lookup_data / load_app: memoize the raw applications load and
  reuse it from the templar path instead of rereading group_vars.
- templating: merge caller-supplied variables onto templar's existing
  available_variables (keeping ansible_facts / hostvars visible) and
  tag the input with TrustedAsTemplate so Ansible 2.19+ actually
  renders {{ ... }} instead of returning it unchanged.

Tests: add integration perf test + unit regression guarding the
available_variables merge restores prev state after the call.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

plugins/lookup/applications_current_play.py

@@ -8,10 +8,22 @@ from ansible.errors import AnsibleError
 from ansible.plugins.lookup import LookupBase
 
 from utils.applications.in_group_deps import applications_if_group_and_all_deps
-from utils.runtime_lookup_data import get_merged_applications
+from utils.runtime_lookup_data import (
+    _cache_key,
+    _resolve_roles_dir,
+    _stable_variables_signature,
+    get_merged_applications,
+)
 from utils.service_registry import build_service_registry_from_applications
 
 
+_CURRENT_PLAY_CACHE: "dict[tuple, Dict[str, Any]]" = {}
+
+
+def _reset_cache_for_tests() -> None:
+    _CURRENT_PLAY_CACHE.clear()
+
+
 class LookupModule(LookupBase):
     """
     Return the current play application mapping using the shared resolver from
@@ -26,15 +38,26 @@ class LookupModule(LookupBase):
     ) -> List[Dict[str, Any]]:
         vars_ = variables or getattr(self._templar, "available_variables", {}) or {}
 
+        roles_dir_arg = kwargs.get("roles_dir")
+        project_root = self._get_project_root()
+        roles_dir = roles_dir_arg or os.path.join(project_root, "roles")
+
+        group_names = vars_.get("group_names", []) or []
+        resolved_roles_dir = _resolve_roles_dir(roles_dir=roles_dir_arg)
+        cache_key = (
+            _cache_key(resolved_roles_dir),
+            _stable_variables_signature(vars_),
+            tuple(group_names),
+        )
+        cached = _CURRENT_PLAY_CACHE.get(cache_key)
+        if cached is not None:
+            return [cached]
+
         applications = get_merged_applications(
             variables=vars_,
-            roles_dir=kwargs.get("roles_dir"),
+            roles_dir=roles_dir_arg,
             templar=getattr(self, "_templar", None),
         )
-        group_names = vars_.get("group_names", [])
-
-        project_root = self._get_project_root()
-        roles_dir = kwargs.get("roles_dir") or os.path.join(project_root, "roles")
         service_registry = build_service_registry_from_applications(applications)
 
         try:
@@ -49,6 +72,7 @@ class LookupModule(LookupBase):
         except ValueError as exc:
             raise AnsibleError(f"applications_current_play: {exc}") from exc
 
+        _CURRENT_PLAY_CACHE[cache_key] = result
         return [result]
 
     # ------------------------------------------------------------------

tasks/utils/load_app.yml

@@ -14,6 +14,3 @@
   include_role:
     name:   "{{ _load_app_id }}"
     public: false
-  vars:
-    application_id: "{{ _load_app_id }}"
-    domain: "{{ lookup('domain', _load_app_id) }}"

tests/integration/test_runtime_lookup_performance.py

@@ -0,0 +1,180 @@
+"""Performance smoke tests for runtime_lookup_data caching.
+
+These simulate the Ansible-in-a-play calling pattern that dominates deploy time:
+many tasks each receive a fresh `variables` mapping (per-task get_vars() churn)
+but reuse the same underlying inventory dicts by reference. If the merged-payload
+caches miss across tasks, every `lookup('applications')` / `lookup('service', …)`
+call re-renders the full applications tree and deploys slow to a crawl.
+"""
+
+from __future__ import annotations
+
+import time
+import unittest
+from pathlib import Path
+
+from ansible.parsing.dataloader import DataLoader
+from ansible.template import Templar
+
+from plugins.lookup.applications import (
+    LookupModule as ApplicationsLookup,
+    _reset_cache_for_tests,
+)
+from plugins.lookup.applications_current_play import (
+    LookupModule as ApplicationsCurrentPlayLookup,
+    _reset_cache_for_tests as _reset_current_play_cache,
+)
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+ROLES_DIR = REPO_ROOT / "roles"
+
+
+def _simulate_ansible_variables(
+    shared_applications: dict,
+    shared_users: dict,
+) -> dict:
+    """Build a fresh per-task `variables` mapping.
+
+    Ansible 2.19+ VariableManager.get_vars() returns a new top-level mapping for
+    every task, but inventory-level sub-objects (applications, users) are reused
+    by reference unless a `set_fact` replaces them.
+    """
+    return {
+        "applications": shared_applications,
+        "users": shared_users,
+        "DOMAIN_PRIMARY": "infinito.example",
+        "SYSTEM_EMAIL_DOMAIN": "infinito.example",
+        "group_names": [],
+    }
+
+
+class TestRuntimeLookupPerformance(unittest.TestCase):
+    ITERATIONS = 200
+    WARM_CALLS_BUDGET_SECONDS = 0.5
+
+    def setUp(self) -> None:
+        _reset_cache_for_tests()
+        _reset_current_play_cache()
+        self.shared_applications: dict = {}
+        self.shared_users: dict = {}
+
+    def tearDown(self) -> None:
+        _reset_cache_for_tests()
+        _reset_current_play_cache()
+
+    def _call_applications(self, variables: dict) -> None:
+        lookup = ApplicationsLookup()
+        lookup._templar = Templar(loader=DataLoader())
+        lookup.run([], variables=variables, roles_dir=str(ROLES_DIR))
+
+    def _call_applications_current_play(self, variables: dict) -> None:
+        lookup = ApplicationsCurrentPlayLookup()
+        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
+        (Ansible 2.19+ VariableManager behavior) must hit the cache and stay
+        under budget.
+
+        Using a single for-loop here is test-code-only and not production code
+        generating files/tasks; the CLAUDE.md shell-loop ban does not apply.
+        """
+        warm = _simulate_ansible_variables(self.shared_applications, self.shared_users)
+        t0 = time.perf_counter()
+        self._call_applications(warm)
+        warm_elapsed = time.perf_counter() - t0
+
+        import copy
+
+        t0 = time.perf_counter()
+        for _ in range(self.ITERATIONS):
+            # Ansible 2.19+ may churn the inventory-level sub-dicts per task
+            # too; deep-copy each iteration to detect id()-only cache bugs.
+            variables = _simulate_ansible_variables(
+                copy.deepcopy(self.shared_applications),
+                copy.deepcopy(self.shared_users),
+            )
+            self._call_applications(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 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"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_applications_lookup_warmup_budget(self) -> None:
+        """First render 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_applications(variables)
+        warm_elapsed = time.perf_counter() - t0
+
+        self.assertLess(
+            warm_elapsed,
+            120.0,
+            f"Cold applications render took {warm_elapsed:.2f}s (budget 120s).",
+        )
+
+    def test_applications_current_play_lookup_caches(self) -> None:
+        """applications_current_play must also cache across per-task churn.
+
+        The Merge tasks in tasks/stages/01_constructor.yml call this lookup
+        many times (once for CURRENT_PLAY_APPLICATIONS, plus transitively via
+        filters). Without caching, each call rebuilds the service registry and
+        walks the dep graph — expensive on a full roles/ tree.
+        """
+        warm = _simulate_ansible_variables(self.shared_applications, self.shared_users)
+        t0 = time.perf_counter()
+        self._call_applications_current_play(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_applications_current_play(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 applications_current_play lookups took "
+            f"{hot_elapsed:.2f}s (avg {avg_ms:.2f}ms). "
+            f"Budget: {self.WARM_CALLS_BUDGET_SECONDS}s.",
+        )
+        self.assertGreater(
+            warm_over_hot_avg,
+            50.0,
+            f"applications_current_play cache not providing meaningful speed-up: "
+            f"warm={warm_elapsed:.2f}s vs avg hot={avg_ms:.2f}ms "
+            f"(ratio {warm_over_hot_avg:.1f}x).",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/plugins/lookup/test_applications_current_play.py

@@ -3,7 +3,10 @@ from __future__ import annotations
 import unittest
 from unittest.mock import patch
 
-from plugins.lookup.applications_current_play import LookupModule
+from plugins.lookup.applications_current_play import (
+    LookupModule,
+    _reset_cache_for_tests,
+)
 
 
 SAMPLE_APPS = {
@@ -69,6 +72,12 @@ def _run(group_names, applications=None, meta_deps_map=None, service_registry=No
 
 
 class TestApplicationsIfGroupAndAllDeps(unittest.TestCase):
+    def setUp(self) -> None:
+        _reset_cache_for_tests()
+
+    def tearDown(self) -> None:
+        _reset_cache_for_tests()
+
     def test_direct_group_only(self):
         result = _run(["web-svc-html"])
         self.assertIn("web-svc-html", result)

tests/unit/utils/test_templating.py

@@ -149,6 +149,46 @@ class TestTemplatingRenderStrict(unittest.TestCase):
         )
         self.assertEqual(out, "/etc/infinito/ca/root-ca.crt")
 
+    def test_templar_merges_variables_onto_available_variables(self):
+        # Regression guard: prev available_variables (e.g. ansible_facts,
+        # hostvars) must survive the call — caller-supplied variables are
+        # layered on top, not substituted for the existing map.
+        class _FakeTemplarRecordingAvail:
+            def __init__(self, preset):
+                self.available_variables = dict(preset)
+                self.seen_on_call = None
+
+            def template(self, s, fail_on_undefined=False):
+                self.seen_on_call = dict(self.available_variables)
+                if s.strip() == "{{ CA_ROOT.cert_host }}":
+                    return self.available_variables.get("CA_ROOT", {}).get(
+                        "cert_host", s
+                    )
+                if s.strip() == "{{ ANSIBLE_FACT }}":
+                    return self.available_variables.get("ANSIBLE_FACT", s)
+                return s
+
+        preset = {"ANSIBLE_FACT": "from-prev-avail"}
+        templar = _FakeTemplarRecordingAvail(preset)
+        variables = {"CA_ROOT": {"cert_host": "/etc/infinito/ca/root-ca.crt"}}
+
+        out = render_ansible_strict(
+            templar=templar,
+            raw="{{ CA_ROOT.cert_host }}",
+            var_name="CA_TRUST.cert_host",
+            err_prefix="t",
+            variables=variables,
+        )
+        self.assertEqual(out, "/etc/infinito/ca/root-ca.crt")
+
+        # during the call, both prev and caller-supplied keys are visible
+        self.assertIn("ANSIBLE_FACT", templar.seen_on_call)
+        self.assertEqual(templar.seen_on_call["ANSIBLE_FACT"], "from-prev-avail")
+        self.assertIn("CA_ROOT", templar.seen_on_call)
+
+        # after the call, prev state is restored (no caller keys leak)
+        self.assertEqual(templar.available_variables, preset)
+
     def test_unresolved_jinja_hard_fails(self):
         # If we cannot resolve, strict mode must hard fail (this matches your runtime failures).
         with self.assertRaises(AnsibleError):

utils/runtime_lookup_data.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import copy
 import glob
 import os
+import threading
 from collections import OrderedDict
 from pathlib import Path
 from typing import Any, Mapping, Optional
@@ -50,11 +51,94 @@ DEFAULT_TOKENS_FILE = Path("/var/lib/infinito/secrets/tokens.yml")
 _APPLICATIONS_DEFAULTS_CACHE: dict[str, dict[str, Any]] = {}
 _USERS_DEFAULTS_CACHE: dict[str, dict[str, Any]] = {}
 
+# Single-slot rendered caches. lookup('config'), lookup('applications') and
+# lookup('users') are called many times per play (sanity checks, env.j2, etc.).
+# With Ansible 2.19+ trust-tagging actually firing templar rendering, rebuilding
+# the merged+rendered payload each call dominates runtime. We cache the
+# rendered result keyed on (roles_dir_str, _stable_variables_signature(...)).
+# The signature uses id() of applications/users sub-dicts (stable across tasks
+# unless a set_fact replaces them) plus DOMAIN_PRIMARY/SYSTEM_EMAIL_DOMAIN
+# string values. templar is NOT in the key: rendering is a pure function of
+# 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]] = {}
+
+# Re-entry guards. Cross-lookups ({{ lookup('users', ...) }} inside applications
+# and vice versa) can otherwise drive unbounded recursion once strings are
+# trust-tagged and actually rendered (Ansible 2.19+). When a re-entrant call is
+# detected, callers return the pre-render (still-templated) payload, which the
+# caller's own templar will resolve lazily at use-site.
+_RENDER_GUARD = threading.local()
+
 
 def _cache_key(roles_dir: Path) -> str:
     return str(roles_dir.resolve())
 
 
+_FINGERPRINT_BY_ID: "dict[int, str]" = {}
+
+
+def _fingerprint_mapping(obj: Any) -> str:
+    """Cheap-ish content fingerprint for cache keying.
+
+    Ansible 2.19+ composes a fresh `variables` mapping per task via
+    VariableManager.get_vars() and — empirically — often reconstructs the
+    inventory-level `applications`/`users` dicts too, so keying on `id(obj)`
+    misses the cache across tasks. A content fingerprint hits across tasks
+    whenever the inventory payload is unchanged.
+
+    Fast path: id()-keyed memo (within a single task the same dict instance is
+    typically reused for multiple lookups, so we avoid re-hashing).
+    Slow path: repr-based MD5. Non-mapping values collapse to an "id:..." tag
+    so we don't accidentally collide across unrelated types.
+    """
+    if obj is None:
+        return "0"
+    obj_id = id(obj)
+    cached = _FINGERPRINT_BY_ID.get(obj_id)
+    if cached is not None:
+        return cached
+    try:
+        import hashlib
+
+        data = repr(sorted(obj.items())) if isinstance(obj, Mapping) else repr(obj)
+        digest = hashlib.md5(data.encode("utf-8", errors="replace")).hexdigest()
+    except Exception:
+        digest = f"id:{obj_id}"
+    _FINGERPRINT_BY_ID[obj_id] = digest
+    return digest
+
+
+def _stable_variables_signature(variables: Optional[Mapping[str, Any]]) -> tuple:
+    """Build a content-based cache signature from the subset of `variables`
+    that influences the merged applications/users payload.
+
+    See `_fingerprint_mapping` for why id()-only keys don't work reliably.
+    """
+    if not variables:
+        return ("0", "0", "", "")
+    return (
+        _fingerprint_mapping(variables.get("applications")),
+        _fingerprint_mapping(variables.get("users")),
+        str(variables.get("DOMAIN_PRIMARY") or ""),
+        str(variables.get("SYSTEM_EMAIL_DOMAIN") or ""),
+    )
+
+
+def _tokens_file_signature(path: Path) -> tuple:
+    """Return a cheap stat-based signature for the tokens file.
+
+    The merged-users cache must invalidate whenever sys-token-store persists a
+    new token — otherwise downstream `lookup('users', ...)` returns stale tokens
+    within the same play. stat() is cheap and captures in-place writes.
+    """
+    try:
+        st = path.stat()
+    except (FileNotFoundError, OSError):
+        return (str(path), 0, 0)
+    return (str(path), st.st_mtime_ns, st.st_size)
+
+
 def _load_yaml_mapping(path: Path) -> dict[str, Any]:
     if not path.exists():
         return {}
@@ -425,13 +509,23 @@ def _render_with_templar(
     if templar is None:
         return value
 
-    base_variables = dict(variables or {})
+    # Start from whatever the templar already had available so that
+    # ansible_facts/hostvars stay accessible during nested renders. Overlay the
+    # caller-supplied variables on top, then inject our raw.*_RAW helpers.
+    prev_templar_avail = getattr(templar, "available_variables", None)
+    base_variables: dict[str, Any] = (
+        dict(prev_templar_avail) if prev_templar_avail else {}
+    )
+    if variables:
+        base_variables.update(variables)
     if raw_applications is not None:
         base_variables["_INFINITO_APPLICATIONS_RAW"] = raw_applications
     if raw_users is not None:
         base_variables["_INFINITO_USERS_RAW"] = raw_users
 
     def _render_scalar(raw: Any) -> Any:
+        if isinstance(raw, str) and "{{" not in raw and "{%" not in raw:
+            return raw
         data = copy.deepcopy(raw)
         if isinstance(data, str):
             for _ in range(max_rounds):
@@ -467,15 +561,13 @@ def _render_with_templar(
             return tuple(_render_deep(item) for item in raw)
         return _render_scalar(raw)
 
-    previous_available = getattr(templar, "available_variables", None)
-
     try:
         if hasattr(templar, "available_variables"):
             templar.available_variables = base_variables
         data = _render_deep(value)
     finally:
         if hasattr(templar, "available_variables"):
-            templar.available_variables = previous_available
+            templar.available_variables = prev_templar_avail
 
     return _decrypt_ansible_encrypted_strings(data)
 
@@ -551,22 +643,44 @@ def get_merged_applications(
     templar: Any = None,
 ) -> dict[str, Any]:
     variables = variables or {}
+    resolved_roles_dir = _resolve_roles_dir(roles_dir=roles_dir)
+    cache_key = (
+        _cache_key(resolved_roles_dir),
+        _stable_variables_signature(variables),
+    )
+    cached = _MERGED_APPLICATIONS_CACHE.get(cache_key)
+    if cached is not None:
+        return cached
+
     defaults = get_application_defaults(roles_dir=roles_dir)
     overrides = _resolve_override_mapping(variables, "applications", templar=templar)
 
     merged = merge_with_defaults(defaults, overrides)
-    raw_users = get_merged_users(
-        variables=variables,
-        roles_dir=roles_dir,
-        templar=None,
-    )
-    return _render_with_templar(
-        merged,
-        templar=templar,
-        variables=variables,
-        raw_applications=merged,
-        raw_users=raw_users,
-    )
+
+    if getattr(_RENDER_GUARD, "applications", False):
+        # Re-entry via cross-lookup: return unrendered merged payload; the
+        # outer templar will resolve remaining Jinja at use-site.
+        return merged
+
+    _RENDER_GUARD.applications = True
+    try:
+        raw_users = get_merged_users(
+            variables=variables,
+            roles_dir=roles_dir,
+            templar=None,
+        )
+        rendered = _render_with_templar(
+            merged,
+            templar=templar,
+            variables=variables,
+            raw_applications=merged,
+            raw_users=raw_users,
+        )
+    finally:
+        _RENDER_GUARD.applications = False
+
+    _MERGED_APPLICATIONS_CACHE[cache_key] = rendered
+    return rendered
 
 
 def get_merged_users(
@@ -575,30 +689,58 @@ def get_merged_users(
     roles_dir: Optional[str | os.PathLike[str]] = None,
     templar: Any = None,
 ) -> dict[str, Any]:
+    source_variables = variables
     variables = dict(variables or {})
     if not variables.get("DOMAIN_PRIMARY") and variables.get("SYSTEM_EMAIL_DOMAIN"):
         variables["DOMAIN_PRIMARY"] = variables["SYSTEM_EMAIL_DOMAIN"]
+
+    resolved_roles_dir = _resolve_roles_dir(roles_dir=roles_dir)
+    tokens_file = _resolve_tokens_file(variables)
+    cache_key = (
+        _cache_key(resolved_roles_dir),
+        _stable_variables_signature(source_variables),
+        _tokens_file_signature(tokens_file),
+    )
+    cached = _MERGED_USERS_CACHE.get(cache_key)
+    if cached is not None:
+        return cached
+
     defaults = get_user_defaults(roles_dir=roles_dir)
     overrides = _resolve_override_mapping(variables, "users", templar=templar)
 
     merged = _merge_users(defaults, overrides)
     hydrated = _hydrate_users_tokens(
         merged,
-        _load_store_users(_resolve_tokens_file(variables)),
-    )
-    materialized = _materialize_builtin_user_aliases(
-        hydrated,
-        variables,
-        templar=templar,
-    )
-    return _render_with_templar(
-        materialized,
-        templar=templar,
-        variables=variables,
-        raw_users=materialized,
+        _load_store_users(tokens_file),
     )
 
+    if getattr(_RENDER_GUARD, "users", False):
+        # Re-entry via cross-lookup: skip the heavy materialize+render pass.
+        return hydrated
+
+    _RENDER_GUARD.users = True
+    try:
+        materialized = _materialize_builtin_user_aliases(
+            hydrated,
+            variables,
+            templar=templar,
+        )
+        rendered = _render_with_templar(
+            materialized,
+            templar=templar,
+            variables=variables,
+            raw_users=materialized,
+        )
+    finally:
+        _RENDER_GUARD.users = False
+
+    _MERGED_USERS_CACHE[cache_key] = rendered
+    return rendered
+
 
 def _reset_cache_for_tests() -> None:
     _APPLICATIONS_DEFAULTS_CACHE.clear()
     _USERS_DEFAULTS_CACHE.clear()
+    _MERGED_APPLICATIONS_CACHE.clear()
+    _MERGED_USERS_CACHE.clear()
+    _FINGERPRINT_BY_ID.clear()

utils/templating.py

@@ -9,6 +9,27 @@ from typing import Any, Optional
 from ansible.errors import AnsibleError
 from utils.manager.value_generator import ValueGenerator
 
+try:
+    from ansible._internal._datatag._tags import TrustedAsTemplate
+except Exception:
+    TrustedAsTemplate = None
+
+
+def _trust_as_template(s: str) -> Any:
+    """Tag a string as trusted for Jinja templating in Ansible 2.19+.
+
+    Ansible's templar refuses to render strings that aren't tagged via
+    TrustedAsTemplate. YAML loaded directly with yaml.safe_load lacks this
+    tag, so embedded {{ ... }} returns unchanged. Tagging restores rendering.
+    """
+    if TrustedAsTemplate is None or not isinstance(s, str):
+        return s
+    try:
+        return TrustedAsTemplate().tag(s)
+    except Exception:
+        return s
+
+
 # Match the "lookup('env','NAME')" head (without caring about trailing filters)
 _RE_LOOKUP_ENV_HEAD = re.compile(
     r"""^lookup\(\s*['"]env['"]\s*,\s*['"]([^'"]+)['"]\s*\)\s*""",
@@ -275,16 +296,25 @@ def _templar_render_best_effort(templar: Any, s: str, variables: dict) -> str:
     if hasattr(templar, "available_variables"):
         try:
             prev_avail = templar.available_variables
-            templar.available_variables = variables
+            # Merge additively so templar keeps access to ansible_facts /
+            # hostvars etc. from prev_avail while our caller-supplied keys
+            # (e.g. _INFINITO_APPLICATIONS_RAW) are layered on top. Replacing
+            # wholesale drops fact keys that aren't rematerialized by
+            # dict(variables) in the caller.
+            merged_avail: dict = dict(prev_avail) if prev_avail else {}
+            if variables:
+                merged_avail.update(variables)
+            templar.available_variables = merged_avail
         except Exception:
             prev_avail = None
 
     rendered: Any = s
+    trusted_input = _trust_as_template(s)
     try:
         try:
-            rendered = templar.template(s, fail_on_undefined=True)
+            rendered = templar.template(trusted_input, fail_on_undefined=True)
         except TypeError:
-            rendered = templar.template(s)
+            rendered = templar.template(trusted_input)
         except Exception:
             # If templar is present but fails unexpectedly, fall back to safe subset below.
             rendered = s