sonata-tasks

Reusable, product-independent tasks for sonata-engine. The base installation contains command execution contracts and common command-line tool tasks. Provider and transport integrations are installed through optional extras.

The base catalogue includes Kubernetes namespace ownership, port forwarding, Pod and OCI image inspection, and Minikube profile preflight in sonata_tasks.kubectl, sonata_tasks.imagetools, and sonata_tasks.minikube. sonata_tasks.vm.logged provides bounded VM command output with a complete log copied to the host. Callers supply their own executor, VM provider, and product-specific image checks.

sonata_tasks.containerd inspects a caller-owned container and its image using caller-supplied commands, then checks the runtime image reference and published manifest digest. The caller controls namespace, rootless session, and evidence storage; the task has no NanoLab or recipe dependency.

With the shell extra, sonata_tasks.shell.SubprocessShell streams stdout and stderr to the active workflow sink, including when no output listener is provided. An explicit listener receives each line as well; command results still contain the complete output and exit code.

sonata_tasks.process also exports OwnedCommandRunner, OwnedCommandResult and run_owned_command for bounded local Linux commands. They require procfs, pidfds and subreaper support. An isolated supervisor owns and reaps descendants, including children that create their own sessions, without changing the caller's subreaper state or signaling unrelated processes. Output goes to an exclusive log file; optional framed summaries share the same byte budget.

import os
import sys
from pathlib import Path
from threading import Event

from sonata_tasks.process import run_owned_command

result = run_owned_command(
    [sys.executable, "-c", "print('ordinary application command')"],
    cwd=Path.cwd(),
    env=os.environ,
    log_path=Path("command.log"),
    timeout_s=30,
    cancelled=Event(),
    output_limit_bytes=1024 * 1024,
)
print(result)

The result records exit status, cancellation, timeout, quota exhaustion, forced stops and confirmed reaping. A zero exit code alone is insufficient to declare success. OwnedCommandRunner.stop(timeout_s) requires at least 0.1 seconds for descendant cleanup; a runner executes only once. Linux restrictions apply to this API, while the existing managed_process_resource remains portable.

The implementation in sonata_tasks.owned_process uses only the standard library. A worker that cannot install the catalogue may copy that module from its installed distribution, preserving the distribution version and verifying the source digest. It does not need the workflow engine at worker runtime.

sonata_tasks.artifacts provides exclusive bounded artifact directories, serialized JSONL append, immutable JSON/raw publication and streaming file descriptors. Acquisition requires directory flock; publication uses hard links, fsync and no-follow append flags. Configurable markers share the same directory lock during acquisition and cannot create independent concurrent owners; the filesystem must support those operations. A fresh writer refuses existing contents, and its ownership marker remains after close. Default accounting charges this writer's own bytes, including partially failed appends. No files are excluded by name. The codec limits each JSON record to 1 MiB; raw writes are bounded by the total quota.

An ordinary command audit can also account for a log written outside the writer:

import os
import sys
from pathlib import Path
from tempfile import TemporaryDirectory
from threading import Event

from sonata_tasks.artifacts import ArtifactWriter, describe_artifact, read_records
from sonata_tasks.process import run_owned_command

with TemporaryDirectory() as temporary:
    root = Path(temporary) / "audit"
    writer = ArtifactWriter(
        root,
        1024 * 1024,
        measure_usage=lambda: sum(
            p.stat().st_size
            for p in root.rglob("*")
            if p.is_file() and not p.is_symlink()
        ),
    )
    result = run_owned_command(
        [sys.executable, "-c", "print('ordinary application command')"],
        cwd=root,
        env=os.environ,
        log_path=root / "command.log",
        timeout_s=30,
        cancelled=Event(),
        output_limit_bytes=1024,
    )
    writer.append(
        "commands",
        {
            "returncode": result.returncode,
            "reaped": result.reaped,
            "errors": list(result.errors),
            "log": describe_artifact(root / "command.log"),
        },
    )
    writer.close()
    print(list(read_records(root / "commands.jsonl")))

