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/

DEVELOPMENT.md

@@ -354,3 +354,47 @@ try:
 except ValueError as e:
     # Error handling...
 ```
+#### Use of `assert` in Runtime Code
+
+This package can legitimately be run under `python -O`, which strips every `assert`
+statement from the bytecode. So the choice between `assert` and `raise` decides
+whether a check exists in production at all.
+
+Use `assert` **only** for type-checker narrowing: a condition that the type system
+cannot see but that is structurally guaranteed by the caller, where losing the check
+costs nothing but a slightly less legible `AttributeError` on the following line.
+Narrowing a schema-version union immediately after the caller has already selected
+that version is the canonical example.
+
+Use an explicit `raise` for anything that is a *runtime invariant* — a condition that
+could actually be false because of a race, a partially-completed construction, an
+unexpected model shape, or a caller error. Concretely, prefer `raise` when any of
+these is true:
+
+- The check is in a **public** method or property. A stripped assert turns a clear
+  library error into `AttributeError: 'NoneType' object has no attribute ...` raised
+  from inside this package.
+- The code runs on a **background thread** — a pool worker, a `threading.Timer`
+  callback, or the stdout-forwarding path. An `AssertionError` there reaches only
+  `threading.excepthook`, so the failure is invisible and the surrounding work
+  (waiting on and reaping the subprocess, arming a grace timer) is silently skipped.
+- The assert **is** the fix. Several checks in `_runner_base.py` exist specifically to
+  hold an invariant that a bug-fix introduced. Under `-O` such an assert reverts the
+  fix without changing any test result.
+- Something **downstream would proceed on bad data** rather than stop, for example
+  materializing an embedded file from an object of an unrecognized schema.
+
+When a `raise` would be noisy in an already-defensive spot, prefer binding the value
+once and handling `None` explicitly (return early, log, skip the notification) over
+either an assert or a raise. That keeps the failure legible and the process owned.
+
+Both forms narrow types for mypy, so there is no typing cost to choosing `raise`:
+
+```py
+# Narrowing only — fine as an assert.
+assert isinstance(script, StepScript_2023_09)
+
+# A runtime invariant — must survive -O.
+if not isinstance(file, EmbeddedFileText_2023_09):
+    raise RuntimeError(f"Unsupported embedded file model '{type(file).__name__}'.")
+```

pyproject.toml

@@ -31,7 +31,13 @@ classifiers = [
   "Topic :: Software Development :: Libraries"
 ]
 dependencies = [
-  "openjd-model >= 0.10,< 0.12",
+  # 0.11.1 is the floor because it is the first release carrying the EXPR/
+  # WRAP_ACTIONS model surface this package uses: FormatString.resolve_value and
+  # .whole_field_expression, evaluate_let_bindings, CancelationMethodDeferred,
+  # SymbolTable.expr_host_rules, and `let` on StepTemplate/StepScript. 0.11.0 has
+  # only StepTemplate.let, StepScript.let and SymbolTable.expr_types, so this
+  # package fails at import against it.
+  "openjd-model >= 0.11.1,< 0.12",
   "pywin32 >= 307; platform_system == 'Windows'",
   "psutil >= 5.9,< 7.3; platform_system == 'Windows'",
 ]
@@ -149,6 +155,11 @@ addopts = [
     "--cov-report=xml:build/coverage/coverage.xml",
     "--cov-report=term-missing",
     "--numprocesses=auto",
+    # loadgroup, not the default load: it honours @pytest.mark.xdist_group, which
+    # the tests that race a real subprocess against the wall clock rely on to run
+    # serially with respect to each other (see conftest.serial_process). Behaves
+    # exactly like `load` for every test that is not in a group.
+    "--dist=loadgroup",
     "--timeout=30",
     "--log-format=%(asctime)s.%(msecs)03d %(levelname)s %(filename)20s:%(lineno)-3s %(message)s",
     "--log-date-format=%H:%M:%S"

src/openjd/sessions/_action_filter.py

@@ -15,6 +15,27 @@
 __all__ = ("ActionMessageKind", "ActionMonitoringFilter", "redact_openjd_redacted_env_requests")
 
 
+def _describe_exception(exc: BaseException) -> str:
+    """Render an exception for a log line without trusting it to be renderable.
+
+    A consumer-supplied callback may raise an exception whose ``__str__`` or
+    ``__repr__`` itself raises. Every containment site in this module puts the
+    exception into a message, so rendering it naively re-creates the exact escape
+    the containment exists to prevent -- the exception then propagates from inside
+    the ``except`` block, out of ``filter()``, and into the thread forwarding the
+    subprocess's stdout.
+    """
+    try:
+        return str(exc)
+    except Exception:
+        pass
+    try:
+        return repr(exc)
+    except Exception:
+        pass
+    return f"<unrenderable {type(exc).__name__}>"
+
+
 def redact_openjd_redacted_env_requests(command_str: str) -> str:
     """Redact sensitive information in command strings before they're processed by the regular redaction mechanism.
 
@@ -78,11 +99,27 @@ class ActionMessageKind(Enum):
 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
-envvar_set_regex_str = "^[A-Za-z_][A-Za-z0-9_]*" "=" ".*$"  # Variable name
-envvar_set_regex_json = '^(")?[A-Za-z_][A-Za-z0-9_]*' "=" ".*$"  # Variable name
+#
+# Note on the anchors: these use \Z, not $, because `$` also matches immediately
+# before a trailing newline -- so an `$`-anchored name pattern accepts "FOO\n",
+# a name no operating system can hold. \Z matches only at the true end of the
+# string.
+#
+# Honest scope: this is defence in depth, not a fix for a reachable defect. The
+# outer `filter_regex` below captures the message body with `(.+)$`, and `.` never
+# crosses a newline, so a trailing newline is already stripped before any name
+# reaches these patterns. \Z is used so that these patterns remain correct on
+# their own terms if that outer capture ever changes.
+#
+# The VALUE half deliberately stays permissive: a multi-line value delivered
+# through the JSON form (openjd_env: "FOO=a\nb") is supported, tested behaviour,
+# so the separator hygiene a value needs belongs at each serialization boundary,
+# not here.
+envvar_set_regex_str = "^[A-Za-z_][A-Za-z0-9_]*" "=" ".*\\Z"  # Variable name
+envvar_set_regex_json = '^(")?[A-Za-z_][A-Za-z0-9_]*' "=" ".*\\Z"  # Variable name
 envvar_set_matcher_str = re.compile(envvar_set_regex_str)
 envvar_set_matcher_json = re.compile(envvar_set_regex_json)
-envvar_unset_regex = "^[A-Za-z_][A-Za-z0-9_]*$"
+envvar_unset_regex = "^[A-Za-z_][A-Za-z0-9_]*\\Z"
 envvar_unset_matcher = re.compile(envvar_unset_regex)
 
 
@@ -191,7 +228,7 @@ def _redactions_enabled(self) -> bool:
             or "REDACTED_ENV_VARS" in self._revision_extensions.extensions
         )
 
-    def apply_message_redaction(self, record: logging.LogRecord):
+    def apply_message_redaction(self, record: logging.LogRecord) -> None:
         """Redact the log message if it contains any substrings which have been registered for redaction
 
         Args:
@@ -200,21 +237,36 @@ def apply_message_redaction(self, record: logging.LogRecord):
         # Check if we need to redact any sensitive values from the log message
         if (self._redacted_values or self._redacted_lines) and isinstance(record.msg, str):
 
-            # If we have args, first do string formatting, then redact
-            try:
-                record.msg = record.msg % record.args
-                record.args = ()  # Clear args since we've done the formatting
-            except Exception:
-                # If string formatting fails, fall back to just redacting the message
-                LOG.warning(
-                    "Failed to format log message for redaction. Proceeding with redaction on unformatted message."
-                )
+            # Fold `args` into `msg` and clear them UNCONDITIONALLY before
+            # scanning (R5-1 fix). The redaction below only ever inspects
+            # `record.msg`, and a downstream handler calls record.getMessage(),
+            # which re-runs `msg % args`. So any path that leaves `args`
+            # populated re-interpolates the un-scanned original into the emitted
+            # line -- a secret carried in `args` would be emitted verbatim. The
+            # `finally` is the load-bearing part: `args` must end up empty on the
+            # failure path too, not just on the success path.
+            if record.args:
+                try:
+                    record.msg = record.msg % record.args
+                except Exception:
+                    # The record's own formatting is broken, so getMessage()
+                    # would raise inside the handler and Python's logging would
+                    # dump the raw args to stderr, outside this filter's reach.
+                    # Fold them in textually instead: the record stays emittable
+                    # AND every byte of it goes through the scan below.
+                    record.msg = f"{record.msg} {record.args!r}"
+                    LOG.warning(
+                        "Failed to format log message for redaction. Redacting the message with its "
+                        "arguments appended instead.",
+                        extra=LogExtraInfo(openjd_log_content=LogContent.COMMAND_OUTPUT),
+                    )
+                finally:
+                    record.args = ()
 
             # Check if the entire message matches a line in the redacted_lines set
             if record.msg in self._redacted_lines:
                 record.msg = "*" * 8
-                record.args = ()
-                return True
+                return
 
             # Find all segments that need redaction
             segments_to_redact = []
@@ -254,7 +306,6 @@ def apply_message_redaction(self, record: logging.LogRecord):
                 for start, end in reversed(merged_segments):
                     msg_chars[start:end] = list("*" * 8)  # Always use 8 asterisks for redaction
                 record.msg = "".join(msg_chars)
-                record.args = ()
 
     def filter(self, record: logging.LogRecord) -> bool:
         """Called automatically by Python's logging subsystem when a log record
@@ -323,6 +374,27 @@ def filter(self, record: logging.LogRecord) -> bool:
                     record.msg = record.msg + f" -- ERROR: {str(e)}"
                     # There was an error. Don't suppress the message from the log.
                     return True
+                except Exception as e:
+                    # R5-2 fix: `handler` invokes the consumer-supplied callback,
+                    # which may raise anything. Anything other than ValueError
+                    # used to propagate out of filter(), into Python's logging
+                    # machinery, and into the stdout pump thread that called
+                    # logger.info() -- killing the pump, silently dropping the
+                    # rest of the subprocess's output, and leaving the child
+                    # unreaped. Same isolation as the Session's terminal
+                    # callbacks; keep the record in the log so the failure is
+                    # visible in the action's own output.
+                    #
+                    # _describe_exception, not f"{e}": rendering a hostile
+                    # exception can itself raise, which would re-open this very
+                    # escape from inside the handler for it.
+                    detail = _describe_exception(e)
+                    LOG.error(
+                        f"Open Job Description: Exception in the action message callback: {detail}",
+                        extra=LogExtraInfo(openjd_log_content=LogContent.EXCEPTION_INFO),
+                    )
+                    record.msg = record.msg + f" -- ERROR: {detail}"
+                    return True
                 return not self._suppress_filtered
 
             # Check for "almost" matching openjd_env and openjd_unset_env commands
@@ -335,13 +407,46 @@ def filter(self, record: logging.LogRecord) -> bool:
                 record.msg = record.msg + f" -- ERROR: {err_message}"
 
                 # Callback to cancel the action and mark it as FAILED
-                self._callback(ActionMessageKind.FAIL, err_message, True)
+                # R5-2 fix: isolated for the same reason as the handler call
+                # above -- this is the one consumer-callback invocation in this
+                # method that is not routed through `handler`.
+                self._invoke_callback(ActionMessageKind.FAIL, err_message, True)
                 return True
 
             return True
         finally:
-            # Always check for redaction before returning
-            self.apply_message_redaction(record)
+            # Always check for redaction before returning. Fail closed: if the
+            # redaction control itself raises, blank the message rather than
+            # letting an unscanned record through (and rather than letting the
+            # exception reach the stdout pump thread).
+            try:
+                self.apply_message_redaction(record)
+            except Exception as e:
+                record.msg = "*" * 8
+                record.args = ()
+                LOG.error(
+                    "Open Job Description: Log redaction failed; message suppressed: "
+                    f"{_describe_exception(e)}",
+                    extra=LogExtraInfo(openjd_log_content=LogContent.EXCEPTION_INFO),
+                )
+
+    def _invoke_callback(self, kind: ActionMessageKind, value: Any, cancel_and_fail: bool) -> None:
+        """Invoke the consumer-supplied callback, isolating any exception it
+        raises (R5-2).
+
+        This filter runs on the thread that is forwarding the subprocess's stdout
+        to the log, so an exception escaping here unwinds
+        ``LoggingSubprocess.run()`` before the child is waited on. Consumer bugs
+        must not cost us the output stream or process ownership.
+        """
+        try:
+            self._callback(kind, value, cancel_and_fail)
+        except Exception as e:
+            LOG.error(
+                "Open Job Description: Exception in the action message callback: "
+                f"{_describe_exception(e)}",
+                extra=LogExtraInfo(openjd_log_content=LogContent.EXCEPTION_INFO),
+            )
 
     def _handle_progress(self, message: str) -> None:
         """Local handling of Progress messages. Processes the message and then
@@ -404,33 +509,28 @@ def _parse_env_variable(self, message: str) -> tuple[str, str, bool, int, Option
         if equals_position == -1:
             return "", "", False, -1, None
 
-        # Check if the message is valid
-        is_valid = envvar_set_matcher_str.match(message) or envvar_set_matcher_json.match(message)
-
-        if not is_valid:
+        # Check if the message is valid. Bound once: the plain-string match decides
+        # both validity and which parse branch to take below.
+        str_match = envvar_set_matcher_str.match(message)
+        if not (str_match or envvar_set_matcher_json.match(message)):
             return "", "", False, equals_position, None
 
         # Parse the variable name and value
-        try:
-            original_value = None
-            if envvar_set_matcher_str.match(message):
-                name, _, value = message.partition("=")
-            else:
-                # Handle JSON format
-                try:
-                    # Store the original value before JSON parsing
-                    original_value = message[equals_position + 1 :]
-                    message_json_str = json.loads(message)
-                    name, _, value = message_json_str.partition("=")
-                except json.JSONDecodeError as e:
-                    raise ValueError(
-                        f"Unterminated string starting at: line {e.lineno} column {e.colno} (char {e.pos})"
-                    )
-            return name, value, True, equals_position, original_value
-        except json.JSONDecodeError as e:
-            raise ValueError(
-                f"Unterminated string starting at: line {e.lineno} column {e.colno} (char {e.pos})"
-            )
+        original_value = None
+        if str_match:
+            name, _, value = message.partition("=")
+        else:
+            # Handle JSON format
+            try:
+                # Store the original value before JSON parsing
+                original_value = message[equals_position + 1 :]
+                message_json_str = json.loads(message)
+                name, _, value = message_json_str.partition("=")
+            except json.JSONDecodeError as e:
+                raise ValueError(
+                    f"Unterminated string starting at: line {e.lineno} column {e.colno} (char {e.pos})"
+                )
+        return name, value, True, equals_position, original_value
 
     def _handle_env_error(self, error_message: str, is_redacted: bool = False) -> None:
         """Handle errors in environment variable processing.