Compute

Bounded context for scheduling and tracking compute tasks on external schedulers (Slurm, local runner, Kubernetes, …). The domain vocabulary is scheduler-neutral: TaskRunSucceeded / TaskRunFailed never mention sbatch or squeue. Concrete schedulers live exclusively in the driven adapters.

Overview

Two aggregates

Entity	Purpose
TaskSpec	Immutable specification of what to run — command, resources, retry policy, labels. Carries no runtime state.
TaskRun	A concrete execution of a TaskSpec. Evolves via apply_task_event() through a well-defined lifecycle.

Lifecycle

CREATED
  │  TaskDispatchRequested
  ▼
DISPATCH_REQUESTED
  │  TaskSubmitted
  ▼
SUBMITTED ──► QUEUED ──► RUNNING ──► SUCCEEDED
                                 └──► FAILED / CANCELLED / TIMED_OUT
                                       / OUT_OF_MEMORY / INFRA_FAILED / PREEMPTED

UNKNOWN is used when the scheduler reports a state that cannot be mapped to any of the above. TaskRun.is_terminal returns True for all terminal statuses.

Event flow

TaskDispatchRequested
        │
        ▼
TaskDispatchRequestedHandler
        │  calls
        ▼
JobDispatcher.submit(DispatchRequest)
        │
        ▼
SlurmJobDispatcher → sbatch → ExternalJobRef
        │
        ▼
TaskSubmitted(external_ref=ExternalJobRef(...))

Later — SlurmJobDispatcher.poll() → squeue / sacct
        │
        ▼
JobObservation(state=RUNNING | SUCCEEDED | FAILED | ...)
        │
        ▼
JobObservationHandler → domain_event_from_observation()
        │
        ▼
TaskRunStarted / TaskRunSucceeded / TaskRunFailed

Scheduler vocabulary mapping

Adapters translate native scheduler states before they cross into the domain:

Slurm state	ExternalJobState	Domain event
PD / PENDING	QUEUED	TaskQueued
R / RUNNING, CG	RUNNING	TaskRunStarted
CD / COMPLETED	SUCCEEDED	TaskRunSucceeded
F / FAILED	FAILED	TaskRunFailed(FAILED)
CA / CANCELLED	CANCELLED	TaskRunFailed(CANCELLED)
TO / TIMEOUT	TIMED_OUT	TaskRunFailed(TIMED_OUT)
OOM / OUT_OF_MEMORY	OUT_OF_MEMORY	TaskRunFailed(OUT_OF_MEMORY)
NF / BF / NODE_FAIL	INFRA_FAILED	TaskRunFailed(INFRA_FAILED)
PR / PREEMPTED	PREEMPTED	TaskRunFailed(PREEMPTED)

Design rules

• apply_task_event() is idempotent: re-applying an already-applied event returns the run unchanged.
• Events target a different run_id are silently ignored.
• Events arriving after the run is terminal are silently ignored.
• JobObservation is an integration DTO, not a domain event. Only domain_event_from_observation() promotes it to an event.
• ExternalJobRef is intentionally generic — it never says "Slurm".

Usage example

from pathlib import Path
from iris.domain.compute.entities import TaskRun, TaskSpec
from iris.domain.compute.events import TaskDispatchRequested
from iris.domain.compute.handlers import JobObservationHandler, TaskDispatchRequestedHandler
from iris.domain.compute.objects import CommandSpec, DispatchHints, ResourceRequest
from iris.adapters.driven.dispatchers.slurm import SlurmJobDispatcher
from iris.adapters.driven.repositories.redis_compute import make_redis_task_run_repository

