diff --git a/src/openjd/sessions/_session.py b/src/openjd/sessions/_session.py index 71bcba70..b763f8d3 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,43 @@ 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. + # 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". # - # 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. + # Record it in the session-lifetime map only. _created_env_vars + # 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 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) + # 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 + # 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 +2482,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 +2548,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..f25acba4 100644 --- a/test/openjd/sessions_v0/test_wrap_actions.py +++ b/test/openjd/sessions_v0/test_wrap_actions.py @@ -28,8 +28,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 +750,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_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 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. + # + # 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: + 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 + # 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