From a6af8a372163c1da3aae77c2bfa0b36e043031e3 Mon Sep 17 00:00:00 2001 From: David Leong <116610336+leongdl@users.noreply.github.com> Date: Sun, 26 Jul 2026 17:56:43 -0700 Subject: [PATCH 01/13] fix: Anchor the openjd_env near-miss regex (openjd-rs parity) An unanchored `^(openjd_env|openjd_redacted_env|openjd_unset_env)` treats any line merely starting with one of those tokens as a malformed macro, and the filter then fails the action with cancel_action_mark_failed=True. So a script logging `openjd_environment: ready` or `openjd_env_setup: done` kills its own task, with no adversary involved. openjd-rs does not have this bug: is_malformed_env_command (action_filter.rs) anchors each of the same three tokens on ':', a space, or end-of-string, and its comment names exactly this false positive. In Rust the line falls through parse_directive as an unrecognised kind and passes through as ordinary log output. Adds the equivalent `(?::|\s|\Z)` tail. Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com> (cherry picked from commit ede6cad2a9bac143eb34215e2417817f8760c020) --- src/openjd/sessions/_action_filter.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/src/openjd/sessions/_action_filter.py b/src/openjd/sessions/_action_filter.py index 25d43aa..4c49e37 100644 --- a/src/openjd/sessions/_action_filter.py +++ b/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 From f34e219f0cc1c4ffc794d629d7d880d8ac69d42b Mon Sep 17 00:00:00 2001 From: David Leong <116610336+leongdl@users.noreply.github.com> Date: Sun, 26 Jul 2026 17:56:43 -0700 Subject: [PATCH 02/13] fix: Do not gate terminate on leader liveness (openjd-rs parity) terminate() guarded on `proc.poll() is None`, so once the wrapper leader had exited it did nothing -- even though _sudo_child_process_group_id was recorded and children could still be alive. killpg on a process group still reaches surviving members after the leader is reaped, so the guard made the call a no-op exactly when the recorded group was the only remaining handle on the survivors, orphaning them past terminal publication. A wrap environment is precisely the shape that produces an exited leader with a live workload, so this PR increases exposure to it. openjd-rs never gates: every cancel/timeout/reap-failure path calls send_terminate(pid) -> killpg unconditionally, and its is_process_alive() probe is #[allow(dead_code)] for this reason. Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com> (cherry picked from commit c22621bf5e2a0256a6a0dfa0ea68e65e634ae47e) --- src/openjd/sessions/_subprocess.py | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/src/openjd/sessions/_subprocess.py b/src/openjd/sessions/_subprocess.py index a48bd02..263910b 100644 --- a/src/openjd/sessions/_subprocess.py +++ b/src/openjd/sessions/_subprocess.py @@ -351,7 +351,18 @@ def terminate(self) -> None: """ # Review22-F4 fix: Bind _process once. See notify() for rationale. proc = self._process - if proc is not None and proc.poll() is None: + # Deliberately NOT gated on `proc.poll() is None`. The signal target is a + # process *group*, and killpg still reaches surviving members after the + # group leader has exited and been reaped -- which is precisely the shape + # a wrap environment produces (a wrapper that execs and returns while its + # workload lives on). Gating on leader liveness made this a no-op exactly + # when the recorded _sudo_child_process_group_id was the only handle left + # on the survivors, orphaning them past terminal publication. + # + # openjd-rs never gates: every cancel/timeout/reap-failure path calls + # send_terminate(pid) -> killpg unconditionally (subprocess.rs), and its + # is_process_alive() probe is #[allow(dead_code)] for this reason. + if proc is not None: self._terminate_process(proc) def _terminate_process(self, process: Popen) -> None: From d72674220e3c7939a077f49a4a18bccc76598242 Mon Sep 17 00:00:00 2001 From: David Leong <116610336+leongdl@users.noreply.github.com> Date: Sun, 26 Jul 2026 17:56:44 -0700 Subject: [PATCH 03/13] fix: cancel_info.json handler caught the wrong exception type TTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTuard, and all four call sites run before their runner exists -- so an OSError escaped run_task / enter_environment / exit_environment with no callback and no ActionStatus. Now translated to RuntimeError and routed to _fail_action_before_start at all four sites, which publishes FAILED/READY_ENDING and notifies. enter_environment still returns its identifier on this path, matching its existing resolve-variables failure contract, because the environment is already on the entered stack and the caller must be able to exit it. openjd-rs supplies the contract: std::fs::write maps to SessionError::WorkingDirectory and every call site maps that to fail_action_setup(), which sets Failed + notifies rather than propagating. Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com> (cherry picked from commit c5d29273d4044eb5dae67976adf3dd00ae25dc03) --- src/openjd/sessions/_runner_base.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/src/openjd/sessions/_runner_base.py b/src/openjd/sessions/_runner_base.py index 1d8e917..91a0d31 100644 --- a/src/openjd/sessions/_runner_base.py +++ b/src/openjd/sessions/_runner_base.py @@ -1251,7 +1251,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. From 20f607cb81c0c17d81a4aea7727d2bac60482dba Mon Sep 17 00:00:00 2001 From: David Leong <116610336+leongdl@users.noreply.github.com> Date: Sun, 26 Jul 2026 17:56:57 -0700 Subject: [PATCH 04/13] fix: Give _materialize_path_mapping a failure path (openjd-rs parity) mkstemp + write_file_for_user had no guard, and all four call sites run before their runner exists -- so an OSError escaped run_task / enter_environment / exit_environment with no callback and no ActionStatus. Now translated to RuntimeError and routed to _fail_action_before_start at all four sites, which publishes FAILED/READY_ENDING and notifies. enter_environment still returns its identifier on this path, matching its existing resolve-variables failure contract, because the environment is already on the entered stack and the caller must be able to exit it. openjd-rs supplies the contract: std::fs::write maps to SessionError::WorkingDirectory and every call site maps that to fail_action_setup(), which sets Failed + notifies rather than propagating. Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com> (cherry picked from commit d4748ab457c2aa1626409511cee46876b4304476) --- src/openjd/sessions/_session.py | 44 +++++++++++++++++++++++++++------ 1 file changed, 37 insertions(+), 7 deletions(-) diff --git a/src/openjd/sessions/_session.py b/src/openjd/sessions/_session.py index 73063e1..444b4e9 100644 --- a/src/openjd/sessions/_session.py +++ b/src/openjd/sessions/_session.py @@ -883,7 +883,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 +1052,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 +1233,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 +1371,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. @@ -2135,9 +2154,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 From f874c7c935f2d5747d8ccd6d931751e56e3efc09 Mon Sep 17 00:00:00 2001 From: David Leong <116610336+leongdl@users.noreply.github.com> Date: Sun, 26 Jul 2026 19:19:54 -0700 Subject: [PATCH 05/13] fix: WrappedAction.Environment must be session-lifetime (RFC 0008 MUST) rfcs/0008-environment-wrap-actions.md:404-412 requires that runtimes "MUST include in `WrappedAction.Environment` every `openjd_env`-defined variable emitted by any earlier action in the same session -- regardless of whether that action ran normally or via a wrap hook -- and every entry of each entered environment's declarative `variables:` map." _collect_session_env_list() iterated `_environments_entered`, which exit_environment pops, so an earlier environment's export vanished from the symbol the moment that environment exited. A wrap script forwarding WrappedAction.Environment into a container therefore silently dropped variables the spec guarantees it. Reproduced first: the new test fails against the old code and passes against the new. Fixed by mirroring openjd-rs, which holds two maps deliberately: _session_env_vars session-lifetime, feeds WrappedAction.Environment. Written by declarative `variables:` and by openjd_env; the only remover is an explicit openjd_unset_env, not an environment exit. (Rust: `env_vars`.) _created_env_vars per environment, read through _environments_entered to build the child *process* environment -- this view must shrink on exit, and still does. (Rust: `created_env_vars`.) Following Rust also settles the MUST's ambiguous second clause: Rust writes declarative `variables:` into its session-lifetime map at the same point it writes them per-environment, so both halves are session-scoped here too. Tests, and two pre-existing tests corrected: Five behaviours are now pinned, one per fix on this branch, and all six mutants are caught (revert the fix, a named test fails): - an exited environment's export is still listed, with an explicit unset as the negative control so retention does not become "nothing is removed" - a longer token like `openjd_environment: ready` is ordinary output, with a genuine near-miss still failing the action as the control - terminate() signals after the leader exits, with the released-handle case as the control - a cancel_info write failure is contained for both OSError and RuntimeError - a rules-file write failure publishes FAILED instead of raising test_wrap_actions.py::test_later_set_overrides_earlier_value_without_duplicate and test_wrap_task_run.py::test_injects_wrapped_environment_as_key_value_list both fabricated `_environments_entered` entries and assigned `_created_env_vars` directly, bypassing the production writers entirely -- which is precisely why they kept passing while the MUST was being violated. Both now drive the real path. A test that never exercises the writers cannot notice the writers being wrong. 878 passed / 39 skipped / 16 xfailed (861 before, +17 new). ruff, black and mypy clean on native and --platform win32. Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com> (cherry picked from commit b7f0986e9ec0bac45754c3c2626c2898c5a5d7dd) --- src/openjd/sessions/_session.py | 70 ++++++---- .../test_action_filter_hardening.py | 59 ++++++++ test/openjd/sessions_v0/test_runner_base.py | 39 ++++++ .../sessions_v0/test_session_let_bindings.py | 53 ++++++++ .../sessions_v0/test_subprocess_reaping.py | 45 +++++++ test/openjd/sessions_v0/test_wrap_actions.py | 127 +++++++++++++----- test/openjd/sessions_v0/test_wrap_task_run.py | 22 ++- 7 files changed, 343 insertions(+), 72 deletions(-) diff --git a/src/openjd/sessions/_session.py b/src/openjd/sessions/_session.py index 444b4e9..e950393 100644 --- a/src/openjd/sessions/_session.py +++ b/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 @@ -1902,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 @@ -2250,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: @@ -2272,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) diff --git a/test/openjd/sessions_v0/test_action_filter_hardening.py b/test/openjd/sessions_v0/test_action_filter_hardening.py index 6446dda..7a1b002 100644 --- a/test/openjd/sessions_v0/test_action_filter_hardening.py +++ b/test/openjd/sessions_v0/test_action_filter_hardening.py @@ -363,3 +363,62 @@ def callback(kind: ActionMessageKind, value: Any, fail: bool) -> None: # THEN assert "a perfectly ordinary boom" in record.getMessage() + + +class TestNearMissEnvMacroIsAnchored: + """The near-miss detector must not fire on a *longer* token. + + Its regex had no trailing boundary, so any line merely starting with + `openjd_env`, `openjd_redacted_env` or `openjd_unset_env` was reported as a + malformed macro -- and that path fails the running action. A script logging + `openjd_environment: ready` therefore killed its own task, with no adversary + involved. + + openjd-rs anchors the same three tokens on ':', a space, or end-of-string + (`action_filter.rs` `is_malformed_env_command`), and its comment names this + exact false positive. + """ + + @pytest.mark.parametrize( + "line", + [ + "openjd_environment: ready", + "openjd_env_setup: done", + "openjd_envvar=1", + "openjd_unset_environment_now", + "openjd_redacted_envelope: opened", + ], + ) + def test_a_longer_token_is_ordinary_output(self, line: str) -> None: + # GIVEN a line whose prefix only *looks* like a macro + callback = MagicMock() + f = ActionMonitoringFilter(session_id="foo", callback=callback) + + # WHEN + f.filter(_make_record(line)) + + # THEN: nothing is reported -- it is just a log line + assert callback.call_count == 0 + + @pytest.mark.parametrize( + "line", + [ + "openjd_env FOO=bar", + "openjd_env:FOO=bar", + "OpenJD_Env: FOO=bar", + "openjd_unset_env", + " openjd_env: FOO", + ], + ) + def test_a_genuine_near_miss_still_fails_the_action(self, line: str) -> None: + """Negative control: anchoring must not disarm the detector.""" + # GIVEN a genuinely malformed macro + callback = MagicMock() + f = ActionMonitoringFilter(session_id="foo", callback=callback) + + # WHEN + f.filter(_make_record(line)) + + # THEN: it is still reported as a failure + fails = [c for c in callback.call_args_list if c[0][0] == ActionMessageKind.FAIL] + assert len(fails) == 1, callback.call_args_list diff --git a/test/openjd/sessions_v0/test_runner_base.py b/test/openjd/sessions_v0/test_runner_base.py index 4a993b4..3f8e612 100644 --- a/test/openjd/sessions_v0/test_runner_base.py +++ b/test/openjd/sessions_v0/test_runner_base.py @@ -1233,3 +1233,42 @@ def test_materialize_files_fails( assert not os.path.exists(dest_dir / "test_materialize_files.txt") messages = collect_queue_messages(message_queue) assert any(m.startswith("openjd_fail") for m in messages) + + +class TestCancelInfoWriteFailureIsContained: + """A failed `cancel_info.json` write must not unwind the cancel. + + The guard named `OSError`, but `write_file_for_user` reaches + `WindowsPermissionHelper`, which raises `RuntimeError` when it cannot set an + ACL for an impersonated user -- and `RuntimeError` is not an `OSError`. The + exception therefore escaped and unwound the runtime-limit `Timer` thread: no + notify, no grace timer, a live child, and the timeout silently unenforced. + + openjd-rs cannot hit this: `write_cancel_info` discards the write result and + schedules notify plus the delayed terminate regardless. + """ + + @pytest.mark.parametrize( + "error", + [ + OSError(28, "No space left on device"), + RuntimeError("Could not set the ACL for the session user"), + ], + ids=["oserror", "runtimeerror"], + ) + def test_the_action_still_gets_a_terminate(self, error: Exception, tmp_path: Path) -> None: + # GIVEN a notify-then-terminate cancel whose cancel_info write fails + logger = build_logger(QueueHandler(SimpleQueue())) + with NotifyingRunner(logger=logger, session_working_directory=tmp_path) as runner: + live_child = MagicMock() + live_child.is_running = True + runner._process = live_child + + # WHEN + with patch("openjd.sessions._runner_base.write_file_for_user", side_effect=error): + runner.cancel() + + # THEN: no exception escaped, and we fell back to terminating rather + # than leaving the child running with no scheduled kill. + assert live_child.terminate.called + assert not live_child.notify.called diff --git a/test/openjd/sessions_v0/test_session_let_bindings.py b/test/openjd/sessions_v0/test_session_let_bindings.py index 47ba8ad..2a1dffb 100644 --- a/test/openjd/sessions_v0/test_session_let_bindings.py +++ b/test/openjd/sessions_v0/test_session_let_bindings.py @@ -19,6 +19,7 @@ import sys import time import uuid +from unittest.mock import patch as mock_patch from typing import Any import pytest @@ -521,3 +522,55 @@ def test_task_file_supports_path_property_access( assert status is not None assert status.state == ActionState.SUCCESS assert any("PARENT=" in m and "embedded_files" in m for m in caplog.messages) + + +# --------------------------------------------------------------------------- +# A failure writing the path-mapping rules file must FAIL the action via the +# callback path, never raise out of the public Session API. +# --------------------------------------------------------------------------- +class TestPathMappingRulesFileWriteFailure: + """`_materialize_path_mapping` had no failure path. + + All four call sites run before their runner exists, so an OSError from + `mkstemp` or `write_file_for_user` escaped `enter_environment` / `run_task` / + `exit_environment` with no callback and no ActionStatus -- the caller was + left with no terminal state to observe. + + openjd-rs supplies the contract: `std::fs::write` maps to + `SessionError::WorkingDirectory` and every call site maps that to + `fail_action_setup()`, which publishes Failed and notifies. + """ + + def test_a_write_failure_fails_the_action_instead_of_raising(self) -> None: + # GIVEN: the rules-file write cannot succeed + callback_events: list[ActionStatus] = [] + + def callback(session_id: str, status: ActionStatus) -> None: + callback_events.append(status) + + context = ModelParsingContext_2023_09(supported_extensions=["EXPR"]) + environment = Environment_2023_09.model_validate( + { + "name": "AnyEnv", + "script": {"actions": {"onEnter": {"command": "echo", "args": ["entered"]}}}, + }, + context=context, + ) + with Session( + session_id=uuid.uuid4().hex, job_parameter_values={}, callback=callback + ) as session: + with mock_patch( + "openjd.sessions._session.write_file_for_user", + side_effect=OSError(13, "Permission denied"), + ): + # WHEN: entering must not raise + identifier = session.enter_environment(environment=environment) + + # THEN: a terminal FAILED status naming the working directory, and a + # session the caller can still unwind. + assert identifier is not None + status = session.action_status + assert status is not None + assert status.state == ActionState.FAILED + assert status.fail_message is not None + assert "path mapping rules" in status.fail_message diff --git a/test/openjd/sessions_v0/test_subprocess_reaping.py b/test/openjd/sessions_v0/test_subprocess_reaping.py index 703fb38..8470d93 100644 --- a/test/openjd/sessions_v0/test_subprocess_reaping.py +++ b/test/openjd/sessions_v0/test_subprocess_reaping.py @@ -338,3 +338,48 @@ def test_a_child_that_survives_termination_is_reported_not_hung(self, python_exe finally: popen.kill() popen.wait() + + +class TestTerminateDoesNotDependOnLeaderLiveness: + """`terminate()` must still signal after the group leader has exited. + + The signal target is a process *group*, and killpg still reaches surviving + members once the leader has exited and been reaped. Gating the call on + `proc.poll() is None` made it a no-op in exactly the case where the recorded + group was the only remaining handle on those survivors -- which is the shape + a wrap environment produces, a wrapper that execs and returns while its + workload lives on. The children then outlived terminal publication. + + openjd-rs never gates: every cancel/timeout/reap-failure path calls + `send_terminate(pid)` -> killpg unconditionally, and its `is_process_alive()` + probe is `#[allow(dead_code)]` for this reason. + """ + + def test_signal_is_sent_even_when_poll_reports_the_leader_gone(self) -> None: + # GIVEN a subprocess whose leader has already exited + subproc = LoggingSubprocess(logger=MagicMock(), args=["unused"]) + departed = MagicMock() + departed.poll.return_value = 0 # reaped; a live workload may remain + subproc._process = departed + + # WHEN + with patch.object(subproc, "_terminate_process") as terminate_process: + subproc.terminate() + + # THEN: the group is still signalled + terminate_process.assert_called_once_with(departed) + + def test_nothing_is_signalled_once_the_process_is_released(self) -> None: + """Negative control: `run()`'s finally clears `_process` after reaping, + and terminate() must stay a no-op from then on -- otherwise the pid + could be recycled and an unrelated group signalled.""" + # GIVEN a subprocess that has released its handle + subproc = LoggingSubprocess(logger=MagicMock(), args=["unused"]) + subproc._process = None + + # WHEN + with patch.object(subproc, "_terminate_process") as terminate_process: + subproc.terminate() + + # THEN + terminate_process.assert_not_called() diff --git a/test/openjd/sessions_v0/test_wrap_actions.py b/test/openjd/sessions_v0/test_wrap_actions.py index f3fd5b5..121948e 100644 --- a/test/openjd/sessions_v0/test_wrap_actions.py +++ b/test/openjd/sessions_v0/test_wrap_actions.py @@ -600,43 +600,31 @@ def test_variables_map_included_in_wrapped_environment(self) -> None: assert any(entry == "DECLARED_VAR=from-variables-map" for entry in env_list), env_list def test_later_set_overrides_earlier_value_without_duplicate(self) -> None: - # Cumulative flattening: when a later environment's openjd_env sets - # a name an earlier environment already set, the list carries only - # the effective (later) value — matching the real subprocess env - # and the Rust runtime's single cumulative map. - from openjd.sessions._session import ( - EnvironmentVariableSetChange, - EnvironmentVariableUnsetChange, - SimplifiedEnvironmentVariableChanges, - ) - + """A later set of the same name yields one entry, the effective value. + + Rewritten to drive real environments. The previous version appended + fabricated identifiers to ``_environments_entered`` and assigned + ``_created_env_vars`` directly, which bypassed the production write path + entirely -- so it kept passing while ``WrappedAction.Environment`` + violated RFC 0008's session-lifetime MUST. A test that never exercises + the writers cannot notice the writers being wrong. + """ + # GIVEN: two environments, the second overriding the first's export with Session(session_id=uuid.uuid4().hex, job_parameter_values={}) as session: - scenarios: list[ - tuple[str, list[EnvironmentVariableSetChange | EnvironmentVariableUnsetChange]] - ] = [ - ("env-a", [EnvironmentVariableSetChange(name="FOO", value="1")]), - ("env-b", [EnvironmentVariableSetChange(name="FOO", value="2")]), - ] - for env_id, changes in scenarios: - session._environments_entered.append(env_id) - tracked = SimplifiedEnvironmentVariableChanges({}) - tracked.simplify_ordered_changes(changes) - session._created_env_vars[env_id] = tracked - - assert session._collect_session_env_list() == ["FOO=2"] - - # And a later unset removes the name entirely. - session._environments_entered.append("env-c") - tracked = SimplifiedEnvironmentVariableChanges({}) - tracked.simplify_ordered_changes([EnvironmentVariableUnsetChange(name="FOO")]) - session._created_env_vars["env-c"] = tracked - - assert session._collect_session_env_list() == [] + for value in ("1", "2"): + session.enter_environment( + environment=_env( + f"Setter{value}", + onEnter=_action("sh", "-c", f"echo 'openjd_env: FOO={value}'"), + onExit=_NOOP, + ) + ) + _run_until_ready(session) - # Clear the fake stack so Session.cleanup() doesn't try to - # unwind environments that were never really entered. - session._environments_entered.clear() - session._created_env_vars.clear() + # THEN: one entry, carrying the later value + assert [e for e in session._collect_session_env_list() if e.startswith("FOO=")] == [ + "FOO=2" + ] def test_exit_path_includes_exiting_envs_openjd_env_vars(self) -> None: # On the onWrapEnvExit path, WrappedAction.Environment must include @@ -689,6 +677,75 @@ def test_openjd_env_vars_included_in_wrapped_environment(self) -> None: env_list = session._collect_session_env_list() assert "EMITTED_VAR=from-openjd-env" in env_list + def test_an_exited_environments_export_is_still_listed(self) -> None: + """RFC 0008 makes session-lifetime inclusion a MUST. + + `rfcs/0008-environment-wrap-actions.md:404-412`: runtimes "MUST include + in `WrappedAction.Environment` every `openjd_env`-defined variable + emitted by any earlier action in the same session -- regardless of + whether that action ran normally or via a wrap hook". + + `_collect_session_env_list()` used to iterate `_environments_entered`, + which `exit_environment` pops, so an earlier environment's export + disappeared the moment that environment exited. openjd-rs keeps a + session-lifetime `env_vars` map for exactly this symbol, held separate + from the per-entered view that feeds the child process environment. + """ + # GIVEN: an environment that exports a variable and is then exited + with Session(session_id=uuid.uuid4().hex, job_parameter_values={}) as session: + early = _env( + "Early", + onEnter=_action("sh", "-c", "echo 'openjd_env: EARLY_VAR=early-value'"), + onExit=_NOOP, + ) + early_id = session.enter_environment(environment=early) + _run_until_ready(session) + assert "EARLY_VAR=early-value" in session._collect_session_env_list() + + # WHEN + session.exit_environment(identifier=early_id, keep_session_running=True) + _run_until_ready(session) + + # THEN: the export outlives the environment that emitted it + assert "EARLY_VAR=early-value" in session._collect_session_env_list() + + def test_an_explicit_unset_still_removes_the_name(self) -> None: + """Negative control for the test above. + + Session-lifetime retention must not become "nothing is ever removed". + An explicit `openjd_unset_env` is the one remover -- the same contract + openjd-rs has, where `env_vars` is only ever erased by an unset. + """ + # GIVEN: a variable exported and then explicitly unset by a later env + with Session(session_id=uuid.uuid4().hex, job_parameter_values={}) as session: + setter = _env( + "Setter", + onEnter=_action("sh", "-c", "echo 'openjd_env: DOOMED=value'"), + onExit=_NOOP, + ) + setter_id = session.enter_environment(environment=setter) + _run_until_ready(session) + assert "DOOMED=value" in session._collect_session_env_list() + + unsetter = _env( + "Unsetter", + onEnter=_action("sh", "-c", "echo 'openjd_unset_env: DOOMED'"), + onExit=_NOOP, + ) + unsetter_id = session.enter_environment(environment=unsetter) + _run_until_ready(session) + + # WHEN: both environments exit + session.exit_environment(identifier=unsetter_id, keep_session_running=True) + _run_until_ready(session) + session.exit_environment(identifier=setter_id, keep_session_running=True) + _run_until_ready(session) + + # THEN: the unset wins, and it too outlives its environment + assert not any( + entry.startswith("DOOMED=") for entry in session._collect_session_env_list() + ) + class TestRunWrapHookGuard: def test_unknown_hook_name_raises(self) -> None: diff --git a/test/openjd/sessions_v0/test_wrap_task_run.py b/test/openjd/sessions_v0/test_wrap_task_run.py index 53446eb..1e7803f 100644 --- a/test/openjd/sessions_v0/test_wrap_task_run.py +++ b/test/openjd/sessions_v0/test_wrap_task_run.py @@ -150,22 +150,16 @@ def test_injects_wrapped_environment_as_key_value_list(self) -> None: # simplify_ordered_changes) AND the declarative variables:-map seed. session = Session(session_id=uuid.uuid4().hex, job_parameter_values={}) try: - from openjd.sessions._session import ( - EnvironmentVariableSetChange, - SimplifiedEnvironmentVariableChanges, + # Seeded through the session-lifetime map that now backs this symbol. + # Previously this fabricated an entry in `_environments_entered` plus + # a `_created_env_vars` record, which is the per-entered view that + # feeds the child *process* environment -- not this symbol. That + # bypassed the production writers, which is why it kept passing while + # WrappedAction.Environment violated RFC 0008's session-lifetime MUST. + session._session_env_vars.update( + {"DECLARED": "from-variables-map", "FOO": "bar", "BAZ": "qux"} ) - fake_id = "env-1" - session._environments_entered.append(fake_id) - changes = SimplifiedEnvironmentVariableChanges({"DECLARED": "from-variables-map"}) - changes.simplify_ordered_changes( - [ - EnvironmentVariableSetChange(name="FOO", value="bar"), - EnvironmentVariableSetChange(name="BAZ", value="qux"), - ] - ) - session._created_env_vars[fake_id] = changes - symtab = SymbolTable() script = _step_script("echo", ["hi"]) session._inject_wrapped_task_symbols(symtab, script, "Step1", inner_symtab=symtab) From 9389ae98f46d8279fa89ed651023da14df17b1d5 Mon Sep 17 00:00:00 2001 From: David Leong <116610336+leongdl@users.noreply.github.com> Date: Tue, 28 Jul 2026 16:14:01 -0700 Subject: [PATCH 06/13] fix: run() must own its child from creation, not after logging `LoggingSubprocess.run()` logged "Command started as pid" and "Output:", and did the POSIX process-group lookup, *before* entering the `try` whose `finally` reaps the child. Anything raising in that window escaped `run()` having never waited on the child: a live process, no exit code recorded, and the runner publishing a terminal state for it anyway. The reachable trigger is a `logging.Filter` that raises. Filters propagate out of `Logger.handle`, unlike handlers, whose exceptions are routed to `handleError` -- and installing a filter is a supported use of this library; `Session` installs its own. Reproduced before the fix: the child stayed alive with `exit_code is None`, while a filter raising from inside the pump (already covered by the `finally`) reaped correctly. The ownership `try` now begins immediately after successful process creation, so every exit path from that point goes through `_reap`. The body binds `proc = self._process` once instead of re-reading the attribute, matching the bind-once convention used by `notify()`/`terminate()`. `_reap`'s own logging is now best-effort. It runs inside `run()`'s `finally`, so anything escaping it replaces the exception already propagating and hides the real cause. That containment was in place for the terminate and the wait but not for the three `logger.error` calls between them -- and a raising logging stack is exactly what routes execution into `_reap`, which makes those calls the likeliest thing to fail there, not the least. Reported by mwiebe on #333: https://github.com/OpenJobDescription/openjd-sessions-for-python/pull/333#discussion_r3661659053 Two honest notes on the scope of the fix: - Reordering `_reap` to terminate before logging was tried and dropped. With best-effort logging in place the swallow already lets the terminate run, so the ordering changed nothing an `Exception` could observe; it only guarded a `BaseException` from the logging stack, which cannot reach the pool worker thread `run()` executes on. Mutation testing proved it inert, so it is not shipped rather than shipped behind a contrived test. - When the process-group lookup is *itself* what fails, the boundary buys the reap attempt and the handle release but not the kill: with no group recorded, `_posix_signal_subprocess` declines to signal rather than target a possibly recycled pid (the R5-9 convention). That is a pre-existing limitation of the signalling path, not something this change introduces, and the test for that path asserts only what actually happens. Recorded as a follow-up. Tests: 6 added. Every one mutation-checked -- 5 mutants, 0 survivors: the startup logging moved back outside the boundary, the process-group lookup moved back outside it, and each of the three reap log sites bypassing best-effort. The pre-existing 13 tests in the file pass unmodified. Full suite 884 passed; ruff, black, and mypy clean on native and --platform win32. Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com> (cherry picked from commit acee6551aa8bdbb2485587b9c550ad22dfec680d) --- src/openjd/sessions/_subprocess.py | 161 ++++++---- .../sessions_v0/test_subprocess_reaping.py | 285 +++++++++++++++++- 2 files changed, 382 insertions(+), 64 deletions(-) diff --git a/src/openjd/sessions/_subprocess.py b/src/openjd/sessions/_subprocess.py index 263910b..8ec47d0 100644 --- a/src/openjd/sessions/_subprocess.py +++ b/src/openjd/sessions/_subprocess.py @@ -187,56 +187,71 @@ def run(self) -> None: self._callback() return - self._pid = self._process.pid - - # Would use is_posix(), but it doesn't short-circuit mypy which then complains - # about os.getpgid not being a valid attribute. - if not sys.platform == "win32": - # A trivial command can exit before we get here. Looking up the - # process group of an already-reaped child raises ProcessLookupError - # (ESRCH), which must not fail the action: the child ran, and its - # exit code is still collected below. - try: - if not self._user or self._user.is_process_user(): - self._sudo_child_process_group_id = os.getpgid(self._process.pid) - else: - self._sudo_child_process_group_id = find_sudo_child_process_group_id( - logger=self._logger, - sudo_process=self._process, + # Bind the Popen once, and enter the ownership `try` immediately: from here + # on this method owns a live child, so every exit path -- including an + # exceptional one -- has to go through the `finally` that reaps it. + # + # The process-group lookup and the two startup log lines used to sit + # *before* the `try`, outside that guarantee. A `logging.Filter` that + # raises (filters propagate out of `Logger.handle`, unlike handlers) then + # escaped `run()` having never waited on the child: a live process with no + # exit code recorded, while the runner published a terminal state for it. + # Installing such a filter is a supported use of this library -- `Session` + # installs its own. Reproduced before this change; see + # TestStartupLoggingDoesNotAbandonTheChild. + proc = self._process + try: + self._pid = proc.pid + + # Would use is_posix(), but it doesn't short-circuit mypy which then complains + # about os.getpgid not being a valid attribute. + if not sys.platform == "win32": + # A trivial command can exit before we get here. Looking up the + # process group of an already-reaped child raises ProcessLookupError + # (ESRCH), which must not fail the action: the child ran, and its + # exit code is still collected below. + try: + if not self._user or self._user.is_process_user(): + self._sudo_child_process_group_id = os.getpgid(proc.pid) + else: + self._sudo_child_process_group_id = find_sudo_child_process_group_id( + logger=self._logger, + sudo_process=proc, + ) + except ProcessLookupError: + # R5-9 fix: leave this as None -- "no process group known" -- + # rather than substituting the pid. The lookup failed *because* + # the process is gone, so its pid is dead: signalling it is at + # best a no-op, and after pid recycling on a busy host `killpg` + # would target an unrelated group. In the sudo branch the pid is + # not even the right identifier, since it belongs to `sudo` and + # not to the workload's process group. + # + # None is the established "unknown" value on this field -- + # find_sudo_child_process_group_id returns it for the same reason + # (_linux/_sudo.py), and _posix_signal_subprocess already retries + # the lookup and then declines to signal when it is still unset. + self._sudo_child_process_group_id = None + self._logger.debug( + f"Process {proc.pid} exited before its process group could be " + "determined; no process group will be signalled for it.", + extra=LogExtraInfo(openjd_log_content=LogContent.PROCESS_CONTROL), ) - except ProcessLookupError: - # R5-9 fix: leave this as None -- "no process group known" -- - # rather than substituting the pid. The lookup failed *because* - # the process is gone, so its pid is dead: signalling it is at - # best a no-op, and after pid recycling on a busy host `killpg` - # would target an unrelated group. In the sudo branch the pid is - # not even the right identifier, since it belongs to `sudo` and - # not to the workload's process group. - # - # None is the established "unknown" value on this field -- - # find_sudo_child_process_group_id returns it for the same reason - # (_linux/_sudo.py), and _posix_signal_subprocess already retries - # the lookup and then declines to signal when it is still unset. - self._sudo_child_process_group_id = None - self._logger.debug( - f"Process {self._process.pid} exited before its process group could be " - "determined; no process group will be signalled for it.", - extra=LogExtraInfo(openjd_log_content=LogContent.PROCESS_CONTROL), - ) - self._logger.info( - f"Command started as pid: {self._process.pid}", - extra=LogExtraInfo(openjd_log_content=LogContent.PROCESS_CONTROL), - ) - self._logger.info( - "Output:", - extra=LogExtraInfo(openjd_log_content=LogContent.BANNER | LogContent.COMMAND_OUTPUT), - ) + self._logger.info( + f"Command started as pid: {proc.pid}", + extra=LogExtraInfo(openjd_log_content=LogContent.PROCESS_CONTROL), + ) + self._logger.info( + "Output:", + extra=LogExtraInfo( + openjd_log_content=LogContent.BANNER | LogContent.COMMAND_OUTPUT + ), + ) - try: self._log_subproc_stdout() # Blocking - self._process.wait() - self._returncode = self._process.returncode + proc.wait() + self._returncode = proc.returncode self._log_returncode() if self._callback: self._callback() @@ -245,7 +260,6 @@ def run(self) -> None: # PopenWindowsAsUser and there's stuff to deallocate. `proc` itself is # a function local and dies with the frame; a `del proc` here would be # a no-op, which is what CodeQL's "unnecessary delete" reports. - proc = self._process self._process = None # This method owns the subprocess for its whole life, and clearing # `_process` is the point after which nothing else can reach it: @@ -276,21 +290,20 @@ def _reap(self, proc: Optional[Popen]) -> None: # evil -- the alternative is an orphan holding the session directory # and, for a cross-user action, running as the job user with no # owner. - self._logger.error( + # + # Every log call in this method goes through + # `_log_error_best_effort`: the failure that routes us here is a + # raising `logging.Filter`, so the log call in front of the kill was + # itself the thing that skipped the kill. + self._log_error_best_effort( f"Abandoning the subprocess {proc.pid} before it exited; terminating it so it " - "is not orphaned.", - extra=LogExtraInfo( - openjd_log_content=LogContent.PROCESS_CONTROL | LogContent.EXCEPTION_INFO - ), + "is not orphaned." ) try: self._terminate_process(proc) except Exception as e: # noqa: BLE001 - self._logger.error( - f"Could not terminate the abandoned subprocess {proc.pid}: {e}", - extra=LogExtraInfo( - openjd_log_content=LogContent.PROCESS_CONTROL | LogContent.EXCEPTION_INFO - ), + self._log_error_best_effort( + f"Could not terminate the abandoned subprocess {proc.pid}: {e}" ) try: # Bounded: a SIGKILLed child is reaped promptly, and this runs on @@ -298,13 +311,10 @@ def _reap(self, proc: Optional[Popen]) -> None: # runner's future rather than surface anything useful. proc.wait(timeout=ABANDONED_PROCESS_REAP_TIMEOUT_SECONDS) except TimeoutExpired: - self._logger.error( + self._log_error_best_effort( f"Subprocess {proc.pid} did not exit within " f"{ABANDONED_PROCESS_REAP_TIMEOUT_SECONDS}s of being terminated; it is left " - "unreaped.", - extra=LogExtraInfo( - openjd_log_content=LogContent.PROCESS_CONTROL | LogContent.EXCEPTION_INFO - ), + "unreaped." ) # Record the exit status. Unconditional on purpose: `Popen.returncode` is # populated by the poll()/wait() above and is the authority either way -- @@ -317,6 +327,33 @@ def _reap(self, proc: Optional[Popen]) -> None: # it left every test passing. An unfalsifiable check is worse than none. self._returncode = proc.returncode + def _log_error_best_effort(self, message: str) -> None: + """Log ``message`` at ERROR, swallowing any failure of the logging stack + itself. + + :meth:`_reap` runs inside :meth:`run`'s ``finally``, so anything raised + from it replaces the exception already propagating and hides the real + cause. That containment was already in place for the terminate and the + wait, but not for the log calls between them -- and the failure mode that + routes us into ``_reap`` in the first place is a ``logging.Filter`` that + raises, which makes those calls the *likeliest* thing to raise here, not + the least. + + Silently swallowing is normally the wrong answer. It is right here + because the alternative is strictly worse on both counts: the caller + loses the genuine exception, and the child is left unreaped. + """ + try: + self._logger.error( + message, + extra=LogExtraInfo( + openjd_log_content=LogContent.PROCESS_CONTROL | LogContent.EXCEPTION_INFO + ), + ) + except Exception: # noqa: BLE001 + # Nowhere left to report this: the logger is what failed. + pass + def notify(self) -> None: """The 'Notify' part of Open Job Description's subprocess cancelation method. On Linux/macOS: diff --git a/test/openjd/sessions_v0/test_subprocess_reaping.py b/test/openjd/sessions_v0/test_subprocess_reaping.py index 8470d93..7370421 100644 --- a/test/openjd/sessions_v0/test_subprocess_reaping.py +++ b/test/openjd/sessions_v0/test_subprocess_reaping.py @@ -27,9 +27,10 @@ import os import signal import time +from contextlib import contextmanager from logging.handlers import QueueHandler from queue import SimpleQueue -from typing import Any, Optional +from typing import Any, Generator, Optional from unittest.mock import MagicMock, patch import pytest @@ -37,7 +38,7 @@ from openjd.sessions._os_checker import is_posix from openjd.sessions._subprocess import LoggingSubprocess -from .conftest import build_logger +from .conftest import build_logger, create_unique_logger_name, serial_process # A child that emits one line (to drive the pump), then outlives the pump. _LONG_CHILD = "import time\nprint('MARKER', flush=True)\ntime.sleep(60)\n" @@ -99,6 +100,57 @@ def exploding_logger() -> Any: logger.removeFilter(exploder) +class _FilterRaisingOnAny(logging.Filter): + """Raises on any record whose message contains one of ``markers``. + + Separate from :class:`_ExplodingFilter` so the tests using that keep their + exact single-marker behaviour. + + The raised message names the marker that triggered it, so a test can tell + *which* log call failed -- which is what distinguishes "the startup line + raised and was contained" from "something inside the reap raised and replaced + it". Both are ``RuntimeError`` from the same filter, so nothing else does. + """ + + def __init__(self, *markers: str) -> None: + super().__init__() + self._markers = markers + self.fired = False + + def filter(self, record: logging.LogRecord) -> bool: + message = str(record.msg) + for marker in self._markers: + if marker in message: + self.fired = True + raise RuntimeError(f"filter blew up on: {marker}") + return True + + +@contextmanager +def logger_carrying(filt: logging.Filter) -> Generator[Any, None, None]: + """A uniquely-named logger carrying `filt`, removed again on exit. + + The removal is not tidiness: ``logging.getLogger`` interns loggers in a + process-global registry, so a *raising* filter left installed would outlive + its test and fire inside an unrelated one. + + Yields ``Any`` rather than ``logging.Logger`` because ``LoggingSubprocess`` + declares its ``logger`` parameter as a ``LoggerAdapter``, while these tests + pass a plain ``Logger`` -- as the ``exploding_logger`` fixture above already + does, for the same reason. + """ + logger = logging.getLogger(create_unique_logger_name(prefix="openjd.test.reaping.startup")) + logger.setLevel(logging.INFO) + handler = logging.NullHandler() + logger.addHandler(handler) + logger.addFilter(filt) + try: + yield logger + finally: + logger.removeFilter(filt) + logger.removeHandler(handler) + + @pytest.mark.skipif(not is_posix(), reason="pid liveness/reap checks are POSIX-only here") class TestPumpExceptionDoesNotAbandonTheChild: def test_live_child_is_terminated_and_reaped( @@ -314,6 +366,44 @@ def test_a_terminate_failure_does_not_escape(self, python_exe: str) -> None: popen.kill() popen.wait() + def test_a_logging_failure_during_the_reap_does_not_escape(self, python_exe: str) -> None: + """`_reap`'s own logging must not raise either. + + The containment here already covered the terminate and the wait, but not + the three `logger.error` calls between them -- and a raising + `logging.Filter` is exactly what routes execution into `_reap`, which + makes those calls the *likeliest* thing to fail, not the least. Anything + escaping replaces the exception already propagating out of `run()`. + + This drives all three call sites at once: the abandoning notice, the + terminate-failure notice, and the unreaped-after-timeout notice. + """ + # GIVEN a live child, a termination that fails, a 0s wait budget so the + # timeout branch is reached too, and a logger that raises on every error + logger = MagicMock() + logger.error.side_effect = RuntimeError("the logging stack blew up") + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", _LONG_CHILD]) + popen = proc._start_subprocess() + assert popen is not None + try: + with ( + patch.object( + LoggingSubprocess, + "_terminate_process", + side_effect=OSError("cannot signal"), + ), + patch("openjd.sessions._subprocess.ABANDONED_PROCESS_REAP_TIMEOUT_SECONDS", 0), + ): + # WHEN / THEN: nothing escapes, despite every log call failing + proc._reap(popen) + + # All three messages were attempted, so this really did exercise + # every call site rather than returning early. + assert logger.error.call_count == 3 + finally: + popen.kill() + popen.wait() + def test_a_child_that_survives_termination_is_reported_not_hung(self, python_exe: str) -> None: """The wait is bounded: a child that somehow outlives SIGKILL must not hang the pool worker.""" @@ -383,3 +473,194 @@ def test_nothing_is_signalled_once_the_process_is_released(self) -> None: # THEN terminate_process.assert_not_called() + + +@serial_process +class TestStartupLoggingDoesNotAbandonTheChild: + """The ownership `try/finally` must begin at process creation, not after the + startup logging. + + `run()` logged "Command started as pid" and "Output:" *before* entering the + `try` whose `finally` reaps, and did the POSIX process-group lookup there + too. Anything raising in that window escaped `run()` having never waited on + the child, leaving a live process with no exit code recorded while the runner + published a terminal state for it. + + A `logging.Filter` is the reachable trigger because filters propagate out of + `Logger.handle`, unlike handlers, whose exceptions go to `handleError` -- + and `Session` installs a filter of its own. The process-group lookup is + covered by its own test below, which is why the boundary moved to + immediately after creation rather than merely above the two log calls. + """ + + @pytest.mark.skipif(not is_posix(), reason="pid liveness/reap checks are POSIX-only here") + def test_a_filter_raising_on_the_pid_line_does_not_leak_the_child( + self, python_exe: str + ) -> None: + # GIVEN: a long-lived child, and a filter that raises on the first + # startup log line -- while the child is still alive + exploder = _FilterRaisingOnAny("Command started as pid") + with logger_carrying(exploder) as logger: + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", _LONG_CHILD]) + + # WHEN + with pytest.raises(RuntimeError, match="blew up"): + proc.run() + + # THEN: the filter really did fire on that line... + assert exploder.fired is True + pid = proc.pid + assert pid is not None + # ...the child is gone, not merely signalled... + assert _wait_gone(pid), f"child {pid} was left running" + # ...and its status was recorded rather than lost. + assert proc.exit_code == -signal.SIGKILL # type: ignore + assert proc._process is None + + @pytest.mark.skipif(not is_posix(), reason="pid liveness/reap checks are POSIX-only here") + def test_a_filter_raising_on_the_output_banner_does_not_leak_the_child( + self, python_exe: str + ) -> None: + """A second, separate unprotected line -- not the same one as above.""" + # GIVEN + exploder = _FilterRaisingOnAny("Output:") + with logger_carrying(exploder) as logger: + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", _LONG_CHILD]) + + # WHEN + with pytest.raises(RuntimeError, match="blew up"): + proc.run() + + # THEN + assert exploder.fired is True + pid = proc.pid + # Not decoration: `_wait_gone(None)` returns True, so without this the + # liveness assertion below degrades to a tautology. + assert pid is not None + assert _wait_gone(pid), f"child {pid} was left running" + assert proc.exit_code == -signal.SIGKILL # type: ignore + assert proc._process is None + + @pytest.mark.skipif(not is_posix(), reason="the process-group lookup is POSIX-only") + def test_a_failing_process_group_lookup_does_not_leak_the_child( + self, message_queue: SimpleQueue, queue_handler: QueueHandler, python_exe: str + ) -> None: + """The POSIX process-group lookup was in the unguarded window too. + + `ProcessLookupError` from it is caught deliberately (a trivial command + can exit first), but nothing else was, and it sits between the two log + lines -- so covering only the logging would leave the middle of the same + window unprotected. No filter here: the failure is injected into the + lookup itself, so this is independent of the logging stack. + + What is asserted is only that the reap is *reached*, and deliberately + NOT that the child dies. When the group lookup is itself what failed, + `_sudo_child_process_group_id` stays unset, and + `_posix_signal_subprocess` then declines to signal at all rather than + target a possibly-recycled pid (the R5-9 "no process group known" + convention). So on this one path the boundary buys the attempt and the + handle release, not the kill. + + That is a real, pre-existing limitation of the signalling path -- not + something this change introduces or claims to fix -- and it is recorded + as a follow-up. Asserting a dead child here would require standing in for + the signal, which would make this test assert a kill the shipped code + does not perform. + """ + # GIVEN a live child and a process-group lookup that fails in a way the + # production code does not catch + logger = build_logger(queue_handler) + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", _LONG_CHILD]) + + # WHEN + with ( + patch( + "openjd.sessions._subprocess.os.getpgid", + side_effect=RuntimeError("getpgid blew up"), + ), + # Stubbed rather than stood-in-for: the real call would spend a second + # scanning for a sudo child it will not find, and this test is about + # whether the reap is reached at all. + patch.object( + LoggingSubprocess, "_terminate_process", autospec=True + ) as terminate_process, + # No wait budget: the child cannot have been killed by the stub, so + # the bounded wait would otherwise be spent in full. + patch("openjd.sessions._subprocess.ABANDONED_PROCESS_REAP_TIMEOUT_SECONDS", 0), + ): + with pytest.raises(RuntimeError, match="getpgid blew up"): + proc.run() + + pid = proc.pid + try: + # THEN: ownership was discharged -- the reap ran, instead of the + # failure escaping in front of it + terminate_process.assert_called_once() + assert proc._process is None + finally: + # The stubbed terminate killed nothing, so this test owns the child. + if pid is not None and _pid_alive(pid): + os.kill(pid, signal.SIGKILL) # type: ignore[attr-defined] + try: + os.waitpid(pid, 0) + except ChildProcessError: + pass + + @pytest.mark.skipif(not is_posix(), reason="pid liveness/reap checks are POSIX-only here") + def test_the_reap_kills_the_child_even_when_its_own_logging_raises( + self, python_exe: str + ) -> None: + """`_reap` must terminate before it logs. + + With the boundary moved, a raising filter now routes *into* `_reap` -- so + if that same filter also matches `_reap`'s own "Abandoning the subprocess" + message, a log-then-terminate order meant the `logger.error` raised and the + kill never ran. The leak would have survived the boundary fix. + + Two properties, both load-bearing and each failing for a different + mutation: the child dies (log-then-terminate breaks this), and the caller + still sees the *startup* failure rather than one raised from inside the + `finally` (a reap log that propagates breaks this). + """ + # GIVEN: a filter that raises on the startup line AND on the message + # `_reap` emits while cleaning up. "Running command" is deliberately not + # matched, so the child is really created. + exploder = _FilterRaisingOnAny("Command started as pid", "Abandoning the subprocess") + with logger_carrying(exploder) as logger: + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", _LONG_CHILD]) + + # WHEN + with pytest.raises(RuntimeError) as excinfo: + proc.run() + + # THEN: the child is reaped despite the logging stack being broken... + pid = proc.pid + assert pid is not None + assert _wait_gone(pid), f"child {pid} was left running" + assert proc.exit_code == -signal.SIGKILL # type: ignore + assert proc._process is None + # ...and the exception is the startup failure. Naming the marker matters: + # both failures are RuntimeError from the same filter, so only the marker + # distinguishes "contained" from "replaced the propagating exception". + assert "filter blew up on: Command started as pid" in str(excinfo.value) + assert "Abandoning the subprocess" not in str(excinfo.value) + + def test_the_startup_lines_are_still_logged_on_a_healthy_run( + self, message_queue: SimpleQueue, queue_handler: QueueHandler, python_exe: str + ) -> None: + """Negative control. Deleting the startup logging would satisfy every + assertion above, so pin that it still happens -- and that a healthy run + keeps its own exit code.""" + # GIVEN / WHEN + logger = build_logger(queue_handler) + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", _SHORT_CHILD]) + proc.run() + + # THEN + lines = [] + while not message_queue.empty(): + lines.append(message_queue.get().getMessage()) + assert any(f"Command started as pid: {proc.pid}" in line for line in lines), lines + assert any(line == "Output:" for line in lines), lines + assert proc.exit_code == 7 + assert proc.failed_to_start is False From cc5b6172f73b0b159b8aeae577e4ee7c46e64260 Mon Sep 17 00:00:00 2001 From: David Leong <116610336+leongdl@users.noreply.github.com> Date: Tue, 28 Jul 2026 16:19:37 -0700 Subject: [PATCH 07/13] fix: A failed launch must release wait_until_started() `_start_subprocess()` catches every `Exception` from the launch and reports it by logging "Process failed to start" -- but that log can itself raise, and then the exception propagates out of `run()` before `self._has_started.set()` is reached. `ScriptRunnerBase._run` waits on `wait_until_started()` with **no timeout** (`_runner_base.py:869`), on the thread that called the public `Session` API. So a launch that had already definitively failed hung its caller forever. Reproduced: with a `logging.Filter` raising on the pre-launch "Running command" line, `run()` raised, `_has_started.is_set()` was False, and a waiter on `wait_until_started()` never returned. The handler's own log raises for a mundane reason -- it interpolates the original error, so any filter matching the first message matches the second too. `_has_started.set()` now happens in a `finally`, so it is reached however `_start_subprocess` leaves. Deliberately unchanged: `failed_to_start` stays False on the raising path, because the runner learns about it from the future's exception (`_on_process_exit` logs "Error running subprocess" and publishes a terminal state). Widening that flag would change what `state` reports for a launch that raised rather than returned None, which is a separate question from the hang. Found by an independent audit of the preceding commit, which made this window reachable to inspection; the defect itself is pre-existing. Tests: 3 added, in the same file as the ownership fix they sit beside -- the event is set, the public `wait_until_started()` really does return, and a negative control that an ordinary failure-to-start (where `_start_subprocess` returns None rather than raising) still reports `failed_to_start` with no exception. Both mutants caught: setting the event only on the non-raising path, and removing the set entirely. The bounded wait on a daemon thread is deliberate, so a regression fails on an assertion instead of hanging the suite. Full suite 887 passed; ruff, black, and mypy clean on native and --platform win32. Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com> (cherry picked from commit b81ef1634ab19f490b1612da9b27d25bf47f2003) --- src/openjd/sessions/_subprocess.py | 20 +++-- .../sessions_v0/test_subprocess_reaping.py | 84 +++++++++++++++++++ 2 files changed, 99 insertions(+), 5 deletions(-) diff --git a/src/openjd/sessions/_subprocess.py b/src/openjd/sessions/_subprocess.py index 8ec47d0..d6bc85c 100644 --- a/src/openjd/sessions/_subprocess.py +++ b/src/openjd/sessions/_subprocess.py @@ -175,11 +175,21 @@ def run(self) -> None: if self._has_started.is_set(): raise RuntimeError("The process has already been run") - self._process = self._start_subprocess() - # Set _has_started regardless of whether we started the process successfully or - # not. That will wake up anyone waiting on wait_until_started() to know whether - # we've gotten this far. - self._has_started.set() + try: + self._process = self._start_subprocess() + finally: + # Set _has_started regardless of whether we started the process successfully or + # not. That will wake up anyone waiting on wait_until_started() to know whether + # we've gotten this far. + # + # In a `finally` because `_start_subprocess` can also *raise*: its + # `except Exception` handler logs "Process failed to start", and that + # log can itself fail (a raising `logging.Filter`), which propagates + # out past this point. `ScriptRunnerBase._run` waits on + # `wait_until_started()` with no timeout, so leaving the event unset + # hangs the thread that called the public `Session` API -- forever, + # on a launch that has already definitively failed. + self._has_started.set() if self._process is None: # We failed to start the subprocess self._start_failed = True diff --git a/test/openjd/sessions_v0/test_subprocess_reaping.py b/test/openjd/sessions_v0/test_subprocess_reaping.py index 7370421..51a12b6 100644 --- a/test/openjd/sessions_v0/test_subprocess_reaping.py +++ b/test/openjd/sessions_v0/test_subprocess_reaping.py @@ -26,8 +26,10 @@ import logging import os import signal +import threading import time from contextlib import contextmanager +from pathlib import Path from logging.handlers import QueueHandler from queue import SimpleQueue from typing import Any, Generator, Optional @@ -664,3 +666,85 @@ def test_the_startup_lines_are_still_logged_on_a_healthy_run( assert any(line == "Output:" for line in lines), lines assert proc.exit_code == 7 assert proc.failed_to_start is False + + +class TestStartFailureAlwaysReleasesWaiters: + """`_has_started` must be set however `_start_subprocess` returns. + + `_start_subprocess` catches every `Exception` from the launch and logs + "Process failed to start" -- but that log can itself raise, and then the + exception propagates out of `run()` before `_has_started.set()` was reached. + `ScriptRunnerBase._run` waits on `wait_until_started()` with **no timeout**, + so the thread that called the public `Session` API blocked forever on a + launch that had already definitively failed. + + Platform-agnostic: no child is ever created (the failing log precedes + `Popen`), so there is nothing to signal or reap and no POSIX guard is needed. + """ + + def test_a_raising_log_in_the_failure_handler_still_releases_waiters( + self, python_exe: str + ) -> None: + # GIVEN a filter that raises on the pre-launch "Running command" line. + # The failure handler then logs "Process failed to start: ", + # whose text contains the marker too, so the handler's own log raises as + # well -- which is the case that used to escape. + exploder = _FilterRaisingOnAny("Running command") + with logger_carrying(exploder) as logger: + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", "pass"]) + + # WHEN + with pytest.raises(RuntimeError, match="blew up"): + proc.run() + + # THEN: waiters are released rather than blocked forever + assert proc._has_started.is_set() is True + # No child was created, so nothing leaked. + assert proc.pid is None + + def test_wait_until_started_returns_after_a_raising_failure_handler( + self, python_exe: str + ) -> None: + """The observable form of the above, through the public API that hung. + + Uses a bounded wait on a daemon thread deliberately: if this regresses, + the test fails on the flag rather than hanging the suite. + """ + # GIVEN + exploder = _FilterRaisingOnAny("Running command") + with logger_carrying(exploder) as logger: + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", "pass"]) + with pytest.raises(RuntimeError, match="blew up"): + proc.run() + + returned = threading.Event() + + def waiter() -> None: + proc.wait_until_started() # as ScriptRunnerBase._run calls it: no timeout + returned.set() + + # WHEN + threading.Thread(target=waiter, daemon=True).start() + + # THEN + assert returned.wait(timeout=10.0) is True, "wait_until_started() did not return" + + def test_a_normal_start_failure_is_still_reported( + self, message_queue: SimpleQueue, queue_handler: QueueHandler + ) -> None: + """Negative control: the ordinary failure-to-start path is unchanged -- + `_start_subprocess` returns None, `run()` does not raise, and the object + reports `failed_to_start`.""" + # GIVEN a command that cannot be launched, and a working logger + logger = build_logger(queue_handler) + proc = LoggingSubprocess( + logger=logger, args=[str(Path("this-command-does-not-exist-openjd-test"))] + ) + + # WHEN + proc.run() + + # THEN + assert proc.failed_to_start is True + assert proc._has_started.is_set() is True + assert proc.is_running is False From 08d6d2efdb3c3dfadbf88b482f74e734bfb1ad47 Mon Sep 17 00:00:00 2001 From: David Leong <116610336+leongdl@users.noreply.github.com> Date: Tue, 28 Jul 2026 16:22:57 -0700 Subject: [PATCH 08/13] fix: Send the kill signal before announcing it `_posix_signal_subprocess` logged 'INTERRUPT: Sending signal "kill" to process group N' and only then called `os.killpg`, with both inside a `try` that catches `OSError`. A `logging.Filter` that raises is not an `OSError`, so the exception left the method having never signalled anything: the kill was gated behind a log call. This matters because of the commit two before this one. `run()`'s reap now reliably runs when a raising filter aborts an action, and its whole purpose is to not abandon a live child -- but the reap's terminate went through this helper, so the child survived anyway. Reproduced: filter raising on the startup line and on the announcement gave a reaped-and-released `LoggingSubprocess` with `exit_code=None` and the child still alive. The ownership fix was only as good as the signal path underneath it. Swapping the two statements is the whole change. A raising filter still propagates -- this is not an attempt to make the helper exception-proof -- but the child is dead before it does, which is the property that matters. The announcement is now emitted after a successful `killpg`; on the `OSError` path the existing handler still logs, so no failure goes unreported. Found by an independent audit of the ownership fix. Tests: 2 added. The child dies even when the announcement raises, plus a negative control that a healthy cancel still emits the announcement and still reaps with -SIGKILL -- deleting the log line satisfies the first test alone. Both mutants caught: the pre-fix ordering, and deleting the announcement. The filter helper now records which markers fired, so the first test can prove the announcement really was the thing that raised rather than passing on a tree where the signal path was never reached. Full suite 889 passed; ruff, black, and mypy clean on native and --platform win32. Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com> (cherry picked from commit 243e4e6e2f328f45f7d9cc0ed61d4bd38412c81a) --- src/openjd/sessions/_subprocess.py | 12 +++- .../sessions_v0/test_subprocess_reaping.py | 63 +++++++++++++++++++ 2 files changed, 74 insertions(+), 1 deletion(-) diff --git a/src/openjd/sessions/_subprocess.py b/src/openjd/sessions/_subprocess.py index d6bc85c..6814562 100644 --- a/src/openjd/sessions/_subprocess.py +++ b/src/openjd/sessions/_subprocess.py @@ -730,11 +730,21 @@ def _posix_signal_subprocess( with ctx_mgr as has_cap_kill: if has_cap_kill or not self._user or self._user.is_process_user(): try: + # Signal first, announce second. The announcement used to come + # first, which put the kill behind a log call: a raising + # `logging.Filter` skipped `killpg` entirely and the exception + # left this method without a signal ever being sent. That is + # reachable from LoggingSubprocess.run()'s reap, whose whole + # purpose is to not abandon a live child. + # + # Only OSError is caught here, so a raising filter propagates + # either way; the difference is whether the child was killed + # before it did. + os.killpg(self._sudo_child_process_group_id, numeric_signal) self._logger.info( f'INTERRUPT: Sending signal "{signal_name}" to process group {self._sudo_child_process_group_id}', extra=LogExtraInfo(openjd_log_content=LogContent.PROCESS_CONTROL), ) - os.killpg(self._sudo_child_process_group_id, numeric_signal) except OSError: self._logger.info( "Could not directly send signal {signal_name} to {self._posix_signal_target.pid}, trying sudo.", diff --git a/test/openjd/sessions_v0/test_subprocess_reaping.py b/test/openjd/sessions_v0/test_subprocess_reaping.py index 51a12b6..bc840f6 100644 --- a/test/openjd/sessions_v0/test_subprocess_reaping.py +++ b/test/openjd/sessions_v0/test_subprocess_reaping.py @@ -118,12 +118,14 @@ def __init__(self, *markers: str) -> None: super().__init__() self._markers = markers self.fired = False + self.fired_markers: list[str] = [] def filter(self, record: logging.LogRecord) -> bool: message = str(record.msg) for marker in self._markers: if marker in message: self.fired = True + self.fired_markers.append(marker) raise RuntimeError(f"filter blew up on: {marker}") return True @@ -748,3 +750,64 @@ def test_a_normal_start_failure_is_still_reported( assert proc.failed_to_start is True assert proc._has_started.is_set() is True assert proc.is_running is False + + +@serial_process +@pytest.mark.skipif(not is_posix(), reason="POSIX process-group signalling") +class TestTheKillIsNotGatedBehindItsOwnLog: + """`_posix_signal_subprocess` must send the signal before announcing it. + + The announcement came first, inside a `try` that catches only `OSError`, so a + raising `logging.Filter` skipped `os.killpg` and left the method without ever + signalling. That is reachable from `run()`'s reap -- which exists precisely so + a live child is not abandoned -- so the reap could be entered, attempt the + kill, and still leave the child running. + """ + + def test_the_child_dies_even_when_the_interrupt_log_raises(self, python_exe: str) -> None: + # GIVEN a live child, and a filter that raises both on the startup line + # (to reach the reap with the child alive) and on the signal announcement + exploder = _FilterRaisingOnAny("Command started as pid", 'INTERRUPT: Sending signal "kill"') + with logger_carrying(exploder) as logger: + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", _LONG_CHILD]) + + # WHEN + with pytest.raises(RuntimeError, match="blew up"): + proc.run() + + # THEN: the announcement really did raise -- without this the test could + # pass on a tree where the signal path was never reached at all... + assert 'INTERRUPT: Sending signal "kill"' in exploder.fired_markers + pid = proc.pid + assert pid is not None + # ...and the child was killed regardless + assert _wait_gone(pid), f"child {pid} was left running" + + def test_the_signal_is_still_announced_on_a_healthy_cancel( + self, message_queue: SimpleQueue, queue_handler: QueueHandler, python_exe: str + ) -> None: + """Negative control: deleting the announcement would satisfy the test + above, so pin that a working logger still gets it -- and that the child + still dies.""" + # GIVEN a running child + logger = build_logger(queue_handler) + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", _LONG_CHILD]) + runner = threading.Thread(target=proc.run, daemon=True) + runner.start() + proc.wait_until_started() + deadline = time.time() + 10.0 + while proc._sudo_child_process_group_id is None and time.time() < deadline: + time.sleep(0.02) + + # WHEN + proc.terminate() + runner.join(timeout=15.0) + + # THEN + lines = [] + while not message_queue.empty(): + lines.append(message_queue.get().getMessage()) + assert any( + 'INTERRUPT: Sending signal "kill" to process group' in line for line in lines + ), lines + assert proc.exit_code == -signal.SIGKILL # type: ignore From df4cb542a1fc5a87fc27b1e6314f997efe753026 Mon Sep 17 00:00:00 2001 From: David Leong <116610336+leongdl@users.noreply.github.com> Date: Tue, 28 Jul 2026 19:22:53 -0700 Subject: [PATCH 09/13] test: Pin the observable state the ownership fix changed Three behaviours around `run()`'s abandoned path were observable but asserted nowhere, which is how a later change flips one without anyone noticing. `is_running` is `_has_started and _process is not None`. Before the ownership fix an abort in the startup window left `_process` set forever, so the property reported True for a subprocess nobody would ever wait on. It now reports False, which is accurate -- pinned along with both of its inputs so a change to either is visible in one place, and with `failed_to_start` asserted False because the child did start before it was abandoned. The subprocess callback is *not* invoked on the abandoned path, even though `_reap` now records a real exit code. That is deliberate: `ScriptRunnerBase` registers `_on_process_exit` as the future's done-callback, so the Session is notified via the future's exception either way, and invoking this one too would notify twice for a single action. Pinned so it stays deliberate, with a negative control that a healthy run still invokes it -- otherwise "assert not called" would be satisfied by a callback that is never invoked at all. Also completes the bind-once convention in `_log_returncode`, which still read `self._process` while running on the pool worker, where `run()`'s `finally` is the thing that clears it. Behaviour-neutral: the pre-existing tests pass unmodified, and a mutant that clears `_process` before the log and reads the attribute again is caught by them. `_log_error_best_effort` now documents why it catches `Exception` and not `BaseException`: a KeyboardInterrupt or SystemExit must still propagate, and neither is reachable on a ThreadPoolExecutor worker anyway. This records the reasoning behind an ordering change that was tried, proved inert by mutation testing, and dropped. Not done, and the review prescription was wrong about it: completing the same bind-once refactor in `_log_subproc_stdout` would delete a `_process is None` check that is *not* dead code -- it is the R5-6 `python -O` guard, and `test_optimized_mode_invariants.py` calls that method directly to pin its error message. Removing it would require changing an existing test, which makes it a behaviour change rather than a refactor. Left alone. Tests: 3 added. 4 mutants, 0 survivors -- `_process` left set, the callback invoked on the abandoned path, the callback never invoked, and the returncode log reading a cleared attribute. Full suite 892 passed; ruff, black, and mypy clean on native and --platform win32. Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com> (cherry picked from commit b2a8768a8d3f904e66150228e52e0521aa009e1a) --- src/openjd/sessions/_subprocess.py | 25 ++++-- .../sessions_v0/test_subprocess_reaping.py | 83 +++++++++++++++++++ 2 files changed, 103 insertions(+), 5 deletions(-) diff --git a/src/openjd/sessions/_subprocess.py b/src/openjd/sessions/_subprocess.py index 6814562..d1c7335 100644 --- a/src/openjd/sessions/_subprocess.py +++ b/src/openjd/sessions/_subprocess.py @@ -262,7 +262,7 @@ def run(self) -> None: self._log_subproc_stdout() # Blocking proc.wait() self._returncode = proc.returncode - self._log_returncode() + self._log_returncode(proc) if self._callback: self._callback() finally: @@ -352,6 +352,14 @@ def _log_error_best_effort(self, message: str) -> None: Silently swallowing is normally the wrong answer. It is right here because the alternative is strictly worse on both counts: the caller loses the genuine exception, and the child is left unreaped. + + Catches ``Exception``, deliberately not ``BaseException``. A + ``KeyboardInterrupt`` or ``SystemExit`` must still propagate, and neither + is reachable here anyway: ``run()`` executes on a ``ThreadPoolExecutor`` + worker, and CPython delivers those to the main thread. The kill is + therefore not ordered around this call -- an ordering that would only + matter for a ``BaseException`` was tried, proved inert by mutation + testing, and dropped rather than shipped unfalsifiable. """ try: self._logger.error( @@ -626,13 +634,20 @@ def _enqueue_stdout(): line, extra=LogExtraInfo(openjd_log_content=LogContent.COMMAND_OUTPUT) ) - def _log_returncode(self): - """Logs the return code of the exited subprocess""" + def _log_returncode(self, proc: Popen) -> None: + """Logs the return code of the exited subprocess. + + Takes the ``Popen`` rather than reading ``self._process``, for the same + reason ``notify()``/``terminate()``/``exit_code`` bind it once: this runs + on the pool worker while ``run()``'s ``finally`` is the thing that clears + the attribute, and an ``AttributeError`` on ``None`` here would unwind + ``run()`` before the child was waited on. + """ if self._returncode is not None: # Print out the signed representation of returncodes that would be negative as a 32-bit signed integer if self._returncode < 0x7FFFFFFF: self._logger.info( - f"Process pid {self._process.pid} exited with code: {self._returncode} (unsigned) / {hex(self._returncode)} (hex)", + f"Process pid {proc.pid} exited with code: {self._returncode} (unsigned) / {hex(self._returncode)} (hex)", extra=LogExtraInfo(openjd_log_content=LogContent.PROCESS_CONTROL), ) else: @@ -642,7 +657,7 @@ def _tosigned(n: int) -> int: return int.from_bytes(b, "big", signed=True) self._logger.info( - f"Process pid {self._process.pid} exited with code: {self._returncode} (unsigned) / {hex(self._returncode)} (hex) / {_tosigned(self._returncode)} (signed)", + f"Process pid {proc.pid} exited with code: {self._returncode} (unsigned) / {hex(self._returncode)} (hex) / {_tosigned(self._returncode)} (signed)", extra=LogExtraInfo(openjd_log_content=LogContent.PROCESS_CONTROL), ) diff --git a/test/openjd/sessions_v0/test_subprocess_reaping.py b/test/openjd/sessions_v0/test_subprocess_reaping.py index bc840f6..4b05c59 100644 --- a/test/openjd/sessions_v0/test_subprocess_reaping.py +++ b/test/openjd/sessions_v0/test_subprocess_reaping.py @@ -811,3 +811,86 @@ def test_the_signal_is_still_announced_on_a_healthy_cancel( 'INTERRUPT: Sending signal "kill" to process group' in line for line in lines ), lines assert proc.exit_code == -signal.SIGKILL # type: ignore + + +@serial_process +@pytest.mark.skipif(not is_posix(), reason="pid liveness/reap checks are POSIX-only here") +class TestObservableStateOnTheAbandonedPath: + """Two things the ownership fix changed, or deliberately left alone, that no + test asserted either way. + + Neither is a defect. They are pinned because they are *observable* -- one is a + public property, the other a consumer callback -- and an unasserted observable + is one a later change can flip without anyone noticing. + """ + + def test_is_running_reports_false_once_the_child_is_abandoned(self, python_exe: str) -> None: + """`is_running` is `_has_started and _process is not None`. + + Before the ownership fix, an abort in the startup window left `_process` + set forever, so this property reported **True** for a subprocess nobody + would ever wait on -- indefinitely, since only `run()` clears it. Now the + `finally` clears it, so the answer is False and it is accurate. + """ + # GIVEN a child abandoned by a raising filter in the startup window + exploder = _FilterRaisingOnAny("Command started as pid") + with logger_carrying(exploder) as logger: + proc = LoggingSubprocess(logger=logger, args=[python_exe, "-c", _LONG_CHILD]) + with pytest.raises(RuntimeError, match="blew up"): + proc.run() + + # THEN: it does not claim to be running + assert proc.is_running is False + # ...and the two inputs to that answer are what we think they are, so a + # future change to either is visible here. + assert proc.has_started is True + assert proc._process is None + # Not a failure to *start*: the child did start, and then was abandoned. + assert proc.failed_to_start is False + assert _wait_gone(proc.pid), f"child {proc.pid} was left running" + + def test_the_subprocess_callback_is_not_invoked_on_the_abandoned_path( + self, python_exe: str + ) -> None: + """Deliberate, and pinned so it stays deliberate. + + `run()` invokes its callback on the two paths that end normally: a failure + to start, and a completed `wait()`. The abandoned path does neither, even + though `_reap` now records a real exit code. That is correct rather than an + omission: `ScriptRunnerBase` registers `_on_process_exit` as the future's + done-callback, so the Session is notified through the future's exception + regardless, and invoking this callback too would notify twice for one + action. + """ + # GIVEN a child abandoned by a raising filter, and a callback + callback = MagicMock() + exploder = _FilterRaisingOnAny("Command started as pid") + with logger_carrying(exploder) as logger: + proc = LoggingSubprocess( + logger=logger, args=[python_exe, "-c", _LONG_CHILD], callback=callback + ) + with pytest.raises(RuntimeError, match="blew up"): + proc.run() + + # THEN: not invoked from this path... + callback.assert_not_called() + # ...even though a terminal exit code was recorded. + assert proc.exit_code == -signal.SIGKILL # type: ignore + assert _wait_gone(proc.pid), f"child {proc.pid} was left running" + + def test_the_subprocess_callback_is_still_invoked_on_a_healthy_run( + self, message_queue: SimpleQueue, queue_handler: QueueHandler, python_exe: str + ) -> None: + """Negative control: the assertion above must not be satisfiable by a + callback that is simply never invoked at all.""" + # GIVEN / WHEN + callback = MagicMock() + logger = build_logger(queue_handler) + proc = LoggingSubprocess( + logger=logger, args=[python_exe, "-c", _SHORT_CHILD], callback=callback + ) + proc.run() + + # THEN + callback.assert_called_once_with() + assert proc.exit_code == 7 From ce863b77e6fbdc71ae7a90ade8741c0278dae717 Mon Sep 17 00:00:00 2001 From: David Leong <116610336+leongdl@users.noreply.github.com> Date: Wed, 29 Jul 2026 16:22:59 -0700 Subject: [PATCH 10/13] fix: Do not load the native EXPR extension unless EXPR is used openjd.expr is a thin facade over the native openjd._openjd_rs extension. _runner_base imported ExprValue and TypeCode from it at module level, and openjd.sessions.__init__ reaches _runner_base via _session, so `import openjd.sessions` loaded the extension unconditionally -- including for a consumer that only ever runs non-EXPR templates. This was new in 0.10.11; 0.10.10 has no such import. Verified against both released versions: after `import openjd.sessions`, openjd._openjd_rs is absent from sys.modules on 0.10.10 and present on 0.10.11. Reach the two EXPR types lazily instead. The guard is on sys.modules rather than on the value's type: FormatString.resolve_value is not limited to str and ExprValue -- a legacy whole-field interpolation returns the symbol's own value, so an int reaches _is_expr_null on a purely non-EXPR path and a "not a str" test would still load the extension for an integer-valued timeout. An ExprValue cannot exist unless the extension that defines it has been loaded, so the sys.modules check is exact rather than heuristic. openjd.model already keeps the Rust surface off its import path deliberately (openjd.model._format_strings._parser duplicates a constant rather than import it); this restores the same discipline here. The remaining module-level _openjd_rs imports are all under openjd.sessions._v1, the opt-in Rust-backed API surface, which openjd.sessions does not reach. Behaviour is unchanged: the existing EXPR null/list tests pass unmodified. Adds test_import_purity.py, pinning purity at import time and while resolving a non-EXPR template. Mutation-checked: restoring the module-level import is caught by 7 tests; dropping the sys.modules guard, or weakening it to isinstance(value, str), is caught by test_resolving_a_non_expr_field_does_not_load_native_extension. Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com> --- src/openjd/sessions/_runner_base.py | 63 ++++++- test/openjd/test_import_purity.py | 283 ++++++++++++++++++++++++++++ 2 files changed, 337 insertions(+), 9 deletions(-) create mode 100644 test/openjd/test_import_purity.py diff --git a/src/openjd/sessions/_runner_base.py b/src/openjd/sessions/_runner_base.py index 91a0d31..0efd7b9 100644 --- a/src/openjd/sessions/_runner_base.py +++ b/src/openjd/sessions/_runner_base.py @@ -5,6 +5,7 @@ import re import stat import shlex +import sys from abc import ABC, abstractmethod from concurrent.futures import Future, ThreadPoolExecutor from dataclasses import dataclass @@ -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,56 @@ 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" + + +def _is_expr_null(value: Any) -> bool: + """True if ``value`` is the EXPR engine's typed null. + + ``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 False without importing anything. Once it *has* been loaded -- + i.e. an EXPR expression has been evaluated in this process -- the import + below is a ``sys.modules`` hit. + + ``ExprValue`` is then 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 False + from openjd.expr import ExprValue + + return isinstance(value, ExprValue) and value.is_null + + +def _is_expr_list(value: Any) -> bool: + """True if ``value`` is an EXPR list value, whose elements flatten into + one argument each (RFC 0005 §1.3.2). + + Only reached with an ``ExprValue`` in hand (the caller has already read + ``.is_null`` off it), so the extension is loaded by definition here. + """ + from openjd.expr import TypeCode + + return bool(value.type.type_code == TypeCode.LIST) + + def _timeout_from_seconds(seconds: int, logger: LoggerAdapter) -> Optional[timedelta]: """Convert a resolved timeout in seconds into an enforceable time limit. @@ -240,7 +285,7 @@ def resolve_action_arg_values(args: Optional[Sequence], symtab: SymbolTable) -> resolved.append(value) elif value.is_null: continue - elif value.type.type_code == TypeCode.LIST: + elif _is_expr_list(value): resolved.extend(str(element) for element in value) else: resolved.append(str(value)) @@ -302,7 +347,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 +426,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 diff --git a/test/openjd/test_import_purity.py b/test/openjd/test_import_purity.py new file mode 100644 index 0000000..e057c66 --- /dev/null +++ b/test/openjd/test_import_purity.py @@ -0,0 +1,283 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +"""``openjd.sessions`` must not load the native EXPR extension unless used. + +``openjd.expr`` is a thin facade over the native ``openjd._openjd_rs`` +extension. A module-level import of it anywhere on the pure-Python session +path makes the extension a *load-time* requirement of ``openjd.sessions``, so +a consumer (e.g. the Deadline Cloud worker agent) that only runs non-EXPR +templates can no longer start without a working native extension for its +platform. openjd.model keeps the Rust surface off its import path deliberately +(see ``openjd.model._format_strings._parser``, which duplicates a constant +rather than import it); these tests hold openjd.sessions to the same contract, +at import time *and* while resolving a non-EXPR template. + +Each check runs in a fresh interpreter, because by the time any given test runs +the extension may already have been loaded by another test in the same worker. + +Note for coverage readers: these probes run in child processes with no +``COVERAGE_PROCESS_START``, so the guard in ``_runner_base._is_expr_null`` +that they exercise still reports as uncovered. Do not "fix" that by deleting +the guard -- it is what these tests exist to protect. +""" + +from __future__ import annotations + +import json +import subprocess +import sys +import textwrap +from pathlib import Path + +import pytest + +# Kept below the suite-wide ``--timeout=30`` (see pyproject addopts) so that a +# hung probe is reported as a probe timeout with its output, rather than +# pytest-timeout killing the test first and losing it. +_PROBE_TIMEOUT_SECONDS = 20 + +# Injected ahead of every probe body, so a body never needs interpolation -- +# probe bodies contain OpenJD ``{{ ... }}`` format strings, which an f-string +# or ``str.format`` would silently mangle into ``{ ... }``. +_PROBE_PREAMBLE = """\ +import json +import sys + +sys.path[:] = json.loads(sys.argv[1]) + +RS = "openjd._openjd_rs" +""" + + +def _probe_python() -> str: + """The interpreter to run a probe with. + + Mirrors the ``python_exe`` fixture in ``sessions_v0/conftest.py``: under + Windows Session 0 ``sys.executable`` is ``pythonservice.exe``, which cannot + run a script. + """ + if sys.platform == "win32" and "pythonservice.exe" in sys.executable.lower(): + return sys.executable.lower().replace("pythonservice.exe", "python.exe") + return sys.executable + + +def _run_probe(tmp_path: Path, body: str) -> str: + """Run ``body`` in a fresh interpreter and return its stdout, stripped. + + ``body`` may use the name ``RS`` (the native extension's module name) and + ``sys``. It is written to a script file rather than passed with ``-c`` so + that a traceback is inspectable, and ``sys.path`` is handed over as a JSON + argument -- not embedded in the source -- so that a path which is not + encodable as UTF-8 source cannot break the probe. + """ + script = tmp_path / "probe.py" + script.write_text(_PROBE_PREAMBLE + textwrap.dedent(body), encoding="utf-8") + completed = subprocess.run( + [_probe_python(), str(script), json.dumps(sys.path)], + capture_output=True, + encoding="utf-8", + errors="replace", + timeout=_PROBE_TIMEOUT_SECONDS, + ) + assert completed.returncode == 0, ( + f"probe exited {completed.returncode}\n" + f"--- script ---\n{script.read_text(encoding='utf-8')}\n" + f"--- stdout ---\n{completed.stdout}\n--- stderr ---\n{completed.stderr}" + ) + return completed.stdout.strip() + + +def test_importing_sessions_does_not_load_native_extension(tmp_path: Path) -> None: + """The 0.10.11 regression: a module-level ``from openjd.expr import ...`` + on the pure-Python path makes ``import openjd.sessions`` load the native + extension.""" + # WHEN + loaded = _run_probe( + tmp_path, + """ + import openjd.sessions + + print(RS in sys.modules) + """, + ) + + # THEN + assert loaded == "False", ( + "`import openjd.sessions` loaded the native extension. Something on the " + "pure-Python session path imports openjd.expr (or openjd._openjd_rs) at " + "module level; it must be imported lazily instead." + ) + + +def test_importing_sessions_does_not_resolve_openjd_expr(tmp_path: Path) -> None: + """``openjd.expr`` is the facade the extension gets pulled in through, so + it must not appear either -- this still fails on a module-level import if + some future build made the extension itself lazy.""" + # WHEN + resolved = _run_probe( + tmp_path, + """ + import openjd.sessions + + print("openjd.expr" in sys.modules) + """, + ) + + # THEN + assert resolved == "False", ( + "`import openjd.sessions` resolved openjd.expr at import time; it must " + "only be imported from inside a function on the EXPR path." + ) + + +def test_native_extension_is_available(tmp_path: Path) -> None: + """Negative control: the extension really is installed here. Without this, + every "must not be loaded" test above would pass trivially in an + environment where it simply is not present.""" + # WHEN + loaded = _run_probe( + tmp_path, + """ + import openjd.expr + + print(RS in sys.modules) + """, + ) + + # THEN + assert loaded == "True", ( + "openjd.expr did not load the native extension, so the purity tests in " + "this file prove nothing. Check the openjd-model install." + ) + + +def test_probe_detects_a_loaded_extension(tmp_path: Path) -> None: + """Negative control for the probe harness: an extension loaded *before* + ``openjd.sessions`` must be observed as loaded. Guards against the checks + above passing because ``sys.modules`` is being read wrongly.""" + # WHEN + loaded = _run_probe( + tmp_path, + """ + import openjd.expr + import openjd.sessions + + print(RS in sys.modules) + """, + ) + + # THEN + assert loaded == "True", "the probe cannot observe a loaded extension at all" + + +def test_resolving_a_non_expr_field_does_not_load_native_extension( + tmp_path: Path, +) -> None: + """Import-time purity is not enough: *running* a non-EXPR template must + not load the extension either. + + ``_is_expr_null`` runs for every optional int field (an action's + ``timeout``, a cancelation's ``notifyPeriodInSeconds``). Its + ``sys.modules`` guard is what keeps that path extension-free. A type-based + guard is not sufficient and this test is why: a legacy whole-field + interpolation returns the symbol's own value, so ``resolve_value`` here + returns an ``int`` -- not a ``str`` and not an ``ExprValue``. + """ + # WHEN: a legacy (non-EXPR) interpolation resolves against an int symbol. + loaded = _run_probe( + tmp_path, + """ + from openjd.model import SymbolTable + from openjd.model.v2023_09 import Action, ModelParsingContext + from openjd.sessions._runner_base import resolve_optional_int_field + + context = ModelParsingContext(supported_extensions=["FEATURE_BUNDLE_1"]) + action = Action.model_validate( + {"command": "echo", "timeout": "{{ Param.Seconds }}"}, context=context + ) + symtab = SymbolTable() + symtab["Param.Seconds"] = 30 + + resolved = resolve_optional_int_field( + action.timeout, symtab, ge=1, description="timeout" + ) + assert resolved == 30, resolved + print(RS in sys.modules) + """, + ) + + # THEN + assert loaded == "False", ( + "resolving a non-EXPR timeout loaded the native extension. The " + "`sys.modules` guard in _runner_base._is_expr_null is the thing that " + "prevents this; check it has not been replaced by a type-based test." + ) + + +def test_resolving_an_expr_field_still_resolves_null(tmp_path: Path) -> None: + """Negative control for over-correction: the EXPR path must still work. + + Deliberately asserts the resolved *value*, not ``sys.modules``: parsing an + EXPR-extension template already loads the extension inside openjd.model, + so a ``sys.modules`` assertion here would hold no matter what + ``_is_expr_null`` did. A whole-field EXPR expression resolving to null must + yield ``None`` ("field omitted"), which is exactly the branch the lazy + ``ExprValue`` import serves -- if it stopped working, the strict integer + parse would raise instead. + """ + # WHEN + resolved = _run_probe( + tmp_path, + """ + from openjd.model import SymbolTable + from openjd.model.v2023_09 import Action, ModelParsingContext + from openjd.sessions._runner_base import resolve_optional_int_field + + context = ModelParsingContext(supported_extensions=["FEATURE_BUNDLE_1", "EXPR"]) + action = Action.model_validate( + {"command": "echo", "timeout": "{{ Seconds }}"}, context=context + ) + symtab = SymbolTable() + symtab["Seconds"] = None + + print(repr(resolve_optional_int_field( + action.timeout, symtab, ge=1, description="timeout" + ))) + """, + ) + + # THEN + assert resolved == "None", ( + "a whole-field EXPR expression resolving to null must be treated as " + f"'field omitted' (None), got {resolved}" + ) + + +@pytest.mark.parametrize( + "module", + [ + "openjd.sessions._runner_base", + "openjd.sessions._session", + "openjd.sessions._runner_step_script", + "openjd.sessions._runner_env_script", + ], +) +def test_pure_path_module_does_not_load_native_extension(tmp_path: Path, module: str) -> None: + """Importing any individual pure-Python-path module must stay extension + free, so the contract does not silently depend on ``__init__`` ordering. + + ``openjd.sessions._v1`` is deliberately excluded: it is the opt-in + Rust-backed API surface, is not reachable from ``openjd.sessions``, and is + expected to load the extension. + """ + # WHEN + loaded = _run_probe( + tmp_path, + f""" + import {module} + + print(RS in sys.modules) + """, + ) + + # THEN + assert loaded == "False", f"importing {module} loaded the native extension" From f3579ecec5cb532ac2b0f70703882dc7437973ef Mon Sep 17 00:00:00 2001 From: David Leong <116610336+leongdl@users.noreply.github.com> Date: Thu, 30 Jul 2026 00:21:17 -0700 Subject: [PATCH 11/13] fix: Resolve a legacy non-string argument instead of crashing resolve_action_arg_values read `.is_null` off whatever resolve_value returned, on the assumption that a non-str result is always an ExprValue. It is not. A legacy (non-EXPR) whole-field interpolation returns the symbol's own value, which is whatever the caller put in the symbol table. openjd.model.ParameterValue types `value` as Any. It documents INT/FLOAT/ PATH values as strings but does not validate that, and it carries BOOL values natively by design (RFC 0007). So both an int and a bool genuinely reach here, and the result was AttributeError ('int' object has no attribute 'is_null') raised out of the public Session.run_task rather than a failed action. Reproduced end-to-end through the public API before fixing: an INT job parameter carrying an int, and a BOOL job parameter carrying True, each crash run_task on a non-EXPR step script. Dispatch on _as_expr_value() so typed null/list semantics apply only to an actual ExprValue, and everything else resolves to its string form -- which is what the function's docstring already promised for legacy expressions, and what a multi-segment arg over the same symbol has always produced. Verified that agreement directly: whole-field and "pre-{{ V }}" render identically for int, bool, float and None. _is_expr_null delegates to the same helper, so there is one Rust-boundary predicate rather than two. Which values reach the branch is decided by openjd-model, not by the symbol table (SymbolTable validates nothing): FullNameNode.allows_nonscalar is False, so a legacy whole-field result that is not a numbers.Real or a str raises FormatStringError and takes the pre-existing plain-resolution fallback. A list therefore never reaches the branch, before or after this change. Mutation-checked. Restoring the original ladder is caught by test_int_job_parameter_value_reaches_the_subprocess_as_a_string and the parametrized test_legacy_whole_field_value_resolves_to_its_string_form; dropping the argument is caught by the same set; and the str-vs-repr coercion is caught by the [str-not-repr] case. That last case exists because an earlier mutant appeared to cover the coercion but did not: it replaced the whole line including the isinstance(value, str) fast path, so it failed via broken string handling instead. str(v) == repr(v) for every plain int, bool and float, so pinning the coercion needs a value where they differ. It uses an int subclass overriding __str__ rather than an IntEnum, because str(SomeIntEnum.MEMBER) is 'E.MEMBER' on Python 3.9-3.10 and '3' from 3.11 on; the targeted tests were run on 3.9 to confirm. The import-purity tests from the previous commit were re-checked against the moved sys.modules guard. Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com> --- src/openjd/sessions/_runner_base.py | 53 ++++-- .../sessions_v0/test_session_let_bindings.py | 158 +++++++++++++++++- 2 files changed, 192 insertions(+), 19 deletions(-) diff --git a/src/openjd/sessions/_runner_base.py b/src/openjd/sessions/_runner_base.py index 0efd7b9..cfd9262 100644 --- a/src/openjd/sessions/_runner_base.py +++ b/src/openjd/sessions/_runner_base.py @@ -145,36 +145,47 @@ def _over_range_message(description: str, value: int) -> str: _EXTENSION_MODULE = "openjd._openjd_rs" -def _is_expr_null(value: Any) -> bool: - """True if ``value`` is the EXPR engine's typed null. +def _as_expr_value(value: Any) -> Optional[Any]: + """``value`` if it is the EXPR engine's typed value, else ``None``. ``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 False without importing anything. Once it *has* been loaded -- + answer is ``None`` without importing anything. Once it *has* been loaded -- i.e. an EXPR expression has been evaluated in this process -- the import below is a ``sys.modules`` hit. ``ExprValue`` is then 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". + + Returning the value rather than a bool matters at the call sites: typed + null/list handling applies *only* to an ``ExprValue``, and anything else + must fall through to plain string resolution rather than have ``.is_null`` + read off it. """ if _EXTENSION_MODULE not in sys.modules: - return False + return None from openjd.expr import ExprValue - return isinstance(value, ExprValue) and value.is_null + return value if isinstance(value, ExprValue) else None + + +def _is_expr_null(value: Any) -> bool: + """True if ``value`` is the EXPR engine's typed null.""" + expr_value = _as_expr_value(value) + return expr_value is not None and bool(expr_value.is_null) -def _is_expr_list(value: Any) -> bool: - """True if ``value`` is an EXPR list value, whose elements flatten into - one argument each (RFC 0005 §1.3.2). +def _is_expr_list(expr_value: Any) -> bool: + """True if ``expr_value`` is an EXPR list value, whose elements flatten + into one argument each (RFC 0005 §1.3.2). - Only reached with an ``ExprValue`` in hand (the caller has already read - ``.is_null`` off it), so the extension is loaded by definition here. + Takes an already-identified ``ExprValue``, so the extension is loaded by + definition here. """ from openjd.expr import TypeCode - return bool(value.type.type_code == TypeCode.LIST) + return bool(expr_value.type.type_code == TypeCode.LIST) def _timeout_from_seconds(seconds: int, logger: LoggerAdapter) -> Optional[timedelta]: @@ -281,14 +292,22 @@ 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: + expr_value = _as_expr_value(value) + if expr_value is None: + # Not a typed EXPR result: 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 expr_value.is_null: continue - elif _is_expr_list(value): - resolved.extend(str(element) for element in value) + elif _is_expr_list(expr_value): + resolved.extend(str(element) for element in expr_value) else: - resolved.append(str(value)) + resolved.append(str(expr_value)) return resolved diff --git a/test/openjd/sessions_v0/test_session_let_bindings.py b/test/openjd/sessions_v0/test_session_let_bindings.py index 2a1dffb..f6a1166 100644 --- a/test/openjd/sessions_v0/test_session_let_bindings.py +++ b/test/openjd/sessions_v0/test_session_let_bindings.py @@ -11,7 +11,10 @@ enter and exit sides); - binding-RHS parsing is memoized across applications; - the unified optional int-or-format-string field resolver - (``resolve_optional_int_field``) enforces consistent bounds. + (``resolve_optional_int_field``) enforces consistent bounds; +- and, as the mirror of the EXPR cases above, a LEGACY (non-EXPR) + whole-field interpolation of a non-string symbol value resolves to its + string form instead of raising ``AttributeError``. """ from __future__ import annotations @@ -24,7 +27,7 @@ import pytest -from openjd.model import SymbolTable +from openjd.model import ParameterValue, ParameterValueType, SymbolTable from openjd.model.v2023_09 import ( Action as Action_2023_09, ArgString as ArgString_2023_09, @@ -78,6 +81,37 @@ def _format_string_field(raw: str) -> Any: return action.timeout +def _legacy_args_action(*args: str) -> Action_2023_09: + """An action with FormatString-typed ``args`` and EXPR *not* enabled. + + Parsed through ``model_validate`` with an explicit context rather than + built like :func:`_action`, because only the FEATURE_BUNDLE_1 parse gives + the args the FormatString behaviour whose whole-field resolution is under + test here. + """ + context = ModelParsingContext_2023_09(supported_extensions=["FEATURE_BUNDLE_1"]) + return Action_2023_09.model_validate({"command": "echo", "args": list(args)}, context=context) + + +class _StrDiffersFromRepr(int): + """An int-valued symbol whose ``str`` and ``repr`` differ. + + Needed to pin *which* coercion the string-form branch applies: for a plain + int, bool or float ``str(v) == repr(v)``, so none of those can tell ``str`` + from ``repr``. + + Overrides ``__str__`` rather than ``__repr__`` on purpose. ``int`` has no + separate ``__str__`` slot, so overriding ``__repr__`` would change ``str()`` + too and discriminate nothing. An ``IntEnum`` would not work either: + ``str(SomeIntEnum.MEMBER)`` is ``'E.MEMBER'`` on Python 3.9-3.10 and + ``'3'`` from 3.11 on, which would fail on 6 of the 18 CI jobs. This class + behaves identically on 3.9 through 3.14 (verified). + """ + + def __str__(self) -> str: + return "str-form" + + # --------------------------------------------------------------------------- # A failing extra `let` binding must FAIL the action via the callback path, # never raise out of enter_environment()/exit_environment(). @@ -250,6 +284,126 @@ def test_let_bound_list_flattens_into_args(self) -> None: assert resolved == ["front", "alpha beta", "gamma", "back"] +# --------------------------------------------------------------------------- +# The mirror of the case above (RFC 0005 §1.3.2, the LEGACY path): a legacy +# (non-EXPR) whole-field interpolation returns the symbol's OWN value, not an +# ExprValue. Typed null/list handling must therefore be reached only for a real +# ExprValue -- reading `.is_null` off an int or a bool raised AttributeError +# straight out of the public Session API. `ParameterValue` documents +# INT/FLOAT/PATH values as strings but does not enforce it, and it carries BOOL +# values natively *by design* (RFC 0007), so both types genuinely arrive here. +# +# Which values reach the branch is decided by openjd-model, not by the symbol +# table (`SymbolTable` validates nothing -- `symtab["X"] = object()` succeeds): +# `FullNameNode.allows_nonscalar` is False, so a legacy whole-field result that +# is not a `numbers.Real` or a `str` raises FormatStringError and takes the +# pre-existing plain-resolution fallback instead. That is why a `list` never +# reaches the branch, and why None does not either. +# --------------------------------------------------------------------------- + + +class TestLegacyNonStringSymbolValuesInArgs: + @pytest.mark.parametrize( + "value, expected", + [ + pytest.param(30, "30", id="int-INT-parameter"), + pytest.param(True, "True", id="bool-BOOL-parameter"), + pytest.param(1.5, "1.5", id="float-FLOAT-parameter"), + # str() and repr() differ only here, so this is the one case that + # pins which coercion the branch applies. + pytest.param(_StrDiffersFromRepr(3), "str-form", id="str-not-repr"), + pytest.param("30", "30", id="str-control"), + # None does NOT reach the branch: the whole-field arg raises + # FormatStringError and takes the pre-existing fallback (the + # multi-segment arg still enters the branch, as an ordinary str). + # Kept as a control that the fallback agrees, not as cover for the + # branch. + pytest.param(None, "", id="none-via-fallback"), + ], + ) + def test_legacy_whole_field_value_resolves_to_its_string_form( + self, value: Any, expected: str + ) -> None: + """A non-string symbol value resolves to its string form, and agrees + with what a multi-segment arg over the same symbol produces. + + The agreement is the invariant that makes string-form resolution the + right answer: ``pre-{{ Param.V }}`` has always gone through plain + string resolution, so a whole-field ``{{ Param.V }}`` must render the + same way or the two disagree within one legacy template. + + Both are asserted against a literal rather than against each other -- + ``multi_segment == f"pre-{whole_field}"`` would hold if both were wrong + in the same way, or if both were empty. + + Holds for the scalar types that reach the branch. A ``str`` *subclass* + is the known exception: the branch passes it through untouched while + the multi-segment path applies ``str()`` to it. That divergence + predates this fix and is not addressed here. + """ + # GIVEN: a non-EXPR action with a whole-field arg and a multi-segment + # arg over the same symbol, whose value is not a string. + action = _legacy_args_action("{{ Param.V }}", "pre-{{ Param.V }}") + symtab = SymbolTable() + symtab["Param.V"] = value + + # WHEN + whole_field, multi_segment = resolve_action_arg_values(action.args, symtab) + + # THEN: the string form. Before the fix an int/bool/float raised + # AttributeError ('is_null') here. + assert whole_field == expected + assert multi_segment == f"pre-{expected}" + + def test_int_job_parameter_value_reaches_the_subprocess_as_a_string( + self, caplog: pytest.LogCaptureFixture, python_exe: str + ) -> None: + """End-to-end through the public API, asserting the argv the process + actually received -- not merely that nothing crashed. + + ``ParameterValue`` documents INT values as strings but types the field + ``Any`` and does not validate it, so a caller passing the int itself + reached ``resolve_action_arg_values``. The failure was an + ``AttributeError`` escaping ``run_task`` -- not a failed action, an + unhandled crash in the caller's thread. + """ + # GIVEN: an INT job parameter carrying an int, consumed as a + # whole-field argument by a non-EXPR step script. + context = ModelParsingContext_2023_09(supported_extensions=["FEATURE_BUNDLE_1"]) + step = StepScript_2023_09.model_validate( + { + "actions": { + "onRun": { + "command": python_exe, + "args": [ + "-c", + "import sys; print('ARGV=' + repr(sys.argv[1:]))", + "{{ Param.N }}", + ], + } + } + }, + context=context, + ) + job_parameters = {"N": ParameterValue(type=ParameterValueType.INT, value=30)} + + with Session(session_id=uuid.uuid4().hex, job_parameter_values=job_parameters) as session: + # WHEN: this raised AttributeError before the fix. + session.run_task(step_script=step, task_parameter_values={}) + _run_until_ready(session) + + # THEN: the action succeeded AND passed the single argument "30". An + # empty or dropped argument would also avoid the crash, so the content + # is what pins the behaviour. + status = session.action_status + assert status is not None + assert status.state == ActionState.SUCCESS + expected = "ARGV=" + repr(["30"]) + assert any(expected in m for m in caplog.messages), [ + m for m in caplog.messages if "ARGV=" in m + ] + + # --------------------------------------------------------------------------- # resolve_optional_int_field: one implementation for the three # "optional int-or-format-string" call sites, with consistent bounds. From d77d7f6cb6ea506caf5f5f60fd04a5f055356a15 Mon Sep 17 00:00:00 2001 From: David Leong <116610336+leongdl@users.noreply.github.com> Date: Thu, 30 Jul 2026 01:15:41 -0700 Subject: [PATCH 12/13] docs: Explain the ignored ChildProcessError in a test cleanup Addresses the CodeQL py/empty-except finding on PR #341: the `except ChildProcessError: pass` in this test's cleanup had no comment saying why swallowing it is correct. It is correct because the child being already reaped between the liveness check and the waitpid is the outcome the cleanup wants -- there is simply nothing left to wait for. Comment only; no behaviour change. Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com> --- test/openjd/sessions_v0/test_subprocess_reaping.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/test/openjd/sessions_v0/test_subprocess_reaping.py b/test/openjd/sessions_v0/test_subprocess_reaping.py index 4b05c59..a5afe62 100644 --- a/test/openjd/sessions_v0/test_subprocess_reaping.py +++ b/test/openjd/sessions_v0/test_subprocess_reaping.py @@ -608,6 +608,8 @@ def test_a_failing_process_group_lookup_does_not_leak_the_child( try: os.waitpid(pid, 0) except ChildProcessError: + # Already reaped between the liveness check and here -- the + # desired end state, so there is nothing left to wait for. pass @pytest.mark.skipif(not is_posix(), reason="pid liveness/reap checks are POSIX-only here") From c1239c6f387d4e66eaf5d442f980e526d3bf762f Mon Sep 17 00:00:00 2001 From: David Leong <116610336+leongdl@users.noreply.github.com> Date: Thu, 30 Jul 2026 11:03:59 -0700 Subject: [PATCH 13/13] refactor: Put every openjd.expr crossing behind the one sys.modules guard Addresses jericht's review on PR #341, which asked _is_expr_list to guard on sys.modules before importing openjd.expr, the way _is_expr_null does. Adding that guard literally would have been dead code: _is_expr_list had one call site, reached only after _as_expr_value had already returned non-None (which required the extension loaded) and after .is_null had been read off the value. Instrumenting it confirmed this -- 4 real calls, all with an ExprValue, all with the extension already loaded, none where the guard would fire. Worse, its `return False` would mean "not a list", so if it ever did fire the caller would stringify an EXPR list instead of flattening it into separate argv entries: a silently wrong answer, not a safe fallback. The concern underneath is real though. The asymmetry was hard to read, and a future caller reaching _is_expr_list with an arbitrary value would have loaded the extension through an unguarded import -- reintroducing the very leak ce863b7 fixed. So remove the asymmetry instead of documenting it. _as_expr_value and _is_expr_list are replaced by a single _classify_expr_value returning an _ExprKind (NOT_EXPR / NULL / LIST / SCALAR). It is now the only function in the module that imports openjd.expr, and it sits behind the one guard, so the property is structural rather than a docstring promise. _is_expr_null becomes a one-line wrapper over it. Behaviour is unchanged and the existing tests pass unmodified: 909 before, 909 after, same set. Mutation-checked the new dispatch seam -- dropping or weakening the guard, suppressing each of the four arms, misclassifying SCALAR as LIST or NULL, and breaking either call-site arm are all caught by name. Two notes: - Classifying SCALAR as LIST initially SURVIVED. A scalar ExprValue is not iterable, so that misclassification raises TypeError out of the runner, and no test passed a scalar whole-field EXPR argument through resolve_action_arg_values -- list and null coverage left that arm open. Adds test_let_bound_scalar_becomes_exactly_one_argument to close it. - Inverting the NULL/LIST check order survives and should: a null value is TypeCode.NULLTYPE and an empty list is not null, so the two conditions are mutually exclusive and the order cannot matter. Verified rather than assumed; recorded here as an equivalent mutant, not a gap. Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com> --- src/openjd/sessions/_runner_base.py | 97 +++++++++++-------- .../sessions_v0/test_session_let_bindings.py | 34 +++++++ 2 files changed, 88 insertions(+), 43 deletions(-) diff --git a/src/openjd/sessions/_runner_base.py b/src/openjd/sessions/_runner_base.py index cfd9262..c98c13c 100644 --- a/src/openjd/sessions/_runner_base.py +++ b/src/openjd/sessions/_runner_base.py @@ -10,7 +10,7 @@ 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 @@ -145,47 +145,59 @@ def _over_range_message(description: str, value: int) -> str: _EXTENSION_MODULE = "openjd._openjd_rs" -def _as_expr_value(value: Any) -> Optional[Any]: - """``value`` if it is the EXPR engine's typed value, else ``None``. +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 ``None`` without importing anything. Once it *has* been loaded -- - i.e. an EXPR expression has been evaluated in this process -- the import - below is a ``sys.modules`` hit. + 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 then imported concretely (rather than duck-typed with + ``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". - - Returning the value rather than a bool matters at the call sites: typed - null/list handling applies *only* to an ``ExprValue``, and anything else - must fall through to plain string resolution rather than have ``.is_null`` - read off it. """ if _EXTENSION_MODULE not in sys.modules: - return None - from openjd.expr import ExprValue + return _ExprKind.NOT_EXPR + from openjd.expr import ExprValue, TypeCode - return value if isinstance(value, ExprValue) else None + 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.""" - expr_value = _as_expr_value(value) - return expr_value is not None and bool(expr_value.is_null) - - -def _is_expr_list(expr_value: Any) -> bool: - """True if ``expr_value`` is an EXPR list value, whose elements flatten - into one argument each (RFC 0005 §1.3.2). - - Takes an already-identified ``ExprValue``, so the extension is loaded by - definition here. - """ - from openjd.expr import TypeCode - - return bool(expr_value.type.type_code == TypeCode.LIST) + return _classify_expr_value(value) is _ExprKind.NULL def _timeout_from_seconds(seconds: int, logger: LoggerAdapter) -> Optional[timedelta]: @@ -292,22 +304,21 @@ def resolve_action_arg_values(args: Optional[Sequence], symtab: SymbolTable) -> # argument is genuinely unresolvable. resolved.append(arg.resolve(symtab=symtab)) continue - expr_value = _as_expr_value(value) - if expr_value is None: - # Not a typed EXPR result: 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. + 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 expr_value.is_null: + elif kind is _ExprKind.NULL: continue - elif _is_expr_list(expr_value): - resolved.extend(str(element) for element in expr_value) - else: - resolved.append(str(expr_value)) + elif kind is _ExprKind.LIST: + resolved.extend(str(element) for element in value) + else: # _ExprKind.SCALAR + resolved.append(str(value)) return resolved diff --git a/test/openjd/sessions_v0/test_session_let_bindings.py b/test/openjd/sessions_v0/test_session_let_bindings.py index f6a1166..2b3c5dd 100644 --- a/test/openjd/sessions_v0/test_session_let_bindings.py +++ b/test/openjd/sessions_v0/test_session_let_bindings.py @@ -283,6 +283,40 @@ def test_let_bound_list_flattens_into_args(self) -> None: # whitespace preserved — not rendered as a single stringified list. assert resolved == ["front", "alpha beta", "gamma", "back"] + def test_let_bound_scalar_becomes_exactly_one_argument(self) -> None: + """The counterpart to list flattening: a non-null, non-list typed value + is ONE argument and must not be flattened. + + Pins the scalar arm of the typed-argument dispatch, which list and null + coverage alone leaves open. A scalar ``ExprValue`` is not iterable, so + misclassifying one as a list raises TypeError out of the runner -- and + with only list/null cases covered, nothing noticed. + """ + # GIVEN: a step script binding an int and a string with embedded + # whitespace, both consumed as whole-field argument expressions. + context = ModelParsingContext_2023_09(supported_extensions=["EXPR"]) + script = StepScript_2023_09.model_validate( + { + "let": ["count = 7", "label = 'solo value'"], + "actions": { + "onRun": { + "command": "echo", + "args": ["front", "{{ count }}", "{{ label }}", "back"], + } + }, + }, + context=context, + ) + + # WHEN + symtab = SymbolTable() + apply_let_bindings(symtab=symtab, let_bindings=script.let or []) + resolved = resolve_action_arg_values(script.actions.onRun.args, symtab) + + # THEN: one argument each -- the int rendered, and the string kept whole + # rather than split on its space or flattened character-wise. + assert resolved == ["front", "7", "solo value", "back"] + # --------------------------------------------------------------------------- # The mirror of the case above (RFC 0005 §1.3.2, the LEGACY path): a legacy