spec = TaskSpec(
    name="vcf_norm_chr22",
    command=CommandSpec(
        argv=["bcftools", "norm", "-f", "/ref/GRCh38.fa", "-Oz", "-o", "out.vcf.gz", "in.vcf.gz"],
        workdir=Path("/scratch/project/run_001"),
        env={"OMP_NUM_THREADS": "4"},
    ),
    resources=ResourceRequest(ranks=1, cpus_per_rank=4, memory_mb=16_000, time_limit_s=7200),
    hints=DispatchHints(queue="short", project="bioinfo"),
    labels={"workflow": "prs-preprocess", "chrom": "22"},
)

repo = make_redis_task_run_repository()
dispatcher = SlurmJobDispatcher(submit_dir=Path("/scratch/project/.dispatch"))

run = TaskRun(spec=spec)
repo.save(run)

# Dispatch
event = TaskDispatchRequested(task_id=spec.task_id, run_id=run.run_id)
new_events = TaskDispatchRequestedHandler(repo=repo, dispatcher=dispatcher).handle(event)

# Poll (from a poller / cron)
run = repo.get(run.run_id)
for obs in dispatcher.poll([run.external_ref]):
    JobObservationHandler(repo=repo).handle(obs)

Entities

iris.domain.compute.entities

Compute domain entities: TaskSpec (intent) and TaskRun (execution).

TaskSpec

Immutable specification of what to run.

A TaskSpec carries no runtime state; it defines intent. Each concrete execution is a :class:TaskRun.

TaskRun

A concrete execution of a :class:TaskSpec.

Immutable like every domain object: state changes are produced with attrs.evolve() inside :func:~iris.domain.compute.services.apply_task_event, which keeps event handling idempotent and reproducible.

is_terminal property

is_terminal

Return True if the run has reached a terminal status.

Objects

iris.domain.compute.objects

Compute domain value objects: enums, resource requests, command specs, and helpers.

This module models scheduler-neutral concepts for running compute tasks. The domain vocabulary (TaskStatus, ExternalJobState) is deliberately generic: concrete schedulers such as Slurm live in the driven adapters and translate their native states into these domain terms.

TaskStatus

Bases: str, Enum

Lifecycle status of a :class:~iris.domain.compute.entities.TaskRun.

ExternalJobState

Bases: str, Enum

Scheduler-neutral state of an external job as reported by a dispatcher.

Adapters translate their native scheduler states (e.g. Slurm PD, R, CD) into these values inside a :class:JobObservation.

CommandSpec

An immutable command to execute: argv, optional workdir, and environment.

ResourceRequest

Scheduler-neutral compute resource request for a task.

ranks is the domain-level process/rank count. In Slurm this usually maps to --ntasks.

DispatchHints

Scheduler-neutral scheduling hints.

A Slurm adapter may map queue → --partition, project → --account, and quality → --qos. extra is an escape hatch for rare, adapter-scoped options (e.g. ("slurm.exclusive", "true")) and should be used sparingly.

RetryPolicy

Retry policy governing when a failed task run should be retried.

ExternalJobRef

Opaque reference to a job in an external scheduler.

Generic by design: the domain never names it SlurmJobRef. The dispatcher field identifies which adapter owns the reference.

ExitStatus

Exit information for a finished job: exit code, signal, and/or message.

JobObservation

Integration DTO describing an external job's observed state at a point in time.

Not a domain event. Adapters (e.g. SlurmJobDispatcher) build this from squeue / sacct output; the observation handler translates it into domain events.

utcnow

utcnow()

Return the current time as a timezone-aware UTC datetime.

Source code in src/iris/domain/compute/objects.py

def utcnow() -> datetime:
    """Return the current time as a timezone-aware UTC datetime."""
    return datetime.now(timezone.utc)

new_id

new_id(prefix)

Generate a unique identifier of the form {prefix}_{hex}.

Source code in src/iris/domain/compute/objects.py

def new_id(prefix: str) -> str:
    """Generate a unique identifier of the form ``{prefix}_{hex}``."""
    return f"{prefix}_{uuid.uuid4().hex}"

as_str_tuple

as_str_tuple(value)

Convert an iterable to a tuple of strings, or empty tuple when None.

