Tasks and Artifacts

Report units of work with rb.task, parallelize them with threads or Function.map, and register durable outputs with rb.artifact.

Tasks and artifacts answer two different questions about a run:

A task says what unit of work ran, whether it succeeded, and what happened to it.
An artifact says where a durable output was written.

Neither API moves your workload or your data by itself. rb.task(...) is an inline status scope, and rb.artifact(...) registers a URI after your code has written the output. The run's [ Tasks ] and [ Artifacts ] views then make both visible without requiring someone to reconstruct them from logs or a result JSON document.

Report Inline Work with `rb.task`

Use a task context around one named unit of work:

import rebase as rb


@rb.workflow(project="energy")
def capture_all(datasets: list[str]) -> dict:
    import rebase as rb
    from energy_capture import fetch_and_store

    captured = []
    failures = []

    for dataset in datasets:
        try:
            with rb.task(
                f"Capture {dataset}",
                key=dataset,
                parameters={"dataset": dataset},
            ) as task:
                uri, rows = fetch_and_store(dataset)
                task.set_result({"outcome": "captured", "rows": rows, "uri": uri})
                captured.append(uri)
        except Exception as exc:
            # Catch outside the context. __exit__ has already reported this task as failed.
            failures.append({"dataset": dataset, "error": str(exc)})

    if failures:
        raise RuntimeError(f"{len(failures)} captures failed: {failures}")
    return {"captured": captured}

The examples use energy_capture.fetch_and_store as an application-owned storage helper. Replace it with your implementation and include that module in the deployed source or image. Imports used at runtime belong inside the decorated body because Rebase can ship that body independently.

Entering the context reports running. A normal exit reports succeeded; an exception reports failed and is re-raised. task.set_result(...) is optional and accepts a JSON object. Domain outcomes such as pending, skipped, or reused normally belong in that result while the lifecycle status remains succeeded.

rb.task does not create a worker, retry the body, or make the loop parallel. It records work that is already executing in the current function or workflow process.

Catch exceptions outside the with rb.task(...) block when later units should continue. If the overall run should also fail, collect the failures and raise after the remaining tasks have finished, as in the example above.

Outside a hosted run, the same context manager validates its inputs and behaves as an in-memory no-op. That keeps the function ordinary Python during local execution.

Parallel Inline Tasks with Threads

For a modest amount of I/O-bound work, a thread pool can run several inline tasks in the same container. Create the rb.task inside each worker so every unit gets its own lifecycle record:

import rebase as rb


@rb.function(project="energy", run_type="long")
def capture_threaded(datasets: list[str], max_workers: int = 8) -> dict:
    from concurrent.futures import ThreadPoolExecutor, as_completed
    from contextvars import copy_context

    import rebase as rb
    from energy_capture import fetch_and_store

    def capture_one(dataset: str) -> dict:
        with rb.task(
            f"Capture {dataset}",
            key=dataset,
            parameters={"dataset": dataset},
        ) as task:
            uri, rows = fetch_and_store(dataset)
            artifact = rb.artifact(
                f"Captured dataset {dataset}",
                uri=uri,
                key=f"capture/{dataset}",
                media_type="application/parquet",
                metadata={"dataset": dataset, "rows": rows},
            )
            result = {"dataset": dataset, "rows": rows, "artifact_id": artifact.id, "uri": uri}
            task.set_result(result)
            return result

    completed = []
    failures = []
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = {}
        for dataset in datasets:
            # Use a fresh snapshot once per worker. It carries this hosted run's
            # request-local reporting identity into the new thread.
            context = copy_context()
            futures[pool.submit(context.run, capture_one, dataset)] = dataset

        for future in as_completed(futures):
            dataset = futures[future]
            try:
                completed.append(future.result())
            except Exception as exc:
                failures.append({"dataset": dataset, "error": str(exc)})

    if failures:
        raise RuntimeError(f"{len(failures)} captures failed: {failures}")
    return {"completed": completed}

copy_context() is load-bearing. Hosted service runners keep run credentials and IDs in Python context variables so concurrent requests cannot see one another's identities. ThreadPoolExecutor workers do not reliably inherit that request context. Taking a fresh snapshot for each submitted worker preserves task reporting and lets rb.artifact inherit the task created inside that worker. A single Context object must not be entered by two threads simultaneously, which is why the loop calls copy_context() once per submission.

Threads are a good fit for bounded, I/O-heavy work that shares one environment. They still share the process, memory limit, CPU allocation, and failure domain. For CPU-heavy work, stronger isolation, or wider fan-out, deploy the unit as a function and use Function.map.

Platform Fan-out with `Function.map`

Function.map submits one deployed function run per input. When issued inside a workflow step, every map item is automatically recorded as a task of the parent workflow run. You do not need to wrap each item in rb.task.

import rebase as rb

project = rb.project("energy")


@project.function(name="capture-one", run_type="quick")
def capture_one(dataset: str) -> dict:
    import rebase as rb
    from energy_capture import fetch_and_store

    uri, rows = fetch_and_store(dataset)
    artifact = rb.artifact(
        f"Captured dataset {dataset}",
        uri=uri,
        key=f"capture/{dataset}",
        disposition="created",
        media_type="application/parquet",
        metadata={"dataset": dataset, "rows": rows},
    )
    return {"dataset": dataset, "rows": rows, "artifact_id": artifact.id, "uri": uri}


