feat: accept and forward resolvedSymbolTable to the Rust session runtime

pyproject.toml

@@ -16,7 +16,7 @@ dependencies = [
     "deadline-job-attachments == 0.1.3",
     # Pinned to patch version due to Host Config Script runner usage of private OpenJD Sessions API.
     "openjd-sessions == 0.10.14",
-    "openjd-model >= 0.11.3, < 0.12",
+    "openjd-model >= 0.11.4, < 0.12",
     # tomli became tomllib in standard library in Python 3.11
     "tomli == 2.0.* ; python_version<'3.11'",
     "tomlkit >= 0.13,< 0.16",

src/deadline_worker_agent/api_models.py

@@ -218,6 +218,9 @@ class StepDetailsData(StepDetailsIdentifierFields):
     dependencies: NotRequired[list[str]]
     """A list of step identifiers that this step depends on"""
 
+    resolvedSymbolTable: NotRequired[str]
+    """Pre-resolved symbol table as a JSON string, forwarded to the Rust session runtime."""
+
     extensions: NotRequired[list[str]]
     """The extensions enabled for the job, as supplied by the service"""
 
@@ -401,6 +404,8 @@ class EnvironmentDetailsData(EnvironmentDetailsIdentifierFields):
     """The Open Job Description schema version"""
     template: dict[str, Any]
     """The template of the environment."""
+    resolvedSymbolTable: NotRequired[str]
+    """Pre-resolved symbol table as a JSON string, forwarded to the Rust session runtime."""
     extensions: NotRequired[list[str]]
     """The extensions enabled for the job, as supplied by the service"""
 

src/deadline_worker_agent/scheduler/scheduler.py

@@ -1227,6 +1227,8 @@ def _create_new_sessions(
                 region=self._boto_session.region_name,
             )
 
+            resolved_symbol_table_json = queue.peek_resolved_symbol_table_json()
+
             try:
                 session = Session(
                     id=new_session_id,
@@ -1244,6 +1246,7 @@ def _create_new_sessions(
                     session_root_dir=self._session_root_dir,
                     farm_id=self._farm_id,
                     region=self._boto_session.region_name,
+                    resolved_symbol_table_json=resolved_symbol_table_json,
                 )
             except (ValueError, NotImplementedError, OSError) as e:
                 # Runtime construction can fail per-session (e.g. the selected runtime's

src/deadline_worker_agent/scheduler/session_queue.py

@@ -452,6 +452,7 @@ def dequeue(self) -> SessionActionDefinition | None:
                     next_action = ExitEnvironmentAction(
                         id=action_id,
                         environment_id=environment_id,
+                        details=environment_details,
                     )
                 else:
                     raise ValueError(f'Unknown action type "{action_type}".')
@@ -557,3 +558,51 @@ def dequeue(self) -> SessionActionDefinition | None:
                     f'Unknown action type "{action_type}". Complete action = {action_definition}'
                 )
         return next_action
+
+    def peek_resolved_symbol_table_json(self) -> str | None:
+        """Inspect the first queued action's resolved symbol table without consuming it.
+
+        This accessor is non-consuming: the queue state is not mutated, and a
+        subsequent ``dequeue`` call will still yield the same front action.
+
+        Entity resolution results are cached by ``JobEntities``, so the later
+        ``dequeue`` issues no additional service request for the same entity.
+
+        Returns
+        -------
+        str | None
+            The ``resolved_symbol_table_json`` from the first action's entity,
+            or None when the queue is empty or the action type has no table.
+        """
+        if not self._actions:
+            return None
+
+        action_queue_entry = self._actions[0]
+        action_type = action_queue_entry.definition["actionType"]
+
+        try:
+            if action_type.startswith("ENV_"):
+                action_queue_entry = cast(EnvironmentQueueEntry, action_queue_entry)
+                environment_id = action_queue_entry.definition["environmentId"]
+                environment_details = self._job_entities.environment_details(
+                    environment_id=environment_id
+                )
+                return environment_details.resolved_symbol_table_json
+            elif action_type == "TASK_RUN":
+                action_queue_entry = cast(TaskRunQueueEntry, action_queue_entry)
+                step_id = action_queue_entry.definition["stepId"]
+                step_details = self._job_entities.step_details(step_id=step_id)
+                return step_details.resolved_symbol_table_json
+            else:
+                return None
+        except Exception:
+            # This accessor only seeds session-scoped symbols (e.g. Job.Name),
+            # so a failure must not break session creation. The subsequent
+            # dequeue surfaces the real error through the normal action-failure
+            # path.
+            logger.warning(
+                "Failed to prefetch resolved symbol table for the first queued action "
+                "(type=%s); proceeding without it.",
+                action_type,
+            )
+            return None

src/deadline_worker_agent/sessions/actions/enter_env.py

@@ -109,4 +109,5 @@ def start(
             job_env_id=self._job_env_id,
             environment=self._details.environment,
             os_env_vars={"DEADLINE_SESSIONACTION_ID": self._id},
+            resolved_symbol_table_json=self._details.resolved_symbol_table_json,
         )

src/deadline_worker_agent/sessions/actions/exit_env.py

@@ -8,6 +8,7 @@
 from .openjd_action import OpenjdAction
 
 if TYPE_CHECKING:
+    from ..job_entities import EnvironmentDetails
     from ..session import Session
 
 
@@ -20,26 +21,32 @@ class ExitEnvironmentAction(OpenjdAction):
         A unique identifier for the session action
     environment_id : str
         The job environment identifier
+    details : EnvironmentDetails | None
+        Optional environment details carrying the pre-resolved symbol table
     """
 
     _environment_id: str
+    _details: EnvironmentDetails | None
 
     def __init__(
         self,
         *,
         id: str,
         environment_id: str,
+        details: EnvironmentDetails | None = None,
     ) -> None:
         super(ExitEnvironmentAction, self).__init__(
             id=id, action_log_kind=SessionActionLogKind.ENV_EXIT
         )
         self._environment_id = environment_id
+        self._details = details
 
     def __eq__(self, other: Any) -> bool:
         return (
             type(self) is type(other)
             and self._id == other._id
             and self._environment_id == other._environment_id
+            and self._details == other._details
         )
 
     def start(
@@ -58,5 +65,9 @@ def start(
             An executor for running futures
         """
         session.exit_environment(
-            job_env_id=self._environment_id, os_env_vars={"DEADLINE_SESSIONACTION_ID": self._id}
+            job_env_id=self._environment_id,
+            os_env_vars={"DEADLINE_SESSIONACTION_ID": self._id},
+            resolved_symbol_table_json=self._details.resolved_symbol_table_json
+            if self._details is not None
+            else None,
         )

src/deadline_worker_agent/sessions/actions/run_step_task.py

@@ -85,4 +85,5 @@ def start(self, *, session: Session, executor: Executor) -> None:
             task_parameter_values=self._task_parameter_values,
             os_env_vars=env_vars,
             step_name=self._details.step_template.name,
+            resolved_symbol_table_json=self._details.resolved_symbol_table_json,
         )

