provrun.live

A live feed of a run, for a program that shows the run as it goes.

For example, a FastAPI app that sends it to a browser does this:

live = start_run(RunFile.load("run.yaml"), catalog=Catalog.open())
async for item in live.follow():
    ...                       # send each item on
outcome = await live.result()

The feed gives numbered items, in the order the run made them:

{"n": 1, "type": "event", "event": "run_created", "seq": 0, ...} for each evidence event, as provevi’s shown_at cuts it;
{"n": 7, "type": "text", "text": "Revenue "} for each piece of a streamed reply, at the bodies level only;
{"n": 12, "type": "done", "status": "completed", "run_id": ...} last. The status is completed, failed or cancelled, as the run’s last event says, so a run cancelled while it was sealed shows as it was sealed. A run that ended before it opened has no events; its status is that of its task, and its run ID is None.

The run’s telemetry policy sets the level, as the level of its destination of kind live, which a run file’s telemetry section can lower. A policy with no such destination shows a viewer no event and no text, only the end. The event items are those that provevi’s replay gives once the run has ended, so a viewer that comes late can read the evidence instead. A text item comes from each piece of a streamed reply; a reply that does not stream comes as one text item after its model_called event, as replay gives it. At bodies the feed keeps only the name of each payload, and reads the payload from the run’s evidence as it hands the item to a viewer, so a large payload is not held in memory. The feed keeps every other item, with the pieces of each streamed reply, as long as the program runs.

A viewer that falls behind by more than QUEUE items is dropped: it gets each item it was given, then {"type": "dropped"}, and the run never waits for it. It can come back with follow(after=n), n the last item it got. A viewer that stops following only stops: the run goes on, and its evidence is sealed as always. follow(after=n) starts after item n, so a viewer that comes back gets what it missed. The evidence is the lasting record; the feed lives only as long as the program.

class LiveFeed

source

LiveFeed(loop: asyncio.AbstractEventLoop, level: str | None = None)

The items of one run, kept in order, and handed to each viewer.

It is an evidence sink, so it gets each event as it is written, from any thread; each item goes to the event loop, so items keep their order.

Parameter	Type	Default	Description
`loop`	asyncio.AbstractEventLoop	required	The event loop that the viewers follow on
`level`	str \| None	`None`	The level shown, one of provevi’s LEVELS; the run sets it from its policy as it starts

LiveFeed.append

source

LiveFeed.append(line: str) -> None

Take an evidence line, from the recorder, in any thread.

A viewer must never fail a run, so the feed keeps its own errors.

Parameter	Type	Default	Description
`line`	str	required	One evidence line, as the recorder writes it

LiveFeed.text

source

LiveFeed.text(piece: str) -> None

A piece of a streamed reply, which only the bodies level shows.

Parameter	Type	Default	Description
`piece`	str	required	One piece of a streamed reply; an empty piece is dropped

LiveFeed.end

source

LiveFeed.end(item: dict) -> None

The last item, once the run is over.

Parameter	Type	Default	Description
`item`	dict	required	The last item, {“type”: “done”, “status”: …, “run_id”: …}

LiveFeed.follow

source

LiveFeed.follow(after: int = 0) -> AsyncIterator[dict]

The items after number after, as they come, up to the last.

Parameter	Type	Default	Description
`after`	int	`0`	The number of the last item the viewer has; 0 starts at the first
returns	AsyncIterator[dict]		Each item after it, up to the done item, or a dropped item for a viewer that fell behind

class LiveRun

source

LiveRun(task: asyncio.Task, feed: LiveFeed)

A run that has started, with its feed.

Parameter	Type	Default	Description
`task`	asyncio.Task	required	The task that runs the run file
`feed`	LiveFeed	required	The run’s feed

LiveRun.follow

source

LiveRun.follow(after: int = 0) -> AsyncIterator[dict]

The run’s items after number after, as they come; see the module.

Parameter	Type	Default	Description
`after`	int	`0`	The number of the last item the viewer has; 0 starts at the first
returns	AsyncIterator[dict]		Each item after it, as the feed gives it

LiveRun.result

source

async LiveRun.result() -> RunOutcome

Wait for the run, and give its outcome, as run_file does.

Parameter	Type	Default	Description
returns	RunOutcome		The run’s outcome; it raises what the run raised, such as RunSettingError or CancelledError

LiveRun.cancel

source

LiveRun.cancel() -> None

Cancel the run.

A run that has opened ends with run_failed, unless it was being sealed, which ends as it was; result() then raises CancelledError. A run cancelled before it opened leaves no evidence.

start_run

source

start_run(rf: RunFile, **kw: Any) -> LiveRun

Start a run file as a task in the running event loop, with a live feed.

The keyword arguments, **kw, are those of run_file, but not feed, which start_run makes.

Parameter	Type	Default	Description
`rf`	RunFile	required	The run file, as RunFile.load returns it
`kw`	Any		The keyword arguments of run_file, such as catalog; not feed, which start_run makes
returns	LiveRun		The run, started, with its feed