Hints serverside

.pre-commit-config.yaml

@@ -10,7 +10,6 @@ repos:
       - id: trailing-whitespace
       - id: end-of-file-fixer
       - id: check-yaml
-        exclude: ^test/workflows/test_meta/override_dirac_hints_twice\.yaml$
       - id: check-added-large-files
 
   - repo: https://github.com/astral-sh/ruff-pre-commit

pyproject.toml

@@ -110,8 +110,8 @@ python scripts/generate_schemas.py \
   --individual \
   --unified && \
 echo "Copying schemas to test locations..." && \
-cp generated_schemas/dirac-metadata.json test/workflows/test_meta/schemas/dirac-metadata.json && \
-cp generated_schemas/plugins-summary.json test/workflows/test_meta/schemas/plugins-summary.json
+cp generated_schemas/dirac-metadata.json test/schemas/dirac-metadata.json && \
+cp generated_schemas/plugins-summary.json test/schemas/plugins-summary.json
 """
 
 schemas-yaml = """
@@ -128,8 +128,8 @@ schemas = { depends-on = ["schemas-json", "schemas-yaml"] }
 clean-schemas = """
 echo "Cleaning generated schemas..." && \
 rm -rf generated_schemas/ && \
-rm -f test/workflows/test_meta/schemas/dirac-metadata.json \
-      test/workflows/test_meta/schemas/plugins-summary.json
+rm -f test/schemas/dirac-metadata.json \
+      test/schemas/plugins-summary.json
 """
 
 # Schema validation

scripts/generate_schemas.py

@@ -47,9 +47,9 @@ def collect_pydantic_models() -> Dict[str, Any]:
         models.update(
             {
                 "ExecutionHooksBasePlugin": ExecutionHooksBasePlugin,
-                "ExecutionHooksHint": ExecutionHooksHint,
-                "SchedulingHint": SchedulingHint,
-                "TransformationExecutionHooksHint": TransformationExecutionHooksHint,
+                "ExecutionHooks": ExecutionHooksHint,
+                "Scheduling": SchedulingHint,
+                "TransformationExecutionHooks": TransformationExecutionHooksHint,
             }
         )
         logger.info("Collected core metadata models")

src/dirac_cwl_proto/execution_hooks/core.py

@@ -9,7 +9,7 @@
 import logging
 from abc import ABC, abstractmethod
 from pathlib import Path
-from typing import Any, ClassVar, Dict, List, Mapping, Optional, TypeVar, Union
+from typing import Any, ClassVar, Dict, List, Mapping, Optional, Self, TypeVar, Union
 
 from pydantic import BaseModel, ConfigDict, Field, PrivateAttr
 
@@ -366,7 +366,7 @@ def from_cwl(cls: type[T], cwl_object: Any) -> T:
 
         hints = getattr(cwl_object, "hints", []) or []
         for hint in hints:
-            if hint.get("class") == "dirac:scheduling":
+            if hint.get("class") == "dirac:Scheduling":
                 hint_data = {k: v for k, v in hint.items() if k != "class"}
                 descriptor = descriptor.model_copy(update=hint_data)
 
@@ -407,7 +407,7 @@ def model_copy(
         update: Optional[Mapping[str, Any]] = None,
         *,
         deep: bool = False,
-    ) -> "ExecutionHooksHint":
+    ) -> Self:
         """Enhanced model copy with intelligent merging of dict fields (including configuration)."""
         if update is None:
             update = {}
@@ -489,12 +489,12 @@ def _dash_to_snake(s: str) -> str:
         return get_registry().instantiate_plugin(descriptor)
 
     @classmethod
-    def from_cwl(cls, cwl_object: Any) -> "ExecutionHooksHint":
+    def from_cwl(cls, cwl_object: Any) -> Self:
         """Extract metadata descriptor from CWL object using Hint interface."""
         descriptor = cls()
         hints = getattr(cwl_object, "hints", []) or []
         for hint in hints:
-            if hint.get("class") == "dirac:execution-hooks":
+            if hint.get("class") == "dirac:ExecutionHooks":
                 hint_data = {k: v for k, v in hint.items() if k != "class"}
                 descriptor = descriptor.model_copy(update=hint_data)
         return descriptor

src/dirac_cwl_proto/job/__init__.py

@@ -28,7 +28,6 @@
 from dirac_cwl_proto.submission_models import (
     JobInputModel,
     JobSubmissionModel,
-    extract_dirac_hints,
 )
 
 app = AsyncTyper()
