[Design]: CWL Job Submission in DiracX

[ryuwd]

This task/design documents the first steps towards integrating CWL into DiracX.

cc @aldbr

Status: Draft for Discussion

Context: The Three-Level CWL Model

DIRAC CWL workflows operate at three levels, determined by which hints are present (not by CWL class type):

Level	DIRAC Concept	CWL Hint	Determines
Production	Production/Request	dirac:Production	Input dataset sourcing, orchestrates transformations
Transformation	Transformation	dirac:Transformation	Job template — grouping, input queries
Job	Job	dirac:Job (new, replaces dirac:Scheduling + dirac:ExecutionHooks)	Single execution: scheduling, I/O, hooks

A CWL Workflow with dirac:Production creates Transformations from its steps. Each Transformation is a job template that creates many Jobs. The dirac:Job hint lives at the Job level — it tells DIRAC how to schedule and manage a single execution.

This design focuses on the Job level: the dirac:Job hint and the diracX submission endpoint.

Problem Statement

At the moment in DIRAC, CWL jobs are submitted to DIRAC by:

1. Setting Executable = "dirac-cwl-exec" in the JDL
2. Shipping the CWL definition (job.json) as an InputSandbox file
3. At runtime on the worker node, __createCWLJobWrapper clones the dirac-cwl repo, installs pixi, downloads the sandbox, and runs the CWL

This has several problems:

• The CWL workflow definition is opaque to DIRAC — it's just a sandbox blob, not queryable or inspectable
• dirac:Scheduling and dirac:ExecutionHooks are separate hints that duplicate concepts already present in JDL (Site, Priority, OutputSandbox, etc.)
• The existing convert_to_jdl() in submission_clients.py only maps ~40% of available JDL fields
• There is no native diracX API for CWL submission — everything goes through the JDL path
• The worker-node shim git clones dirac-cwl and installs pixi on every job

Goals

1. Unified dirac:Job hint — replace dirac:Scheduling and dirac:ExecutionHooks with a single versioned hint
2. Dedicated workflows table — store CWL definitions once, content-addressed by SHA-256; jobs reference a workflow, not embed it
3. job_workflow_params table — per-job parameters stored separately, lightweight; 10k parametric jobs = 1 workflow row + 10k param rows
4. New diracX endpoint — POST /api/jobs/ accepts CWL + input YAML(s) directly
5. Models in diracX — JobSubmissionModel, JobHint, and related types live in diracX (migrated from dirac-cwl)
6. No git clone on worker nodes — dirac-cwl is installed in the diracX environment; job wrapper is accessed via importlib.resources
7. Fail fast — strict validation of all CWL ID references, types, and hint fields at submission time

Design

1. The dirac:Job Hint

Design principles

1. Use standard CWL where possible — don't duplicate what CWL already provides natively via requirements
2. Execution hooks are not user-configured — they are determined automatically by type; the submitter doesn't choose them
3. Derive what you can — job_name from CWL label/id, processors from ResourceRequirement, etc.
4. Reference CWL I/O by source ID — instead of duplicating file lists, use source: to point to CWL input/output IDs
5. Versioned schema — the hint carries a schema_version to enable forward-compatible evolution

What CWL already provides (via requirements)

These standard CWL constructs map directly to JDL fields without needing a dirac:Job field:

CWL Requirement	CWL Field	JDL Equivalent	Notes
ResourceRequirement	coresMin	MinNumberOfProcessors
ResourceRequirement	coresMax	MaxNumberOfProcessors
ResourceRequirement	ramMin	MinRAM
ResourceRequirement	ramMax	MaxRAM
ToolTimeLimit	timelimit	—	Wall-clock seconds; see CPUTime
DockerRequirement	dockerPull	—	Container support TBD (unrelated to Platform)
CUDARequirement	(presence)	Tags: ["GPU"]	Implies GPU tag
MPIRequirement	(presence)	—	Not supported — raises NotImplementedError
(CWL task)	label or id	JobName	Derived automatically

CPUTime and CPU work

DIRAC's CPUTime is normalized CPU work in HS06-seconds (wall_time * CPUNormalizationFactor), not wall-clock time. CWL's ToolTimeLimit is wall-clock seconds. These are fundamentally different units.

Approach: The dirac:Job hint provides an explicit cpu_work field representing normalized HS06-seconds — the same unit DIRAC uses internally. This avoids ambiguity:

• cpu_work in dirac:Job → maps directly to JDL CPUTime (HS06-seconds)
• ToolTimeLimit (if present) → used by cwltool for local execution; not translated to CPUTime

The normalization factor itself can be calculated by DB12 on the worker node. Users who think in wall-clock terms can compute cpu_work = wall_seconds * estimated_HS06_factor.

What dirac:Job adds (DIRAC-specific, no CWL equivalent)

cwlVersion: v1.2
class: CommandLineTool

requirements:
  - class: ResourceRequirement
    coresMin: 1
    coresMax: 4
    ramMin: 2048        # MB

hints:
  - class: dirac:Job
    schema_version: "1.0"

    # --- Scheduling ---
    priority: 5
    cpu_work: 864000              # HS06-seconds (= CPUTime in JDL)
    platform: "x86_64-el9"
    sites:
      - LCG.CERN.cern
      - LCG.IN2P3.fr
    banned_sites:
      - LCG.RAL.uk
    tags: ["GPU"]                 # additional tags beyond auto-derived ones

    # --- Job metadata ---
    type: "User"                  # determines execution hooks automatically
    group: "lhcb_analysis"
    log_level: "INFO"

    # --- I/O: reference CWL inputs/outputs by source ID ---
    input_sandbox:
      - source: helper_script               # CWL input ID (type: File) → job root
      - source: config_files                # CWL input ID (type: File[])
        path: "conf/"                       # relative to job working directory
    input_data:
      - source: input_lfns                  # CWL input ID (type: File[])
    output_sandbox:
      - source: stderr_log                  # CWL output ID
    output_data:
      - source: result_file                 # CWL output ID
        output_path: "/lhcb/user/r/roneil/output/"
        output_se: ["SE-USER"]
      - source: histogram                   # CWL output ID
        output_path: "/lhcb/user/r/roneil/histos/"
        output_se: ["SE-AUXILIARY"]

