fix: Subprocess ownership on every exit path, plus five RFC 0008 cross-check fixes

src/openjd/sessions/_action_filter.py

@@ -95,7 +95,14 @@ class ActionMessageKind(Enum):
 )
 filter_matcher = re.compile(filter_regex)
 
-openjd_env_actions_filter_regex = "^(openjd_env|openjd_redacted_env|openjd_unset_env)"
+# The near-miss detector for env macros. The trailing `(?::|\s|\Z)` is
+# load-bearing: without it any line merely *starting with* one of these tokens
+# is reported as a malformed macro and FAILS the action, so a script logging
+# `openjd_environment: ready` or `openjd_env_setup: done` kills its own task.
+# openjd-rs anchors the same three tokens on `:`, a space, or end-of-string
+# (action_filter.rs is_malformed_env_command, whose comment names exactly this
+# false positive), so this is parity as well as a bug fix.
+openjd_env_actions_filter_regex = r"^(openjd_env|openjd_redacted_env|openjd_unset_env)(?::|\s|\Z)"
 openjd_env_actions_filter_matcher = re.compile(openjd_env_actions_filter_regex)
 
 # A regex for matching the assignment of a value to an environment variable

src/openjd/sessions/_runner_base.py

@@ -5,11 +5,12 @@
 import re
 import stat
 import shlex
+import sys
 from abc import ABC, abstractmethod
 from concurrent.futures import Future, ThreadPoolExecutor
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
-from enum import Enum
+from enum import Enum, auto
 from pathlib import Path
 from threading import Lock, Timer
 from typing import Any, Callable, Literal, Optional, Sequence, Type, cast
@@ -20,12 +21,6 @@
 from openjd.model import FormatStringError
 from openjd.model import evaluate_let_bindings
 
-# The EXPR engine's typed value. Imported concretely (rather than duck-typed
-# with getattr) so that a model API change fails loudly at import time instead
-# of silently mis-classifying every optional integer field as "omitted".
-# openjd.expr ships in the same distribution as openjd.model, which this module
-# already hard-imports unreleased API from.
-from openjd.expr import ExprValue, TypeCode
 from openjd.model.v2023_09 import Action as Action_2023_09
 from openjd.model.v2023_09 import CancelationMethodDeferred as CancelationMethodDeferred_2023_09
 from openjd.model.v2023_09 import CancelationMode as CancelationMode_2023_09
@@ -132,6 +127,79 @@ def _over_range_message(description: str, value: int) -> str:
     return f"{description} must be at most {MAX_INT_FIELD_VALUE}, got '{value}'"
 
 