@project.step(name="capture-datasets")
def capture_datasets(datasets: list[str], max_concurrency: int = 20) -> dict:
    import rebase as rb

    # A step body is shipped independently, so resolve the deployed function by name.
    capture = rb.get_function("energy", "capture-one")
    items = [{"dataset": dataset} for dataset in datasets]
    outcomes = list(
        capture.map(
            items,
            max_concurrency=max_concurrency,
            return_exceptions=True,
        )
    )

    completed = []
    failures = []
    for item, outcome in zip(items, outcomes):
        if isinstance(outcome, Exception):
            failures.append({"dataset": item["dataset"], "error": str(outcome)})
        else:
            completed.append(outcome)

    if failures:
        raise RuntimeError(f"{len(failures)} captures failed: {failures}")
    return {"completed": completed}


@project.workflow(name="daily-capture")
def daily_capture(datasets: list[str]) -> dict:
    return capture_datasets(datasets)

The map call returns results in input order by default, which makes zip(items, outcomes) safe. Set ordered=False when you want results as soon as they complete and each result already carries enough identity to stand on its own. With return_exceptions=True, one failed item is yielded as an exception instead of stopping iteration at the first failure.

The runtime passes each map item's task ID into the child function. Consequently an rb.artifact(...) call inside capture_one appears in the parent workflow's [ Artifacts ] view, linked to the correct map task, while retaining the child function run as its producer.

A map uses platform compute for every item and is subject to workspace concurrency and compute limits. max_concurrency is a ceiling, not a request to bypass those limits.

Threads or `Function.map`?

	Thread pool + `rb.task`	`Function.map`
Where work runs	Threads in one function/workflow container	Separate deployed function runs
Task records	Created explicitly with `rb.task`	One task per map item automatically
Context handling	Propagate with a fresh `copy_context()` per worker	Injected by the Rebase runtime
Best for	Modest I/O-bound concurrency, shared in-process state	Wider fan-out, CPU work, isolation, per-item run visibility
Failure domain	One process and one memory/CPU allocation	Each item has its own function run
Environment	Exactly the caller's environment	The mapped function's deployed image, secrets, env, and attachments

Register Durable Outputs with `rb.artifact`

Write the object first, then register the location:

uri = upload_parquet(frame, "gs://forecast-archive/2026-08-11/SE3.parquet")

artifact = rb.artifact(
    "SE3 day-ahead forecast",
    uri=uri,
    key="forecast/SE3",
    disposition="created",
    media_type="application/parquet",
    size_bytes=218_420,
    version="1741632447112345",
    digest="md5:8d777f385d3dfec8815d20f7496026dc",
    metadata={"zone": "SE3", "rows": 24},
)

Rebase stores this record; it does not upload, download, proxy, or validate the bytes at the URI. The URI must be absolute and include a scheme, such as gs://, s3://, or https://.

Argument	Description
`name`	Human-readable output name shown in the run.
`uri`	Durable absolute URI. Required.
`key`	Optional logical identity within the run. Retrying the same key and URI is idempotent; using the key for another URI is rejected.
`disposition`	`"created"` by default, or `"reused"` when this run found an existing output.
`media_type`	Optional MIME type such as `application/json` or `application/parquet`.
`size_bytes`	Optional non-negative object size.
`version`	Optional provider version, generation, snapshot, or ETag.
`digest`	Optional checksum with an explicit algorithm prefix, such as `md5:…`.
`metadata`	Optional JSON object with domain metadata.

if object_exists(uri):
    artifact = rb.artifact(
        "SE3 day-ahead forecast",
        uri=uri,
        key="forecast/SE3",
        disposition="reused",
        media_type="application/parquet",
    )
else:
    write_forecast(uri)
    artifact = rb.artifact(
        "SE3 day-ahead forecast",
        uri=uri,
        key="forecast/SE3",
        disposition="created",
        media_type="application/parquet",
    )

Inside rb.task, an artifact inherits the inline task ID. Inside a mapped function, it inherits the map task ID. Otherwise it is attached directly to the active run and step.

Hosted registration is strict. If Rebase cannot store the record, ArtifactReportingError is raised so a successful upload is not silently missing from the run's provenance. Outside a hosted run, the call validates its arguments and returns an Artifact with id is None without making an API request.

Inspect Tasks and Artifacts

The TUI interleaves tasks and artifacts with run events, steps, and logs. Use the [ Tasks ] and [ Artifacts ] filters for focused tables.

From Python:

run = rb.Run(run_id)

for task in run.tasks():
    print(task["name"], task["status"], task.get("error"))

for artifact in run.artifacts():
    print(artifact["name"], artifact["disposition"], artifact["uri"])

The lower-level client can filter by step or task:

client = rb.Client()

tasks = client.list_run_tasks(run_id, step_run_id=step_run_id)
artifacts = client.list_run_artifacts(run_id, step_run_id=step_run_id)
artifacts = client.list_run_artifacts(run_id, task_id=task_id)

See the rb.task reference, rb.artifact reference, and Function.map reference.

Tasks and Artifacts

Report Inline Work with rb.task

Parallel Inline Tasks with Threads

Platform Fan-out with Function.map

Threads or Function.map?

Register Durable Outputs with rb.artifact

Inspect Tasks and Artifacts

On this page

Report Inline Work with `rb.task`

Platform Fan-out with `Function.map`

Threads or `Function.map`?

Register Durable Outputs with `rb.artifact`