operonx.app.jobs

Jobs run a graph over data that does not talk back — a list, a .jsonl file, a function that yields items — one run per item, every result kept, with a record per run. steps=[...] runs jobs in order. The guide: Jobs.

Job

Job

Job(
    name: str,
    *,
    graph: Any = None,
    items: Any = None,
    input: Optional[str] = None,
    inputs: Optional[Dict[str, Any]] = None,
    key: Union[str, KeyFn, None] = None,
    output: Union[str, Path, Callable[..., Any], None] = None,
    reduce: Any = None,
    concurrency: int = 4,
    on_error: str = "skip",
    retry: Optional[Retry] = None,
    timeout: Optional[float] = None,
    preflight: Optional[Sequence[str]] = None,
    steps: Optional[Sequence["Job"]] = None,
    trace: Any = None,
    keep_results: bool = True,
    record_dir: Union[str, Path, None] = None,
    on_item: Optional[Callable[[Any], Any]] = None,
    description: str = "",
    fail_run: Optional[bool] = None,
    schedule: Any = None,
)

See the module docstring.

Parameters:

Name	Type	Description	Default
name	str	The job's name: the CLI argument and the record's folder.	required
graph	Any	A module-level @graph, a GraphOp, an Operon, or "module:attr" naming one. Runs once per item.	None
items	Any	What to loop over: a function returning an iterable (called on every run), an iterable or async iterable, or a .jsonl path. None runs the graph once.	None
input	Optional[str]	The graph parameter that receives the whole item. Without it, a dict item fills parameters by name and anything else goes to the graph's only free parameter. A graph with ingress takes the item through its door instead.	None
inputs	Optional[Dict[str, Any]]	Fixed graph inputs for every item (--set adds to them).	None
key	Union[str, KeyFn, None]	The item field that identifies it, or a function of the item. Without one each item gets a random id and the job cannot resume.	None
output	Union[str, Path, Callable[..., Any], None]	Also export each successful result as it finishes: a .jsonl path ({"key", "result"} lines), or a function (key, result), sync or async.	None
reduce	Any	A module-level @graph taking results (the list of every successful result, in key order) plus inputs. Runs once, after the last item; its outputs are run.reduced.	None
concurrency	int	Items in flight at once.	4
on_error	str	"skip" (carry on) or "stop" (start nothing new, and do not reduce). Either way the run is failed when an item failed — unless fail_run=False.	'skip'
fail_run	Optional[bool]	False for a batch where a failed item is an outcome, not a broken run: each failure is still in the record (on_item sees it), and the run — and the next step — goes on ok. Default: the class's (True; False for an Eval).	None
retry	Optional[Retry]	Retry(max_attempts=N): a failed or timed-out item runs again, with the policy's backoff between attempts.	None
timeout	Optional[float]	Seconds one item's run may take; past it, the run is cancelled and the item recorded timeout.	None
preflight	Optional[Sequence[str]]	Resource keys ("llm:x") that must answer before any item runs, so a dead endpoint fails the job once, in seconds.	None
steps	Optional[Sequence['Job']]	Instead of graph: jobs to run in order, as one command. The first step whose run is not ok stops the rest.	None
trace	Any	Trace consumers for the runs ([tracing] in operonx.toml overrides it). Ignored when graph is already an Operon.	None
keep_results	bool	Keep each result in the record's results.jsonl. False for results too big to keep; reduce then has nothing to read and is refused.	True
on_item	Optional[Callable[[Any], Any]]	Called with each item's ItemResult as it is recorded (sync or async): progress lines, events for a host. It never breaks the run. run(on_item=...) replaces it for that run.	None
record_dir	Union[str, Path, None]	Where runs are recorded (<record_dir>/<name>/<run>). Default: the project's — [jobs] dir in operonx.toml, else .operonx/jobs under the project root.	None
description	str	One line, for --list and the studio.	''
schedule	Any	schedule(every="1h" | at="07:00", port=...): the job also runs on that clock, inside operonx serve on that port. Each tick is a fresh run (no resume) recording what triggered it; a tick while the last run is going is skipped and counted. operonx run <job> still runs it on demand.	None

Source code in operonx/app/jobs/job.py