Source code in src/iris/domain/compute/objects.py

def as_str_tuple(value: Iterable[object] | None) -> tuple[str, ...]:
    """Convert an iterable to a tuple of strings, or empty tuple when None."""
    if value is None:
        return ()
    return tuple(str(x) for x in value)

as_kv_tuple

as_kv_tuple(value)

Convert a mapping or pair-iterable to a sorted tuple of (str, str) pairs.

Source code in src/iris/domain/compute/objects.py

def as_kv_tuple(
    value: Mapping[str, object] | Iterable[tuple[str, object]] | None,
) -> tuple[KV, ...]:
    """Convert a mapping or pair-iterable to a sorted tuple of ``(str, str)`` pairs."""
    if value is None:
        return ()
    items = value.items() if isinstance(value, Mapping) else value
    return tuple(sorted((str(k), str(v)) for k, v in items))

positive_int

positive_int(instance, attribute, value)

Validate that value is an integer greater than or equal to 1.

Source code in src/iris/domain/compute/objects.py

def positive_int(instance: object, attribute: attrs.Attribute[int], value: int) -> None:
    """Validate that ``value`` is an integer greater than or equal to 1."""
    if not isinstance(value, int) or value < 1:
        raise ValueError(f"{attribute.name} must be an integer >= 1, got {value!r}")

non_negative_int

non_negative_int(instance, attribute, value)

Validate that value is an integer greater than or equal to 0.

Source code in src/iris/domain/compute/objects.py

def non_negative_int(instance: object, attribute: attrs.Attribute[int], value: int) -> None:
    """Validate that ``value`` is an integer greater than or equal to 0."""
    if not isinstance(value, int) or value < 0:
        raise ValueError(f"{attribute.name} must be an integer >= 0, got {value!r}")

optional_positive_int

optional_positive_int(instance, attribute, value)

Validate that value is None or an integer greater than or equal to 1.

Source code in src/iris/domain/compute/objects.py

def optional_positive_int(
    instance: object, attribute: attrs.Attribute[int | None], value: int | None
) -> None:
    """Validate that ``value`` is None or an integer greater than or equal to 1."""
    if value is not None and (not isinstance(value, int) or value < 1):
        raise ValueError(f"{attribute.name} must be None or an integer >= 1, got {value!r}")

non_empty_argv

non_empty_argv(instance, attribute, value)

Validate that a command argv tuple is not empty.

Source code in src/iris/domain/compute/objects.py

def non_empty_argv(
    instance: object, attribute: attrs.Attribute[tuple[str, ...]], value: tuple[str, ...]
) -> None:
    """Validate that a command argv tuple is not empty."""
    if not value:
        raise ValueError("command argv must not be empty")

Events

iris.domain.compute.events

Domain events for the compute bounded context.

Events are informative: they carry only the relevant identifiers plus event metadata (occurred_at, correlation and causation IDs). They never embed a snapshot of the aggregate. The domain speaks in terms of TaskRunSucceeded / TaskRunFailed; scheduler-specific states (Slurm PD, R, CD, …) never appear here — adapters translate them before events are emitted.

DomainEvent

Base class for compute domain events.

TaskRegistered

Bases: DomainEvent

Emitted when a task specification is registered.

TaskDispatchRequested

Bases: DomainEvent

Emitted when dispatch of a task run is requested.

TaskSubmitted

Bases: DomainEvent

Emitted when a task run has been submitted to an external scheduler.

TaskQueued

Bases: DomainEvent

Emitted when a submitted task run is queued by the scheduler.

TaskRunStarted

Bases: DomainEvent

Emitted when a task run starts executing.

TaskRunSucceeded

Bases: DomainEvent

Emitted when a task run completes successfully.

TaskRunFailed

Bases: DomainEvent

Emitted when a task run reaches a terminal failure state.

Services

iris.domain.compute.services

Compute domain services: aggregate transitions and observation translation.