src/deadline_worker_agent/sessions/job_entities/environment_details.py

@@ -24,6 +24,9 @@ class EnvironmentDetails:
     environment: EnvironmentModel
     """The environment"""
 
+    resolved_symbol_table_json: str | None = None
+    """Pre-resolved symbol table JSON from the service, forwarded to the Rust session runtime."""
+
     @classmethod
     def from_boto(cls, environment_details_data: EnvironmentDetailsData) -> EnvironmentDetails:
         """Converts an environmentDetails entity received from BatchGetJobEntity API response into
@@ -59,7 +62,10 @@ def from_boto(cls, environment_details_data: EnvironmentDetailsData) -> Environm
         else:
             raise UnsupportedSchema(schema_version.value)
 
-        return EnvironmentDetails(environment=environment)
+        return EnvironmentDetails(
+            environment=environment,
+            resolved_symbol_table_json=environment_details_data.get("resolvedSymbolTable", None),
+        )
 
     @classmethod
     def validate_entity_data(cls, entity_data: dict[str, Any]) -> EnvironmentDetailsData:
@@ -90,6 +96,7 @@ def validate_entity_data(cls, entity_data: dict[str, Any]) -> EnvironmentDetails
                 Field(key="environmentId", expected_type=str, required=True),
                 Field(key="jobId", expected_type=str, required=True),
                 Field(key="schemaVersion", expected_type=str, required=True),
+                Field(key="resolvedSymbolTable", expected_type=str, required=False),
                 Field(key="extensions", expected_type=list, required=False),
             ),
         )

src/deadline_worker_agent/sessions/job_entities/step_details.py

@@ -37,6 +37,9 @@ class StepDetails:
     dependencies: list[str] = field(default_factory=list)
     """The dependencies (a list of IDs) that the step depends on"""
 
+    resolved_symbol_table_json: str | None = None
+    """Pre-resolved symbol table JSON from the service, forwarded to the Rust session runtime."""
+
     @classmethod
     def from_boto(cls, step_details_data: StepDetailsData) -> StepDetails:
         """Converts an stepDetails entity received from BatchGetJobEntity API response into a
@@ -86,6 +89,7 @@ def from_boto(cls, step_details_data: StepDetailsData) -> StepDetails:
             step_template=step_template,
             step_id=step_details_data["stepId"],
             dependencies=step_details_data["dependencies"],
+            resolved_symbol_table_json=step_details_data.get("resolvedSymbolTable", None),
         )
 
     @classmethod
@@ -118,6 +122,7 @@ def validate_entity_data(cls, entity_data: dict[str, Any]) -> StepDetailsData:
                 Field(key="template", expected_type=dict, required=True),
                 Field(key="stepId", expected_type=str, required=True),
                 Field(key="dependencies", expected_type=list, required=False),
+                Field(key="resolvedSymbolTable", expected_type=str, required=False),
                 Field(key="extensions", expected_type=list, required=False),
             ),
         )