@@ -80,16 +79,6 @@ async def submit_job_client(
         return typer.Exit(code=1)
 
     console.print(f"\t[green]:heavy_check_mark:[/green] Task {task_path}")
-
-    # Extract and validate dirac hints; unknown hints are logged as warnings.
-    try:
-        job_metadata, job_scheduling = extract_dirac_hints(task)
-    except Exception as exc:
-        console.print(
-            f"[red]:heavy_multiplication_x:[/red] [bold]CLI:[/bold] Invalid DIRAC hints:\n{exc}"
-        )
-        return typer.Exit(code=1)
-
     console.print("\t[green]:heavy_check_mark:[/green] Hints")
 
     # Extract parameters if any
@@ -104,17 +93,6 @@ async def submit_job_client(
                 )
                 return typer.Exit(code=1)
 
-            overrides = parameter.pop("cwltool:overrides", {})
-            if overrides:
-                override_hints = overrides[next(iter(overrides))].get("hints", {})
-                if override_hints:
-                    job_scheduling = job_scheduling.model_copy(
-                        update=override_hints.pop("dirac:scheduling", {})
-                    )
-                    job_metadata = job_metadata.model_copy(
-                        update=override_hints.pop("dirac:execution-hooks", {})
-                    )
-
             # Prepare files for the ISB
             isb_file_paths = prepare_input_sandbox(parameter)
 
@@ -131,12 +109,7 @@ async def submit_job_client(
                 f"\t[green]:heavy_check_mark:[/green] Parameter {parameter_p}"
             )
 
-    job = JobSubmissionModel(
-        task=task,
-        parameters=parameters,
-        scheduling=job_scheduling,
-        execution_hooks=job_metadata,
-    )
+    job = JobSubmissionModel(task=task, parameters=parameters)
     console.print(
         "[green]:heavy_check_mark:[/green] [bold]CLI:[/bold] Job(s) validated."
     )
@@ -175,8 +148,6 @@ def validate_jobs(job: JobSubmissionModel) -> list[JobSubmissionModel]:
                 JobSubmissionModel(
                     task=job.task,
                     parameters=[parameter],
-                    scheduling=job.scheduling,
-                    execution_hooks=job.execution_hooks,
                 )
             )
     console.print(

src/dirac_cwl_proto/job/job_wrapper.py

@@ -21,6 +21,7 @@
 from rich.text import Text
 from ruamel.yaml import YAML
 
+from dirac_cwl_proto.execution_hooks import ExecutionHooksHint
 from dirac_cwl_proto.execution_hooks.core import ExecutionHooksBasePlugin
 from dirac_cwl_proto.submission_models import (
     JobInputModel,
@@ -161,8 +162,9 @@ def run_job(self, job: JobSubmissionModel) -> bool:
         logger = logging.getLogger("JobWrapper")
         # Instantiate runtime metadata from the serializable descriptor and
         # the job context so implementations can access task inputs/overrides.
+        job_execution_hooks = ExecutionHooksHint.from_cwl(job.task)
         self.runtime_metadata = (
-            job.execution_hooks.to_runtime(job) if job.execution_hooks else None
+            job_execution_hooks.to_runtime(job) if job_execution_hooks else None
         )
 
         # Isolate the job in a specific directory

src/dirac_cwl_proto/job/submission_clients.py

@@ -14,6 +14,7 @@
 from diracx.client.aio import AsyncDiracClient
 from rich.console import Console
 
+from dirac_cwl_proto.execution_hooks import SchedulingHint
 from dirac_cwl_proto.submission_models import JobSubmissionModel
 
 console = Console()
@@ -169,11 +170,12 @@ def convert_to_jdl(self, job: JobSubmissionModel, sandbox_ids: list[str]) -> str
         jdl_lines.append("JobName = test;")
         jdl_lines.append("OutputSandbox = {std.out, std.err};")
 
-        if job.scheduling.priority:
-            jdl_lines.append(f"Priority = {job.scheduling.priority};")
+        job_scheduling = SchedulingHint.from_cwl(job.task)
+        if job_scheduling.priority:
+            jdl_lines.append(f"Priority = {job_scheduling.priority};")
 
-        if job.scheduling.sites:
-            jdl_lines.append(f"Site = {job.scheduling.sites};")
+        if job_scheduling.sites:
+            jdl_lines.append(f"Site = {job_scheduling.sites};")
 
         jdl_lines.append(f"InputSandbox = {sandbox_ids};")
 

src/dirac_cwl_proto/production/__init__.py

@@ -4,7 +4,7 @@
 
 import logging
 from concurrent.futures import ThreadPoolExecutor
-from typing import Any, List, Optional
+from typing import List, Optional
 
 import typer
 from cwl_utils.pack import pack
@@ -18,13 +18,8 @@
 )
 from rich import print_json
 from rich.console import Console
-from ruamel.yaml import YAML
 from schema_salad.exceptions import ValidationException
 
-from dirac_cwl_proto.execution_hooks import (
-    SchedulingHint,
-    TransformationExecutionHooksHint,
-)
 from dirac_cwl_proto.submission_models import (
     ProductionSubmissionModel,
     TransformationSubmissionModel,
@@ -45,10 +40,6 @@
 @app.command("submit")
 def submit_production_client(
     task_path: str = typer.Argument(..., help="Path to the CWL file"),
-    steps_metadata_path: str = typer.Option(
-        None,
-        help="Path to metadata file used to generate the transformations (one entry per step)",
-    ),
     # Specific parameter for the purpose of the prototype
     local: Optional[bool] = typer.Option(
         True, help="Run the job locally instead of submitting it to the router"
@@ -78,33 +69,10 @@ def submit_production_client(
         )
         return typer.Exit(code=1)
     console.print(f"\t[green]:heavy_check_mark:[/green] Task {task_path}")
-
-    # Load the metadata: at this stage, only the structure is validated, not the content
-    steps_metadata = {}
-    if steps_metadata_path:
-        with open(steps_metadata_path, "r") as file:
-            steps_metadata = YAML(typ="safe").load(file)
-
-    production_step_execution_hooks = {}
-    production_step_scheduling = {}
-    for step_name, step_data in steps_metadata.items():
-        # Extract metadata and scheduling from step_data
-        metadata_config = step_data.get("execution-hooks", {})
-        scheduling_config = step_data.get("scheduling", {})
-
-        # Create TransformationExecutionHooksHint with the metadata
-        production_step_execution_hooks[step_name] = TransformationExecutionHooksHint(
-            **metadata_config
-        )
-        production_step_scheduling[step_name] = SchedulingHint(**scheduling_config)
     console.print("\t[green]:heavy_check_mark:[/green] Metadata")
 
     # Create the production
-    transformation = ProductionSubmissionModel(
-        task=task,
-        steps_execution_hooks=production_step_execution_hooks,
-        steps_scheduling=production_step_scheduling,
-    )
+    production = ProductionSubmissionModel(task=task)
     console.print(
         "[green]:heavy_check_mark:[/green] [bold]CLI:[/bold] Production validated."
     )
@@ -113,8 +81,8 @@ def submit_production_client(
     console.print(
         "[blue]:information_source:[/blue] [bold]CLI:[/bold] Submitting the production..."
     )
-    print_json(transformation.model_dump_json(indent=4))
-    if not submit_production_router(transformation):
+    print_json(production.model_dump_json(indent=4))
+    if not submit_production_router(production):
         console.print(
             "[red]:heavy_multiplication_x:[/red] [bold]CLI:[/bold] Failed to run production."
         )
@@ -170,31 +138,15 @@ def _get_transformations(
     """
     # Create a subworkflow and a transformation for each step
     transformations = []
+
     for step in production.task.steps:
         step_task = _create_subworkflow(
             step, str(production.task.cwlVersion), production.task.inputs
         )
-        configuration = _get_configuration(production.task)
-
-        # Get the execution_hooks & description for the step
-        step_id = step.id.split("#")[-1]
-        step_data: TransformationExecutionHooksHint = (
-            production.steps_execution_hooks.get(
-                step_id,
-                TransformationExecutionHooksHint(),
-            )
-        )
-        step_scheduling: SchedulingHint = production.steps_scheduling.get(
-            step_id,
-            SchedulingHint(),
-        )
-        step_data.configuration.update(configuration)
 
         transformations.append(
             TransformationSubmissionModel(
                 task=step_task,
-                execution_hooks=step_data,
-                scheduling=step_scheduling,
             )
         )
     return transformations
@@ -263,16 +215,3 @@ def _create_subworkflow(
                 break
 
     return new_workflow
-
-
-def _get_configuration(task: Workflow) -> dict[str, Any]:
-    """Get the external inputs of a step.
-
-    :param task: The task to get the query params for
-
-    :return: A dictionary of query params
-    """
-    configuration = {}
-    for input in task.inputs:
-        configuration[input.id.split("#")[-1]] = input.default
-    return configuration