label: "my-analysis-job"   # → becomes JobName

inputs:
  - id: helper_script
    type: File
  - id: config_files
    type: File[]
  - id: input_lfns
    type: File[]
  - id: config_param
    type: string

outputs:
  - id: result_file
    type: File
    outputBinding:
      glob: "result.root"
  - id: histogram
    type: File
    outputBinding:
      glob: "histos.root"
  - id: stderr_log
    type: File
    outputBinding:
      glob: "std.err"

$namespaces:
  dirac: "schemas/dirac-metadata.json#/$defs/"

Note: no baseCommand — the executor is always the dirac-cwl runner. The CWL task defines what to run; DIRAC handles how to run it.

Key design decisions

type instead of job_type: Shorter, cleaner, and consistent with CWL's own class: convention. Maps to JDL JobType. Determines execution hooks automatically (see Execution hooks).

Sites: sites + banned_sites as flat lists:

# Run only at these sites:
sites:
  - LCG.CERN.cern
  - LCG.IN2P3.fr

# Exclude specific sites:
banned_sites:
  - LCG.RAL.uk

DIRAC computes the effective set as Sites - BannedSites. If sites is omitted or empty, the job can run anywhere (equivalent to Site = ANY). Both fields are optional flat lists — simple to read and write, no nesting required. The semantics mirror DIRAC's native model directly.

I/O by CWL source ID using CWL-idiomatic source: syntax:

Each I/O entry uses source: to reference a CWL input or output by its id, mirroring CWL's own outputSource convention:

• input_sandbox: [{source: helper_script}] — the CWL input with id: helper_script (must be type: File or File[]) will be uploaded to the DIRAC sandbox store and delivered to the worker node. An optional path: specifies a relative directory within the job working directory (e.g., path: "conf/" places the file(s) in <job_root>/conf/). If omitted, files land in the job root
• input_data: [{source: input_lfns}] — the CWL input with id: input_lfns will be resolved as LFN paths and registered as InputData in the JDL for data-driven scheduling
• output_sandbox: [{source: stderr_log}] — the CWL output with id: stderr_log will be uploaded to the sandbox store after execution
• output_data: [{source: result_file, output_path: "/lhcb/...", output_se: ["SE-USER"]}] — the CWL output with id: result_file will be registered in the file catalog at the given LFN path, on the specified storage element(s)

The source: syntax is:

• CWL-idiomatic — consistent with how CWL references inputs/outputs elsewhere
• Extensible — per-entry metadata (like output_se, output_path, path) lives alongside the source reference
• Per-output SE — each output_data entry specifies its own output_se, allowing different outputs to go to different storage elements (e.g., large data to tape, small histograms to disk)

All referenced IDs are strictly validated at submission time — the translation layer verifies that each source ID exists in the CWL task's inputs/outputs and has a compatible type (File or File[]). Invalid references fail the submission immediately.

Schema versioning: Every dirac:Job hint must carry a schema_version field. This enables the system to:

• Reject hints with unsupported versions
• Evolve the schema without breaking existing workflows
• Provide clear error messages when a workflow targets a newer schema version

Field mapping summary

Source	Field	JDL Equivalent	In dirac:Job?
CWL label/id	(auto)	JobName	No — derived
ResourceRequirement.coresMin	(auto)	MinNumberOfProcessors	No — CWL native
ResourceRequirement.coresMax	(auto)	MaxNumberOfProcessors	No — CWL native
ResourceRequirement.ramMin	(auto)	MinRAM	No — CWL native
ResourceRequirement.ramMax	(auto)	MaxRAM	No — CWL native
CUDARequirement	(auto)	Tags += ["GPU"]	No — CWL native
MPIRequirement	—	—	NotImplementedError
dirac:Job	schema_version	—	Yes — required
dirac:Job	cpu_work	CPUTime	Yes — HS06-seconds
dirac:Job	priority	Priority	Yes
dirac:Job	platform	Platform	Yes
dirac:Job	sites	Site	Yes
dirac:Job	banned_sites	BannedSites	Yes
dirac:Job	tags	Tags (merged with auto)	Yes
dirac:Job	type	JobType	Yes
dirac:Job	group	JobGroup	Yes
dirac:Job	log_level	LogLevel	Yes
dirac:Job	input_sandbox[].source	InputSandbox	Yes — CWL input IDs
dirac:Job	input_sandbox[].path	(worker-side)	Yes — relative directory
dirac:Job	input_data[].source	InputData	Yes — CWL input IDs
dirac:Job	output_sandbox[].source	OutputSandbox	Yes — CWL output IDs
dirac:Job	output_data[].source	OutputData	Yes — CWL output ID
dirac:Job	output_data[].output_path	OutputPath	Yes — per-output LFN path
dirac:Job	output_data[].output_se	OutputSE	Yes — per-output SE list
(system)	Executable	dirac-cwl-exec	No — always set
(system)	Owner, OwnerGroup, VO	(from auth)	No — injected
(system)	Status, MinorStatus	(managed)	No
(system)	JobID	(auto)	No

Execution hooks: automatic, not user-configured

Execution hooks are derived from type:

• type: "User" → QueryBasedPlugin (default)
• type: "MCSimulation" → VO-specific simulation plugin
• etc.

The hook plugin registry (currently in dirac-cwl, eventually migrated to diracX as entrypoints) handles discovery by VO and type. The dirac:Job hint does not expose hook_plugin or hook_config — these are internal to the system.