def __init__(
    self,
    name: str,
    *,
    graph: Any = None,
    items: Any = None,
    input: Optional[str] = None,  # noqa: A002 — the graph parameter's name
    inputs: Optional[Dict[str, Any]] = None,
    key: Union[str, KeyFn, None] = None,
    output: Union[str, Path, Callable[..., Any], None] = None,
    reduce: Any = None,
    concurrency: int = 4,
    on_error: str = "skip",
    retry: Optional[Retry] = None,
    timeout: Optional[float] = None,
    preflight: Optional[Sequence[str]] = None,
    steps: Optional[Sequence["Job"]] = None,
    trace: Any = None,
    keep_results: bool = True,
    record_dir: Union[str, Path, None] = None,
    on_item: Optional[Callable[[Any], Any]] = None,
    description: str = "",
    fail_run: Optional[bool] = None,
    schedule: Any = None,
):
    if not name or not isinstance(name, str):
        raise ValueError("a job needs a name")
    if schedule is not None and getattr(schedule, "kind", None) != "schedule":
        raise TypeError(
            f"job {name!r}: schedule= is a schedule(every=... | at=..., port=...), "
            f"not {type(schedule).__name__}"
        )
    if fail_run is not None:
        self.items_fail_run = fail_run
    self.name = name
    self.description = description
    self.steps: Optional[List[Job]] = None
    if steps is not None:
        self._check_steps(steps, graph=graph, items=items, reduce=reduce, output=output)
        self.steps = list(steps)
    elif graph is None:
        raise ValueError(f"job {name!r} needs a graph (or steps=[...])")
    check_items(items, name)
    if input is not None and not (isinstance(input, str) and input):
        raise TypeError(f"job {name!r}: input is the name of a graph parameter")
    if key is not None and not (isinstance(key, str) or callable(key)):
        raise TypeError(f"job {name!r}: key must be a field name or a function of the item")
    if output is not None and not callable(output):
        if Path(output).suffix.lower() != ".jsonl":
            raise ValueError(
                f"job {name!r}: output={str(output)!r} — a path must be a .jsonl file; "
                "for anything else pass a function (key, result)"
            )
    if int(concurrency) < 1:
        raise ValueError(f"job {name!r}: concurrency must be at least 1")
    if on_error not in ON_ERROR:
        raise ValueError(f"job {name!r}: on_error must be 'skip' or 'stop', not {on_error!r}")
    if retry is not None and not isinstance(retry, Retry):
        raise TypeError(f"job {name!r}: retry is a Retry(max_attempts=N), not {retry!r}")
    if timeout is not None and not float(timeout) > 0:
        raise ValueError(f"job {name!r}: timeout must be a positive number of seconds")
    if reduce is not None and not keep_results:
        raise ValueError(
            f"job {name!r}: reduce reads the results the record keeps; "
            "it cannot run with keep_results=False"
        )

    self.graph = graph
    self.items = items
    self.input = input
    self.inputs: Dict[str, Any] = dict(inputs or {})
    self.key = key
    self.output = output
    self.reduce = reduce
    self.concurrency = int(concurrency)
    self.on_error = on_error
    self.retry = retry
    self.timeout = float(timeout) if timeout is not None else None
    self.preflight: List[str] = list(preflight or [])
    self.trace = trace
    self.keep_results = bool(keep_results)
    #: Where runs are recorded (``<record_dir>/<name>/<run>``). ``None``
    #: is the project's: see :func:`default_record_dir`.
    self.record_dir: Optional[Path] = Path(record_dir) if record_dir is not None else None
    self.on_item = on_item
    #: When the job runs by itself: a ``schedule()`` listener. ``operonx
    #: serve`` runs its clock in the server on that port.
    self.schedule = schedule
    self._engine: Any = None
    self._reducer: Any = None
    self._doors: Optional[bool] = None

Methods:

engine

engine() -> Any

The compiled Operon, built once.

Source code in operonx/app/jobs/job.py

def engine(self) -> Any:
    """The compiled Operon, built once."""
    if self._engine is None:
        if self.steps is not None:
            raise TypeError(f"job {self.name!r} runs steps; it has no graph")
        self._engine = _compile(self.name, "graph", self.graph, self.trace)
    return self._engine

reducer

reducer() -> Any

The compiled reduce graph, built once (None without one).

Source code in operonx/app/jobs/job.py

def reducer(self) -> Any:
    """The compiled ``reduce`` graph, built once (``None`` without one)."""
    if self.reduce is not None and self._reducer is None:
        self._reducer = _compile(self.name, "reduce", self.reduce, self.trace)
        if "results" not in self._reducer.graph.inputs:
            raise TypeError(
                f"job {self.name!r}: the reduce graph needs a `results` parameter "
                "(the list of every result)"
            )
    return self._reducer

has_doors

has_doors() -> bool

True when the graph reads items through an ingress door — the serving shape (:func:operonx.app.doors.has_doors).

Source code in operonx/app/jobs/job.py