+# Why the EXPR engine's types are reached lazily, from here down:
+#
+# ``openjd.expr`` is a thin facade over the native ``openjd._openjd_rs``
+# extension, so a module-level import of it makes ``import openjd.sessions``
+# load the extension unconditionally -- including for a worker that only ever
+# runs non-EXPR templates. openjd.model deliberately avoids that (see
+# ``openjd.model._format_strings._parser``, which duplicates a constant rather
+# than import the Rust surface), and this module matches that discipline.
+#
+# The guard is on ``sys.modules`` rather than on the value's type, because
+# ``FormatString.resolve_value`` is not limited to ``str`` and ``ExprValue``:
+# a legacy (non-EXPR) whole-field interpolation returns the symbol's own value,
+# so an ``int`` reaches here on a purely non-EXPR path. Testing "not a str"
+# would therefore still load the extension for e.g. an integer-valued
+# ``timeout``.
+_EXTENSION_MODULE = "openjd._openjd_rs"
+
+
+class _ExprKind(Enum):
+    """How a resolved format-string value relates to the EXPR type system."""
+
+    NOT_EXPR = auto()
+    """Not an ``ExprValue``: a plain string, or a legacy interpolation's own
+    value. Resolves to its string form; typed semantics do not apply."""
+
+    NULL = auto()
+    """The engine's typed null -- "field omitted" / "argument skipped"."""
+
+    LIST = auto()
+    """A list, whose elements flatten to one argument each (RFC 0005 §1.3.2)."""
+
+    SCALAR = auto()
+    """Any other typed value; resolves to its string form."""
+
+
+def _classify_expr_value(value: Any) -> _ExprKind:
+    """Classify ``value`` against the EXPR type system.
+
+    This is the *only* place in this module that imports ``openjd.expr``, so
+    that every crossing into the native extension sits behind the one
+    ``sys.modules`` guard below. Doing the whole classification here rather
+    than exposing a separate "is it a list" predicate keeps that property
+    structural instead of merely documented: there is no second, unguarded
+    entry point for a future caller to reach with an arbitrary value.
+
+    ``ExprValue`` instances are created only by the native extension, so if
+    that extension has not been loaded then ``value`` cannot be one and the
+    answer is ``NOT_EXPR`` without importing anything. Once it *has* been
+    loaded -- i.e. an EXPR expression has been evaluated in this process --
+    the import is a ``sys.modules`` hit.
+
+    ``ExprValue`` is imported concretely (rather than duck-typed with
+    ``getattr``) so that a model API change fails loudly here instead of
+    silently mis-classifying every optional integer field as "omitted".
+    """
+    if _EXTENSION_MODULE not in sys.modules:
+        return _ExprKind.NOT_EXPR
+    from openjd.expr import ExprValue, TypeCode
+
+    if not isinstance(value, ExprValue):
+        return _ExprKind.NOT_EXPR
+    if value.is_null:
+        return _ExprKind.NULL
+    if value.type.type_code == TypeCode.LIST:
+        return _ExprKind.LIST
+    return _ExprKind.SCALAR
+
+
+def _is_expr_null(value: Any) -> bool:
+    """True if ``value`` is the EXPR engine's typed null."""
+    return _classify_expr_value(value) is _ExprKind.NULL
+
+
 def _timeout_from_seconds(seconds: int, logger: LoggerAdapter) -> Optional[timedelta]:
     """Convert a resolved timeout in seconds into an enforceable time limit.
 
@@ -236,13 +304,20 @@ def resolve_action_arg_values(args: Optional[Sequence], symtab: SymbolTable) ->
                 # argument is genuinely unresolvable.
                 resolved.append(arg.resolve(symtab=symtab))
                 continue
-            if isinstance(value, str):
-                resolved.append(value)
-            elif value.is_null:
+            kind = _classify_expr_value(value)
+            if kind is _ExprKind.NOT_EXPR:
+                # A plain string, or -- for a legacy (non-EXPR) whole-field
+                # interpolation -- the symbol's own value, which may be any
+                # Python type a caller put in the symbol table (an ``int`` for
+                # an INT job parameter, a ``bool`` for a BOOL one). Typed
+                # null/list semantics exist only under EXPR whole-field
+                # resolution, so everything here resolves to its string form.
+                resolved.append(value if isinstance(value, str) else str(value))
+            elif kind is _ExprKind.NULL:
                 continue
-            elif value.type.type_code == TypeCode.LIST:
+            elif kind is _ExprKind.LIST:
                 resolved.extend(str(element) for element in value)
-            else:
+            else:  # _ExprKind.SCALAR
                 resolved.append(str(value))
     return resolved
 
@@ -302,7 +377,7 @@ def resolve_optional_int_field(
     # to plain string resolution — correct, since typed nulls only exist
     # under EXPR whole-field semantics (Template Schemas 5.3).
     resolved_value = value.resolve_value(symtab=symtab)
-    if isinstance(resolved_value, ExprValue) and resolved_value.is_null:
+    if _is_expr_null(resolved_value):
         return None
     resolved = str(resolved_value)
     # Strict ASCII integer grammar, matching the Rust runtime's str::parse
@@ -381,7 +456,7 @@ def resolve_period(period: Any) -> Optional[int]:
         # openjd-rs runtime, which errors on any non-null, non-mode-name
         # result).
         mode_value = cancelation.mode.resolve_value(symtab=symtab)
-        if isinstance(mode_value, ExprValue) and mode_value.is_null:
+        if _is_expr_null(mode_value):
             # Null mode drops the ENTIRE cancelation object: mode is the
             # object's required discriminator, so an "omitted" mode cannot
             # leave a partial object behind. The action behaves exactly as
@@ -1251,7 +1326,14 @@ def _cancel(
                     write_file_for_user(
                         self._session_working_directory / "cancel_info.json", notify_end, self._user
                     )
-                except OSError as err:
+                except Exception as err:
+                    # `Exception`, not `OSError`: write_file_for_user reaches
+                    # WindowsPermissionHelper, which raises RuntimeError (not an
+                    # OSError) when it cannot set an ACL for an impersonated user.
+                    # With the narrower handler that escaped here and unwound the
+                    # runtime-limit Timer thread -- no notify, no grace timer, a
+                    # live child, and the timeout silently unenforced.
+                    #
                     # F6 fix: If we cannot write the cancel_info.json (disk full, permission
                     # denied, etc.), log and fall back to immediate termination. A script
                     # waiting on that file would hang forever otherwise.

src/openjd/sessions/_session.py

@@ -279,6 +279,23 @@ class Session(object):
     """OS environment variables defined by Open Job Description Environments
     """
 
+    _session_env_vars: dict[str, str]
+    """Session-defined environment variables, for the lifetime of the Session.
+
+    Deliberately separate from :attr:`_created_env_vars`, which is keyed by
+    environment and read through ``_environments_entered`` to build the child
+    *process* environment -- that view must shrink when an environment exits.
+    This one must not: RFC 0008 (``rfcs/0008-environment-wrap-actions.md``,
+    "MUST include in ``WrappedAction.Environment`` every ``openjd_env``-defined
+    variable emitted by any earlier action in the same session") makes
+    session-lifetime inclusion a requirement, so a wrap script can forward a
+    variable exported by an environment that has since exited.
+
+    The only remover is an explicit ``openjd_unset_env``. openjd-rs holds the
+    same split -- its session-lifetime ``env_vars`` beside its per-environment
+    ``created_env_vars`` -- and this mirrors it.
+    """
+
     _wrap_env_file_records: dict[EnvironmentIdentifier, list["_FileRecord"]]
     """RFC 0008: per wrap environment, the embedded-file records whose on-disk
     paths were allocated on the environment's first wrap-hook invocation and
@@ -437,6 +454,7 @@ def __init__(
         self._running_environment_identifier = None
         self._process_env = dict(os_env_vars) if os_env_vars else dict()
         self._created_env_vars = dict()
+        self._session_env_vars = dict()
         self._wrap_env_file_records = dict()
         self._retain_working_dir = retain_working_dir
         self._user = user
@@ -872,6 +890,10 @@ def enter_environment(
                 )
             env_var_changes = SimplifiedEnvironmentVariableChanges(resolved_variables)
             self._created_env_vars[identifier] = env_var_changes
+            # Session-lifetime copy for WrappedAction.Environment. openjd-rs
+            # writes declarative `variables:` into its session-lifetime
+            # `env_vars` alongside `created_env_vars` at the same point.
+            self._session_env_vars.update(resolved_variables)
         else:
             # Running the environment may define environment variable
             # mutations via its stdout. We create an empty env changes
@@ -883,7 +905,14 @@ def enter_environment(
         # Must be called _after_ we append to _environments_entered
         action_env_vars = self._evaluate_current_session_env_vars(os_env_vars)
 
-        self._materialize_path_mapping(environment.revision, action_env_vars, symtab)
+        try:
+            self._materialize_path_mapping(environment.revision, action_env_vars, symtab)
+        except RuntimeError as e:
+            self._fail_action_before_start(str(e))
+            # The identifier is still returned on a pre-start failure: the
+            # environment is on the entered stack, so the caller must be able to
+            # exit it. Same contract as the resolve-variables failure above.
+            return identifier
 
         # Note: RUNNING is set below, immediately before the runner is asked to
         # start, and never before `self._runner` exists. `cancel_action()` is a
@@ -1045,7 +1074,11 @@ def exit_environment(
         self._running_environment_identifier = identifier
 
         symtab = self._symbol_table(environment.revision)
-        self._materialize_path_mapping(environment.revision, action_env_vars, symtab)
+        try:
+            self._materialize_path_mapping(environment.revision, action_env_vars, symtab)
+        except RuntimeError as e:
+            self._fail_action_before_start(str(e))
+            return
 
         # Re-seed the owning step's name (Step.Name, RFC 0005 EXPR) and
         # re-apply the extra `let` bindings this environment was entered with
@@ -1222,7 +1255,11 @@ def run_task(
         if step_name is not None:
             symtab["Step.Name"] = step_name
         action_env_vars = self._evaluate_current_session_env_vars(os_env_vars)
-        self._materialize_path_mapping(step_script.revision, action_env_vars, symtab)
+        try:
+            self._materialize_path_mapping(step_script.revision, action_env_vars, symtab)
+        except RuntimeError as e:
+            self._fail_action_before_start(str(e))
+            return
 
         # Note: the step script's EXPR `let` bindings (RFC 0005) are evaluated
         # by the script runner, after embedded-file paths are allocated (so
@@ -1356,7 +1393,11 @@ def _run_task_without_session_env(
         if os_env_vars:
             action_env_vars.update(**os_env_vars)
 
-        self._materialize_path_mapping(step_script.revision, action_env_vars, symtab)
+        try:
+            self._materialize_path_mapping(step_script.revision, action_env_vars, symtab)
+        except RuntimeError as e:
+            self._fail_action_before_start(str(e))
+            return
 
         # Note: the step script's EXPR `let` bindings (RFC 0005) are evaluated
         # by the script runner, after embedded-file paths are allocated.
@@ -1883,31 +1924,26 @@ def _try_inject_wrapped_symbols(
         return True
 
     def _collect_session_env_list(self) -> list[str]:
-        """Collect all session-defined variables across the active
-        environment stack as ``["KEY=value", ...]`` for
+        """The session-defined variables as ``["KEY=value", ...]`` for
         ``WrappedAction.Environment``.
 
-        Session-defined variables are ``openjd_env`` stdout definitions
-        *and* entered environments' declarative ``variables:`` maps
-        (openjd-rs #277); host-inherited variables remain intentionally
-        excluded per RFC 0008.
-
-        The per-environment changes are flattened cumulatively, in
-        environment-entry order, so the list reflects the *effective*
-        state exactly like the real subprocess environment does: a later
-        set overrides an earlier value for the same name (one entry, not
-        two), and a later unset removes the name entirely. This mirrors
-        the Rust runtime's single cumulative ``env_vars`` map."""
-        effective: dict[str, Optional[str]] = {}
-        for env_id in self._environments_entered:
-            if env_id in self._created_env_vars:
-                changes = self._created_env_vars[env_id]
-                # effective_items() is insertion-ordered, holding both the
-                # environment's declarative `variables:` seed and any
-                # openjd_env sets/unsets, so the list order is deterministic.
-                for key, value in changes.effective_items().items():
-                    effective[key] = value
-        return [f"{key}={value}" for key, value in effective.items() if value is not None]
+        Session-defined variables are ``openjd_env`` stdout definitions *and*
+        entered environments' declarative ``variables:`` maps (openjd-rs #277);
+        host-inherited variables remain intentionally excluded per RFC 0008.
+
+        Read from :attr:`_session_env_vars`, which is **session-lifetime**, not
+        from ``_environments_entered``. This used to walk the entered stack and
+        flatten each environment's changes, which meant an export vanished from
+        this symbol the moment its environment exited -- a violation of RFC
+        0008's MUST that every ``openjd_env`` variable from any earlier action
+        in the session be included. The child *process* environment is the view
+        that correctly shrinks on exit; see ``_evaluate_current_session_env_vars``.
+
+        ``_session_env_vars`` is insertion-ordered and already holds the
+        effective value per name -- a later set overwrites in place, and an
+        explicit ``openjd_unset_env`` removes the name -- so this is a
+        formatting step only."""
+        return [f"{name}={value}" for name, value in self._session_env_vars.items()]
 
     def _resolve_action_timeout(self, action: Any, symtab: SymbolTable) -> Optional[int]:
         """Return the wrapped action's timeout as an int (seconds), or
@@ -2135,9 +2171,20 @@ def _materialize_path_mapping(
             ParameterValueType.BOOL.value
         )
         rules_json = json.dumps(rules_dict)
-        file_handle, filename = mkstemp(dir=self.working_directory, suffix=".json", text=True)
-        os.close(file_handle)
-        write_file_for_user(Path(filename), rules_json, self._user)
+        # An OSError here used to escape run_task/enter_environment/exit_environment
+        # with no callback and no ActionStatus, because all three call this before
+        # their runner exists. openjd-rs propagates the same failure as
+        # SessionError::WorkingDirectory and every call site maps it to
+        # fail_action_setup(), which publishes FAILED and notifies -- so translate
+        # to a RuntimeError the callers already turn into that terminal status.
+        try:
+            file_handle, filename = mkstemp(dir=self.working_directory, suffix=".json", text=True)
+            os.close(file_handle)
+            write_file_for_user(Path(filename), rules_json, self._user)
+        except Exception as err:
+            raise RuntimeError(
+                f"Could not write the path mapping rules file in {self.working_directory}: {err}"
+            ) from err
         symtab[ValueReferenceConstants_2023_09.PATH_MAPPING_RULES_FILE.value] = str(filename)
         symtab.expr_types[ValueReferenceConstants_2023_09.PATH_MAPPING_RULES_FILE.value] = (
             ParameterValueType.PATH.value
@@ -2220,6 +2267,10 @@ def _action_log_filter_callback(
             env_vars.simplify_ordered_changes(
                 changes=[EnvironmentVariableSetChange(name=value["name"], value=value["value"])]
             )
+            # Session-lifetime copy for WrappedAction.Environment; see
+            # _session_env_vars. Runs on the LoggingSubprocess IO thread, like
+            # the line above -- a plain dict assignment, so no new hazard.
+            self._session_env_vars[value["name"]] = value["value"]
             return
         elif kind == ActionMessageKind.UNSET_ENV:
             if self._running_environment_identifier is None:
@@ -2242,6 +2293,9 @@ def _action_log_filter_callback(
             assert isinstance(value, str)
             env_vars = self._created_env_vars[self._running_environment_identifier]
             env_vars.simplify_ordered_changes(changes=[EnvironmentVariableUnsetChange(name=value)])
+            # An explicit unset is the one remover from the session-lifetime
+            # map, matching openjd-rs. Environment exit is not.
+            self._session_env_vars.pop(value, None)
             return
         else:  # ActionMessageKind.SESSION_RUNTIME_LOGLEVEL
             assert isinstance(value, int)