feat: RFC 0008 environment wrap actions

.gitignore

@@ -23,3 +23,6 @@ __pycache__/
 /dist
 _version.py
 *.log
+
+# Local one-off helper scripts (not part of the project)
+/tmp/

pyproject.toml

@@ -31,6 +31,11 @@ classifiers = [
   "Topic :: Software Development :: Libraries"
 ]
 dependencies = [
+  # NOTE (version-skew guard): this branch requires openjd-model features that
+  # only exist as of openjd-model-for-python PR #318 (FormatString.resolve_value,
+  # FormatString.whole_field_expression, evaluate_let_bindings, Step.let).
+  # Before release, bump the version floor below to the first openjd-model
+  # release that includes PR #318's surface.
   "openjd-model >= 0.10,< 0.12",
   "pywin32 >= 307; platform_system == 'Windows'",
   "psutil >= 5.9,< 7.3; platform_system == 'Windows'",

src/openjd/sessions/_embedded_files.py

@@ -209,30 +209,91 @@ def __init__(
         self._user = user
 
     def materialize(self, files: EmbeddedFilesListType, symtab: SymbolTable) -> None:
+        records = self.allocate_file_paths(files, symtab)
+        self.write_file_contents(records, symtab)
+
+    def allocate_file_paths(
+        self, files: EmbeddedFilesListType, symtab: SymbolTable
+    ) -> list[_FileRecord]:
+        """Allocate the on-disk paths for the embedded files and define their
+        ``Env.File.*``/``Task.File.*`` symbols in ``symtab``, without writing
+        the file contents.
+
+        Splitting allocation from :meth:`write_file_contents` lets the runner
+        evaluate EXPR ``let`` bindings between the two phases (RFC 0007): a
+        file's *path* never depends on ``let`` values (``filename`` is a plain
+        string), so the ``Env.File.*``/``Task.File.*`` symbols are available
+        to the bindings, while a file's ``data`` is written afterwards so it
+        can reference let-bound values. Mirrors the openjd-rs runners.
+        """
         if self._scope == EmbeddedFilesScope.ENV:
             self._logger.info("Writing embedded files for Environment to disk.")
         else:
             self._logger.info("Writing embedded files for Task to disk.")
 
+        records = self.allocate_records(files)
+        self._define_symbols(records, symtab)
+        return records
+
+    def allocate_records(self, files: EmbeddedFilesListType) -> list["_FileRecord"]:
+        """Allocate the on-disk paths for the embedded files, without logging
+        or defining any symbols.
+
+        Used by the Session to pre-allocate a wrap environment's file records
+        once (RFC 0008); each wrap-hook invocation then defines the symbols
+        and logs through :meth:`register_file_paths` against its own symbol
+        table.
+
+        Raises:
+            RuntimeError: If a file path could not be allocated.
+        """
         try:
             records = list[_FileRecord]()
             # Generate the symbol table values and filenames
             for file in files:
                 # Raises: OSError
                 symbol, filename = self._get_symtab_entry(file)
                 records.append(_FileRecord(symbol=symbol, filename=filename, file=file))
+            return records
+        except (OSError, ValueError) as err:
+            raise RuntimeError(f"Could not write embedded file: {err}")
 
-            # Add symbols to the symbol table
-            for record in records:
-                symtab[record.symbol] = str(record.filename)
-                self._logger.info(
-                    f"Mapping: {record.symbol} -> {record.filename}",
-                    extra=LogExtraInfo(
-                        openjd_log_content=LogContent.FILE_PATH | LogContent.PARAMETER_INFO
-                    ),
-                )
+    def register_file_paths(self, records: list["_FileRecord"], symtab: SymbolTable) -> None:
+        """Define the ``Env.File.*``/``Task.File.*`` symbols in ``symtab``
+        for already-allocated records, without allocating new paths.
 
-            # Write the files to disk.
+        Used to reuse a wrap environment's embedded-file records across
+        wrap-hook invocations (RFC 0008): the on-disk paths are allocated
+        once so the symbols stay stable and unnamed files do not accumulate,
+        while the contents are re-resolved and rewritten per invocation via
+        :meth:`write_file_contents`.
+        """
+        if self._scope == EmbeddedFilesScope.ENV:
+            self._logger.info("Writing embedded files for Environment to disk.")
+        else:
+            self._logger.info("Writing embedded files for Task to disk.")
+        self._define_symbols(records, symtab)
+
+    def _define_symbols(self, records: list["_FileRecord"], symtab: SymbolTable) -> None:
+        # Add symbols to the symbol table. For EXPR evaluation the
+        # Env.File.*/Task.File.* symbols are host-format path values
+        # (property access like `.parent` works), matching openjd-rs;
+        # the legacy (non-EXPR) interpolation path ignores the type and
+        # keeps the string form.
+        for record in records:
+            symtab[record.symbol] = str(record.filename)
+            symtab.expr_types[record.symbol] = "PATH"
+            self._logger.info(
+                f"Mapping: {record.symbol} -> {record.filename}",
+                extra=LogExtraInfo(
+                    openjd_log_content=LogContent.FILE_PATH | LogContent.PARAMETER_INFO
+                ),
+            )
+
+    def write_file_contents(self, records: list["_FileRecord"], symtab: SymbolTable) -> None:
+        """Resolve each allocated file's ``data`` against ``symtab`` and write
+        it to disk. See :meth:`allocate_file_paths`."""
+        try:
             for record in records:
                 # Raises: OSError
                 self._materialize_file(record.filename, record.file, symtab)

src/openjd/sessions/_path_mapping.py

@@ -4,27 +4,43 @@
 from enum import Enum
 from os import name as os_name
 from pathlib import PurePath, PurePosixPath, PureWindowsPath
+from typing import Optional, Union
 
 
 class PathFormat(str, Enum):
     POSIX = "POSIX"
     WINDOWS = "WINDOWS"
+    # RFC 0006 §2.3.2 (EXPR extension): URI-form source paths
+    # (e.g. "s3://bucket/prefix") that map to local filesystem paths.
+    URI = "URI"
 
 
 @dataclass(frozen=True)
 class PathMappingRule:
     source_path_format: PathFormat
-    source_path: PurePath
+    # URI-format rules keep the source as the raw string: URIs are not
+    # filesystem paths, and PurePath normalization would corrupt the
+    # "scheme://" separator.
+    source_path: Union[PurePath, str]
     destination_path: PurePath
 
     def __init__(
-        self, *, source_path_format: PathFormat, source_path: PurePath, destination_path: PurePath
+        self,
+        *,
+        source_path_format: PathFormat,
+        source_path: Union[PurePath, str],
+        destination_path: PurePath,
     ):
         if source_path_format == PathFormat.POSIX:
             if not isinstance(source_path, PurePosixPath):
                 raise ValueError(
                     "Path mapping rule source_path_format does not match source_path type"
                 )
+        elif source_path_format == PathFormat.URI:
+            if not isinstance(source_path, str):
+                raise ValueError(
+                    "Path mapping rule source_path must be a string for the URI source_path_format"
+                )
         else:
             if not isinstance(source_path, PureWindowsPath):
                 raise ValueError(
@@ -49,9 +65,12 @@ def from_dict(rule: dict[str, str]) -> "PathMappingRule":
                 raise ValueError(f"Path mapping rule requires the following fields: {field_names}")
 
         source_path_format = PathFormat(rule["source_path_format"].upper())
-        source_path: PurePath
+        source_path: Union[PurePath, str]
         if source_path_format == PathFormat.POSIX:
             source_path = PurePosixPath(rule["source_path"])
+        elif source_path_format == PathFormat.URI:
+            # Keep URIs verbatim; PurePath would collapse "scheme://".
+            source_path = rule["source_path"]
         else:
             source_path = PureWindowsPath(rule["source_path"])
         destination_path = PurePath(rule["destination_path"])
@@ -76,25 +95,90 @@ def to_dict(self) -> dict[str, str]:
             "destination_path": str(self.destination_path),
         }
 
+    @staticmethod
+    def _uri_path_start(uri: str) -> Optional[int]:
+        """Byte offset where the path component begins in a URI (after
+        ``scheme://authority``), or None when there is no ``://`` separator.
+        Mirrors openjd-rs's ``uri_path_start``."""
+        scheme_sep = uri.find("://")
+        if scheme_sep == -1:
+            return None
+        authority_start = scheme_sep + 3
+        slash = uri.find("/", authority_start)
+        return len(uri) if slash == -1 else slash
+
+    def _apply_uri(self, path: str) -> tuple[bool, str]:
+        """Apply a URI-format rule, mirroring openjd-rs's ``apply_uri``.
+
+        Per RFC 3986 the scheme and authority match case-insensitively while
+        the path portion matches case-sensitively, on whole path components.
+        The result is a local path in the host's format.
+        """
+        sep = "/" if os_name == "posix" else "\\"
+        source = str(self.source_path)
+        src_path_start = self._uri_path_start(source)
+        src_path_start = 0 if src_path_start is None else src_path_start
+        inp_path_start = self._uri_path_start(path)
+        inp_path_start = 0 if inp_path_start is None else inp_path_start
+
+        # Scheme+authority must match case-insensitively.
+        if path[:inp_path_start].lower() != source[:src_path_start].lower():
+            return False, path
+        # Path portion must match case-sensitively, on a component boundary.
+        src_path = source[src_path_start:]
+        inp_path = path[inp_path_start:]
+        if not inp_path.startswith(src_path):
+            return False, path
+        remainder = inp_path[len(src_path) :]
+        if remainder and not remainder.startswith("/"):
+            return False, path
+
+        child_parts = remainder[1:].split("/") if remainder else []
+        result = str(self.destination_path)
+        for part in child_parts:
+            result += sep + part
+        if path.endswith("/") and not result.endswith(sep):
+            result += sep
+        return True, result
+
+    def source_path_component_count(self) -> int:
+        """Number of components in the source path, used to order rules from
+        most to least specific. For URI sources the ``scheme://authority``
+        counts as one component plus one per path segment."""
+        if isinstance(self.source_path, PurePath):
+            return len(self.source_path.parts)
+        source = str(self.source_path)
+        path_start = self._uri_path_start(source)
+        if path_start is None:
+            return 1
+        path_portion = source[path_start:].strip("/")
+        return 1 + (len(path_portion.split("/")) if path_portion else 0)
+
     def apply(self, *, path: str) -> tuple[bool, str]:
         """Applies the path mapping rule on the given path, if it matches the rule.
         Does not collapse ".." since symbolic paths could be used.
 
         Returns: tuple[bool, str] - indicating if the path matched the rule and the resulting
         mapped path. If it doesn't match, then it returns the original path unmodified.
         """
+        if self.source_path_format == PathFormat.URI:
+            return self._apply_uri(path)
+        source_path = self.source_path
+        if not isinstance(source_path, PurePath):
+            # The constructor guarantees non-URI rules carry PurePath sources.
+            raise TypeError(
+                "Path mapping rule source_path must be a PurePath for filesystem source formats"
+            )
         pure_path: PurePath
         if self.source_path_format == PathFormat.POSIX:
             pure_path = PurePosixPath(path)
         else:
             pure_path = PureWindowsPath(path)
 
-        if not pure_path.is_relative_to(self.source_path):
+        if not pure_path.is_relative_to(source_path):
             return False, path
 
-        remapped_parts = (
-            self.destination_path.parts + pure_path.parts[len(self.source_path.parts) :]
-        )
+        remapped_parts = self.destination_path.parts + pure_path.parts[len(source_path.parts) :]
         if os_name == "posix":
             result = str(PurePosixPath(*remapped_parts))
             if self._has_trailing_slash(self.source_path_format, path):