def has_doors(self) -> bool:
    """True when the graph reads items through an ingress door — the
    serving shape (:func:`operonx.app.doors.has_doors`)."""
    if self._doors is None:
        from operonx.app.doors import has_doors

        self._doors = has_doors(self.engine())
    return self._doors

bind

bind(item: Any) -> Dict[str, Any]

The run inputs for one item of a graph without doors (:func:operonx.app.doors.bind_item).

Source code in operonx/app/jobs/job.py

def bind(self, item: Any) -> Dict[str, Any]:
    """The run inputs for one item of a graph without doors
    (:func:`operonx.app.doors.bind_item`)."""
    from operonx.app.doors import bind_item

    return bind_item(
        list(self.engine().graph.inputs), item, fixed=self.inputs, input=self.input
    )

item_of

item_of(raw: Any) -> Any

What the graph receives for one raw item — the item itself; an Eval hands over a case's input.

Source code in operonx/app/jobs/job.py

def item_of(self, raw: Any) -> Any:
    """What the graph receives for one raw item — the item itself; an
    `Eval` hands over a case's ``input``."""
    return raw

key_of

key_of(item: Any) -> str

The item's identity, as a non-empty string.

Source code in operonx/app/jobs/job.py

def key_of(self, item: Any) -> str:
    """The item's identity, as a non-empty string."""
    if self.key is None:
        return uuid.uuid4().hex[:12]
    if callable(self.key):
        value = self.key(item)
    elif isinstance(item, Mapping):
        if self.key not in item:
            raise KeyError(f"item has no field {self.key!r}")
        value = item[self.key]
    else:
        if not hasattr(item, self.key):
            raise KeyError(f"item has no attribute {self.key!r}")
        value = getattr(item, self.key)
    text = "" if value is None else str(value)
    if not text:
        raise ValueError(f"item key {self.key!r} is empty")
    return text

run async

run(
    *,
    resume: bool = False,
    record_dir: Union[str, Path, None] = None,
    on_item: Optional[Callable[[Any], Any]] = None,
    trigger: Optional[Mapping[str, Any]] = None,
) -> JobRun

