ofplang run

A runner for Object-flow Programming Language v0 — a YAML-based dataflow workflow IR with linear Object tracking. The language is defined in the ofplang/spec repository.

The runner drives an ofplang v0 workflow to completion against an execution backend, emitting an execution status document (spec §6/§7) as it progresses and routing typed view values through the workflow. It runs on a simulator — a simulated physical backend — so a full workflow can be exercised end to end without real hardware; the same dispatch contract targets real hardware later.

Status: the simulator, the runner, and a typed (dummy) value layer are implemented.

• Simulator (ofplang.run.simulator) — a physical backend: devices, spots, transporters, and timed operations advanced on a clock. It validates every dispatch (an inconsistent plan is rejected), models timed up/down for a device or a transporter and injected operation failure, and at completion produces each operation's output view values via a device model (with none injected, the built-in script_device_model runs a script process — see below — and otherwise falls back to default_device_model, which fills type defaults and carries Object outputs through from their objects.map inputs; a custom / real model computes them). Simulator is an abstract base over the runner's Backend contract; the concrete VirtualTimeSimulator advances time instantly (deterministic; the default) and RealTimeSimulator paces it to a wall clock (a hardware-free stand-in for a real backend).
• Runner (ofplang.run.runner) — two ways to drive a backend:
  • replay runs a given execution plan (spec §6) on the backend verbatim.
  • run is a rolling-horizon loop: it calls ofplang.schedule each tick, dispatches the work that can start now, advances the clock, and polls — replanning from the committed history as it goes. It re-routes around a downed machine — a device (its process modes and, by default, its spots' transports and the refills of it), a transporter (the transports it carries) or a replenisher (the refills it performs) — polls at a fixed interval with completion-time estimation, absorbs duration variance (an operation running longer or shorter than planned), and stops the whole run if any activity fails (marking the abandoned work cancelled). Machine up/down, operation failure, and duration variance are scenario concerns injected from Python (not CLI flags).
• Value layer — the runner resolves each port's type and view schema (§7), routes typed view values along the workflow's arcs (producer output → consumer input, across nested composites and the invocations of a map / fold), contract-checks them, and assembles the whole-workflow outputs. A caller supplies the whole-workflow I/O as a single run boundary (--boundary): one document with a per-port {spot, view} descriptor — spot places a boundary Object (§6.8), view supplies an input value — for the workflow's entry inputs and final outputs. An Array of Objects takes a list of spots and, if supplied, a list of views in the same order. An unsupplied entry input runs on its type's default, and the run warns that it did; nothing inside the workflow is ever defaulted — an input with no source is refused before the run starts. A workflow-embedded static literal (bind: {port: {value: …}}, §11) is that consumer input's value. At run end the produced output views are echoed back into a result boundary of the same schema (--boundary-out). Non-script values are typed but still dummy — a real device backend plugs into the same seam later. Those dummies are the built-in simulator's, a mock device's answer: the runner itself fills no output, and a backend that leaves a declared output unset stops the job (backend_output_missing).
• Device-local consumables (ofplang-schedule §4.7) — a device may declare a stock it holds and a process mode what it draws per run. What each stock holds at the start of the run is a property of the run, so it goes in the run boundary (boundary.inventories.levels), beside the input views; the level at any later moment is never stated, it is replayed by the scheduler from those levels and the consumption each completed activity carries. --ignore-resources switches the model off (§4.7.3) for a lab that declares stocks nobody is tracking. The starting levels are not echoed into the result boundary — that document is written to be fed back, and a second run replays no history, so echoing them would hand it stock the first run already spent; feed back the status instead.
• Replenishment (ofplang-schedule §4.7.1) — where the environment says a replenisher can reach a device, a stock that would run out is topped up rather than ending the run. The scheduler places the refill; the runner dispatches it like any other activity, and it holds both machines while it works — the device being filled and the replenisher filling it — so it can never overlap the work it feeds. It moves no material and reports no level: a level is derived from what the run started with plus its history, never observed. A refill that fails stops the run like any activity failure. Backend gained dispatch_replenishment for this, which is why 0.3.0 is a breaking release for a custom backend.
• The backend is told, not asked — nothing in the contract reports the state of the world back. The runner says what to do and learns whether it worked; where material is and what a stock holds are derived from that. Both follow from the same fact: a real laboratory keeps no ledger of its spots or its levels, so such a query has no truthful implementation. 0.7.0 made that true of material as it had always been of stock — spot_state left the contract when the runner stopped asking, and clear joined it as the counterpart of place, for saying that the material on a spot has been taken away. Shrinking a structural contract breaks nobody; clear costs the backends here nothing, all of them deriving from Simulator.
• Provenance is optional, one keyword at a time (0.10.0) — a backend is told where an activity came from only if its signature asks: declare a node parameter to be told which workflow node it is, and a job parameter to be told which job of a joint run (--jobs) it belongs to. A backend that predates an extension is driven exactly as before, and one that wants only the node gets only the node. job is passed only where there is one — a single-workflow run has no roster and says nothing about jobs — which matters because two jobs of one workflow render the same node paths and can move between the same pair of spots: a backend that keeps a record or mints identities from provenance cannot tell them apart without it. The same pair reaches the device model too, by the same rule: a record is made when an operation is dispatched, but an identity is minted when it completes, so a model declaring node / job is told there as well and the historical 5-positional model is called unchanged. A backend wrapping the built-in simulator can declare **kwargs and forward whatever is offered.
• A refusal is a failure, not a crash (BackendRefused, 0.9.0) — when the runner's derivation and the backend's reality disagree, the backend refuses the dispatch (or the placement): the destination spot is really full, the source really empty, the machine really busy or really down. That is now recorded as the activity having failed — which stops its job, leaves the rest of the laboratory running, and makes what it was holding derivable — instead of escaping run() and taking every other job's status with it. A backend raises ofplang.run.backend.BackendRefused to say it, and only before it has started the operation: there is no handle for a refused one, so trouble discovered once something is under way belongs in state(handle) instead. Nothing structural is required, so a backend that raises something else is exactly as fatal as before.
• Python script processes (spec §22, python_script_processes) — an atomic Pure-Data process may carry a script: {language: python, code: …} section. The built-in device model runs it: the input port values are bound as locals, the code returns a mapping of the declared outputs, and those become the operation's computed output values — the first genuine (non-dummy) computation in the value layer. A script that raises, returns the wrong output names, or returns a non-conformant value fails runtime verification (§22.2) and stops the run gracefully like any other activity failure (the failed activity is marked failed, its unstarted successors cancelled, exit 1). The script runs inline in real time but advances no simulation time; its outputs appear when the operation completes at its end, and the environment mode's duration is the scheduler's estimate of the compute cost. Scripts run with full Python builtins and no import restriction (§22.3 permits an implementation to restrict these; this one does not).
• Contracts (spec §9) — an atomic process may declare contracts with requires (preconditions over inputs.*.view) and ensures (postconditions over inputs.*.view and outputs.*.view). The runner evaluates them at runtime against the actual view values: requires before the operation runs (a violation stops it before dispatch), ensures after it completes. An atomic requires that references only run/graph-phase inputs (§5.6) is knowable at run start, so it is checked there as a preflight — before any work is dispatched — rather than waiting for that (possibly late) operation. A violation — or a runtime evaluation error — stops the run gracefully like any activity failure (failed + downstream cancelled, exit 1). Static / graph-time contract checking (a fully-constant contract that is statically false, type errors, bad references) is ofplang-validate's job and is not repeated here. Contracts are checked on atomic processes and on the top-level entry composite (main): the entry composite's requires is evaluated at run start over the whole-workflow inputs (a violation stops the run before any work runs) and its ensures at run end over the whole-workflow inputs and outputs (a violation marks the completed run failed). Nested composite contracts are checked too, at each composite invocation's value boundary: its requires once its inputs are available (for an input fed by an upstream process this is mid-run, before the composite's body runs) and its ensures once its outputs are, a violation stopping the run gracefully at the composite boundary (its not-yet-run body cancelled). Because non-script process outputs are still typed defaults, ensures bites mainly on script processes and boundary-supplied inputs until a real device backend computes physical outputs.
• Failure observability — when a run stops (an injected activity failure, a script error, or a contract violation), the reason is exposed as a structured RollingRunner.failure (a machine-readable kind code, e.g. contract_requires / script_error, plus a human-readable detail, the subject, and the time), and the CLI prints it to stderr. The status document itself stays a valid §6 document (the reason is out of band). An optional contract_observer callback is invoked for every contract check (held or violated) — a trace hook for debugging.
• Static view values (spec §7.4) — a type whose view field declares a value: fixes that field to a constant for every value of the type. The runner projects it onto every view record it routes, so a Python script reading the field and a contract referencing it both see the static value (not a runtime default). A value that carries a conflicting one is forced to the static value.

