Running and managing jobs

Every job runs on the project’s Postgres database, configured by fxtr init (see the Quickstart). Launchers talk to it through the project client, which stores entities and manages jobs, project arrays, and the cache.

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.
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, and set it back for real runs.

Launch from Python

The simplest launcher calls launch_cli, which opens the client, submits the job, prints its viewer link, waits, and reports the result:

launch.py

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)

anchor=__file__ tells fxtr which project the launcher belongs to. max_concurrency caps how many steps run at once. If the job doesn’t succeed, launch_cli exits with status 1. For more control, such as choosing the root, loading datasets, or reading results, write an async launcher with open_local_client and run_job:

launch.py

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 prints the viewer link before waiting, then the job’s status and result. 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 Caching and reruns.
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).

Read results

run_job returns the root workflow’s result as an Array. Entity values in it are bare IDs, so load them with their type:

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

Keep the client open while loading. Alternatively, pass an async on_result(storage, result) callback to run_job or launch_cli; it replaces the default result printing and runs while storage is open.

Launch from the command line

fxtr run launches a workflow by its registered name:

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

Inputs are JSON scalars, or @NAME for a snapshot of a project array. Anything richer needs a launcher script.

Resume a stopped job

A job stops when it can’t make further progress: a step raised, a step is held by the cache, or the job was cancelled. Resume it with its ID, which the launcher prints:

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

Resuming uses the job’s stored inputs and your registered modules, records the current commit, reuses every result the job already has, and retries unfinished work, including steps that raised. From Python:

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)

Resume refuses a job that already succeeded. If another process is running the job, resume waits until that process has been silent for [tool.fxtr] stale_after seconds (30 by default), then takes the job over.

Inspect and cancel jobs

Command	Python	Purpose
`fxtr jobs list`	`await client.list_jobs()`	Every job, newest first.
`fxtr jobs status JOB`	`await handle.status()`	Where a job is, and why it stopped.
`fxtr jobs cancel JOB`	`await handle.cancel()`	Ask the process running a job to stop it.
`fxtr cache list`	`await client.list_cache_entries(prefix=...)`	The cache’s entries.

In Python, handle = await client.job(job_id) attaches to an existing job without starting it.

Watch jobs in the viewer

Start the viewer from the project with uv run fxtr view, then open the link a launcher printed, or pick a job in the viewer. Jobs appear as they run. See Viewing results.

Get started

Concepts

Guides

Reference

Running and managing jobs

Before you launch

Launch from Python

Read results

Launch from the command line

Resume a stopped job

Inspect and cancel jobs

Watch jobs in the viewer

​Before you launch

​Launch from Python

​Read results

​Launch from the command line

​Resume a stopped job

​Inspect and cancel jobs

​Watch jobs in the viewer

Before you launch

Launch from Python

Read results

Launch from the command line

Resume a stopped job

Inspect and cancel jobs

Watch jobs in the viewer