Run the job once and return its record. record_dir overrides where it is written (for this run, and every step's); on_item is called with each item's ItemResult as it is recorded; trigger says what started it (a schedule's tick), kept in the record.

Source code in operonx/app/jobs/job.py

async def run(
    self,
    *,
    resume: bool = False,
    record_dir: Union[str, Path, None] = None,
    on_item: Optional[Callable[[Any], Any]] = None,
    trigger: Optional[Mapping[str, Any]] = None,
) -> JobRun:
    """Run the job once and return its record. *record_dir* overrides
    where it is written (for this run, and every step's); *on_item* is
    called with each item's ``ItemResult`` as it is recorded; *trigger*
    says what started it (a schedule's tick), kept in the record."""
    from .runner import TRIGGER, run_job, run_steps

    kept = (self.record_dir, self.on_item)
    if record_dir is not None:
        self.record_dir = Path(record_dir)
    if on_item is not None:
        self.on_item = on_item
    token = TRIGGER.set(dict(trigger)) if trigger else None
    try:
        if self.steps is not None:
            return await run_steps(self, resume=resume, record_dir=record_dir)
        return await run_job(self, resume=resume)
    finally:
        self.record_dir, self.on_item = kept
        if token is not None:
            TRIGGER.reset(token)

run_sync

run_sync(**kwargs: Any) -> JobRun

run() from synchronous code — a script, a cron entry.

Source code in operonx/app/jobs/job.py

def run_sync(self, **kwargs: Any) -> JobRun:
    """`run()` from synchronous code — a script, a cron entry."""
    return asyncio.run(self.run(**kwargs))

main

main(argv: Optional[Sequence[str]] = None, *, doc: Optional[str] = None) -> int

This job as a command line; returns the exit status. Takes the flags operonx run has, minus the job name.

Source code in operonx/app/jobs/job.py

def main(self, argv: Optional[Sequence[str]] = None, *, doc: Optional[str] = None) -> int:
    """This job as a command line; returns the exit status. Takes the
    flags ``operonx run`` has, minus the job name."""
    from operonx.cli.run import main_for

    return main_for(self, argv, doc=doc)

records

records() -> Path

Where this job's runs are recorded.

Source code in operonx/app/jobs/job.py

def records(self) -> Path:
    """Where this job's runs are recorded."""
    return (
        Path(self.record_dir)
        if self.record_dir is not None
        else default_record_dir(self.folder)
    )

describe

describe() -> Dict[str, Any]

What the job is, for run.json and --list. No secrets: only the names of things.

Source code in operonx/app/jobs/job.py

def describe(self) -> Dict[str, Any]:
    """What the job is, for run.json and ``--list``. No secrets: only
    the names of things."""

    def shown(value: Any) -> Any:
        if value is None or isinstance(value, str):
            return value
        if isinstance(value, Path):
            return str(value)
        if isinstance(value, (list, tuple)):
            return f"{len(value)} items"
        if callable(value) and hasattr(value, "__qualname__"):
            from operonx.app.declare import ref_name

            return ref_name(value)
        return type(value).__name__

    clock = None
    if self.schedule is not None:
        clock = {**self.schedule.options, "port": self.schedule.port}
    if self.steps is not None:
        out = {
            "kind": "steps",
            "steps": [s.name for s in self.steps],
            "description": self.description,
        }
        if clock:
            out["schedule"] = clock
        return out
    out = {
        "kind": "job",
        "graph": self.graph
        if isinstance(self.graph, str)
        else getattr(self.graph, "name", None)
        or getattr(self.graph, "__name__", type(self.graph).__name__),
        "items": shown(self.items),
        "input": self.input,
        "key": self.key
        if isinstance(self.key, str)
        else (getattr(self.key, "__name__", "fn") if self.key else None),
        "output": shown(self.output),
        "reduce": None
        if self.reduce is None
        else getattr(self.reduce, "name", None) or getattr(self.reduce, "__name__", "reduce"),
        "concurrency": self.concurrency,
        "on_error": self.on_error,
        "retry": self.retry.max_attempts if self.retry is not None else None,
        "timeout": self.timeout,
        "preflight": list(self.preflight) or None,
        "description": self.description,
    }
    if clock:
        out["schedule"] = clock
    return out

The record

JobRun dataclass

JobRun(
    job: str,
    run_id: str,
    path: Path,
    status: str,
    started: str,
    ended: Optional[str],
    counts: Dict[str, int],
    items: List[ItemResult] = list(),
    resume_from: Optional[str] = None,
    meta: Dict[str, Any] = dict(),
)

A finished (or in-flight) run, read back from its two files.

Attributes

results property

results: Dict[str, Any]

{key: result} for every item that produced something — this run's and, after --resume, the earlier runs' too. Read from results.jsonl; empty when the job kept no results.

errors property

errors: Dict[str, str]

{key: error} for every item that failed or timed out.

reduced property

reduced: Any

What the job's reduce graph returned (None without one, or when it did not run).

Methods:

load classmethod

load(path: str | Path, items: bool = True) -> 'JobRun'

A run from its directory. items=False reads run.json alone: no items, and the counts as run.json wrote them.

Source code in operonx/app/jobs/record.py

@classmethod
def load(cls, path: str | Path, items: bool = True) -> "JobRun":
    """A run from its directory. ``items=False`` reads ``run.json``
    alone: no items, and the counts as ``run.json`` wrote them."""
    path = Path(path)
    meta = json.loads((path / "run.json").read_text(encoding="utf-8"))
    read_items, items = items, []
    items_file = path / "items.jsonl"
    if read_items and items_file.exists():
        with items_file.open("r", encoding="utf-8") as fh:
            for line in fh:
                line = line.strip()
                if line:
                    items.append(ItemResult.from_dict(json.loads(line)))
    # The item statuses are recounted from items.jsonl, which is the
    # truth; anything else run.json counted is kept as written.
    if read_items:
        counts = {s: 0 for s in _COUNTED}
        counts.update(
            {k: v for k, v in (meta.get("counts") or {}).items() if k not in _COUNTED}
        )
        for item in items:
            counts[item.status] = counts.get(item.status, 0) + 1
    else:
        counts = dict(meta.get("counts") or {})
    return cls(
        job=meta["job"],
        run_id=meta["run_id"],
        path=path,
        status=meta["status"],
        started=meta["started"],
        ended=meta.get("ended"),
        counts=counts,
        items=items,
        resume_from=meta.get("resume_from"),
        meta={
            k: v
            for k, v in meta.items()
            if k not in ("job", "run_id", "status", "started", "ended", "resume_from")
        },
    )

ItemResult dataclass

ItemResult(
    key: str,
    status: str,
    error: Optional[str] = None,
    trace_id: Optional[str] = None,
    ms: float = 0.0,
    sent: int = 0,
    attempts: int = 1,
    verdict: Optional[Dict[str, Any]] = None,
    trace: Any = None,
    result: Any = None,
)

Methods:

as_dict

as_dict() -> Dict[str, Any]

The record line's fields. Shallow: the line is serialized at once, and a deep copy of an eval's verdict (asdict) was most of what writing an item cost (scripts/bench_eval_overhead.py).

Source code in operonx/app/jobs/record.py

def as_dict(self) -> Dict[str, Any]:
    """The record line's fields. Shallow: the line is serialized at
    once, and a deep copy of an eval's verdict (``asdict``) was most of
    what writing an item cost (`scripts/bench_eval_overhead.py`)."""
    out = {
        name: getattr(self, name)
        for name in self.__dataclass_fields__
        if name not in _TRANSIENT
    }
    if out["verdict"] is None:
        del out["verdict"]
    return out

RunRecord

RunRecord(
    root: str | Path,
    job: str,
    *,
    meta: Optional[Dict[str, Any]] = None,
    resume_from: Optional[str] = None,
    keep_results: bool = True,
)

The writer. Opened when the run starts, finished exactly once.

Source code in operonx/app/jobs/record.py

def __init__(
    self,
    root: str | Path,
    job: str,
    *,
    meta: Optional[Dict[str, Any]] = None,
    resume_from: Optional[str] = None,
    keep_results: bool = True,
):
    self.job = job
    for _ in range(3):
        self.run_id = new_run_id()
        self.path = Path(root) / job / self.run_id
        try:
            self.path.mkdir(parents=True, exist_ok=False)
            break
        except FileExistsError:
            continue
    else:
        raise FileExistsError(f"could not mint a fresh run directory under {Path(root) / job}")
    self.started = _now()
    self.resume_from = resume_from
    self.meta = dict(meta or {})
    self.counts: Dict[str, int] = {s: 0 for s in _COUNTED}
    self.error: Optional[str] = None
    self._write_run(RUN_RUNNING, ended=None)
    self._items = (self.path / "items.jsonl").open("a", encoding="utf-8")
    self._results = (
        (self.path / "results.jsonl").open("a", encoding="utf-8") if keep_results else None
    )

Methods:

result

result(key: str, value: Any) -> None

Keep one item's result in results.jsonl (when the job keeps results). Flushed at once, like the item line.

Source code in operonx/app/jobs/record.py

def result(self, key: str, value: Any) -> None:
    """Keep one item's result in ``results.jsonl`` (when the job keeps
    results). Flushed at once, like the item line."""
    if self._results is None:
        return
    self._results.write(
        json.dumps({"key": key, "result": value}, ensure_ascii=False, default=str)
    )
    self._results.write("\n")
    self._results.flush()

finish

finish(
    status: str,
    error: Optional[str] = None,
    *,
    counts: Optional[Dict[str, int]] = None,
    extra: Optional[Dict[str, Any]] = None,
) -> JobRun

Close the record. counts adds to the item counts; extra lands in run.json beside the job's description (what reduce returned, a steps job's steps, an eval's numbers).

Source code in operonx/app/jobs/record.py

def finish(
    self,
    status: str,
    error: Optional[str] = None,
    *,
    counts: Optional[Dict[str, int]] = None,
    extra: Optional[Dict[str, Any]] = None,
) -> JobRun:
    """Close the record. ``counts`` adds to the item counts; ``extra``
    lands in run.json beside the job's description (what ``reduce``
    returned, a ``steps`` job's steps, an eval's numbers)."""
    if error:
        self.error = error
    if counts:
        self.counts.update(counts)
    if extra:
        self.meta.update(extra)
    self._items.close()
    if self._results is not None:
        self._results.close()
    self._write_run(status, ended=_now())
    return JobRun.load(self.path)

Items

iter_items async

iter_items(items: Any) -> AsyncIterator[Any]

Every item, in order, as an async iterator.

Source code in operonx/app/jobs/items.py

async def iter_items(items: Any) -> AsyncIterator[Any]:
    """Every item, in order, as an async iterator."""
    if items is None:
        yield {}
        return
    source: Optional[Any] = items
    if isinstance(items, (str, Path)):
        source = _jsonl(Path(items))
    elif callable(items) and not (hasattr(items, "__iter__") or hasattr(items, "__aiter__")):
        source = items()
        if inspect.isawaitable(source):
            source = await source
    if hasattr(source, "__aiter__"):
        async for item in source:
            yield item
        return
    it = iter(source)
    while True:
        item = await asyncio.to_thread(next, it, _END)
        if item is _END:
            return
        yield item

The session

JobSession

JobSession(key: Optional[str], *, meta: Optional[Mapping[str, Any]] = None)

Bases: BoundedSession

One item in; what egress sends, kept as the item's result.

Source code in operonx/app/jobs/session.py

def __init__(self, key: Optional[str], *, meta: Optional[Mapping[str, Any]] = None):
    super().__init__(meta={"key": key, **dict(meta or {})})
    self.key = key
    self.sent_items: List[Any] = []