Running and viewing jobs

​

Before you launch

• Commit first. A launch records the commit it runs and checks that the virtual environment matches uv.lock, so it refuses uncommitted changes. Pass allow_staged=True (or --allow-staged on the command line) to include staged changes; run_job, launch_cli, client.submit, and client.resume all take it. verify_environment=False on open_local_client or launch_cli (--no-verify on the command line) skips the check of the environment, but never the pinning of the commit. Each attempt at a job records the commit it ran and whether the environment was checked.
• Register your modules. Resuming a job, fxtr run, and the viewer import the modules in [tool.fxtr] modules, not your launcher, so every step and workflow a job uses must be defined in one of them.
• Use a scratch schema for trials. To keep trial jobs and cache entries apart from real results, set schema in fxtr.local.toml to another name while experimenting (or run fxtr init again with the same --database-url, another --schema, and --force), and set it back for real runs.
• Start the viewer and leave it running, so each launch opens its job in it (see Watch jobs in the viewer).

​

Launch from Python

from fxtr.entity_defns.array import Array
from fxtr.project.running import launch_cli
from my_project.experiment import experiment

if __name__ == "__main__":
    prompts = Array.from_items(
        [(("greeting",), "Hello there"), (("question",), "Why?")],
        ("[prompt: str]", str),
    )
    launch_cli(experiment, {"prompts": prompts}, anchor=__file__, max_concurrency=8)

import anyio

from fxtr.entity_defns.array import Array
from fxtr.project.running import open_local_client, run_job
from my_project.experiment import experiment


async def main() -> None:
    prompts = Array.from_items(
        [(("greeting",), "Hello there"), (("question",), "Why?")],
        ("[prompt: str]", str),
    )
    async with open_local_client(__file__, max_concurrency=8) as client:
        result = await run_job(client, experiment, {"prompts": prompts}, root="word-count-v1")
        print(list(result.items()))


if __name__ == "__main__":
    anyio.run(main)

• run_job opens the job in a new tab of the project’s running viewer (uv run fxtr view) and prints its link before waiting, then the job’s status and result; with no viewer running, it prints how to start one. open_browser=False, or FXTR_NO_BROWSER=1 in the environment, only prints the link. It raises JobNotFinishedError if the job doesn’t succeed.
• root is the root workflow’s pathname, and so the start of every cache address in the job. See Managing the step cache.
• Use handle = await client.submit(...) when you need the job’s ID or want to control the waiting yourself; await wait_for_job(client, handle) provides the same reporting as run_job.
• In a notebook, call await main() instead of anyio.run(main).

​

Watch jobs in the viewer

uv run fxtr view

• The tree of workflow invocations. The root workflow at the top, with each child workflow one level down. A mapped child appears once per key.
• A card for every step and child workflow a workflow scheduled, showing its docstring, how it was mapped, its progress, and its log lines. A card links to the function’s code in your editor.
• The arrays each operation received and produced, as tables you can page through and slice by dimension. A value that references an entity is a link.
• Entity details, opened from any link, in a pane you can keep alongside the trace. Hovering a link shows a preview.
• Your project’s README, linked from the side rail.

​

Read results

@dataclass(frozen=True)
class Report(DataclassEntity, fxtr_type="org.example.qa.Report.v1"):
    scores: BoundID[Array[float]]
    summary: BoundID[Array[float]]


@workflow(name="qa.evaluate")
async def evaluate(context: WorkflowContext, ...) -> Report:
    """Score every answer and summarize the scores per model."""
    scores = context.run_step("scores", score, {...}, map_over=["model", "task"])
    summary = context.run_step("summary", summarize, {"scores": scores}, map_over=["model"])
    return Report(
        scores=await context.store(await scores.observe()),
        summary=await context.store(await summary.observe()),
    )

report = await client.load(result.item(), as_type=Report)
scores = await client.load(report.scores)  # references inside a loaded entity are bound

​

Launch from the command line

uv run fxtr run word_count.experiment --input prompts=@word_count_prompts --root word-count-v1

​

Resume a stopped job

uv run fxtr resume 3fa3df92-2edb-4088-9e1f-a99d28b7ec77

import uuid

from fxtr.project.client import JobID
from fxtr.project.running import wait_for_job

handle = await client.resume(JobID(uuid.UUID("3fa3df92-2edb-4088-9e1f-a99d28b7ec77")))
result = await wait_for_job(client, handle)

​

Inspect and cancel jobs

​

Share a job with another project

uv run fxtr jobs export 3fa3df92-2edb-4088-9e1f-a99d28b7ec77 results/qa-v1.car
uv run fxtr jobs import results/qa-v1.car --adopt-cache   # in the other project