src/deadline_worker_agent/sessions/runtime/_abc.py

@@ -57,6 +57,7 @@ def enter_environment(
         environment: EnvironmentModel,
         identifier: Optional[EnvironmentIdentifier] = None,
         os_env_vars: Optional[dict[str, str]] = None,
+        resolved_symbol_table_json: str | None = None,
     ) -> EnvironmentIdentifier:
         """Enter an environment; returns its identifier."""
         ...
@@ -68,6 +69,7 @@ def exit_environment(
         identifier: EnvironmentIdentifier,
         os_env_vars: Optional[dict[str, str]] = None,
         keep_session_running: bool = False,
+        resolved_symbol_table_json: str | None = None,
     ) -> None:
         """Exit a previously entered environment."""
         ...
@@ -81,6 +83,7 @@ def run_task(
         os_env_vars: Optional[dict[str, str]] = None,
         log_task_banner: bool = True,
         step_name: str | None = None,
+        resolved_symbol_table_json: str | None = None,
     ) -> None:
         """Run a task within the session's active environment(s)."""
         ...

src/deadline_worker_agent/sessions/runtime/_config.py

@@ -37,3 +37,9 @@ class SessionRuntimeConfig:
     session_root_directory: Path
     spec_revision: str = "2023-09"
     supported_extensions: tuple[str, ...] = ()
+    resolved_symbol_table_json: str | None = None
+    """Pre-resolved symbol table JSON observed on the session's first queued action.
+
+    Adapters that seed session-scoped symbols at construction (e.g. job_name for
+    the classic Python session) consume this at build time.
+    """

src/deadline_worker_agent/sessions/runtime/python.py

@@ -3,9 +3,11 @@
 from __future__ import annotations
 
 from datetime import timedelta
+from logging import getLogger
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Optional
 
+from openjd.expr import SerializedSymbolTable
 from openjd.model import RevisionExtensions, SpecificationRevision
 from openjd.sessions import Session as OpenJDSession
 
@@ -22,6 +24,29 @@
 
 __all__ = ["PythonSessionRuntime"]
 
+logger = getLogger(__name__)
+
+
+def _extract_job_name(json_str: str | None) -> str | None:
+    """Extract the Job.Name value from a resolved symbol table JSON string.
+
+    Returns None when the input is None, the table lacks a Job.Name entry, the
+    entry's value is not a string, or parsing fails (graceful degradation —
+    mirrors _parse_resolved_symtab in the Rust adapter).
+    """
+    if json_str is None:
+        return None
+    try:
+        symtab = SerializedSymbolTable.from_json_str(json_str).to_symtab()
+        entry = symtab.get("Job.Name")
+        if entry is None:
+            return None
+        value = entry.item()
+        return value if isinstance(value, str) else None
+    except Exception as e:
+        logger.warning("Failed to extract Job.Name from resolvedSymbolTable: %s", e)
+        return None
+
 
 class PythonSessionRuntime(SessionRuntime):
     """SessionRuntime backed by openjd.sessions (v0 Python implementation)."""
@@ -38,6 +63,7 @@ def __init__(self, config: SessionRuntimeConfig) -> None:
             callback=config.action_callback,
             os_env_vars=config.os_env_vars,
             session_root_directory=config.session_root_directory,
+            job_name=_extract_job_name(config.resolved_symbol_table_json),
             revision_extensions=RevisionExtensions(
                 # Currently for simplicity request that our session allow all extensions.
                 # This does not obey the spec. It should be changed at a later date to the
@@ -53,7 +79,10 @@ def enter_environment(
         environment: EnvironmentModel,
         identifier: Optional[EnvironmentIdentifier] = None,
         os_env_vars: Optional[dict[str, str]] = None,
+        resolved_symbol_table_json: str | None = None,
     ) -> EnvironmentIdentifier:
+        # resolved_symbol_table_json: not forwarded — the v0 Python session does
+        # not support pre-resolved symbol tables.
         return self._session.enter_environment(
             environment=environment,
             identifier=identifier,
@@ -66,7 +95,10 @@ def exit_environment(
         identifier: EnvironmentIdentifier,
         os_env_vars: Optional[dict[str, str]] = None,
         keep_session_running: bool = False,
+        resolved_symbol_table_json: str | None = None,
     ) -> None:
+        # resolved_symbol_table_json: not forwarded — the v0 Python session does
+        # not support pre-resolved symbol tables.
         self._session.exit_environment(
             identifier=identifier,
             os_env_vars=os_env_vars,
@@ -81,7 +113,10 @@ def run_task(
         os_env_vars: Optional[dict[str, str]] = None,
         log_task_banner: bool = True,
         step_name: str | None = None,
+        resolved_symbol_table_json: str | None = None,
     ) -> None:
+        # resolved_symbol_table_json: not forwarded — the v0 Python session does
+        # not support pre-resolved symbol tables.
         self._session.run_task(
             step_script=step_script,
             task_parameter_values=task_parameter_values,