output_data (with per-output output_se and output_path) and output_sandbox remain in dirac:Job because they are user-specified data management choices, not hook configuration.

2. Storage Model: workflows + job_workflow_params

Instead of embedding CWL into each JDL row (which would duplicate the CWL blob across thousands of parametric jobs), CWL definitions are stored once in a dedicated table, and jobs reference them.

Schema

CREATE TABLE workflows (
    workflow_id  CHAR(64) PRIMARY KEY,  -- SHA-256 of the CWL content
    cwl          MEDIUMTEXT NOT NULL,    -- CWL YAML (original, uncompressed)
    persistent   BOOL NOT NULL DEFAULT FALSE,
    created_at   DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
);

-- New columns on the existing Jobs table:
ALTER TABLE Jobs
    ADD COLUMN workflow_id      CHAR(64) DEFAULT NULL,     -- FK → workflows.workflow_id
    ADD COLUMN workflow_params  JSON DEFAULT NULL,          -- immutable per-job input parameters
    ADD FOREIGN KEY (workflow_id) REFERENCES workflows(workflow_id);

workflow_params is an immutable JSON column — once set at job creation, it is never updated. It holds the per-job CWL input parameters (the content of an input YAML). Non-CWL jobs leave both columns NULL.

How it works

1. Workflow insertion — on submission, the CWL content is SHA-256 hashed. If the hash already exists in workflows, the insert is skipped (content-addressed, immutable). Workflows are never edited — a changed CWL produces a new hash.

2. Job creation — each job row in Jobs gets a workflow_id reference and its own workflow_params JSON. This is where per-job variation lives — co-located with the job, no extra join needed.

3. Parametric jobs — submitting 10k jobs with the same CWL but different parameters produces:

   • 1 row in workflows (insert-if-not-exists)
   • 10k rows in Jobs with the same workflow_id but different workflow_params (lightweight JSON)

4. persistent flag — controls cleanup behavior:

   • persistent = FALSE (default): ad-hoc user jobs; workflow row can be cleaned up when no jobs reference it
   • persistent = TRUE: production/transformation workflows; retained indefinitely

5. Worker-side retrieval — the job wrapper fetches the CWL via diracX API using the workflow_id from its Jobs row, and reads input parameters from workflow_params. No sandbox involved for the workflow definition.

Parameter mapping via dirac:Job hint

The dirac:Job hint tells DIRAC which CWL inputs should be promoted to job-level parameters visible to the scheduler:

hints:
  - class: dirac:Job
    schema_version: "1.0"
    type: User
    # ... scheduling fields ...

    # Which CWL inputs become job-visible parameters
    input_data:
      - source: input_lfns      # CWL input ID → resolved to InputData for scheduling
    input_sandbox:
      - source: helper_script   # CWL input ID → files uploaded to sandbox

At submission time, the translation layer reads these mappings to populate JDL fields (for the transition period) or job attributes (post-JDL) from the per-job parameters.

Why not CWL-in-JDL?

Concern	CWL-in-JDL (previous)	workflows table (current)
10k parametric jobs	10k copies of compressed CWL	1 workflow row + 10k param rows
Storage	~16MB CWL blob per JDL row	CWL stored once
Queryability	Opaque base64 blob	CWL stored as readable YAML
Immutability	Mutable (JDL can be updated)	Content-addressed, immutable
Cleanup	Tied to JDL lifecycle	Independent lifecycle via persistent flag

3. New diracX Endpoint: POST /api/jobs/

Request format

The endpoint accepts a CWL workflow file plus one or more input YAML files. Each input YAML produces a separate job:

POST /api/jobs/
Content-Type: multipart/form-data

workflow: <wf.cwl>           # CWL workflow/tool definition (YAML)
inputs[]: <input1.yaml>      # Input parameters for job 1
inputs[]: <input2.yaml>      # Input parameters for job 2

This produces 2 jobs:

• Job 1: run wf.cwl with input1.yaml
• Job 2: run wf.cwl with input2.yaml

If no input files are provided, a single job is created with no inputs (suitable for tools with no required inputs or all defaults).

Translation flow

POST /api/jobs/
  │
  ▼
Router (diracx-routers)
  │  Parses multipart: CWL YAML + input YAML(s)
  │  Validates CWL via JobSubmissionModel (pydantic)
  │  Validates schema_version
  ▼
Logic (diracx-logic)
  │  SHA-256 hash CWL → INSERT INTO workflows IF NOT EXISTS
  │  For each input YAML:
  │    Extracts dirac:Job hint from CWL task
  │    Extracts ResourceRequirement, CUDARequirement, etc.
  │    Derives JobName from CWL label/id
  │    Resolves I/O: source IDs → file paths/LFNs (strict validation)
  │    Maps all → JDL fields (transition period)
  │    Calls existing submit_jdl_jobs() with generated JDL
  │    Sets workflow_id + workflow_params (JSON) on Jobs row
  ▼
DB
  │  CWL stored once in workflows table
  │  JDL stored in JobJDLs (transition period)
  │  Job attrs + workflow_id + workflow_params in Jobs table
  ▼
Returns list[InsertedJob]

During the transition period, JDL is still generated for compatibility with existing DIRAC infrastructure (matcher, optimizer, etc.). The workflows table + workflow_params column are the source of truth for the CWL definition and per-job parameters. Once JDL is fully retired, the JDL generation step is removed.

Translation logic (new functions in diracx-logic)

import hashlib
import json

from cwl_utils.parser import save
from cwl_utils.parser.cwl_v1_2 import (
    ResourceRequirement, CUDARequirement, MPIRequirement,
)


SUPPORTED_SCHEMA_VERSIONS = {"1.0"}


