Task

def get_arguments(self) -> list[dict] | None:
    return self.arguments

def get_kernels(self) -> list[ir.Method]:
    return self.kernels

def get_metadata(self) -> list[dict] | None:
    return self.metadata

def get_num_shots(self) -> list[int]:
    if isinstance(self.num_shots, int):
        return [self.num_shots] * self.num_subtasks
    return self.num_shots

def summary(self) -> str:
    msg = "=" * 60 + "\n"
    msg += "DRY RUN -- NO PROGRAM WAS ACTUALLY SUBMITTED FOR EXECUTION\n"
    msg += f"Would now submit a task containing {len(self.kernels)} programs:\n"
    for i, kernel in enumerate(self.kernels):
        kernel_print = f"{kernel.sym_name}("
        if self.arguments is not None:
            for arg in self.arguments[i]:
                kernel_print += f"{arg}, "
        kernel_print += ")"
        shots = (
            self.num_shots if isinstance(self.num_shots, int) else self.num_shots[i]
        )
        msg += f"  * {kernel_print} - {shots} shots\n"
    msg += "Set dry_run=False to actually execute the programs.\n"
    msg += "=" * 60 + "\n"
    return msg

def encode(self, encoded_module: Any, /) -> str | bytes:
    """Encode a Kirin serialization module for `Program.content`."""
    ...

def get_arguments(self) -> list[dict]:
    return self.arguments

def get_kernels(self) -> list[ir.Method]:
    return [self.kernel]

def get_metadata(self) -> list[dict] | None:
    return self.metadata

def get_num_shots(self) -> list[int]:
    if isinstance(self.num_shots, int):
        return [self.num_shots] * self.num_subtasks
    return self.num_shots

def program_index_for_subtask(self, i: int) -> int:
    return 0

def summary(self) -> str:
    msg = "=" * 60 + "\n"
    msg += "DRY RUN -- NO PROGRAM WAS ACTUALLY SUBMITTED FOR EXECUTION\n"
    msg += f"Would now submit a task containing {self.num_subtasks} subtasks.\n"
    msg += f"These subtasks correspond to parameter sets of the kernel {self.kernel.sym_name}.\n"
    msg += "Set dry_run=False to actually execute the parameter scan.\n"
    msg += "=" * 60
    return msg

def get_arguments(self) -> list[dict] | None:
    if self.arguments is not None:
        return [self.arguments]

def get_kernels(self) -> list[ir.Method]:
    return [self.kernel]

def get_metadata(self) -> list[dict] | None:
    if self.metadata is not None:
        return [self.metadata]

def get_num_shots(self) -> list[int]:
    return [self.num_shots]

def summary(self) -> str:
    msg = "=" * 60 + "\n"
    msg += "DRY RUN -- NO PROGRAM WAS ACTUALLY SUBMITTED FOR EXECUTION\n"
    msg += "Would now submit a task containing a single subtask for the kernel:\n"
    if self.arguments is not None:
        formatted_arguments = ", ".join(
            f"{key}={value}" for key, value in self.arguments.items()
        )
    else:
        formatted_arguments = ""
    kernel_print = f"{self.kernel.sym_name}({formatted_arguments})"
    shots = self.num_shots if isinstance(self.num_shots, int) else self.num_shots[0]
    msg += f"  * {kernel_print} - {shots} shots\n"
    msg += "Set dry_run=False to actually execute this kernel.\n"
    msg += "=" * 60
    return msg

def create_task_definition(self) -> TaskDefinition:
    """Build a `TaskDefinition` from this task's kernels and subtasks.

    Override this method directly if your use-case doesn't fit the API
    contract.

    Returns:
        TaskDefinition: Definition ready to be submitted.
    """
    programs = self.programs()

    num_shots = self.get_num_shots()

    subtasks = []
    arguments = self.get_arguments()
    metadata = self.get_metadata()
    for i in range(self.num_subtasks):
        if arguments is not None:
            args = arguments[i]
        else:
            args = None

        if metadata is not None:
            subtask_metadata = TaskMetadata(user_metadata=json.dumps(metadata[i]))
        else:
            subtask_metadata = None

        subtasks.append(
            Subtask(
                program_index=self.program_index_for_subtask(i),
                num_shots=num_shots[i],
                arguments=args,
                subtask_metadata=subtask_metadata,
            )
        )

    program_language_with_version = f"{self.program_language}.v{self.program_language_version.removeprefix('v')}"
    return TaskDefinition(
        program_language=program_language_with_version,
        programs=programs,
        subtasks=subtasks,
        group_id=None,
    )

@abstractmethod
def get_arguments(self) -> list[dict] | None:
    """Return per-subtask argument dictionaries.

    Returns:
        list[dict] | None: One argument dictionary per subtask, or None
            when no arguments are set.
    """
    ...

@abstractmethod
def get_kernels(self) -> list[ir.Method]:
    """Return the kernels used to build the task's programs.

    Returns:
        list[ir.Method]: Kernels in program-index order.
    """
    ...

@abstractmethod
def get_metadata(self) -> list[dict] | None:
    """Return per-subtask metadata dictionaries.

    Returns:
        list[dict] | None: One metadata dictionary per subtask, or None
            when no metadata is set.
    """
    ...

@abstractmethod
def get_num_shots(self) -> list[int]:
    """Return the per-subtask shot counts.

    Returns:
        list[int]: Shot count for each subtask, in subtask order.
    """
    ...

def program_index_for_subtask(self, i: int) -> int:
    """Return the program index used by subtask `i`.

    The default implementation maps each subtask to its own program.
    Parameter-scan tasks override this to reuse a single program.

    Args:
        i (int): Subtask index.

    Returns:
        int: Program index.
    """
    return i