Install

pip install ofplang-run

Requires Python 3.10+. Runtime dependencies (pulled in automatically) are PyYAML, the sibling ofplang-schedule that run (rolling-horizon) calls each tick, and ofplang-validate used by the shared front door (ofplang.run.front_door_check), which validates a workflow file or an already-loaded workflow document, so an embedding caller holding one in memory is checked the same way a CLI is. The runner library never imports validate (and the replans never re-validate), so it stays a one-shot front door.

Who checks what. Across the toolchain there are three kinds of check and one owner for each. Whether a workflow is valid v0 is ofplang-validate's alone, asked once at the front door. Whether it can be planned -- the supported subset, lengths known before the run, the boundary bindings -- is ofplang-schedule's. Whether the values are right at run time -- a value's type, a contract, an each length only a value shows, a backend that returned every output -- is this runner's, being the only one that sees values. Neither the scheduler nor the runner re-validates; both read the workflow assuming it is valid v0, and refuse only where their own reading would otherwise drop or make up a value (an input with no source, a final output with no source), answering with the code validate gives the same document where there is one. That is what a run without validation (--no-validate, validate=False) can still rely on.

For development, install editable with the test extra from a clone:

pip install -e ".[test]"

Command line

ofp-run run <workflow> --env <env>
    [--boundary DOC] [--boundary-out FILE] [--observation-out FILE]
    [--poll-interval D] [--margin M] [--seed N] [--max-transport-legs N]
    [--no-validate] [-o OUT]