def compute_workflow_id(cwl_yaml: str) -> str:
    """Content-address a CWL workflow by its SHA-256 hash."""
    return hashlib.sha256(cwl_yaml.encode()).hexdigest()


async def submit_cwl_jobs(
    cwl_yaml: str,
    input_yamls: list[str],
    db: JobDB,
) -> list[InsertedJob]:
    """Submit CWL jobs: store workflow once, create one job per input YAML."""
    workflow_id = compute_workflow_id(cwl_yaml)

    # INSERT IF NOT EXISTS — idempotent, content-addressed
    await db.insert_workflow(workflow_id, cwl_yaml, persistent=False)

    task = parse_cwl(cwl_yaml)
    job_hint = JobHint.from_cwl(task)

    if job_hint.schema_version not in SUPPORTED_SCHEMA_VERSIONS:
        raise ValueError(
            f"Unsupported dirac:Job schema_version '{job_hint.schema_version}'. "
            f"Supported: {SUPPORTED_SCHEMA_VERSIONS}"
        )

    inserted = []
    for input_yaml in input_yamls:
        inputs = parse_inputs(input_yaml) if input_yaml else None
        workflow_params = json.loads(input_yaml) if input_yaml else None

        # Generate JDL for transition period
        jdl = cwl_to_jdl(task, job_hint, inputs)

        # Submit via existing pipeline
        jobs = await submit_jdl_jobs([jdl])

        # Set workflow reference + immutable params on job row
        for job in jobs:
            await db.set_workflow_ref(
                job.job_id,
                workflow_id=workflow_id,
                workflow_params=workflow_params,
            )
        inserted.extend(jobs)

    return inserted


def cwl_to_jdl(
    task: CommandLineTool | Workflow | ExpressionTool,
    job_hint: JobHint,
    inputs: JobInputModel | None,
) -> str:
    """Convert a CWL task with dirac:Job hint into a JDL string.

    This is a transition-period function — once JDL is retired,
    job attributes are populated directly from the hint + CWL.
    """
    jdl_fields = {
        "Executable": "dirac-cwl-exec",
        "JobType": job_hint.type,
        "Priority": job_hint.priority,
        "LogLevel": job_hint.log_level,
    }

    if job_hint.cpu_work:
        jdl_fields["CPUTime"] = job_hint.cpu_work
    if job_hint.platform:
        jdl_fields["Platform"] = job_hint.platform

    # Derive JobName from CWL label/id
    task_label = getattr(task, "label", None)
    task_id = getattr(task, "id", None)
    if task_label:
        jdl_fields["JobName"] = task_label
    elif task_id and task_id != ".":
        jdl_fields["JobName"] = task_id.split("#")[-1].split("/")[-1]

    # Extract from CWL requirements (standard CWL, not dirac:Job)
    tags = set(job_hint.tags or [])
    for req in (getattr(task, "requirements", None) or []):
        if isinstance(req, ResourceRequirement):
            if req.coresMin:
                jdl_fields["MinNumberOfProcessors"] = int(req.coresMin)
            if req.coresMax:
                jdl_fields["MaxNumberOfProcessors"] = int(req.coresMax)
            if req.ramMin:
                jdl_fields["MinRAM"] = int(req.ramMin)
            if req.ramMax:
                jdl_fields["MaxRAM"] = int(req.ramMax)
        elif isinstance(req, CUDARequirement):
            tags.add("GPU")
        elif isinstance(req, MPIRequirement):
            raise NotImplementedError(
                "MPIRequirement is not yet supported for DIRAC CWL jobs"
            )

    # Auto-derive processor tags
    min_proc = jdl_fields.get("MinNumberOfProcessors", 1)
    max_proc = jdl_fields.get("MaxNumberOfProcessors")
    if min_proc and min_proc > 1:
        tags.add("MultiProcessor")
    if min_proc and max_proc and min_proc == max_proc:
        tags.add(f"{min_proc}Processors")

    if tags:
        jdl_fields["Tags"] = list(tags)

    # Sites
    if job_hint.sites:
        jdl_fields["Site"] = job_hint.sites
    if job_hint.banned_sites:
        jdl_fields["BannedSites"] = job_hint.banned_sites

    if job_hint.group:
        jdl_fields["JobGroup"] = job_hint.group

    # Resolve I/O from CWL input/output source IDs
    cwl_input_ids = {_extract_id(inp.id): inp for inp in (task.inputs or [])}
    cwl_output_ids = {_extract_id(out.id): out for out in (task.outputs or [])}

    # InputSandbox
    if job_hint.input_sandbox:
        sandbox_files = []
        for ref in job_hint.input_sandbox:
            _validate_cwl_id(ref.source, cwl_input_ids, "input", ["File", "File[]"])
            if inputs and ref.source in inputs.cwl:
                sandbox_files.extend(_extract_file_paths(inputs.cwl[ref.source]))
        if sandbox_files:
            jdl_fields["InputSandbox"] = sandbox_files

    # InputData
    if job_hint.input_data:
        lfns = []
        for ref in job_hint.input_data:
            _validate_cwl_id(ref.source, cwl_input_ids, "input", ["File", "File[]"])
            if inputs and ref.source in inputs.cwl:
                lfns.extend(_extract_lfn_paths(inputs.cwl[ref.source]))
        if lfns:
            jdl_fields["InputData"] = lfns

    # OutputSandbox
    if job_hint.output_sandbox:
        sandbox_outputs = []
        for ref in job_hint.output_sandbox:
            _validate_cwl_id(ref.source, cwl_output_ids, "output", ["File", "File[]"])
            out = cwl_output_ids[ref.source]
            if hasattr(out, "outputBinding") and out.outputBinding:
                sandbox_outputs.append(out.outputBinding.glob)
        if sandbox_outputs:
            jdl_fields["OutputSandbox"] = sandbox_outputs

    # OutputData (per-output SE and path)
    if job_hint.output_data:
        output_files = []
        all_ses = set()
        for entry in job_hint.output_data:
            _validate_cwl_id(entry.source, cwl_output_ids, "output", ["File", "File[]"])
            out = cwl_output_ids[entry.source]
            if hasattr(out, "outputBinding") and out.outputBinding:
                output_files.append(out.outputBinding.glob)
            all_ses.update(entry.output_se)
        if output_files:
            jdl_fields["OutputData"] = output_files
            jdl_fields["OutputPath"] = job_hint.output_data[0].output_path
            jdl_fields["OutputSE"] = list(all_ses)

    return format_as_jdl(jdl_fields)


