refactor: split scenario instantiation helpers

implementations/python/packages/aces_sdl/_scenario_instantiation.py

@@ -0,0 +1,38 @@
+"""Private helpers for validating instantiated scenarios."""
+
+from collections.abc import Mapping
+
+from ._base import VARIABLE_TOKEN_RE
+
+
+def collect_variable_tokens(value: object) -> list[str]:
+    """Return the names of every ``${name}`` token found in string values.
+
+    Mirrors ``instantiate._substitute_value``: every string is a substitution
+    site and mapping keys are not. An instantiated scenario is fully concrete,
+    so no token may survive in any string value.
+    """
+    found: list[str] = []
+    if isinstance(value, Mapping):
+        for nested in value.values():
+            found.extend(collect_variable_tokens(nested))
+    elif isinstance(value, (list, tuple)):
+        for item in value:
+            found.extend(collect_variable_tokens(item))
+    elif isinstance(value, str):
+        found.extend(VARIABLE_TOKEN_RE.findall(value))
+    return found
+
+
+def resolve_json_pointer(payload: object, pointer: str) -> object:
+    """Resolve an RFC 6901-style JSON pointer against a scenario payload."""
+    current = payload
+    for raw_segment in pointer.split("/")[1:]:
+        segment = raw_segment.replace("~1", "/").replace("~0", "~")
+        if isinstance(current, Mapping):
+            current = current[segment]
+        elif isinstance(current, (list, tuple)):
+            current = current[int(segment)]
+        else:
+            raise TypeError("JSON Pointer traverses a scalar value")
+    return current

implementations/python/packages/aces_sdl/scenario.py

@@ -16,7 +16,7 @@
 
 from pydantic import ConfigDict, Field, PrivateAttr, model_validator
 
-from ._base import VARIABLE_TOKEN_RE, SDLModel
+from ._base import SDLModel
 from ._errors import SDLParseDiagnostic
 from ._identifiers import (
     PortableIdentifier,
@@ -25,6 +25,7 @@
     require_portable_identifier,
 )
 from ._mapping_scopes import HASHMAP_SECTIONS
+from ._scenario_instantiation import collect_variable_tokens, resolve_json_pointer
 from ._stateful_resource_references import validate_stateful_resource_references
 from .accounts import Account
 from .agents import Agent
@@ -100,25 +101,6 @@ def _constraint_node_refs(
     return refs
 
 
-def _collect_variable_tokens(value: object) -> list[str]:
-    """Return the names of every ``${name}`` token found in string *values*.
-
-    Mirrors ``instantiate._substitute_value``: every string is a substitution
-    site and mapping keys are not. An :class:`InstantiatedScenario` is fully
-    concrete, so no token may survive in any string value.
-    """
-    found: list[str] = []
-    if isinstance(value, Mapping):
-        for nested in value.values():
-            found.extend(_collect_variable_tokens(nested))
-    elif isinstance(value, (list, tuple)):
-        for item in value:
-            found.extend(_collect_variable_tokens(item))
-    elif isinstance(value, str):
-        found.extend(VARIABLE_TOKEN_RE.findall(value))
-    return found
-
-
 def _validate_declaration_identifier(
     identifier: object,
     *,
@@ -465,7 +447,7 @@ def explicitness(self) -> dict[str, ExplicitnessRecord]:
     @model_validator(mode="after")
     def _reject_unresolved_variable_references(self) -> "InstantiatedScenario":
         payload = self.model_dump(mode="json", by_alias=True)
-        tokens = sorted(set(_collect_variable_tokens(payload)))
+        tokens = sorted(set(collect_variable_tokens(payload)))
         if tokens:
             joined = ", ".join(tokens)
             raise ValueError(f"InstantiatedScenario must not contain unresolved variable references: {joined}")
@@ -475,7 +457,7 @@ def _reject_unresolved_variable_references(self) -> "InstantiatedScenario":
                 binding_values[(*imported.namespace, *binding.parameter)] = binding.value
         for constraint in self.instantiation_provenance.capability_constraints:
             try:
-                concrete_value = _resolve_json_pointer(payload, constraint.field_pointer)
+                concrete_value = resolve_json_pointer(payload, constraint.field_pointer)
             except (KeyError, IndexError, TypeError, ValueError) as exc:
                 raise ValueError(
                     f"Capability constraint field_pointer does not resolve: {constraint.field_pointer}"
@@ -485,16 +467,3 @@ def _reject_unresolved_variable_references(self) -> "InstantiatedScenario":
             if not _json_value_equal(concrete_value, binding_values[constraint.parameter]):
                 raise ValueError("Capability constraint binding does not match the concrete field value")
         return self
-
-
-def _resolve_json_pointer(payload: object, pointer: str) -> object:
-    current = payload
-    for raw_segment in pointer.split("/")[1:]:
-        segment = raw_segment.replace("~1", "/").replace("~0", "~")
-        if isinstance(current, Mapping):
-            current = current[segment]
-        elif isinstance(current, (list, tuple)):
-            current = current[int(segment)]
-        else:
-            raise TypeError("JSON Pointer traverses a scalar value")
-    return current