ofp-run run --jobs <run doc> --env <env> [--on-job-failure continue|stop] [...]
ofp-run replay <plan> --env <env> [-o OUT]

run drives a v0 workflow to completion by replanning as it goes: each tick it polls the backend and, when anything the scheduler reads has changed -- an operation finished, a machine went down, a pending activity came due -- renders the committed history as a status, calls the scheduler in-process -- handing it the workflow, the environment and that status as documents, so nothing goes through a temporary file -- and dispatches the newly-runnable work. A tick that changed none of those keeps the plan it already has, so a long protocol costs one solve per activity event rather than one per unit of its makespan; what is observed, and so the status produced, is the same either way. --boundary supplies the whole-workflow I/O as one document — a boundary: mapping with a {spot, view} descriptor per entry input / final output port. spot places a boundary Object on an environment spot (spec §6.8; Object ports only); view supplies an input's view value. An Array of Objects is one Object per element, each on a spot of its own: spot is a list of spots in element order, and view, if given, a list of as many views — plates: {spot: [hotel.a, hotel.b], view: [{barcode: X}, {barcode: Y}]}. An entry input given no view runs on its type's default (one per spot for an Array of Objects), and the run says so on stderr (entry_input_defaulted). The runner projects it into the scheduler's interface (spots only, so the scheduler stays value-independent) and the seeded input values. --boundary-out writes the result boundary — the same schema with each produced output's view filled in — a run-local artifact, separate from the value-free status document. On completion each pinned Object output is checked to have reached its declared spot. --observation-out streams the observation document (see docs/OBSERVATION.md): a YAML multi-document stream recording each completed activity's concrete input / output view values (a transport's moved view), appended as each activity finishes — the value-layer companion to the status document, also run-local. --poll-interval sets the fixed polling interval (default 1). --max-ticks is the non-termination guard: a run that takes more than that many ticks is given up on. One tick is one poll interval, so the guard also caps the makespan a run can reach (the default 100000 with the default interval means 100000 time units); pass 0 for no limit when a long virtual run needs it, which also gives up the protection against a backend whose clock does not advance. --margin sets the running-task margin: on each replan a still-running activity is pinned to end at max(reported end, now + margin), so a positive margin is what keeps an overrunning operation's successor from being planned at now and dispatched onto a value that operation has not produced yet. The default is 0, which is only safe with a backend whose operations cannot finish later than planned (the in-process virtual-time simulator); against a wall-clock or real backend set it to at least the poll interval, or a successor is refused with input_not_produced rather than computing on a value its producer has not given it. --on-job-failure decides what one job's failure does to the rest of a --jobs run: continue (the default) stops that job alone and lets the others finish — which is why they were planned together — while stop stops the whole run. A stopped job's remaining work is reported cancelled, and the spots its material is still sitting on are held so the rest of the run is planned around them rather than onto them — worked out from the history the document carries rather than written into it, since every input to the working-out is already there (ofplang.schedule.derived_holds is how to ask what it comes to). A backend refusing a dispatch or a placement is one such failure (BackendRefused, above) rather than an exception out of run(), so one job meeting a world the plan disagrees with no longer costs the others their status. A single workflow is a single job, so this makes no difference to it. --jobs runs several workflows together in one laboratory (schedule SPEC §6.11) in place of the single <workflow> argument. Its run document names each job — an id, the workflow it runs, its own boundary, and the release time before which it may not start — plus the two things that belong to the laboratory rather than to any one job: what its stocks hold at the start of the run (inventories, §6.10) and which spots it is already holding (occupied, §6.12). The jobs are planned together, so they compete for the same machines and draw on the same stocks: a refill neither job needs alone can appear because the pair of them does. Each job's activities carry its id in the status, and the plan's roster reports the completion the scheduler promised each one. See examples/shared_refill.run.yaml.