def _extract_id(cwl_id: str) -> str:
    """Extract short ID from CWL full URI (e.g., 'file.cwl#input1' → 'input1')."""
    return cwl_id.split("#")[-1].split("/")[-1]

4. Changes to DIRAC Worker-Side Execution

Current flow (__createCWLJobWrapper):

git clone dirac-cwl → install pixi → download sandbox (gets job.json) → run wrapper

New flow:

Fetch CWL from workflows table (via diracX API) → read workflow_params from Jobs row → write job.json → run job wrapper (via importlib.resources)

Since dirac-cwl is installed as a package in the diracX environment, the job wrapper template is accessed via importlib.resources — no git clone or pixi install needed.

Changes in __createCWLJobWrapper in Utils.py:

1. Accept jobParams as a parameter (already available in createJobWrapper)
2. Fetch CWL definition from diracX API using workflow_id from the job
3. Read workflow_params (per-job input parameters) from the job
4. Write CWL + params to local files (task.cwl, params.json)
5. Remove the git clone, pixi install, and dirac-wms-job-get-input steps
6. Load the job wrapper template via importlib.resources.files("dirac_cwl.job")
7. InputSandbox is still used for actual input files — just not for the workflow definition

5. Pydantic Models (in diracX)

These models live in diracX (migrated from dirac-cwl). The old SchedulingHint and ExecutionHooksHint classes are removed — there is no backward compatibility layer.

class IOSource(BaseModel):
    """Reference to a CWL input or output by its ID."""
    source: str             # CWL input/output ID
    path: str | None = None # relative path within job working directory (input_sandbox only)


class OutputDataEntry(BaseModel):
    """Output data entry with per-output SE and LFN path."""
    source: str                    # CWL output ID
    output_path: str               # LFN destination path
    output_se: list[str] = ["SE-USER"]


class JobHint(BaseModel, Hint):
    """Unified DIRAC-specific hint for job scheduling and I/O.

    Resource requirements (cores, RAM) are expressed via standard CWL
    requirements, not in this hint.

    Execution hooks are determined automatically by `type`, not
    configured by the submitter.

    I/O fields reference CWL input/output IDs via `source:` syntax,
    consistent with CWL's own referencing conventions.
    """
    schema_version: str  # required, e.g. "1.0"

    # Scheduling (DIRAC-specific, no CWL equivalent)
    priority: int = 5
    cpu_work: int | None = None   # HS06-seconds → JDL CPUTime
    platform: str | None = None
    sites: list[str] | None = None
    banned_sites: list[str] | None = None
    tags: list[str] | None = None  # merged with auto-derived tags

    # Job metadata
    type: str = "User"
    group: str = ""
    log_level: str = "INFO"

    # I/O: reference CWL input/output IDs via source:
    input_sandbox: list[IOSource] = []
    input_data: list[IOSource] = []
    output_sandbox: list[IOSource] = []
    output_data: list[OutputDataEntry] = []

    @classmethod
    def from_cwl(cls, cwl_object) -> "JobHint":
        hints = getattr(cwl_object, "hints", []) or []
        for hint in hints:
            if hint.get("class") == "dirac:Job":
                data = {k: v for k, v in hint.items() if k != "class"}
                return cls(**data)
        raise ValueError("CWL task is missing required dirac:Job hint")

6. Summary of Changes by Repository

diracx (primary)

Change	Location
JobHint, IOSource, OutputDataEntry models	diracx-core/src/diracx/core/models/
JobSubmissionModel, JobInputModel models	diracx-core/src/diracx/core/models/
workflows table schema	diracx-db/src/diracx/db/sql/job/schema.py
workflow_id + workflow_params columns on Jobs table	diracx-db/src/diracx/db/sql/job/schema.py
New router POST /api/jobs/ (multipart CWL + inputs)	diracx-routers/src/diracx/routers/jobs/submission.py
GET /api/workflows/{workflow_id} endpoint	diracx-routers/src/diracx/routers/jobs/
submit_cwl_jobs() + cwl_to_jdl() logic	diracx-logic/src/diracx/logic/jobs/submission.py
Job wrapper template (migrated from dirac-cwl)	diracx-logic/src/diracx/logic/jobs/
dirac-cwl as dependency	pyproject.toml

dirac-cwl

Change	Location
Remove SchedulingHint + ExecutionHooksHint	execution_hooks/core.py
Remove convert_to_jdl()	job/submission_clients.py
Update JobWrapper.run_job() to resolve hooks from type	job/job_wrapper.py
Update schema with Job definition (versioned)	schemas/dirac-metadata.json
Models migrated to diracX — remove from dirac-cwl	submission_models.py

DIRAC

Change	Location
Modify __createCWLJobWrapper to fetch CWL via diracX API + read workflow_params	WorkloadManagementSystem/Utilities/Utils.py
Remove git clone + pixi install from bash wrapper	Same file
Load job wrapper via importlib.resources	Same file

7. Migration Path

Phase 1 (this work):

• Implement workflows table and workflow_id/workflow_params columns on Jobs
• Implement dirac:Job hint and models in diracX
• Implement POST /api/jobs/ endpoint + GET /api/workflows/{workflow_id}
• Implement cwl_to_jdl() transition shim
• Modify DIRAC __createCWLJobWrapper — remove git clone, use importlib.resources, fetch CWL from API
• Remove old hints from dirac-cwl