These are pure functions over immutable aggregates. apply_task_event folds a domain event onto a :class:TaskRun (event-sourcing style), and is idempotent: re-applying an already-applied event returns the run unchanged. domain_event_from_observation maps a scheduler-neutral :class:JobObservation into the appropriate domain event.

apply_task_event

apply_task_event(run, event)

Return a new :class:TaskRun with event applied.

Idempotent and guarded: events targeting a different run_id or arriving after the run is terminal are ignored (the run is returned unchanged).

PARAMETER	DESCRIPTION
run	The current run aggregate. TYPE: TaskRun
event	A compute domain event to fold onto the run. TYPE: DomainEvent

RETURNS	DESCRIPTION
TaskRun	The evolved run, or the same instance if the event is a no-op.

RAISES	DESCRIPTION
ValueError	If a :class:TaskRunFailed carries a non-failure terminal status.

Source code in src/iris/domain/compute/services.py

def apply_task_event(run: TaskRun, event: DomainEvent) -> TaskRun:
    """Return a new :class:`TaskRun` with ``event`` applied.

    Idempotent and guarded: events targeting a different ``run_id`` or arriving
    after the run is terminal are ignored (the run is returned unchanged).

    Args:
        run: The current run aggregate.
        event: A compute domain event to fold onto the run.

    Returns:
        The evolved run, or the same instance if the event is a no-op.

    Raises:
        ValueError: If a :class:`TaskRunFailed` carries a non-failure terminal
            status.
    """
    if isinstance(event, TaskDispatchRequested):
        if event.run_id != run.run_id:
            return run
        if run.status in {TaskStatus.CREATED, TaskStatus.READY}:
            return attrs.evolve(run, status=TaskStatus.DISPATCH_REQUESTED, version=run.version + 1)
        return run

    if isinstance(event, TaskSubmitted):
        if event.run_id != run.run_id:
            return run
        if run.external_ref == event.ref and run.status == TaskStatus.SUBMITTED:
            return run
        return attrs.evolve(
            run,
            status=TaskStatus.SUBMITTED,
            external_ref=event.ref,
            submitted_at=event.occurred_at,
            version=run.version + 1,
        )

    if isinstance(event, TaskQueued):
        if event.run_id != run.run_id or run.is_terminal:
            return run
        if run.status in {TaskStatus.SUBMITTED, TaskStatus.DISPATCH_REQUESTED}:
            return attrs.evolve(
                run, status=TaskStatus.QUEUED, reason=event.reason, version=run.version + 1
            )
        return run

    if isinstance(event, TaskRunStarted):
        if event.run_id != run.run_id or run.is_terminal:
            return run
        if run.status != TaskStatus.RUNNING:
            return attrs.evolve(
                run,
                status=TaskStatus.RUNNING,
                started_at=event.occurred_at,
                version=run.version + 1,
            )
        return run

    if isinstance(event, TaskRunSucceeded):
        if event.run_id != run.run_id or run.is_terminal:
            return run
        return attrs.evolve(
            run,
            status=TaskStatus.SUCCEEDED,
            exit_status=event.exit_status,
            finished_at=event.occurred_at,
            version=run.version + 1,
        )

    if isinstance(event, TaskRunFailed):
        if event.run_id != run.run_id or run.is_terminal:
            return run
        allowed_failure_statuses = TERMINAL_STATUSES - {TaskStatus.SUCCEEDED}
        if event.terminal_status not in allowed_failure_statuses:
            raise ValueError(f"invalid terminal failure status: {event.terminal_status}")
        return attrs.evolve(
            run,
            status=event.terminal_status,
            exit_status=event.exit_status,
            reason=event.reason,
            finished_at=event.occurred_at,
            version=run.version + 1,
        )

    return run

domain_event_from_observation

domain_event_from_observation(run, obs)

Translate a :class:JobObservation into a domain event, or None.