def programs(self) -> list[Program]:
    """Build the program list for the task definition.

    Returns:
        list[Program]: One `Program` per kernel returned by
            `get_kernels`, serialized via `serialize_kernel`.
    """
    kernels = self.get_kernels()
    programs = []
    for kernel in kernels:
        programs.append(Program(content=self.serialize_kernel(kernel)))
    return programs

def run_async(
    self,
    *,
    dry_run: bool,
    storage: StorageBackend | None = None,
    fetch_options: ApiFetchOptions = DEFAULT_FETCH_OPTIONS,
) -> FutureType | None:
    """Validate the task and either dry-run or submit it.

    Keyword Args:
        dry_run (bool): When True, print a summary and return None.
            When False, submit the task and return a future.
        storage (StorageBackend | None): Storage backend that will receive
            the task definition and later fetched shots. When None, a fresh
            `DictStorage` is used (in-memory; not persisted across
            processes). Defaults to None.
        fetch_options (ApiFetchOptions): Pagination and polling options
            attached to the returned future. Defaults to
            `ApiFetchOptions()`.

    Returns:
        FutureType | None: Future attached to the submitted task when
            `dry_run` is False; otherwise None.

    Raises:
        ValueError: If argument or metadata lengths do not match
            `num_subtasks`.
    """
    self.validate_arguments()

    task_def = self.create_task_definition()

    if dry_run:
        print(self.summary())
        return

    return self.submit_task_definition(
        task_definition=task_def,
        storage=storage,
        fetch_options=fetch_options,
    )

def serialize_kernel(self, kernel: ir.Method) -> str:
    """Serialize a kernel into content suitable for the backend.

    The default implementation first converts the kernel to a Kirin
    serialization module using `program_language_version`, then passes
    that module to `kernel_serializer.encode`. Binary serializer output is
    base64-encoded so it can travel through the API's string-valued
    `Program.content` field. String serializer output is returned as
    produced by the serializer.

    Args:
        kernel (ir.Method): Kernel to serialize.

    Returns:
        str: Serialized kernel content for the submitted `Program`.
    """

    encoded_module = kernel.dialects.encode(
        kernel, version=self.program_language_version
    )
    payload = self.kernel_serializer.encode(encoded_module)

    if isinstance(payload, bytes):
        return base64.b64encode(payload).decode("ascii")

    if isinstance(payload, str):
        return payload

    raise TypeError(
        "kernel_serializer.encode must return str or bytes, "
        f"got {type(payload).__name__}"
    )

def submit_task_definition(
    self,
    *,
    task_definition: TaskDefinition,
    storage: StorageBackend | None = None,
    fetch_options: ApiFetchOptions = DEFAULT_FETCH_OPTIONS,
) -> FutureType:
    """Submit a prepared task definition and return a future.

    When the definition does not set a group ID, a task-level group name
    takes precedence over the `~/.qsh` config group (`plugins.tasks.group`,
    then `defaults.group`). The selected name is resolved before
    submission. When neither is set, the group is omitted and QLAM selects
    the backend default group.

    Keyword Args:
        task_definition (TaskDefinition): Task definition to submit.
        storage (StorageBackend | None): Storage backend that will receive
            the task definition. When None, a fresh `DictStorage` is used
            (in-memory; not persisted across processes). Defaults to None.
        fetch_options (ApiFetchOptions): Pagination and polling options
            attached to the returned future. Defaults to
            `ApiFetchOptions()`.

    Returns:
        FutureType: Future attached to the created task ID.

    Raises:
        ValueError: If the backend response is missing a task ID.
    """
    if storage is None:
        storage = DictStorage()

    self.authenticate()

    if task_definition.group_id is None:
        group = self._configured_group()
        if group is not None:
            task_definition = task_definition.model_copy(
                update={"group_id": self._resolve_group_id(group)}
            )

    task_request = TaskCreationRequest(root=task_definition)
    with TasksClient(self.app_context) as tasks_client:
        created_task = self.call_with_auth_refresh(
            lambda: tasks_client.create(body=task_request)
        )

    task_id = created_task.id

    if not task_id:
        raise ValueError(
            f"Couldn't get id of created task {created_task}. Please report this issue!"
        )

    logger.info(f"Submitted task with ID: {task_id}")

    storage.add_task_definition(task_id, task_definition, created_task.created_date)

    return self.future_cls(
        task_id=task_id,
        fetch_options=fetch_options,
        storage=storage,
        context_name=self.context_name,
    )

def summary(self) -> str:
    """Return a human-readable summary printed on dry-run.

    Returns:
        str: Summary describing what would be submitted.
    """
    return f"Would now submit {self.num_subtasks} subtasks"

def validate_arguments(self) -> None:
    """Validate that argument and metadata lengths match subtask count.

    Raises:
        ValueError: If arguments or metadata length differs from
            `num_subtasks`.
    """
    arguments = self.get_arguments()
    if arguments is not None and len(arguments) != self.num_subtasks:
        raise ValueError(
            f"Length mismatch: got {len(arguments)} sets of arguments for {self.num_subtasks} subtasks!"
        )

    metadata = self.get_metadata()
    if metadata is not None and len(metadata) != self.num_subtasks:
        raise ValueError(
            f"Length mismatch: got {len(metadata)} sets of metadata for {self.num_subtasks} subtasks!"
        )

    num_shots = self.get_num_shots()
    if len(num_shots) != self.num_subtasks:
        raise ValueError(
            f"Length mismatch: got {len(num_shots)} shot counts for {self.num_subtasks} subtasks!"
        )