Phase 2 (future, per production-plugin-system.md):

• dirac:Transformation hint → transformation submission endpoint
• dirac:Production hint → production orchestration endpoint
• Migrate execution hooks plugin registry to diracX entrypoints

8. Decisions Made

#	Decision	Rationale
D1	Models live in diracX, not dirac-cwl	Natural migration path; dirac-cwl components follow
D2	Endpoint accepts raw CWL YAML + input YAMLs	User-friendly; wf.cwl + input1.yaml + input2.yaml → N jobs
D3	Endpoint is POST /api/jobs/ (not /api/jobs/cwl)	CWL becomes the primary submission format
D4	Platform is CPU architecture (unrelated to containers)	Container support is a separate discussion
D5	No git clone on worker nodes	dirac-cwl installed in diracX; wrapper via importlib.resources
D6	Hook registry starts in dirac-cwl, migrates to diracX entrypoints	Incremental migration
D7	cpu_work (HS06-seconds) in hint; ToolTimeLimit for local cwltool only	Avoids unit ambiguity; DB12 calculates normalization factor
D8	Strict I/O validation — fail fast at submission	Prevents wasting CPU on 10^3-10^5 jobs with bad references
D9	No backward compatibility with old hints	Nothing in production; simplifies implementation
D10	Hints carry schema_version	Forward-compatible evolution
D11	CWL stored in workflows table, not embedded in JDL	Content-addressed (SHA-256), immutable; 10k parametric jobs share 1 workflow row
D12	Per-job params as immutable JSON column on Jobs table	Co-located with job, no extra join; immutable once set
D13	JDL generation is a transition-period shim, not the target	workflows + workflow_params are source of truth; JDL generated for existing DIRAC infra compatibility

9. Open Questions

1. ToolTimeLimit in cwl_utils: Need to verify cwl_utils parses it. If available, it can be used for local execution wall-clock limits alongside cpu_work for DIRAC scheduling. To be investigated.

2. Container support: How to run containerized jobs within DIRAC. Unrelated to Platform (which is CPU architecture). Separate design needed.

3. Multipart API design: Exact multipart field naming and how to handle optional inputs (no input YAML = single job with defaults). Also: should the endpoint support JSON as an alternative to YAML?

4. Input YAML templating: The endpoint naturally supports wf.cwl + N input YAMLs → N jobs. Future extension could support templating (e.g., parameter sweeps) — to be designed separately.

5. Workflow cleanup policy: When persistent = FALSE, what triggers cleanup of orphaned workflow rows? Options: periodic GC that checks for referencing jobs, TTL-based expiry, or tied to existing job cleanup routines.

[aldbr]

CWL CLI Submission Client

Date: 2026-04-07
Package: diracx-cli
Status: Draft

Overview

Two submission commands for different audiences, sharing a single internal CWL submission pipeline:

• dirac job submit cmd — simple mode for end-users. Takes a shell command + sandbox files, generates CWL behind the scenes.
• dirac job submit cwl — power-user mode. Takes a CWL workflow + input files, cwltool-like interface.

Both live under the dirac job submit resource group, following the DiracX CLI Resource Hierarchy pattern: dirac <resource> <action> [sub-action].

CLI Surface

# Simple users — run a command on the grid
dirac job submit cmd "python my_script.py"                        # auto-adds my_script.py to sandbox
dirac job submit cmd "python my_script.py" --sandbox config.json  # extra files not in the command
dirac job submit cmd "source /cvmfs/lhcb.cern.ch/lib/LbEnv && lb-conda default python my_script.py" --sandbox my_script.py

# CWL power users — submit a workflow
dirac job submit cwl workflow.cwl
dirac job submit cwl workflow.cwl inputs.yaml
dirac job submit cwl workflow.cwl inputs.json
dirac job submit cwl workflow.cwl inputs1.yaml inputs2.yaml inputs3.yaml
dirac job submit cwl workflow.cwl sweep.yaml
dirac job submit cwl workflow.cwl --range seed=1000
dirac job submit cwl workflow.cwl --range seed=0:1000
dirac job submit cwl workflow.cwl --range seed=0:1000:2
dirac job submit cwl workflow.cwl base_inputs.yaml --range seed=0:1000

# CWL with CLI inputs (cwltool-style, after -- separator)
dirac job submit cwl workflow.cwl -- --message "hello" --count 42
dirac job submit cwl workflow.cwl --range seed=0:1000 -- --message "hello"
dirac job submit cwl workflow.cwl base_inputs.yaml -- --message "override"

# Job operations
dirac job search --condition "Status eq Running"

Command: dirac job submit cmd

Simple mode. Takes a shell command string, optional sandbox files. Generates a CWL CommandLineTool in-memory and submits through the CWL pipeline.

dirac job submit cmd <command> [--sandbox FILE]... [-y/--yes]

Typer signature

@app.async_command()
async def cmd(
    command: Annotated[str, Argument(help="Shell command to run on the grid")],
    sandbox: Annotated[list[Path], Option("--sandbox", help="Additional local files to ship")] = [],
    yes: Annotated[bool, Option("-y", "--yes", help="Skip confirmation prompt")] = False,
):

Auto-sandbox detection

The command string is tokenized and each token is checked against the local filesystem. Files that meet all of the following criteria are automatically added to the sandbox:

• Exists as a regular file (not a directory)
• Not a symlink
• Relative path (not absolute)

Files specified via --sandbox are added in addition to auto-detected files (deduplicated).

# my_script.py exists locally → auto-sandboxed
dirac job submit cmd "python my_script.py"

# config.json not in the command → must be explicit
dirac job submit cmd "python my_script.py" --sandbox config.json