Returns None when the observation does not apply (mismatched reference, terminal run, or a state that implies no transition from the run's current status).

PARAMETER	DESCRIPTION
run	The run the observation is believed to belong to. TYPE: TaskRun
obs	The scheduler-neutral observation. TYPE: JobObservation

RETURNS	DESCRIPTION
DomainEvent | None	The corresponding domain event, or None if no transition applies.

Source code in src/iris/domain/compute/services.py

def domain_event_from_observation(run: TaskRun, obs: JobObservation) -> DomainEvent | None:
    """Translate a :class:`JobObservation` into a domain event, or None.

    Returns None when the observation does not apply (mismatched reference,
    terminal run, or a state that implies no transition from the run's current
    status).

    Args:
        run: The run the observation is believed to belong to.
        obs: The scheduler-neutral observation.

    Returns:
        The corresponding domain event, or None if no transition applies.
    """
    if run.external_ref != obs.ref or run.is_terminal:
        return None

    task_id = run.spec.task_id
    run_id = run.run_id
    correlation_id = run.run_id

    if obs.state == ExternalJobState.QUEUED and run.status in {
        TaskStatus.SUBMITTED,
        TaskStatus.DISPATCH_REQUESTED,
    }:
        return TaskQueued(
            task_id=task_id, run_id=run_id, correlation_id=correlation_id, reason=obs.reason
        )

    if obs.state == ExternalJobState.RUNNING and run.status != TaskStatus.RUNNING:
        return TaskRunStarted(task_id=task_id, run_id=run_id, correlation_id=correlation_id)

    if obs.state == ExternalJobState.SUCCEEDED:
        return TaskRunSucceeded(
            task_id=task_id,
            run_id=run_id,
            correlation_id=correlation_id,
            exit_status=obs.exit_status,
        )

    if obs.state in _FAILURE_MAP:
        return TaskRunFailed(
            task_id=task_id,
            run_id=run_id,
            correlation_id=correlation_id,
            terminal_status=_FAILURE_MAP[obs.state],
            exit_status=obs.exit_status,
            reason=obs.reason or obs.raw_state,
        )

    return None

Handlers

iris.domain.compute.handlers

Compute domain event handlers.

Handlers orchestrate the interaction between the aggregate, the repository, and the :class:JobDispatcher port. They translate between the domain vocabulary and the dispatcher boundary without leaking scheduler-specific concepts into the domain.

TaskDispatchRequestedHandler

Handle :class:TaskDispatchRequested by submitting to the dispatcher.

handle

handle(event)

Submit the run to the dispatcher and emit :class:TaskSubmitted.

Idempotent: if the run already carries an external_ref it is not submitted again. Returns the newly produced events (empty if none).

Source code in src/iris/domain/compute/handlers.py

def handle(self, event: DomainEvent) -> list[DomainEvent]:
    """Submit the run to the dispatcher and emit :class:`TaskSubmitted`.

    Idempotent: if the run already carries an ``external_ref`` it is not
    submitted again. Returns the newly produced events (empty if none).
    """
    if not isinstance(event, TaskDispatchRequested):
        return []

    run = self.repo.get(event.run_id)
    if run is None:
        return []

    run = apply_task_event(run, event)
    self.repo.save(run)

    # Idempotency guard: do not submit twice.
    if run.external_ref is not None:
        return []

    ref = self.dispatcher.submit(DispatchRequest.from_run(run))

    submitted = TaskSubmitted(
        task_id=run.spec.task_id,
        run_id=run.run_id,
        ref=ref,
        correlation_id=event.correlation_id or event.run_id,
        causation_id=event.event_id,
    )

    self.repo.save(apply_task_event(run, submitted))
    return [submitted]

JobObservationHandler

Translate a :class:JobObservation into domain events and persist them.

handle

handle(obs)

Route obs to its run, apply the derived event, and return it.

Returns an empty list when the observation does not match any run or implies no transition.

Source code in src/iris/domain/compute/handlers.py

def handle(self, obs: JobObservation) -> list[DomainEvent]:
    """Route ``obs`` to its run, apply the derived event, and return it.

    Returns an empty list when the observation does not match any run or
    implies no transition.
    """
    run = self.repo.find_by_external_ref(obs.ref)
    if run is None:
        return []

    event = domain_event_from_observation(run, obs)
    if event is None:
        return []

    self.repo.save(apply_task_event(run, event))
    return [event]

Repositories

iris.domain.compute.repositories

Repository interface for the compute domain.

TaskRunRepository

Bases: Repository[TaskRun]

Repository for :class:TaskRun aggregates.

Extends the generic read contract (get) with persistence and a lookup by external scheduler reference, used by the observation handler to route incoming :class:~iris.domain.compute.objects.JobObservation DTOs back to their originating run.

get abstractmethod

get(identifier)

Return the run with run_id == identifier, or None if not found.

Source code in src/iris/domain/compute/repositories.py

@abstractmethod
def get(self, identifier: str) -> TaskRun | None:
    """Return the run with ``run_id == identifier``, or None if not found."""

save abstractmethod

save(entity)

Persist entity and return the saved version.

Source code in src/iris/domain/compute/repositories.py

@abstractmethod
def save(self, entity: TaskRun) -> TaskRun:
    """Persist ``entity`` and return the saved version."""

find_by_external_ref abstractmethod

find_by_external_ref(ref)

Return the run whose external_ref matches ref, or None.

Source code in src/iris/domain/compute/repositories.py

@abstractmethod
def find_by_external_ref(self, ref: ExternalJobRef) -> TaskRun | None:
    """Return the run whose ``external_ref`` matches ``ref``, or None."""

Ports

Job dispatcher

iris.domain.compute.ports.driven.job_dispatcher

Driven port for dispatching compute jobs to an external scheduler.

This port is the boundary with concrete schedulers (Slurm, a local runner, Kubernetes, …). It knows nothing about any specific backend: adapters translate native scheduler states into the domain's :class:ExternalJobState / :class:JobObservation vocabulary.

DispatchRequest

Flattened, scheduler-neutral request handed to a :class:JobDispatcher.

from_run classmethod

from_run(run)

Build a DispatchRequest from a :class:TaskRun.

Source code in src/iris/domain/compute/ports/driven/job_dispatcher.py

@classmethod
def from_run(cls, run: TaskRun) -> DispatchRequest:
    """Build a DispatchRequest from a :class:`TaskRun`."""
    return cls(
        task_id=run.spec.task_id,
        run_id=run.run_id,
        name=run.spec.name,
        command=run.spec.command,
        resources=run.spec.resources,
        hints=run.spec.hints,
    )

JobDispatcher

Bases: ABC

Abstract port for submitting, cancelling, and polling external jobs.

Concrete implementations (e.g. SlurmJobDispatcher) live in the driven adapters. The name attribute identifies the dispatcher and is stamped onto every :class:ExternalJobRef it produces.

submit abstractmethod

submit(request)

Submit a job and return a reference to it.

Source code in src/iris/domain/compute/ports/driven/job_dispatcher.py

@abstractmethod
def submit(self, request: DispatchRequest) -> ExternalJobRef:
    """Submit a job and return a reference to it."""

cancel abstractmethod

cancel(ref)

Cancel the job identified by ref.

Source code in src/iris/domain/compute/ports/driven/job_dispatcher.py

@abstractmethod
def cancel(self, ref: ExternalJobRef) -> None:
    """Cancel the job identified by ``ref``."""

poll abstractmethod

poll(refs)

Poll the given references and yield their current observations.

Source code in src/iris/domain/compute/ports/driven/job_dispatcher.py

@abstractmethod
def poll(self, refs: Iterable[ExternalJobRef]) -> Iterable[JobObservation]:
    """Poll the given references and yield their current observations."""