Callers must serialize producers sharing a measure_usage callback; a writer's lock only protects its own writes. An explicit reserve_bytes is available to write_json("summary.json", value, use_reserve=True), with no privileged filename or default reservation. owner_marker selects a safe hidden filename, defaulting to .artifact-owner. read_records yields valid objects and raises a distinct IncompleteRecordError with path and line_number for a final record without a newline. Complete malformed, non-object, non-finite or oversized records raise ArtifactCorruptionError; clients decide how to report missing evidence.

sonata_tasks.sources captures a Git working tree for repeatable local builds: tracked files (including deletions), nonignored untracked files, executable modes and safe relative symlinks. Already dangling internal links are preserved; links outside the checkout or to existing excluded inputs, cycles and submodules fail. This differs from the committed-source remote transfer in sonata_tasks.archive.

from pathlib import Path

from sonata_tasks.artifacts import ArtifactWriter
from sonata_tasks.sources import capture_source_snapshot, materialize_snapshot

writer = ArtifactWriter(Path("/tmp/application-snapshot"), 16 * 1024 * 1024)
try:
    snapshot = capture_source_snapshot(
        Path("/path/to/application"), writer, max_bytes=128 * 1024 * 1024
    )
    writer.write_json("receipt.json", {"fingerprint": snapshot.fingerprint})
finally:
    writer.close()
workspace = materialize_snapshot(snapshot, Path("/tmp/application-build"))

The caller owns/closes storage and chooses its receipt schema. Capture creates tree and source-manifest.jsonl; failed capture retains partial evidence. max_bytes bounds input bytes, including symlink target bytes; the writer's quota separately bounds the manifest, with a 100000-entry inventory limit. SourceSnapshot seals the manifest and inventory with bare-hex SHA-256 identities. verify_snapshot rejects changed inputs/manifest; materialize_snapshot verifies before and after copying to a fresh independent workspace. source_entry supports checking original inputs in a workspace that also contains generated outputs. The caller must control its filesystem namespace and serialize concurrent writes; repeated inventory checks detect ordinary concurrent changes, without making capture atomic against hostile filesystem mutation. Verification ignores empty directories and binds file/link modes, paths and contents, not directory metadata.

sonata_tasks.buildx.buildx_builder_resource can acquire a private application builder with exclusive=True. It refuses existing names, assigns a unique owner node, bootstraps the builder and returns its name. use=False leaves the client's selected builder intact. Default reuse and optional replacement remain available.

from sonata_engine import TaskInputs
from sonata_tasks.buildx import buildx_builder_resource
from sonata_tasks.execution.local import LocalCommandTaskExecutor
from sonata_tasks.execution.models import CommandOptions

inputs = TaskInputs.empty()
builder = buildx_builder_resource(
    name="application-build",
    executor=LocalCommandTaskExecutor(),
    exclusive=True,
    use=False,
    options=CommandOptions(env={"DOCKER_CONFIG": "/tmp/application-docker"}),
)
name = builder.acquire(inputs)
try:
    print(f"Build application images with docker buildx build --builder {name}")
finally:
    builder.release(inputs, name)

Exclusive cleanup verifies the original single node and docker-container driver before removing the builder. Failed creation and validation compensate partial state; the original exception retains cleanup failures as notes. An explicit owner_node lets a caller record the identity. After failed acquisition, release(inputs, "application-build") can retry unresolved cleanup; it does nothing after confirmed removal or absence. Foreign replacement, added nodes or unavailable inspection prevent removal and report an error. Callers serialize other mutations of the same Buildx client store during inspection and removal; Docker's CLI has no atomic compare-and-delete operation. Emulation installation and platform requirements remain caller policy.

A failed buildx rm may erase the client record while leaving the daemon alive. After such a failure, record absence remains unresolved and requires operator reconciliation. A retry succeeds if the same owned builder is still inspectable and its removal succeeds; an empty listing alone cannot confirm daemon cleanup.