# auto-detected + explicit, deduplicated
dirac job submit cmd "python my_script.py --config config.json" --sandbox config.json

Generated CWL

The command generates a minimal CWL CommandLineTool:

cwlVersion: v1.2
class: CommandLineTool
label: <derived from command>

hints:
  - class: dirac:Job
    schema_version: "1.0"
    type: User
    input_sandbox:
      - source: sandbox_files

baseCommand: [bash, -c, "<command>"]

inputs:
  - id: sandbox_files
    type: File[]?

outputs: []

$namespaces:
  dirac: "https://diracgrid.org/cwl#"

Then calls the shared submit_cwl() pipeline with this CWL and the sandbox files.

Command: dirac job submit cwl

Power-user CWL mode. cwltool-like positional interface.

dirac job submit cwl <workflow> [inputs]... [--range PARAM=RANGE] [-y/--yes] [-- CWL_ARGS...]

Typer signature

@app.async_command(
    context_settings={"allow_extra_args": True, "ignore_unknown_options": True},
)
async def cwl(
    ctx: typer.Context,
    workflow: Annotated[Path, typer.Argument(help="CWL workflow file (.cwl)")],
    range: Annotated[str | None, typer.Option("--range", help="Parametric range: PARAM=END, PARAM=START:END, or PARAM=START:END:STEP")] = None,
    yes: Annotated[bool, typer.Option("-y", "--yes", help="Skip confirmation prompt")] = False,
):
    # Input files and CLI args (after --) are both in ctx.args
    # Positional file paths appear before any --option flags

Input methods

Three ways to provide CWL inputs, matching cwltool conventions:

1. YAML/JSON files (positional)

dirac job submit cwl workflow.cwl inputs.yaml
dirac job submit cwl workflow.cwl inputs.json

• Each file = one job's input set
• Multi-document YAML (--- separators) = multiple jobs from one file
• Multiple files are concatenated: a.yaml (2 docs) + b.yaml (3 docs) = 5 jobs
• JSON files always contain a single input set

2. CLI arguments (after --)

dirac job submit cwl workflow.cwl -- --message "hello" --count 42 --input_file path/to/file.txt

Arguments after -- are parsed against the CWL workflow's declared inputs, using the same type mapping as cwltool:

CWL type	CLI format	Example
string	--param value	--message "hello"
int	--param N	--count 42
float	--param N.N	--threshold 0.5
boolean	--param (flag)	--verbose
File	--param path	--input_file data.root
File[]	repeated	--files f1.txt --files f2.txt

CLI inputs produce a single job (one input set).

3. Hybrid (file + CLI overrides)

dirac job submit cwl workflow.cwl base_inputs.yaml -- --message "override"

The file is loaded first, then CLI arguments override matching keys. Useful for tweaking a single parameter without editing the file.

--range PARAM=RANGE

Behaves like Python's range(). Accepted formats:

--range PARAM=END              # range(END) → 0, 1, ..., END-1
--range PARAM=START:END        # range(START, END) → START, ..., END-1
--range PARAM=START:END:STEP   # range(START, END, STEP) → START, START+STEP, ...

• PARAM must be declared as an input in the CWL workflow. Validated before submission.
• Server-side expansion: the client sends the range spec, the server creates the jobs.
• Can combine with a base inputs file or CLI args: each job gets base_inputs | {PARAM: index}.
• Requires API extension (new field in the submission request).

Examples:

# 1000 jobs with seed 0-999
dirac job submit cwl workflow.cwl --range seed=1000

# 1000 jobs with seed 0-999 (explicit start)
dirac job submit cwl workflow.cwl --range seed=0:1000

# 500 jobs with even seeds 0, 2, 4, ..., 998
dirac job submit cwl workflow.cwl --range seed=0:1000:2

# With base inputs merged into each job
dirac job submit cwl workflow.cwl base_inputs.yaml --range seed=0:1000

# With CLI base inputs merged into each job
dirac job submit cwl workflow.cwl --range seed=0:1000 -- --message "MC run"

Help text

dirac job submit cwl --help must clearly explain the -- separator:

Usage: dirac job submit cwl WORKFLOW [INPUTS]... [OPTIONS] [-- CWL_ARGS...]

  Submit a CWL workflow to the grid.

  WORKFLOW is a CWL file (.cwl). INPUTS are optional YAML/JSON files
  providing input values (one job per file, or one job per YAML document).

  Workflow inputs can also be passed as CLI arguments after a -- separator.
  These are parsed against the workflow's declared input parameters:

    dirac job submit cwl workflow.cwl -- --message "hello" --count 42

  Combine file inputs with CLI overrides:

    dirac job submit cwl workflow.cwl base.yaml -- --message "override"

Arguments:
  WORKFLOW  CWL workflow file (.cwl)  [required]

Options:
  --range PARAM=RANGE  Parametric: PARAM=END, PARAM=START:END,
                       or PARAM=START:END:STEP (like Python range())
  -y, --yes            Skip confirmation prompt
  --help               Show this message and exit

Sandbox Handling

Local files

Input values of CWL type File that reference local paths are uploaded as input sandboxes:

1. For each job, collect its local File references (paths that are not LFN: or SB:)
2. Group jobs by their set of local file paths
3. Deduplicate: N jobs may produce M unique sandboxes (M <= N, typically M = 1 for parametric)
4. Per unique group: bundle files into a compressed tarball, upload via create_sandbox(), get PFN
5. Rewrite all File references in the affected jobs to SB: PFNs

Submission confirmation

Before uploading sandboxes or submitting, the client displays a summary and prompts for confirmation when more than 100 jobs are being submitted:

Submission summary:
  Workflow:    workflow.cwl (hello-world)
  Jobs:        1,000
  Source:      --range seed=0:1000
  Sandboxes:   1 unique (45 MB)
  LFN inputs:  0

