From 54b83a309edb29e0be3efb9133bc21f968644b8c Mon Sep 17 00:00:00 2001 From: David Leong <116610336+leongdl@users.noreply.github.com> Date: Mon, 7 Sep 2026 22:43:37 -0700 Subject: [PATCH 1/3] fix: Record task-emitted openjd_env macros for WrappedAction.Environment RFC 0008 requires WrappedAction.Environment to include every openjd_env variable emitted by any earlier action in the session, "regardless of whether that action ran normally or via a wrap hook". A task's onRun and an onWrapTaskRun hook both run with no running-environment identifier, so _action_callback discarded their macros, and the next task's hook was handed an empty WrappedAction.Environment. Record such a macro in _session_env_vars, and remove it there on openjd_unset_env. _created_env_vars is deliberately untouched, so no child process environment changes and a task behaves the same wrapped or unwrapped. That reproduces the released openjd-rs behaviour: openjd-sessions 0.5.5 writes its cumulative env_vars for every SetEnv and passes that map to seed_wrapped_action_symbols, while evaluate_env_vars builds process environments from created_env_vars alone. Attributing the macro to the wrap environment instead would put it in later subprocess environments, which no openjd-rs version does. Malformed macros keep their current handling. Outside an environment there is no environment action to fail, so they are logged at debug and ignored rather than failing the task. Also fix a concurrency defect this change widens. _collect_session_env_list iterated _session_env_vars live while _action_callback writes it on the LoggingSubprocess stdout thread, raising "RuntimeError: dictionary keys changed during iteration" in 3 of 3 probe runs. The reader now iterates a copy. A lock would let this reader block the thread forwarding a live child's output. Verified against the upstream conformance fixture 2023-09/WRAP_ACTIONS/jobs/wrap-openjd-env-task-grand-child-visible-next-task, which fails before this change and passes after. Full sweeps: WRAP_ACTIONS 97/97, EXPR 356/356. Unit suite 957 passed, coverage 75%. The 8 new tests are mutation-checked against 10 mutants, all caught. For review: openjd_redacted_env reaches this callback as an ENV message, so a task's redacted export now also reaches WrappedAction.Environment and the wrap hook's argv. That matches both an environment's redacted export through this same map and released openjd-rs, and log redaction is unaffected. A test pins the behaviour, so excluding redacted values is a one-line change if preferred. Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com> --- src/openjd/sessions/_session.py | 114 ++++--- test/openjd/sessions_v0/test_wrap_actions.py | 325 +++++++++++++++++++ 2 files changed, 402 insertions(+), 37 deletions(-) diff --git a/src/openjd/sessions/_session.py b/src/openjd/sessions/_session.py index 71bcba70..7147bdf2 100644 --- a/src/openjd/sessions/_session.py +++ b/src/openjd/sessions/_session.py @@ -311,6 +311,21 @@ class Session(object): 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. + + Written by every ``openjd_env`` macro, including one emitted by a task's + ``onRun`` when no environment is running. Such a macro is recorded here and + nowhere else, so it reaches ``WrappedAction.Environment`` without altering + any child process environment. + + openjd-cli 0.1.14 / openjd-sessions 0.5.5, the released openjd-rs lineage, + behaves the same way: ``apply_message`` writes the cumulative ``env_vars`` + for every SetEnv regardless of which action is running, that map is what + ``seed_wrapped_action_symbols`` receives, and ``evaluate_env_vars`` builds + process environments from ``created_env_vars`` alone. Read the released tag, + not ``main``: openjd-rs #362 repointed the wrap-hook seeding at + ``live_session_env_vars()``, which drops task-emitted macros, so ``main`` + currently fails the conformance fixture named below. That is tracked as an + openjd-rs regression, not a divergence to copy here. """ _wrap_env_file_records: dict[EnvironmentIdentifier, list["_FileRecord"]] @@ -2106,8 +2121,16 @@ def _collect_session_env_list(self) -> list[str]: ``_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()] + formatting step only. + + Iterates a snapshot: the writer is ``_action_callback`` on the + LoggingSubprocess stdout thread, and iterating the live dict while that + thread inserts raises ``RuntimeError: dictionary keys changed during + iteration``. Reproduced before this snapshot was added. ``dict.copy()`` + rather than a lock, because the writer runs on the thread that forwards + a running child's output and must not be able to block on this reader. + """ + return [f"{name}={value}" for name, value in self._session_env_vars.copy().items()] def _resolve_action_timeout(self, action: Any, symtab: SymbolTable) -> Optional[int]: """Return the wrapped action's timeout as an int (seconds), or @@ -2404,25 +2427,29 @@ def _action_log_filter_callback( elif kind == ActionMessageKind.ENV: if self._running_environment_identifier is None: - # Ignore the message if we're not running an environment. - # - # Per How-Jobs-Are-Run, `openjd_env` "can only be emitted by the - # Action for entering an Environment" โ€” so a task's onRun cannot - # define a session variable, and neither can an RFC 0008 - # onWrapTaskRun hook, which stands in for one. That keeps - # wrapping transparent: a task that prints an `openjd_env:` line - # behaves the same wrapped and unwrapped. + # No environment is running, so a task's onRun emitted this -- + # directly, or through an RFC 0008 onWrapTaskRun hook standing in + # for one. RFC 0008 (ยง"Stdout forwarding and macro propagation") + # requires it in WrappedAction.Environment regardless: runtimes + # "MUST include ... 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". # - # This is deliberate, not an oversight, and it is a known - # divergence from openjd-rs, which records such a variable in the - # map that feeds WrappedAction.Environment while still not - # applying it to any subprocess environment โ€” advertising a - # variable the wrapped context does not have. The spec does not - # settle the case (it also says a wrap script MAY emit these - # macros directly) and no conformance fixture covers it; filed - # upstream. Logged at debug so an author chasing a silent no-op - # has something to find, without adding noise to every task log. - self._log_discarded_env_macro(kind, value) + # Record it in the session-lifetime map only. _created_env_vars + # stays untouched, so no child process environment changes and a + # task behaves the same wrapped or unwrapped. openjd-rs holds the + # same split: its cumulative env_vars feeds the wrap symbols, + # while evaluate_env_vars builds process environments from + # created_env_vars alone, so a task macro never reaches one. + if cancel_action_mark_failed: + # Malformed macro: value is the parse error, not a variable. + # Keep discarding it, and keep not failing the action -- only + # the in-environment path below does that. + self._log_discarded_env_macro(kind, value) + return + # Assert for the type checker; the type is guaranteed by the ActionMonitoringFilter + assert isinstance(value, dict) + self._session_env_vars[value["name"]] = value["value"] return if cancel_action_mark_failed: # Assert for the type checker; the type is guaranteed by the ActionMonitoringFilter @@ -2441,15 +2468,21 @@ def _action_log_filter_callback( 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. + # _session_env_vars. Runs on the LoggingSubprocess IO thread, so the + # reader iterates a snapshot; see _collect_session_env_list. self._session_env_vars[value["name"]] = value["value"] return elif kind == ActionMessageKind.UNSET_ENV: if self._running_environment_identifier is None: - # Ignore the message if we're not running an environment. - # See the ENV branch above for why this is deliberate. - self._log_discarded_env_macro(kind, value) + # See the ENV branch above. An unset is the one remover from the + # session-lifetime map, matching openjd-rs, where UnsetEnv erases + # the cumulative map whether or not an environment owns the name. + if cancel_action_mark_failed: + self._log_discarded_env_macro(kind, value) + return + # Assert for the type checker; the type is guaranteed by the ActionMonitoringFilter + assert isinstance(value, str) + self._session_env_vars.pop(value, None) return if cancel_action_mark_failed: @@ -2501,21 +2534,28 @@ def _cancel_running_action_as_failed(self) -> None: self.cancel_action(mark_action_failed=True) def _log_discarded_env_macro(self, kind: ActionMessageKind, value: Any) -> None: - """Record, at debug level, that an environment-variable stdout macro was - ignored because the running Action is not an Environment's entry Action. - - Debug rather than a warning on purpose: ``_reset_action_state`` clears - the running-environment identifier for every task, and this callback - cannot tell an RFC 0008 wrap hook from an ordinary task action โ€” so a - warning here would fire for every existing job whose task happens to - print an ``openjd_env:`` line. + """Record, at debug level, that a malformed environment-variable stdout + macro was ignored because no Environment's entry Action is running. + + A well-formed macro from a task is kept (see :attr:`_session_env_vars`); + only one that failed to parse lands here, from ``openjd_env``, + ``openjd_redacted_env`` (which the filter re-dispatches as ``ENV``) or + ``openjd_unset_env``. In an environment, the same parse failure cancels + the action and marks it failed. Outside one there is no environment + action to fail, and this callback cannot tell an RFC 0008 wrap hook from + an ordinary task action, so failing every task that prints a malformed + macro would be a behaviour change for existing jobs. Debug keeps it + findable without that. + + ``value`` is the filter's parse-error message: both call sites are + reached only when ``cancel_action_mark_failed`` is set, and the filter + pairs that flag with a ``str`` payload, never the name/value dict. """ - name = value.get("name") if isinstance(value, dict) else value self._logger.debug( - "Ignoring %s for '%s': environment variables can only be defined by " - "the Action that enters an Environment.", + "Ignoring malformed %s macro (%s): no Environment entry Action is running, " + "so there is no action to fail.", kind.name.lower(), - name, + value, ) def _fail_action_before_start(self, message: str) -> None: diff --git a/test/openjd/sessions_v0/test_wrap_actions.py b/test/openjd/sessions_v0/test_wrap_actions.py index 121948e3..e32de0a8 100644 --- a/test/openjd/sessions_v0/test_wrap_actions.py +++ b/test/openjd/sessions_v0/test_wrap_actions.py @@ -12,6 +12,7 @@ from __future__ import annotations +import threading import time import uuid from pathlib import Path @@ -28,8 +29,11 @@ StepActions as StepActions_2023_09, StepScript as StepScript_2023_09, ) +from openjd.model import RevisionExtensions, SpecificationRevision from openjd.sessions import ActionState, ActionStatus, Session, SessionState +from .conftest import serial_process + # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- @@ -747,6 +751,327 @@ def test_an_explicit_unset_still_removes_the_name(self) -> None: ) +def _await_env_entry(session: Session, entry: str, timeout_s: float = 10.0) -> list[str]: + """Wait until ``entry`` appears in the session env list, then return the list. + + ``_run_until_ready`` is not sufficient on its own: macros are applied by + ``_action_callback`` on the LoggingSubprocess stdout thread, which can still + be draining after the session reports READY. Asserting straight after + ``_run_until_ready`` reads the map mid-flight and fails under load. + """ + deadline = time.time() + timeout_s + while time.time() < deadline: + entries = session._collect_session_env_list() + if entry in entries: + return entries + time.sleep(0.05) + return session._collect_session_env_list() + + +def _await_env_absent(session: Session, prefix: str, timeout_s: float = 10.0) -> list[str]: + """The mirror of :func:`_await_env_entry`, for a name being removed.""" + deadline = time.time() + timeout_s + while time.time() < deadline: + entries = session._collect_session_env_list() + if not any(e.startswith(prefix) for e in entries): + return entries + time.sleep(0.05) + return session._collect_session_env_list() + + +@serial_process +class TestTaskEmittedOpenjdEnv: + """An ``openjd_env`` macro emitted while no environment is entering or exiting. + + `rfcs/0008-environment-wrap-actions.md:447-450`: 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". A task's ``onRun`` and an + ``onWrapTaskRun`` hook both run with no running-environment identifier, so + both used to have the macro discarded, and the second task's hook saw an + empty ``WrappedAction.Environment``. + + The macro is recorded in the session-lifetime map only, never in + ``_created_env_vars``, so no child process environment changes. That is the + split the released openjd-rs lineage has (openjd-sessions 0.5.5, shipped in + openjd-cli 0.1.14); see :attr:`Session._session_env_vars` for why ``main`` + reads differently. + + Upstream conformance fixture, on branch ``conformance-wrap-actions-gaps``: + ``2023-09/WRAP_ACTIONS/jobs/wrap-openjd-env-task-grand-child-visible-next-task``. + It emits the macro from a grandchild whose stdout the hook forwards; the + macro reaches this same callback either way, so the tests below emit from + the task and from the hook directly. + + Variable names here are unique to this class. The module-level logger carries + every session's ActionMonitoringFilter, and reusing a name another test in + this file exports makes a failure here impossible to attribute. + """ + + def test_task_export_is_listed_but_absent_from_a_later_child_environment(self) -> None: + # GIVEN: an entered environment, so _created_env_vars holds a key that a + # mis-attributed task macro could be written into, and then a task that + # exports a variable + with Session(session_id=uuid.uuid4().hex, job_parameter_values={}) as session: + trace = _trace_path(session) + session.enter_environment(environment=_env("Base", onEnter=_NOOP, onExit=_NOOP)) + _run_until_ready(session) + session.run_task( + step_script=_step("sh", "-c", "echo 'openjd_env: TASKEMIT_VAR=from-task'"), + task_parameter_values={}, + step_name="Exporter", + ) + _run_until_ready(session) + assert session.state == SessionState.READY + + # THEN: it is carried for WrappedAction.Environment + entries = _await_env_entry(session, "TASKEMIT_VAR=from-task") + assert "TASKEMIT_VAR=from-task" in entries, entries + + # ... and a later task's real environment does not have it. This is + # the negative control: writing _created_env_vars instead would make + # a wrapped task's stdout mutate later tasks' environments. + session.run_task( + step_script=_step( + "sh", "-c", f"echo \"SAW=[$TASKEMIT_VAR]\" >> '{trace.as_posix()}'" + ), + task_parameter_values={}, + step_name="Reader", + ) + _run_until_ready(session) + contents = trace.read_text() + assert contents.strip() == "SAW=[]", contents + + def test_wrap_hook_export_reaches_the_next_tasks_wrapped_environment(self) -> None: + # GIVEN: a wrap hook that exports a variable and records the + # WrappedAction.Environment it was given + with Session(session_id=uuid.uuid4().hex, job_parameter_values={}) as session: + trace = _trace_path(session) + wrapper = _env( + "Wrapper", + onWrapEnvEnter=_NOOP, + onWrapTaskRun=_action( + "sh", + "-c", + "echo 'openjd_env: TASKEMIT_HOOK=from-hook'; " + f"echo \"WAENV=<{{{{WrappedAction.Environment}}}}>\" >> '{trace.as_posix()}'", + ), + onWrapEnvExit=_NOOP, + ) + session.enter_environment(environment=wrapper) + _run_until_ready(session) + + # WHEN: two tasks run, so the second hook sees what the first exported + session.run_task(step_script=_step("true"), task_parameter_values={}, step_name="First") + _run_until_ready(session) + _await_env_entry(session, "TASKEMIT_HOOK=from-hook") + session.run_task( + step_script=_step("true"), task_parameter_values={}, step_name="Second" + ) + _run_until_ready(session) + + lines = [line for line in trace.read_text().splitlines() if line.strip()] + assert len(lines) == 2, lines + # THEN: the first hook ran before its own export was recorded, and + # the second hook sees it. The ordering half matters: the fixture + # forbids a hook seeing its own task's export. + assert "TASKEMIT_HOOK=from-hook" not in lines[0], lines[0] + assert "TASKEMIT_HOOK=from-hook" in lines[1], lines[1] + + def test_task_emitted_unset_removes_only_the_named_variable(self) -> None: + # Negative control for retention: session-lifetime must not become + # "nothing is ever removed". openjd-rs erases the cumulative map on + # UnsetEnv whether or not an environment owns the name. + # + # Two variables, because a single one cannot tell removing a name from + # clearing the whole map -- and clearing it would drop every session + # variable from WrappedAction.Environment. + with Session(session_id=uuid.uuid4().hex, job_parameter_values={}) as session: + trace = _trace_path(session) + session.enter_environment( + environment=_env( + "Setter", + onEnter=_action( + "sh", + "-c", + "echo 'openjd_env: TASKEMIT_DOOMED=doomed-value'; " + "echo 'openjd_env: TASKEMIT_KEEP=keep-value'", + ), + onExit=_NOOP, + ) + ) + _run_until_ready(session) + _await_env_entry(session, "TASKEMIT_KEEP=keep-value") + assert "TASKEMIT_DOOMED=doomed-value" in session._collect_session_env_list() + + # WHEN: a task unsets one of them + session.run_task( + step_script=_step("sh", "-c", "echo 'openjd_unset_env: TASKEMIT_DOOMED'"), + task_parameter_values={}, + step_name="Unsetter", + ) + _run_until_ready(session) + + # THEN: that name goes, and only that name + entries = _await_env_absent(session, "TASKEMIT_DOOMED=") + assert not any(e.startswith("TASKEMIT_DOOMED=") for e in entries), entries + assert "TASKEMIT_KEEP=keep-value" in entries, entries + + # ... and the environment's own declaration still reaches a child + # process, because _created_env_vars was not touched. + session.run_task( + step_script=_step( + "sh", "-c", f"echo \"SAW=[$TASKEMIT_DOOMED]\" >> '{trace.as_posix()}'" + ), + task_parameter_values={}, + step_name="Reader", + ) + _run_until_ready(session) + contents = trace.read_text() + assert contents.strip() == "SAW=[doomed-value]", contents + + def test_task_export_does_not_displace_an_environments_value_in_a_child(self) -> None: + # A task and an entered environment define the same name. The session map + # takes the later write (the task), while the child process environment + # keeps the environment's value -- the two views can disagree, which is + # the documented cost of not writing _created_env_vars. + with Session(session_id=uuid.uuid4().hex, job_parameter_values={}) as session: + trace = _trace_path(session) + session.enter_environment( + environment=_env( + "Declarer", + onEnter=_action("sh", "-c", "echo 'openjd_env: TASKEMIT_BOTH=from-env'"), + onExit=_NOOP, + ) + ) + _run_until_ready(session) + _await_env_entry(session, "TASKEMIT_BOTH=from-env") + + session.run_task( + step_script=_step("sh", "-c", "echo 'openjd_env: TASKEMIT_BOTH=from-task'"), + task_parameter_values={}, + step_name="Overwriter", + ) + _run_until_ready(session) + + entries = _await_env_entry(session, "TASKEMIT_BOTH=from-task") + assert [e for e in entries if e.startswith("TASKEMIT_BOTH=")] == [ + "TASKEMIT_BOTH=from-task" + ], entries + + session.run_task( + step_script=_step( + "sh", "-c", f"echo \"SAW=[$TASKEMIT_BOTH]\" >> '{trace.as_posix()}'" + ), + task_parameter_values={}, + step_name="Reader", + ) + _run_until_ready(session) + contents = trace.read_text() + assert contents.strip() == "SAW=[from-env]", contents + + @pytest.mark.parametrize( + "macro", + [ + pytest.param("openjd_env: no-equals-sign", id="env-missing-equals"), + pytest.param("openjd_unset_env: 9bad-name", id="unset-invalid-name"), + ], + ) + def test_malformed_task_macro_is_ignored_without_failing_the_task(self, macro: str) -> None: + # Unchanged behaviour, pinned because the fix edits both branches: + # outside an environment there is no environment action to fail, and + # failing the task instead would break existing jobs whose tasks print a + # malformed macro. The parse error must also not be recorded as if it + # were a variable name. + with Session(session_id=uuid.uuid4().hex, job_parameter_values={}) as session: + session.run_task( + step_script=_step("sh", "-c", f"echo '{macro}'"), + task_parameter_values={}, + step_name="Malformed", + ) + _run_until_ready(session) + + assert session.action_status == ActionStatus(state=ActionState.SUCCESS, exit_code=0) + entries = session._collect_session_env_list() + assert not any("parse" in entry.lower() for entry in entries), entries + assert not any(entry.startswith("no-equals-sign") for entry in entries), entries + + def test_collect_session_env_list_tolerates_a_concurrent_writer(self) -> None: + # _action_callback writes _session_env_vars on the LoggingSubprocess + # stdout thread, while _collect_session_env_list reads it on the caller's + # thread to build WrappedAction.Environment. Iterating the live dict + # raises "RuntimeError: dictionary keys changed during iteration"; before + # the reader took a snapshot this reproduced in 3 of 3 runs. + # + # Drives the map directly rather than through real subprocess macros: + # the behaviour under test is the reader's iteration, and 20k real macros + # would be slow without stressing it any harder. Task-path writes are + # what made this reachable during a wrap hook, which is why it is pinned + # with the rest of that change. + with Session(session_id=uuid.uuid4().hex, job_parameter_values={}) as session: + errors: list[BaseException] = [] + stop = threading.Event() + + def writer() -> None: + for i in range(20000): + if stop.is_set(): + return + session._session_env_vars[f"TASKEMIT_RACE_{i}"] = "x" + if i >= 50: + session._session_env_vars.pop(f"TASKEMIT_RACE_{i - 50}", None) + + def reader() -> None: + try: + while not stop.is_set(): + session._collect_session_env_list() + except BaseException as e: # noqa: BLE001 - the failure being pinned + errors.append(e) + + threads = [threading.Thread(target=writer), threading.Thread(target=reader)] + threads[1].start() + threads[0].start() + threads[0].join(timeout=20.0) + stop.set() + for t in threads: + t.join(timeout=20.0) + + assert errors == [], f"{type(errors[0]).__name__}: {errors[0]}" + + def test_task_emitted_redacted_env_is_listed_like_any_other_export(self) -> None: + # openjd_redacted_env reaches _action_callback as an ENV message with a + # name/value dict (_action_filter._handle_redacted_env), so a task's + # redacted export is now recorded like a plain one and reaches + # WrappedAction.Environment -- which interpolates it into the hook's + # argv. Pinned deliberately rather than left implicit: + # - it matches what an *environment's* redacted export has always done + # through this same map, so excluding it here would be inconsistent, + # - the released openjd-rs lineage does the same (RedactedEnv writes the + # cumulative env_vars that seeds the wrap symbols), + # - log redaction is unaffected: the filter holds the value in + # _redacted_values either way. + # If the maintainers want redacted values kept out of the symbol, that is + # a deliberate divergence and this test is where it gets inverted. + with Session( + session_id=uuid.uuid4().hex, + job_parameter_values={}, + revision_extensions=RevisionExtensions( + spec_rev=SpecificationRevision.v2023_09, + supported_extensions=["REDACTED_ENV_VARS"], + ), + ) as session: + session.run_task( + step_script=_step( + "sh", "-c", "echo 'openjd_redacted_env: TASKEMIT_SECRET=hunter2'" + ), + task_parameter_values={}, + step_name="Redactor", + ) + _run_until_ready(session) + + entries = _await_env_entry(session, "TASKEMIT_SECRET=hunter2") + assert "TASKEMIT_SECRET=hunter2" in entries, entries + + class TestRunWrapHookGuard: def test_unknown_hook_name_raises(self) -> None: from openjd.sessions._runner_env_script import EnvironmentScriptRunner From 973a78d063c524d77e3d26c1784a687861ec05c4 Mon Sep 17 00:00:00 2001 From: David Leong <116610336+leongdl@users.noreply.github.com> Date: Tue, 8 Sep 2026 14:10:42 -0700 Subject: [PATCH 2/3] test: Pin the snapshot read deterministically, and narrow the caught exception Two review findings on the tests, no production change. CodeQL flagged the concurrency test for catching BaseException. RuntimeError, the failure being pinned, is an Exception, so narrowing loses no coverage and clears the alert. Narrowing it exposed the larger problem: the test was not pinning anything. With the reader's dict.copy() removed, it passed 3 of 3 runs on its own. It had only failed inside the full class, where sibling tests supplied the contention that made the interleaving land inside the iteration -- so the earlier "mutant caught" result was an artifact of load, not of the assertion. Replaced with a deterministic version. A value whose __format__ inserts into the map stands in for the IO thread, so the write happens during the read by construction: iterating a snapshot tolerates it, iterating the live mapping raises on the next step. Now fails 3 of 3 against the un-snapshotted reader, and it also asserts the insert landed, so the value cannot be inert. The threaded version is gone rather than kept alongside, since it demonstrated the race without pinning it, and the docstring records that the original failure was observed with real threads. Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com> --- test/openjd/sessions_v0/test_wrap_actions.py | 73 ++++++++++---------- 1 file changed, 36 insertions(+), 37 deletions(-) diff --git a/test/openjd/sessions_v0/test_wrap_actions.py b/test/openjd/sessions_v0/test_wrap_actions.py index e32de0a8..f25acba4 100644 --- a/test/openjd/sessions_v0/test_wrap_actions.py +++ b/test/openjd/sessions_v0/test_wrap_actions.py @@ -12,7 +12,6 @@ from __future__ import annotations -import threading import time import uuid from pathlib import Path @@ -996,46 +995,46 @@ def test_malformed_task_macro_is_ignored_without_failing_the_task(self, macro: s assert not any("parse" in entry.lower() for entry in entries), entries assert not any(entry.startswith("no-equals-sign") for entry in entries), entries - def test_collect_session_env_list_tolerates_a_concurrent_writer(self) -> None: + def test_collect_session_env_list_reads_a_snapshot(self) -> None: # _action_callback writes _session_env_vars on the LoggingSubprocess # stdout thread, while _collect_session_env_list reads it on the caller's - # thread to build WrappedAction.Environment. Iterating the live dict - # raises "RuntimeError: dictionary keys changed during iteration"; before - # the reader took a snapshot this reproduced in 3 of 3 runs. + # thread to build WrappedAction.Environment. Iterating the live dict while + # that thread inserts raises "RuntimeError: dictionary keys changed during + # iteration", reproduced with real threads in 3 of 3 runs before the + # reader took a snapshot. # - # Drives the map directly rather than through real subprocess macros: - # the behaviour under test is the reader's iteration, and 20k real macros - # would be slow without stressing it any harder. Task-path writes are - # what made this reachable during a wrap hook, which is why it is pinned - # with the rest of that change. + # Two threads cannot pin that: whether the interleaving lands inside the + # iteration is up to the scheduler, and a first version of this test + # passed against a reader that did NOT snapshot. So the write is triggered + # from inside the read instead. A value that mutates the map while being + # rendered stands in for the IO thread, which makes the failure + # deterministic: iterating a snapshot tolerates it, iterating the live + # mapping raises on the next step. Two entries are needed so there IS a + # next step. + class _MutatesWhenRendered(str): + """A value that inserts into `target` while being formatted.""" + + target: dict[str, str] + + def __format__(self, spec: str) -> str: + self.target[f"INSERTED_{len(self.target)}"] = "x" + return str.__format__(self, spec) + with Session(session_id=uuid.uuid4().hex, job_parameter_values={}) as session: - errors: list[BaseException] = [] - stop = threading.Event() - - def writer() -> None: - for i in range(20000): - if stop.is_set(): - return - session._session_env_vars[f"TASKEMIT_RACE_{i}"] = "x" - if i >= 50: - session._session_env_vars.pop(f"TASKEMIT_RACE_{i - 50}", None) - - def reader() -> None: - try: - while not stop.is_set(): - session._collect_session_env_list() - except BaseException as e: # noqa: BLE001 - the failure being pinned - errors.append(e) - - threads = [threading.Thread(target=writer), threading.Thread(target=reader)] - threads[1].start() - threads[0].start() - threads[0].join(timeout=20.0) - stop.set() - for t in threads: - t.join(timeout=20.0) - - assert errors == [], f"{type(errors[0]).__name__}: {errors[0]}" + first = _MutatesWhenRendered("value-one") + first.target = session._session_env_vars + session._session_env_vars["TASKEMIT_FIRST"] = first + session._session_env_vars["TASKEMIT_SECOND"] = "value-two" + + entries = session._collect_session_env_list() + + assert "TASKEMIT_FIRST=value-one" in entries, entries + assert "TASKEMIT_SECOND=value-two" in entries, entries + # The insert landed in the real map, so the read genuinely raced a + # write rather than the value being inert. + assert any( + name.startswith("INSERTED_") for name in session._session_env_vars + ), session._session_env_vars def test_task_emitted_redacted_env_is_listed_like_any_other_export(self) -> None: # openjd_redacted_env reaches _action_callback as an ENV message with a From 9529ab076fefcacb679ac60cbdc5a32aaf501c37 Mon Sep 17 00:00:00 2001 From: David Leong <116610336+leongdl@users.noreply.github.com> Date: Tue, 8 Sep 2026 20:36:37 -0700 Subject: [PATCH 3/3] docs: Correct the wrapped/unwrapped equivalence claim on the SET branch The comment asserted that a task behaves the same wrapped or unwrapped. The PR's own test_task_export_does_not_displace_an_environments_value_in_a_child disproves it: a task exporting a name an entered environment also declares wins in WrappedAction.Environment while an unwrapped task's real environment keeps the environment's value. What holds is the narrower claim: no process environment changes, so an unwrapped task's environment is exactly what it was before. The comment now says that, and names the consequence it was papering over -- a task's stdout can choose names and values a later wrapped task's hook is handed, through a hook that forwards the symbol. That is new for tasks, matches the released openjd-rs cumulative map, and belongs with the redacted-value and unbounded-growth threads as one spec question: whether a task may write this symbol at all. Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com> --- src/openjd/sessions/_session.py | 24 +++++++++++++++++++----- 1 file changed, 19 insertions(+), 5 deletions(-) diff --git a/src/openjd/sessions/_session.py b/src/openjd/sessions/_session.py index 7147bdf2..b763f8d3 100644 --- a/src/openjd/sessions/_session.py +++ b/src/openjd/sessions/_session.py @@ -2436,11 +2436,25 @@ def _action_log_filter_callback( # that action ran normally or via a wrap hook". # # Record it in the session-lifetime map only. _created_env_vars - # stays untouched, so no child process environment changes and a - # task behaves the same wrapped or unwrapped. openjd-rs holds the - # same split: its cumulative env_vars feeds the wrap symbols, - # while evaluate_env_vars builds process environments from - # created_env_vars alone, so a task macro never reaches one. + # stays untouched, so no process environment changes: an + # unwrapped task's environment is exactly what it was before this + # branch existed. openjd-rs holds the same split, its cumulative + # env_vars feeding the wrap symbols while evaluate_env_vars builds + # process environments from created_env_vars alone. + # + # This is NOT full wrapped/unwrapped equivalence, and the earlier + # comment here overclaimed it. The two views can disagree, in this + # direction: a task exporting a name an entered environment also + # declares wins here, so a wrap hook is handed the task's value + # while an unwrapped task's real environment keeps the + # environment's. test_task_export_does_not_displace_an_environments + # _value_in_a_child pins both halves. That also means a task's + # stdout chooses names and values a later wrapped task's hook is + # handed -- reachable only through a hook that forwards the symbol, + # but new for tasks, and matching what the released openjd-rs + # cumulative map already did. Filed upstream for a spec ruling + # alongside the redacted-value and unbounded-growth cases; it is + # one question, whether a task may write this symbol at all. if cancel_action_mark_failed: # Malformed macro: value is the parse error, not a variable. # Keep discarding it, and keep not failing the action -- only