A job can also leave a --jobs run while it is going. The roster is the set of jobs something of which is still in the laboratory — unfinished work, or material nobody has collected — so RollingRunner.withdraw(job_id) is how a program driving the run says that a finished job's material has been collected and its entry may go. It takes effect on the next replan, and what makes it more than dropping an entry is the arithmetic: a job's history is part of what the current stock levels are made of, so the levels are carried forward to that moment (inventories.at, §6.10) rather than handing the stock back everything that job drew. A final output the job's boundary bound to a spot is taken as collected from there; one it left unbound is not — the schedule chose where that came to rest and nobody was told — so that spot is written into occupied as the job goes, dated when the plate was actually left there. That is the one moment an occupancy is written down: everything else a job holds follows from its history, and stops following from anything the instant that history leaves with it.

A freeze can be lifted. RollingRunner.free_spot(spot) says a spot this run was keeping frozen may be used again. occupied says a spot is not to be used, not that something is on it — a failed transport freezes both of its ends, though its plate is at one of them — so lifting one is a declaration rather than a report, and why (it was collected, it was never there, it was looked at and found fine) is not the document's business. Only a freeze the document states can be lifted: what a stopped job is holding is derived afresh on every solve, so there is no entry to remove and removing one would not stop the next solve deriving it again. What ends that hold is the job leaving the plan.

And a job can arrive into one. RollingRunner.admit(request) takes a job into a run already under way — the same JobRequest the run document describes, handed over later. It joins the end of the roster, which is the end of the priority order: an arrival owes the jobs already being planned the completions they were promised, so it is fitted around them rather than the other way round, and is promised a completion of its own by the first plan that includes it. Its entry material appears on its spots when its release comes, which for an arrival is the moment it arrives unless it says otherwise — a job that did not exist cannot have been released earlier. A job that cannot run is refused rather than admitted and then stopped: its own preconditions are checked before it joins, so a refused call leaves the run exactly as it was.

--max-transport-legs N is how many transport activities one Object-bearing arc may be carried in (schedule SPEC §6.4.1), joined by relay activities. It is 1 by default — the single hop this has always planned. Raise it for a device the transporter reaches at one position only, or a plate that has to cross a hand-off station; only the fewest possible moves are offered, so an arc one move apart is never sent round by way of somewhere else. --no-validate skips the one-shot ofplang-validate front-door check of the workflow — use it when the workflow was already validated upstream (e.g. by the ofp umbrella CLI); $import is still resolved and the capability gate still runs, since both are structural rather than validation. replay runs a plan produced by ofp-schedule verbatim on the simulator (no value layer). Both write the final execution status as YAML (-o, else stdout). Exit codes: 0 success, 1 execution failed (an activity failed, or a replan is infeasible), 2 usage/input error.

This tool is also the run subcommand of the umbrella ofp CLI (ofplang), which forwards to it in-process with this CLI's own subcommands intact: ofp run run …, ofp run replay …, each with the same options and exit codes as above.

The package lives under the ofplang PEP 420 namespace (ofplang.run), shared across the organization's tools.

Feature support

v0 defines seven optional features (spec §4.2), and a document requiring one an implementation does not have "is valid v0 but unsupported by that implementation" (§4.1). So ofp-validate accepting a workflow does not mean this runner can execute it:

Examples

examples/ holds runnable scenarios — supplied inputs and computed outputs, views routed across a composite boundary, a script process with contracts checked at runtime, re-routing around a device that goes down, the drift fixed-interval polling costs, two jobs run together needing a refill neither needs alone, and one job of three failing while the others finish. Most are Python scripts rather than CLI invocations, because what they demonstrate is injected from code: a device model, a machine fault, a polling interval. Their output is committed under examples/outputs/, so an example can be read without being run.

Tests

pytest