Proceed? [y/N]

The -y / --yes flag skips confirmation for all submissions.

LFN passthrough

File inputs with LFN: prefix are passed through without upload. These are resolved worker-side via the replica map and data management system.

input_files:
  - class: File
    path: "LFN:/lhcb/data/file1.root"

SB: passthrough

File inputs already using SB: references (pre-uploaded sandboxes) are passed through without modification.

Input Validation

Before submission, the client validates:

1. CWL structure: valid CWL document with dirac:Job hint present
2. Required inputs: all required CWL inputs have values provided
3. Type checking: input values match declared CWL types (string, int, File, array, etc.)
4. --range parameter: the named parameter exists in the CWL inputs and is a numeric type
5. Source references: dirac:Job hint source fields reference valid CWL input/output IDs

Validation errors are reported with clear messages before any sandbox upload or submission occurs.

Architecture

Package structure

diracx-cli/src/diracx/cli/
├── __init__.py              # main app, loads subgroups via entry points
├── job/
│   ├── __init__.py          # job app, adds submit subgroup + search
│   ├── search.py            # dirac job search
│   └── submit/
│       ├── __init__.py      # submit app, registers cmd/cwl/jdl subcommands
│       ├── cmd.py           # dirac job submit cmd (simple mode)
│       ├── cwl.py           # dirac job submit cwl (CWL power-user mode)
│       └── jdl.py           # dirac job submit jdl (legacy JDL)
├── _submission/             # shared logic, not a command group
│   ├── __init__.py
│   ├── pipeline.py          # submit_cwl() — core CWL submission pipeline
│   ├── simple.py            # detect_sandbox_files(), generate_cwl()
│   ├── sandbox.py           # sandbox scanning, dedup, upload
│   ├── inputs.py            # CWL input parsing: YAML, JSON, CLI args, merge
│   └── confirm.py           # submission confirmation prompt
├── executor/                # dirac-cwl-run entry point
└── internal/                # legacy commands (hidden)

Shared pipeline

Both commands converge on a single submit_cwl() function in _submission/pipeline.py:

dirac job submit cmd "cmd" --sandbox ...  →  generate_cwl() + submit_cwl()
dirac job submit cwl wf.cwl inputs ...    →  submit_cwl()

submit_cwl() handles: input parsing, validation, sandbox processing, confirmation prompt, API call, result reporting.

Entry points (pyproject.toml)

[project.scripts]
dirac = "diracx.cli:app"
dirac-cwl-run = "diracx.cli.executor.__main__:cli"

[project.entry-points."diracx.cli"]
job = "diracx.cli.job:app"

[project.entry-points."diracx.cli.hidden"]
internal = "diracx.cli.internal:app"

The job entry point registers the dirac job subgroup. submit is a nested AsyncTyper under job, with cmd, cwl, and jdl as its subcommands.

Submission Flow

1. Parse CWL
   - cmd: generate CWL from command string + sandbox files
   - cwl: read CWL file
   - Validate structure, extract dirac:Job hint, extract input schema

2. Parse inputs
   - cmd: no user inputs (sandbox files handled via auto-detect + --sandbox)
   - cwl: YAML/JSON files, CLI args (after --), or hybrid
   - Apply --range if present
   - Result: list of input dicts (one per job)

3. Validate inputs against CWL schema

4. Sandbox processing
   - Scan all input dicts for local File references
   - Group jobs by unique file set
   - Confirm if > 100 jobs (unless -y); warn if egress > 10 GB
   - Upload unique sandboxes via create_sandbox()
   - Rewrite File references to SB: PFNs

5. Submit
   - POST /api/jobs/ with {workflow: cwl_yaml, inputs: [...]}
   - For --range: send range spec for server-side expansion

6. Report
   - Print job IDs, workflow ID, job count
   - Format: JSON or rich table based on DIRACX_OUTPUT_FORMAT

API Changes Required

--range support (server-side expansion)

The CWLJobSubmission model needs extension to accept a range spec:

class CWLJobSubmission(BaseModel):
    workflow: str
    inputs: list[dict] = []
    range_param: str | None = None    # CWL input parameter name
    range_start: int | None = None    # default 0
    range_end: int | None = None
    range_step: int | None = None     # default 1
    base_inputs: dict | None = None   # Merged with each generated input

When range_param is set, the server generates jobs for range(range_start, range_end, range_step), each with base_inputs | {range_param: index}.

inputs and range_param are mutually exclusive.

Error Handling

Error	When	Message
No dirac:Job hint	CWL parsing	"CWL workflow missing required dirac:Job hint"
Missing required input	Validation	"Missing required input 'X'. Declared inputs: ..."
--range param not in CWL	Validation	"Parameter 'X' not found in workflow inputs. Declared inputs: ..."
--range param not numeric	Validation	"Parameter 'X' must be int or long type for --range, got 'string'"
--range invalid format	Validation	"Invalid range format. Use PARAM=END, PARAM=START:END, or PARAM=START:END:STEP"
Unknown CWL arg after --	CLI parsing	"Unknown input '--foo'. Declared inputs: message, count, input_file"
Sandbox > 100 MB	Upload (server)	"Sandbox exceeds 100 MB server limit. Use LFN references for large files."
Authentication failure	Submission	Handled by existing AsyncTyper error handling

Out of Scope

• CWL-native scatter + chunking: server-side support for class: Workflow with CWL scatter semantics. Significant complexity — tracked as a separate task.
• Array input chunking (--chunk): client-side splitting of array inputs into per-job chunks. Complex future enhancement, ties in input data resolution complexities.
• --conda-lock: shipping a conda lock file for worker environment setup. Requires infrastructure work not yet available.
• Output sandbox download: commands for retrieving job outputs. Separate feature.
• Additional CWL subcommands: validate, show, etc. Future enhancement.