feat: deliver step-scope let bindings to run_task

src/openjd/sessions/_session.py

@@ -1210,6 +1210,7 @@ def run_task(
         os_env_vars: Optional[dict[str, str]] = None,
         log_task_banner: bool = True,
         step_name: Optional[str] = None,
+        extra_let_bindings: Optional[list[str]] = None,
     ) -> None:
         """Run a Task within the Session.
         This method is non-blocking; it will exit when the subprocess is either confirmed to have
@@ -1231,6 +1232,22 @@ def run_task(
             step_name (Optional[str]): The name of the step whose task is being run.
                 Used by RFC 0008 to populate ``WrappedStep.Name`` in wrap hooks.
                 Required when a wrap Environment is active.
+            extra_let_bindings (Optional[list[str]]): Additional EXPR ``let``
+                bindings (RFC 0005) evaluated into the symbol table before the
+                step script's own bindings and actions resolve. This is the
+                step-template-scope ``let`` (``Step.let`` on the instantiated
+                Job), which resolves at job instantiation and so is not part of
+                the step script — the ``run_task`` counterpart of
+                :meth:`enter_environment`'s parameter of the same name, and the
+                v0 counterpart of the per-step resolved symbol table that
+                openjd-rs threads into ``run_task``.
+
+                A caller that obtained its step script from
+                ``create_job`` does not need this: job instantiation folds the
+                step-scope bindings into the script's own ``let``
+                (``StepTemplate.resolve_syntax_sugar``). It is required by a
+                caller that is handed an *un-instantiated* ``StepTemplate``,
+                where ``let`` and ``script.let`` are still separate fields.
 
         Raises:
             RuntimeError: If the Session is not in the READY state.
@@ -1280,6 +1297,31 @@ def run_task(
         # not change non-EXPR behavior.
         if step_name is not None:
             symtab["Step.Name"] = step_name
+
+        # Step-template-scope `let` bindings (RFC 0005 §3.6) accompany the task:
+        # evaluate them into the session-scope table so the step script's own
+        # bindings and its actions can reference them. Seeded after Step.Name so
+        # a step-level binding may reference it, and before path mapping so
+        # {{Session.PathMappingRulesFile}} and the env-var evaluation below see a
+        # complete table -- the same ordering enter_environment uses.
+        #
+        # Script-scope bindings shadow these rather than colliding with them:
+        # StepScriptRunner evaluates `script.let` into a CHILD table sourced from
+        # this one, so a same-named script binding takes precedence.
+        if extra_let_bindings:
+            try:
+                apply_let_bindings(symtab=symtab, let_bindings=extra_let_bindings)
+            except ValueError as e:
+                # ExpressionError and FormatStringError subclass ValueError: a
+                # binding failed to evaluate (e.g. it referenced an undefined
+                # symbol). Fail the action through the normal failure path
+                # rather than raising out of the public API, matching how
+                # enter_environment reports the same failure.
+                self._fail_action_before_start(
+                    f"Failed to evaluate the extra `let` bindings for the task: {e}"
+                )
+                return
+
         action_env_vars = self._evaluate_current_session_env_vars(os_env_vars)
         try:
             self._materialize_path_mapping(step_script.revision, action_env_vars, symtab)

test/openjd/sessions_v0/test_session_let_bindings.py

@@ -207,6 +207,174 @@ def test_step_name_resolvable_in_bindings_and_actions(
             assert any("exit:step is MyStep" in m for m in caplog.messages)
 
 
+# ---------------------------------------------------------------------------
+# run_task(extra_let_bindings=...) delivers step-template-scope `let`
+# (RFC 0005 §3.6) to the task, the counterpart of enter_environment's
+# parameter of the same name.
+#
+# Why this needs its own coverage: step-scope bindings resolve at job
+# instantiation, so a caller holding a Job from create_job never sees the
+# problem -- instantiation folds them into the script's own `let`. A caller
+# handed an un-instantiated StepTemplate (the Deadline Cloud worker agent,
+# which receives one from the service) has `let` and `script.let` as separate
+# fields, and without this parameter the step-scope names are simply absent
+# from the table and every reference fails with "Undefined variable".
+# ---------------------------------------------------------------------------
+
+
+class TestRunTaskExtraLetBindings:
+    def test_step_scope_binding_resolvable_in_action(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        # GIVEN: a step script whose onRun references a name defined only by
+        # the step-template-scope bindings.
+        script = StepScript_2023_09(
+            actions={"onRun": _action("echo", "task:{{ from_step }}")},  # type: ignore[arg-type]
+        )
+        with Session(session_id=uuid.uuid4().hex, job_parameter_values={}) as session:
+            # WHEN
+            session.run_task(
+                step_script=script,
+                task_parameter_values={},
+                extra_let_bindings=["from_step = 'step value'"],
+            )
+            _run_until_ready(session)
+
+            # THEN
+            assert session.state == SessionState.READY
+            status = session.action_status
+            assert status is not None
+            assert status.state == ActionState.SUCCESS
+            assert any("task:step value" in m for m in caplog.messages)
+
+    def test_step_scope_binding_can_reference_step_name(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        # GIVEN: a step-scope binding referencing Step.Name. Pins the seeding
+        # order -- Step.Name must be in the table before the bindings evaluate,
+        # matching enter_environment.
+        script = StepScript_2023_09(
+            actions={"onRun": _action("echo", "task:{{ msg }}")},  # type: ignore[arg-type]
+        )
+        with Session(session_id=uuid.uuid4().hex, job_parameter_values={}) as session:
+            # WHEN
+            session.run_task(
+                step_script=script,
+                task_parameter_values={},
+                step_name="MyStep",
+                extra_let_bindings=["msg = 'step is ' + Step.Name"],
+            )
+            _run_until_ready(session)
+
+            # THEN
+            status = session.action_status
+            assert status is not None
+            assert status.state == ActionState.SUCCESS
+            assert any("task:step is MyStep" in m for m in caplog.messages)
+
+    def test_script_scope_binding_shadows_step_scope(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        # GIVEN: the same name bound at both scopes. RFC 0005 §3.6 scoping
+        # requires the narrower (script) scope to win rather than the two
+        # colliding, which holds because the runner evaluates script bindings
+        # into a child table sourced from the session-scope one.
+        script = StepScript_2023_09(
+            actions={"onRun": _action("echo", "task:{{ shared }}")},  # type: ignore[arg-type]
+            let=["shared = 'from script'"],
+        )
+        with Session(session_id=uuid.uuid4().hex, job_parameter_values={}) as session:
+            # WHEN
+            session.run_task(
+                step_script=script,
+                task_parameter_values={},
+                extra_let_bindings=["shared = 'from step'"],
+            )
+            _run_until_ready(session)
+
+            # THEN
+            status = session.action_status
+            assert status is not None
+            assert status.state == ActionState.SUCCESS
+            assert any("task:from script" in m for m in caplog.messages)
+            assert not any("task:from step" in m for m in caplog.messages)
+
+    def test_script_scope_binding_can_reference_step_scope(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        # GIVEN: a script-scope binding building on a step-scope one. This is
+        # the shape the failing conformance fixtures use, and it only works if
+        # the step bindings are in the parent of the runner's child table.
+        script = StepScript_2023_09(
+            actions={"onRun": _action("echo", "task:{{ derived }}")},  # type: ignore[arg-type]
+            let=["derived = base + '/leaf'"],
+        )
+        with Session(session_id=uuid.uuid4().hex, job_parameter_values={}) as session:
+            # WHEN
+            session.run_task(
+                step_script=script,
+                task_parameter_values={},
+                extra_let_bindings=["base = '/root'"],
+            )
+            _run_until_ready(session)
+
+            # THEN
+            status = session.action_status
+            assert status is not None
+            assert status.state == ActionState.SUCCESS
+            assert any("task:/root/leaf" in m for m in caplog.messages)
+
+    def test_failing_binding_fails_action_cleanly(self) -> None:
+        # GIVEN: a step-scope binding referencing an undefined symbol. It must
+        # fail the action through the callback path, never raise out of the
+        # public API -- the same contract enter_environment holds.
+        callback_events: list[ActionStatus] = []
+
+        def callback(session_id: str, status: ActionStatus) -> None:
+            callback_events.append(status)
+
+        script = StepScript_2023_09(
+            actions={"onRun": _action("echo", "unreachable")},  # type: ignore[arg-type]
+        )
+        with Session(
+            session_id=uuid.uuid4().hex, job_parameter_values={}, callback=callback
+        ) as session:
+            # WHEN: this must not raise.
+            session.run_task(
+                step_script=script,
+                task_parameter_values={},
+                extra_let_bindings=["msg = NoSuchSymbol"],
+            )
+            _run_until_ready(session)
+
+            # THEN
+            status = session.action_status
+            assert status is not None
+            assert status.state == ActionState.FAILED
+            assert status.fail_message is not None
+            assert "let" in status.fail_message
+            assert callback_events and callback_events[-1].state == ActionState.FAILED
+
+    def test_omitting_the_parameter_changes_nothing(self, caplog: pytest.LogCaptureFixture) -> None:
+        # GIVEN: the negative control. The parameter is additive and optional,
+        # so a task that does not use it must behave exactly as before -- this
+        # is what makes the change safe for every existing caller.
+        script = StepScript_2023_09(
+            actions={"onRun": _action("echo", "task:{{ own }}")},  # type: ignore[arg-type]
+            let=["own = 'script only'"],
+        )
+        with Session(session_id=uuid.uuid4().hex, job_parameter_values={}) as session:
+            # WHEN
+            session.run_task(step_script=script, task_parameter_values={})
+            _run_until_ready(session)
+
+            # THEN
+            status = session.action_status
+            assert status is not None
+            assert status.state == ActionState.SUCCESS
+            assert any("task:script only" in m for m in caplog.messages)
+
+
 # ---------------------------------------------------------------------------
 # Binding-RHS parsing is memoized: re-applying the same bindings (per task,
 # per env enter/exit) must not re-parse through the engine each time.