videoflow.core package

Submodules

videoflow.core.compiler module

Turns a videoflow.core.flow.Flow (a built, validated graph) into a list of per-node NodeSpec``s: everything an execution engine needs to launch one worker per node, without any of the live ``Node objects. A NodeSpec is fully JSON-serializable, which is what lets it cross into a separate process or a Kubernetes pod as environment variables / a ConfigMap.

class videoflow.core.compiler.NodeSpec(name: str, node_class: str | None, params: Dict[str, Any], parents: List[str], kind: str, has_children: bool, nb_tasks: int, device_type: str, is_finite: bool, image: str | None = None, partition_by: str | None = None, join_policy: Dict[str, Any] | None = None, component_ref: str | None = None, descriptor: Dict[str, Any] | None = None, command: List[str] | None = None, protocol_version: int | None = None, gpu_count: int = 1, gpu_resource_name: str | None = None, blob_readers: int | None = None, gpu_memory_gib: float | None = None, delivery: Dict[str, Any] | None = None)

Bases: object

A flat, serializable description of one node’s deployment. Here a node refers to a videoflow graph node (producer/processor/consumer), not a physical machine in a Kubernetes cluster.

• Attributes:

  • name: node’s stable name (unique in the flow).

  • node_class: fully-qualified import path, e.g. videoflow.processors.basic.IdentityProcessor.

  • params: dict from node.get_params() — the kwargs to reconstruct it.

  • parents: list of parent node names, in process() positional order.

  • kind: one of producer/processor/consumer.

  • has_children: whether anything downstream consumes this node’s output.

  • nb_tasks: desired replica count (processors only; 1 otherwise).

  • device_type: ‘cpu’ or ‘gpu’ (processors only).

  • gpu_count: whole physical GPUs each replica requests (GPU processors only; default 1).

  • gpu_memory_gib: declared GPU memory demand in GiB (RFC 0004) — drives the mix strategy’s MIG slice choice; None when undeclared.

  • gpu_resource_name: internal — the resolved extended-resource name, set only by a GPU strategy (e.g. the mix solver’s chosen MIG profile); None means the deploy-time default (nvidia.com/gpu).

  • is_finite: for producers, whether next() self-terminates.

  • image: the container image ref declared on the node, or None (the deploy-time default/override supplies it — see videoflow.deploy.images).

  • blob_readers: how many downstream reads each message this node publishes receives (Σ over children of nb_tasks if partitioned else 1); drives refcounted blob reclamation (PROTOCOL.md BLOB-5). 0 for leaves; None when unknown (a legacy spec), which disables reclamation.

  • delivery: this node’s delivery=/on_error= overrides as a dict, or None to inherit the flow type’s preset. Read by provisioning (it sets the durables’ max_deliver) and by the worker.

The field order below is the constructor signature — callers pass these positionally (NodeSpec('n', 'pkg.Cls', {}, [], 'processor', ...)), so reordering or inserting a field is a breaking change. Not frozen: a spec is a plain mutable record, and to_dict/from_dict stay explicit because they are the VF_FLOW_SPECS_JSON serialization boundary — asdict() would deep-copy and rewrite the nested params/descriptor values.

blob_readers: int | None = None

command: List[str] | None = None

component_ref: str | None = None

delivery: Dict[str, Any] | None = None

descriptor: Dict[str, Any] | None = None

device_type: str

classmethod from_dict(d: Dict[str, Any]) → NodeSpec

gpu_count: int = 1

gpu_memory_gib: float | None = None

gpu_resource_name: str | None = None

has_children: bool

image: str | None = None

is_finite: bool

property is_native: bool

runs its own image entrypoint (no node_class) and must speak the protobuf wire.

Type:

A non-Python component

property is_remote: bool

Loaded from a component descriptor (Python or native), vs a native Python graph node.

join_policy: Dict[str, Any] | None = None

kind: str

name: str

nb_tasks: int

node_class: str | None

params: Dict[str, Any]

parents: List[str]

partition_by: str | None = None

protocol_version: int | None = None

to_dict() → Dict[str, Any]

videoflow.core.compiler.batching_policies(flow: Any) → Dict[str, Dict[str, Any]]

The dynamic-batching contracts a flow’s nodes declare (Node.batching_policy, RUN-036): node -> policy.

videoflow.core.compiler.blob_reader_ids(spec: NodeSpec, specs: List[NodeSpec]) → List[str]

The reader obligations every payload spec publishes is held for (RFC 0006 BLOB-13, VF_BLOB_READER_IDS): <child> for a child whose replicas compete on one durable, <child>/p<i> per replica of a partitioned child — the identities the children release under after a confirmed settlement. The same arithmetic as blob_readers, by name instead of by count; not a NodeSpec field, so the specs document is unchanged.

videoflow.core.compiler.compile_flow(flow: Flow, envelope_version: int | None = None) → List[NodeSpec]

• Arguments:

  • flow: a built videoflow.core.flow.Flow (do NOT call .run() on it first).

  • envelope_version: the wire version this flow will deploy with. The only supported version is 4 (protobuf); an explicit incompatible pin is rejected. Defaults to the ambient DEFAULT_ENVELOPE_VERSION.

• Returns:

  • list of NodeSpec, one per node in the flow’s topological sort.

videoflow.core.compiler.execution_groups(flow: Any) → Dict[str, tuple[str, ...]]

The fused execution groups a flow’s nodes declare (Node.execution_group, RUN-035): group name -> member node names, sorted. Empty for a flow that declares none, so the compiled document is unchanged.

videoflow.core.compiler.gpu_provenance(flow: Flow) → Dict[str, Dict[str, str]]

Where every processor’s GPU requirement came from, keyed by node name — {'detector': {'device_type': 'node', 'gpu_count': 'descriptor', 'gpu_memory_gib': 'default'}} (the sources are those of videoflow.core.provenance). The compiler’s companion document to compile_flow: deliberately not a NodeSpec field, because a spec’s serialized form is every specs ConfigMap byte and to_dict is asdict, so it is returned alongside for whoever asks. compile_to_dict may emit it under its own key only when explicitly requested, never in the default document. Producers and consumers carry no GPU knobs and are omitted.

• Arguments:

  • flow: a built videoflow.core.flow.Flow.

• Returns:

  • node name -> {field: source} for every ProcessorNode in the flow.

videoflow.core.compiler.has_native_components(specs: List[NodeSpec]) → bool

Whether any node is a native (non-Python) component — the ones that force the protobuf wire.

videoflow.core.compiler.has_remote_components(specs: List[NodeSpec]) → bool

Whether any node came from a component descriptor (Python or native).

videoflow.core.compiler.parent_replicas(spec: NodeSpec, specs: List[NodeSpec]) → List[int]

VF_PARENT_REPLICAS (RFC 0006 ENV-11): each parent’s nb_tasks, in spec.parents order — how many terminators the EOS-7 barrier expects from it.

videoflow.core.compiler.sink_guarantees(flow: Any) → Dict[str, str]

The non-default effect declarations of a flow’s sinks (ConsumerNode.effect_guarantee, RUN-017): what the planner admits exactly_once_effects against. Empty for a flow whose sinks declare nothing, so the compiled document is unchanged.

videoflow.core.compiler.specs_from_tasks_data(tasks_data: List[tuple]) → List[NodeSpec]

Converts build_tasks_data output — tuples of (node, parent_names, is_last) — into a list of serializable NodeSpec.

It defaults GPU scheduling knobs to 1 GPU per replica, and leaves the GPU resource name unset (None) so the deploy-time default is used. The mix strategy’s solver will override those fields with the chosen MIG profile if the node is a sharer.

It defaults is_finite to True for non-producers, and leaves blob_readers unset (None) for non-leaves, so the engine can compute it once all specs are known.

It defaults component_ref, descriptor, command, and protocol_version to None for native Python nodes, and fills them in for remote components (Python or native) from the node’s descriptor.

videoflow.core.compiler.validate_wire_compatibility(specs: List[NodeSpec], envelope_version: int | None) → None

The wire is the single language-neutral protobuf envelope (version 4) for every flow. Reject an explicit pin to any other version at compile/deploy time so the failure is actionable here rather than a worker refusing to start. (PROTOCOL.md §4.)

videoflow.core.constants module

videoflow.core.context module

Runtime context optionally handed to a node’s lifecycle/processing methods.

A node method (open/next/process/consume/close) may declare a final ctx (or context) parameter; if it does, the task passes a RuntimeContext so the node can read run identity and set a partition key on its output without depending on any global state. Methods that don’t declare it are called exactly as before, so this is fully backward compatible with existing nodes.

videoflow.core.context.CHECKPOINT_METADATA_KEY = '_checkpoint'

Reserved metadata key under which a task hands the node’s pending checkpoint to Messenger.publish_message (RFC 0006 CTRL-4; RUN-003/RUN-022): the state rides with the output it belongs to, so the messenger commits both in one ledger write and the key never reaches the wire. Like _partition_key, a leading underscore marks it as the runtime’s, not the node’s.

class videoflow.core.context.RuntimeContext(flow_id: str, run_id: str, node_name: str, replica_id: int, logger: Logger, messenger: Messenger | None = None)

Bases: object

• Attributes:

  • flow_id / run_id / node_name / replica_id: identity of this running node.

  • logger: a standard library logger scoped to the node.

  • emits_output: whether an output publication follows each processed input (set by the task from has_children). It decides when a checkpoint becomes durable: with an output, together with that output; without one, at once.

checkpoint(state: bytes) → None

Record state — the node’s own serialized state — in the run ledger, together with the identity of the input group being processed, in one write (RFC 0006 CTRL-4; RUN-022): whatever a replacement restores with restore_checkpoint describes exactly the inputs up to and including that group, and the runtime acknowledges that group without handing it to the node again. Call it from process/consume after the state is updated and before returning. A no-op without a messenger.

For a node whose input produces an output (emits_output), the write is deferred and committed by the messenger in the same ledger write as the output it belongs to (the task passes it under CHECKPOINT_METADATA_KEY): a crash between the state and the output can then never leave one without the other (RUN-003). A sink or a leaf node has no output to wait for, so its checkpoint is written at once.

emits_output: bool

property input_info: Dict[str, Any] | None

Per-parent envelope info for the input group currently being processed: {parent_name: {'event_ts': ..., 'metadata': ..., 'trace_id': ..., 'seq': ...}} (None values for parents missing from a quorum emission; lists for collect parents). None for producers. Lets fusion code read each input’s exact event time without changing process() signatures.

property input_key: str | None

A stable identity for the input group being processed (the same across a redelivery or a restart): what an idempotent sink keys its external effect on (effect_guarantee = 'idempotent_key'). None for producers.

restore_checkpoint() → bytes | None

The state bytes of the last checkpoint of this node in the run, or None. Call it from open().

set_event_timestamp(value: float) → None

Set the event time (epoch seconds) stamped on this node’s next published output — when the underlying real-world event was captured (a frame’s capture time, a sensor sample’s timestamp). Producers of time-sensitive data should call this from next(); downstream nodes inherit their input group’s event time automatically, so they rarely need to. Time- aligned joins (JoinPolicy(mode='time')) group on this value.

set_partition_key(value: Any) → None

Set the partition key carried on this node’s next published output, so a downstream partitioned node can route by a business key. Applied to the message metadata under the reserved field _partition_key.

take_pending_checkpoint() → bytes | None

The state a deferred checkpoint left for the next output publication, cleared once taken (the task’s call).

videoflow.core.engine module

class videoflow.core.engine.ExecutionEngine

Bases: object

Defines the interface of the execution environment — how tasks are physically started (as local OS processes for development, or as Kubernetes pods in production) and how flow-wide termination and completion are observed.

allocate_and_run_tasks(tasks_data: List[tuple] | None, flow_id: str, flow_type: str, run_id: str) → None

Defines a template with the order of methods that need to run in order to allocate and run tasks.

• Arguments:

  • tasks_data: list of tuples (node, parent_names : [str], is_last : bool), or None when the engine already holds pre-compiled specs.

  • flow_id: stable identifier for this flow (constant across runs).

  • flow_type: ‘realtime’ or ‘batch’.

  • run_id: per-run identifier that scopes this execution’s broker streams.

join_task_processes() → None

Blocking method. It is supposed to make the calling process sleep until all task processes have finished processing.

signal_flow_termination() → None

Signals the execution environment that the flow needs to stop.

class videoflow.core.engine.Messenger

Bases: object

Utility class that tasks use to receive input and write output, over a message broker (see videoflow.messaging.nats_messenger.NATSMessenger for the concrete implementation). A Messenger is bound to exactly one node in the graph. It knows that node’s own broker subject (for publishing) and its real parents’ subjects (for receiving) — routing is by node name, not by position in a topological sort, so it works correctly for arbitrary DAGs (multi-parent joins, multi-producer graphs).

STOP_AUTHORITY_LOST = 'authority-lost'

STOP_CONTROL = 'control'

the flow-wide control stop (the flow is ending — a producer still closes its stream with EOS, CTRL-2), a quiesce (this process is being stopped; a replacement continues the stream, so no EOS), or the loss of this process’s authority over its partition to a replacement that already continues it (no EOS either).

Type:

Why check_for_termination is true

STOP_QUIESCE = 'quiesce'

ack_inputs() → None

Acknowledge the input group last returned by receive_message — called by the task only after the node has processed it (and, for a processor, published its output). This ack-after-process ordering is what makes a crash mid-processing safe: the un-acked message is redelivered instead of lost.

check_for_termination() → bool

Returns true if a flow-wide termination signal has been received on the control channel. Used by videoflow.core.task.ProducerTask to stop pulling new input even before it naturally reaches StopIteration.

checkpoint(state: bytes) → None

Record the node’s state together with the identity of the input group being processed, in one write (RFC 0006 CTRL-4). Default: no-op.

close() → None

Release any broker resources held by the messenger. Default: no-op.

fail_inputs(exc: BaseException) → None

Report that the node raised while processing the last input group. What that costs is decided by the node’s videoflow.core.policies.DeliveryPolicy together with how the error classified (see videoflow.core.errors): a poison message is dead-lettered immediately, a transient one is redelivered until its budget runs out, and a worker-fatal one is handed back for another replica and never blamed.

last_input_info() → Dict[str, Any] | None

Per-parent envelope info (event_ts, metadata, trace_id, seq) for the input group last returned by receive_message (exposed to nodes as ctx.input_info). Default: None.

last_input_key() → str | None

A stable identity for the input group last returned by receive_message, used as a sink idempotency key. Default: None (no idempotency).

pending_count() → int

How many messages are waiting for this node across its parents. Used by videoflow.core.supervision.ProgressDeadline to tell a stalled node (work available, nothing acked) from an idle one. Default: 0, which reads as “idle” and so never trips the deadline.

pending_observation() → Any

pending_count as an observation: Known(count) when the broker answered, Unknown(reason) when it did not. The progress deadline reads this one, because “the query failed” and “nothing is pending” must lead to different decisions. Default: wraps pending_count as known.

publish_abort(error: Any) → None

Publishes an abnormal termination marker carrying why this node died. A clean end-of-stream and a crash are different facts, and only the first one used to exist on the wire — so a node that died mid-run left every child blocking forever on an EOS that was never coming. Downstream treats ABORT as end-of-stream-with-an-error: it stops, propagates the marker to its own children, and exits non-zero.

• Arguments:

  • error: a videoflow.core.errors.VideoflowError, a bare exception, or an already-normalized error dict being relayed from further upstream.

publish_message(message: Any, metadata: Dict[str, Any] | None = None) → None

Publishes this node’s own output message. Depending on the flow’s configured retention policy (REALTIME vs BATCH), this may drop the message if downstream consumers are behind.

publish_stop_signal() → None

Publishes a termination marker on this node’s own subject. Unlike publish_message, this is never dropped regardless of retention policy — every downstream consumer must observe it exactly once.

quiesce() → None

Stop receiving new input; keep settling what is already held. Called on SIGTERM and by rollout/scale-down drains. Default: no-op.

receive_message() → Dict[str, Dict[str, Any]]

Blocks until this node has received a complete input: one message from every real parent, all derived from the same upstream originating event (see the trace_id propagation scheme in the concrete implementation) — or until every parent has signaled termination.

• Returns:

  • a dict {parent_name: entry} with exactly one entry per real parent of this node. Each entry carries message, metadata, event_ts, and two termination flags: is_stop_signal (that parent ended cleanly) and is_abort (it died). An aborted entry also carries abort_origin and abort_error — the originating node and its error record — so the failure can be reported and relayed downstream rather than degenerating into a hang.

restore_checkpoint() → bytes | None

The last checkpointed state of this node in the run, or None. Default: None.

resume_offset() → int

For a replayable producer (RFC 0006 MSGID-6): the last source offset whose publication was accepted before a restart, so the source resumes from the next one. Default: 0 (start from the beginning; a live source).

set_output_event_timestamp(value: float) → None

Set the event time (epoch seconds) stamped on this node’s next published output (via RuntimeContext.set_event_timestamp) — when the underlying real-world event was captured. Producers of time-sensitive data (cameras, sensors) should set this; downstream nodes inherit their input group’s event time automatically. Default: no-op.

set_output_partition_key(value: Any) → None

Set the partition key attached to this node’s next published output (via RuntimeContext.set_partition_key), so a downstream partitioned node can route by a business key. Default: no-op.

stop_reason() → str | None

Which stop check_for_termination reports, one of the STOP_* constants, or None while the flow runs. Default: the control stop whenever a termination was signalled — the only kind a bare messenger has.

take_drops() → Dict[str, int]

Inputs the messenger gave up on since the last call, by reason — the drops decided below this seam (a retry budget exhausted, undecodable bytes, a join group evicted, a live publication the broker refused) that the health counters could not otherwise see. Default: none.

videoflow.core.errors module

The error taxonomy every other module classifies against.

An exception hierarchy earns its keep only if something branches on it, so each level here drives one specific decision:

Level	Decides
Top-level class	who is at fault, and therefore who sees it and where
disposition	what the messenger does with the in-flight message
code	what the metric is labelled with, and what the DLQ is queryable by
remedy	what the CLI prints under the error

The three branches of the tree map onto the three boundaries of the framework:

• VideoflowUserError — the graph or its configuration is invalid. Raised while the graph is being built, on the machine that builds it; the CLI turns it into exit code 2.

• VideoflowEnvironmentError — the world is not as required (broker down, cluster unreachable, no GPU capacity). Exit code 3.

• VideoflowRuntimeError — something failed while messages were flowing. Its three leaves are the dispositions, and they are the only part of this module the hot path reads.

The disposition is what fixes the framework’s oldest blind spot: with a single except Exception there is no way to tell a bad message (retrying is waste) from a temporarily unavailable world (retrying is exactly right) from a sick worker (retrying is actively harmful — it shreds a healthy stream into the DLQ one message at a time). See videoflow.core.policies.DeliveryPolicy for what each disposition actually causes.

remedy deserves a note: naming the fix rather than only the problem is the strongest convention in this codebase’s error strings. Making it a field rather than prose means the CLI, the DLQ inspector and the Kubernetes termination log can all render it the same way, and that it cannot be quietly forgotten.

exception videoflow.core.errors.ActiveRunConflict(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowEnvironmentError

Under --single-run, another run of the same flow already holds workloads in the namespace (RFC 0006 §10, RUN-047). Refused before anything of the new run is created, so the active run is never reconfigured or overwritten.

code: str = 'VF_ACTIVE_RUN'

exception videoflow.core.errors.BrokerUnavailable(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowEnvironmentError

NATS could not be reached, or a stream/consumer could not be created.

code: str = 'VF_BROKER_UNAVAILABLE'

exception videoflow.core.errors.CapabilityError(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowUserError

The flow asks a component for something it declares it cannot do.

code: str = 'VF_CAPABILITY'

exception videoflow.core.errors.ClusterError(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowEnvironmentError

kubectl is missing, an apply was rejected, or the cluster refused the work.

code: str = 'VF_CLUSTER'

exception videoflow.core.errors.ConfigError(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowUserError

An invalid configuration value: flow type, join policy, mount spec, image ref.

code: str = 'VF_CONFIG'

videoflow.core.errors.DEFAULT_DISPOSITION = 'transient'

Disposition assumed for an exception nothing has classified. TRANSIENT reproduces the framework’s historical behaviour (retry, then dead-letter), so adopting the taxonomy changes nothing until a node or a classifier opts in.

exception videoflow.core.errors.DecodeError(message: str, remedy: str | None = None, **context: Any)

Bases: PoisonMessage

The envelope or payload could not be decoded off the wire.

code: str = 'VF_POISON_DECODE'

exception videoflow.core.errors.DeviceError(message: str, remedy: str | None = None, **context: Any)

Bases: WorkerFatal

The accelerator is unusable: out of memory, fell off the bus, wrong driver.

code: str = 'VF_DEVICE'

class videoflow.core.errors.Diagnostic(severity: str, code: str, node: str | None, message: str, remedy: str | None = None)

Bases: object

One problem found by videoflow.core.graph.validate. Validation collects these rather than raising on the first one, so a flow with three mistakes reports three — a compiler’s contract, not an interpreter’s.

• Attributes:

  • severity: error (the flow cannot run) or warning (it can, but something is probably not what the author meant).

  • code: the VideoflowError.code this would be raised as.

  • node: the node name the problem belongs to, or None if it is graph-wide.

  • message: what is wrong.

  • remedy: what to do about it.

code: str

message: str

node: str | None

remedy: str | None = None

render() → str

One line, as the CLI prints it.

severity: str

videoflow.core.errors.EXIT_USER = 2

Process exit statuses, by fault class. A uniform exit 1 is untriageable in CI; these let a wrapper script tell “your flow is wrong” from “the cluster is wrong” from “the flow ran and lost nodes” without parsing stderr.

exception videoflow.core.errors.FlowFailed(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowEnvironmentError

The flow ran and one or more nodes failed. Distinct exit code so CI can tell it from a bad deploy.

code: str = 'VF_FLOW_FAILED'

exit_code: int = 4

exception videoflow.core.errors.FlowStalled(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowEnvironmentError

The flow can never finish: unschedulable pods, or a node that stopped making progress.

code: str = 'VF_FLOW_STALLED'

exit_code: int = 5

exception videoflow.core.errors.GraphError(message: str, remedy: str | None = None, diagnostics: List[Diagnostic] | None = None, **context: Any)

Bases: VideoflowUserError

The graph itself is invalid: a cycle, an unreachable consumer, duplicate node names, a replicated join without a partition key.

• Arguments:

  • diagnostics: every problem found in one validation pass (see videoflow.core.graph.validate). Reporting them together is the difference between one fix-rerun cycle and four.

code: str = 'VF_GRAPH'

diagnostics: List[Diagnostic]

exception videoflow.core.errors.IdentityCollision(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowUserError

Two distinct logical names encode to the same physical broker or cluster name (a.b and a_b both sanitize to a_b; a 70-character node name truncates onto another). Rejected at compile time: a collision at run time would silently route one node’s messages to another.

code: str = 'VF_IDENTITY_COLLISION'

exception videoflow.core.errors.IncompatibleProfile(message: str, remedy: str | None = None, diagnostics: List[Diagnostic] | None = None, **context: Any)

Bases: VideoflowUserError

The flow requests a guarantee the composed backends cannot provide — a reliable_work channel on a transport with no retained backlog, a restart-safe join without a durable runtime store, an exclusive device from an allocator that only accounts for shares. Raised by the composition planner before anything is provisioned, started or published: the alternative, a silent downgrade to whatever the backend does offer, is precisely the failure the profiles exist to make impossible.

• Arguments:

  • diagnostics: every incompatibility found in one planning pass, so the fix-and-retry loop is one iteration rather than one per channel.

code: str = 'VF_INCOMPATIBLE_PROFILE'

diagnostics: List[Diagnostic]

exception videoflow.core.errors.NodeContractError(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowUserError

A node violates the contract a worker relies on to rebuild and run it.

code: str = 'VF_NODE_CONTRACT'

exception videoflow.core.errors.OwnershipConflict(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowEnvironmentError

A compare-and-swap on shared cluster state lost to a concurrent writer: the node owner label, the shared MIG configuration, the ClusterPolicy pointer. Nothing was mutated on the losing side; re-plan against the current state.

code: str = 'VF_OWNERSHIP_CONFLICT'

videoflow.core.errors.POISON = 'poison'

What the messenger does with the input group a node just failed on. These are the values VideoflowRuntimeError.disposition takes and the keys DeliveryPolicy dispatches on.

exception videoflow.core.errors.PartitionKeyError(message: str, remedy: str | None = None, **context: Any)

Bases: PoisonMessage

A partitioned node received a record whose partition key is unusable — absent, None, empty, or not a scalar — and its policy is to reject such records (RFC 0006, RUN-020). Dead-lettered by the node’s first replica, never hashed as the string "None" into an undeclared hot partition.

code: str = 'VF_POISON_PARTITION_KEY'

exception videoflow.core.errors.PoisonMessage(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowRuntimeError

The message itself is bad. Retrying is waste: it will fail identically every time, so it is dead-lettered on the first failure.

code: str = 'VF_POISON'

disposition: str = 'poison'

exception videoflow.core.errors.ProgressStalled(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowRuntimeError

Raised when a node has acked nothing for the progress deadline while work was pending. Distinct from WorkerUnhealthy: nothing raised, the node simply stopped making progress.

code: str = 'VF_PROGRESS_STALLED'

disposition: str = 'worker_fatal'

exit_code: int = 5

exception videoflow.core.errors.ResourceExhausted(message: str, remedy: str | None = None, **context: Any)

Bases: WorkerFatal

The worker is out of a host resource — memory, disk, file descriptors.

code: str = 'VF_RESOURCE_EXHAUSTED'

exception videoflow.core.errors.ResourceUnavailable(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowEnvironmentError

A resource the flow needs does not exist or cannot be obtained.

code: str = 'VF_RESOURCE_UNAVAILABLE'

exception videoflow.core.errors.SchemaError(message: str, remedy: str | None = None, **context: Any)

Bases: PoisonMessage

The payload decoded but is not what this node requires.

code: str = 'VF_POISON_SCHEMA'

exception videoflow.core.errors.StaleAuthority(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowRuntimeError

This worker tried to commit under an ownership epoch a newer owner has superseded (a partition transferred, a replacement replica started, a fencing token expired). The commit was refused; the worker stops so the current owner proceeds alone. Worker-fatal on purpose: the message is fine, this writer is not the one allowed to decide it.

code: str = 'VF_STALE_AUTHORITY'

disposition: str = 'worker_fatal'

exception videoflow.core.errors.TransientFailure(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowRuntimeError

Something outside the worker was briefly unavailable. Retry with backoff.

code: str = 'VF_TRANSIENT'

disposition: str = 'transient'

exception videoflow.core.errors.UnobservableState(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowEnvironmentError

A read the decision depended on could not be made — a broker query timed out, a pod listing was denied, a node object was malformed — and the code refused to treat “unknown” as “zero”, “empty” or “complete”. The remedy names the dependency to restore; the decision is retried, never guessed.

code: str = 'VF_STATE_UNKNOWN'

exception videoflow.core.errors.UpstreamAborted(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowRuntimeError

An upstream node terminated abnormally and published an ABORT marker. This node stops too, carrying the originating error, instead of waiting forever for an end-of-stream that is never coming.

code: str = 'VF_UPSTREAM_ABORTED'

disposition: str = 'worker_fatal'

exit_code: int = 4

exception videoflow.core.errors.UpstreamUnavailable(message: str, remedy: str | None = None, **context: Any)

Bases: TransientFailure

A service the node depends on (database, API, model server) is down or throttling.

code: str = 'VF_UPSTREAM_UNAVAILABLE'

exception videoflow.core.errors.VideoflowEnvironmentError(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowError

The code is fine; the machine, cluster, broker or registry is not.

code: str = 'VF_ENVIRONMENT'

exit_code: int = 3

exception videoflow.core.errors.VideoflowError(message: str, remedy: str | None = None, **context: Any)

Bases: Exception

Base of every error videoflow raises on purpose.

• Arguments:

  • message: what went wrong.

  • remedy: what the reader should do about it. Kept out of message so every renderer can present it consistently.

  • context: structured key/values (node, trace_id, replica, path). Never interpolated into message either — they are emitted as log fields and proto fields so they stay queryable.

• Class attributes:

  • code: stable, greppable identifier (VF_GRAPH_CYCLE). The message may be reworded freely; the code may not, because metrics and DLQ queries key on it.

  • exit_code: the process exit status when this reaches the CLI.

code: str = 'VF_UNKNOWN'

context: Dict[str, Any]

exit_code: int = 1

render() → str

The operator-facing rendering every entrypoint prints — the CLI, the provision Job — so a failure reads the same wherever it surfaces: what broke, then what to do about it, then the structured context. Never a traceback, which is a stack of framework internals the reader did not write and cannot act on.

to_dict() → Dict[str, Any]

JSON-safe form, used by the Kubernetes termination log and by tests. Kept explicit (not asdict) because this crosses a process boundary and its shape is a contract.

to_json() → str

exception videoflow.core.errors.VideoflowRuntimeError(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowError

A failure that happened while messages were flowing. The disposition is the only thing the hot path reads — see DeliveryPolicy.action_for.

code: str = 'VF_RUNTIME'

disposition: str = 'transient'

exit_code: int = 4

exception videoflow.core.errors.VideoflowUserError(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowError

Something the author of the flow can fix by editing their code or config.

code: str = 'VF_USER'

exit_code: int = 2

exception videoflow.core.errors.WorkerFatal(message: str, remedy: str | None = None, **context: Any)

Bases: VideoflowRuntimeError

This worker cannot process any message.

Two things follow, and both matter. The in-flight message is handed back to the broker for a healthy replica rather than blamed — it is never dead-lettered, because the data is fine. And the worker then stops: raising this is the node asserting that nothing it is given will succeed, so continuing would only nak the rest of the stream one message at a time on the way to the same conclusion.

code: str = 'VF_WORKER_FATAL'

disposition: str = 'worker_fatal'

exception videoflow.core.errors.WorkerUnhealthy(message: str, remedy: str | None = None, **context: Any)

Bases: WorkerFatal

Raised by the task loop when the circuit breaker trips: this worker has failed threshold messages in a row, which is the signature of a sick worker rather than of bad data. It stops the run loop so the un-acked inputs go back to a healthy replica.

code: str = 'VF_WORKER_UNHEALTHY'

videoflow.core.errors.as_runtime_error(exc: BaseException, default: str = 'transient', **context: Any) → VideoflowRuntimeError

Wraps whatever a node raised into a classified VideoflowRuntimeError carrying the message’s identity, so everything downstream of the node-call boundary sees a typed error with a disposition, a code and a context.

One of ours is enriched in place rather than re-wrapped: re-wrapping would bury the original code, which is the thing metrics and the DLQ key on.

• Arguments:

  • exc: the original exception.

  • default: disposition for an exception nothing classifies.

  • context: node, trace_id, seq, replica — whatever identifies the message.

videoflow.core.errors.classify(exc: BaseException, default: str = 'transient') → str

The disposition of exc: its own if it is one of ours, else the most recently registered classifier that matches, else default.

• Arguments:

  • exc: the exception a node raised.

  • default: what an unclassified exception means. Per-node overridable via ProcessorNode(on_error = ...).

videoflow.core.errors.error_to_dict(error: Any) → Dict[str, Any]

Normalizes anything that describes a failure into the JSON-safe record the ABORT marker and the termination log carry: one of ours keeps its structure, a bare exception gets a code derived from its type, and an already-normalized dict passes through (which is how an ABORT is relayed hop to hop without the origin’s detail being rewritten at every step).

videoflow.core.errors.raise_for_diagnostics(diagnostics: List[Diagnostic]) → None

Raises a single GraphError listing every error-severity diagnostic, or returns quietly if there are none. Warnings never raise.

• Raises:

  • GraphError: carrying all error diagnostics in .diagnostics.

videoflow.core.errors.register_classifier_for(name: str, disposition: str, resolver: Callable[[], type | None] | None = None) → bool

Registers a classifier for a type that may not be importable, e.g. a CUDA error class that only exists when torch is installed. Returns whether the type resolved and was registered, so callers can register opportunistically without guarding every import themselves.

• Arguments:

  • name: human name of the type, for the log line when it does not resolve.

  • disposition: one of DISPOSITIONS.

  • resolver: returns the class, or None when the package is absent.

videoflow.core.errors.register_error_classifier(exc_type: Type[BaseException], disposition: str) → None

Maps a third-party exception type onto a disposition, so the retry ladder can reason about failures from libraries videoflow does not own — a component cannot subclass torch.cuda.OutOfMemoryError, but it can register it:

register_error_classifier(torch.cuda.OutOfMemoryError, WORKER_FATAL)

Register on import of the package that raises the type; every flow using that component then inherits the correct behaviour.

• Arguments:

  • exc_type: the exception class to classify. Subclasses match too.

  • disposition: one of DISPOSITIONS.

• Raises:

  • ValueError: if disposition is not a known one (the message names them).

videoflow.core.errors.registered_error_classifiers() → List[tuple]

The (exception type, disposition) pairs currently registered, in registration order.

videoflow.core.flow module

class videoflow.core.flow.Flow(consumers: List[ConsumerNode], flow_type: str = 'realtime', flow_id: str | None = None)

Bases: object

Represents a flow of data from producer nodes to consumer nodes, over the directed acyclic graph formed by however you’ve wired up Node instances via child(*parents).

• Arguments:

  • consumers: a list of consumer nodes of type videoflow.core.node.ConsumerNode. Producers are discovered automatically by walking parents back from these.

  • flow_type: one of ‘realtime’ or ‘batch’. Controls the message broker’s retention/discard policy for every edge in the flow (drop-when-full vs. block/at-least-once).

  • flow_id: a stable identifier for this flow, used to namespace broker subjects and Kubernetes resources. Auto-generated if not given.

property flow_id: str

property flow_type: str

join() → None

Blocking method. Will make the process that calls this method block until the flow finishes running naturally.

run(execution_engine: ExecutionEngine, run_id: str | None = None) → None

Starts the flow using the given ExecutionEngine (e.g. videoflow.engines.local.LocalProcessEngine or videoflow.engines.kubernetes.KubernetesExecutionEngine). Non-blocking: returns once tasks have been allocated/started.

• Arguments:

  • run_id: optional stable per-run id (auto-generated if omitted). Every broker stream/subject for this run is namespaced by it, so re-running the same flow never collides with a previous run’s streams.

property run_id: str | None

The id of the most recent (or in-progress) run, or None before run().

stop() → None

Blocking method. Stops the flow. Makes the execution environment send a flow termination signal.

tasks_data() → List[tuple]

topological_sort() → list

Returns the topologically-sorted list of nodes in this flow. Exposed so the Kubernetes manifest-generation CLI (videoflow.deploy.cli) can inspect the graph without needing to call .run().

videoflow.core.flow.build_tasks_data(graph_engine: GraphEngine) → List[tuple]

Turns a validated GraphEngine into the list of (node, parent_names, is_last) tuples that both the local execution engine and the Kubernetes compiler (videoflow.core.compiler) use to allocate one task per node.

• Returns:

  • tasks_data: list of tuples (node, parent_names : [str], is_last : bool)

videoflow.core.graph module

Validation and topological sorting of a computation graph.

Validation collects every problem in one pass and reports them together, rather than raising on the first. The difference is not cosmetic: a graph is built on one machine and run on many, so a mistake here is cheap to find now and expensive to find later — and making the author fix one problem, re-run, and discover the next is the slowest possible way to spend that cheapness. This is the contract a compiler has and an interpreter does not.

class videoflow.core.graph.GraphEngine(producers: List[ProducerNode], consumers: List[ConsumerNode])

Bases: object

Validates and topologically sorts a computation graph.

• Arguments:

  • producers: list of ProducerNode instances that are the roots of the graph. Any number of producers is supported (a flow may ingest from several independent sources, e.g. multiple cameras, and fan them into shared downstream processors).

  • consumers: list of ConsumerNode instances that are the leaves of the graph.

• Raises:

  • videoflow.core.errors.GraphError listing every problem found: a root that is not a ProducerNode, a cycle, a consumer unreachable from any producer, duplicate node names, or a replicated join with no partition key. Warnings are logged rather than raised.

topological_sort() → list

videoflow.core.graph.validate(producers: List[ProducerNode], consumers: List[ConsumerNode], tsort: List[Node] | None = None) → List[Diagnostic]

Every problem with a graph, in one pass.

• Arguments:

  • producers: the roots.

  • consumers: the leaves.

  • tsort: the topological sort, when the caller already has one. Omit and it is computed — but only if the graph is acyclic, since a cycle makes the sort meaningless.

• Returns:

  • diagnostics in a stable order: structural errors first (they make the later checks unreliable), then per-node ones, then warnings. Empty means the graph is sound.

videoflow.core.node module

class videoflow.core.node.AssetRequirement(path: str, sha256: str, portable: bool = True)

Bases: object

One file a node depends on: path on the worker’s filesystem, the sha256 hex digest its bytes must have, and whether the asset is portable — available on every host through a claim, an image layer or a download — or local-only (a hostPath that exists on one machine). A hostPath is never evidence that the same bytes exist everywhere; a local-only asset constrains where the node may run, and a relocated worker finds out at open time, explicitly.

path: str

portable: bool = True

sha256: str

class videoflow.core.node.ConsumerNode(metadata: bool = False, name: str | None = None, join_policy: JoinPolicy | dict | None = None, idempotent: bool = False, delivery: str | None = None, on_error: str | None = None, **kwargs: Any)

Bases: ErrorHandlingMixin, Leaf

• Arguments:

  • metadata (boolean): By default is False. If True, instead of receiving output of parent nodes, receives metadata produced by parent nodes.

  • delivery (str): 'at-least-once' or 'best-effort' — see ErrorHandlingMixin. A sink is the usual reason to override the flow’s preset: a REALTIME flow may want freshest-wins frames but a durable alert sink.

  • on_error (str): see ErrorHandlingMixin.

  • name (str): see Node.

Class-level declaration effect_guarantee (RFC 0006, RUN-017): what this sink can promise about its external effects — 'at_least_once' (the default: a redelivery may repeat an effect) or 'idempotent_key' (the sink applies each effect through its external system’s own idempotency key or transaction, keyed by ctx.input_key, so one logical effect occurs). A runtime marker alone never certifies exactly-once; only the second declaration lets the planner admit exactly_once_effects for the sink.

consume(item: Any) → None

Method definition that needs to be implemented by subclasses.

• Arguments:

  • item: the item being received as input (or consumed).

effect_guarantee: str = 'at_least_once'

property idempotent: bool

If True (and the flow has a Redis blob/idempotency store), the consumer’s side effects are deduplicated across redelivery.

property join_policy: JoinPolicy | None

property metadata: bool

property partition_by: str | None

partition_key_policy: Dict[str, Any] | None = None

How a partitioned replica treats a record whose partition key is unusable (RFC 0006, RUN-020): None is the default PartitionKeyPolicy (reject: dead-lettered as VF_POISON_PARTITION_KEY); a class may declare {'invalid': 'fallback', 'fallback_partition': 0} instead. A class attribute, so get_params() and the compiled spec are unchanged.

class videoflow.core.node.ErrorHandlingMixin

Bases: object

The two knobs that let a node opt out of its flow’s failure defaults.

They live on the node types that receive messages (processors and consumers); a producer has no input to retry or dead-letter. Both are plain strings so get_params() stays JSON-serializable across the worker boundary, and both default to None meaning “inherit the flow type’s preset” (see videoflow.core.policies.DeliveryPolicy.default_for).

The reason these exist at all: loss tolerance is a property of what a node does with a message, not of the flow it happens to live in. A REALTIME flow whose frame pipeline wants freshest-wins may still have a final sink writing alerts to a database, where a dropped message is a missed incident rather than a stale frame.

property delivery: str | None

This node’s delivery-mode override, or None to inherit the flow’s.

delivery_policy() → Dict[str, Any] | None

The overrides as a dict for NodeSpec.delivery, or None when the node overrides nothing — so a flow that never touches these ships exactly the manifests and env it always did.

property on_error: str | None

This node’s default disposition for unclassified exceptions, or None.

class videoflow.core.node.FunctionProcessorNode(processor_function: Callable[[Any], Any], nb_tasks: int = 1, device_type: str = 'cpu', name: str | None = None, **kwargs: Any)

Bases: ProcessorNode

get_params() → NoReturn

Returns a JSON-serializable dict of keyword arguments sufficient to reconstruct an equivalent node via type(node)(**node.get_params()). This is what lets a worker process (running in its own container, with no access to the object that built the graph) recreate the exact node it is responsible for running.

Default implementation: walks every class in this node’s MRO that declares its own __init__, inspects that __init__’s named parameters (skipping *args/**kwargs), and for each parameter name looks up self._<name> and then self.<name>. This matches the prevailing convention in this codebase of storing constructor arguments verbatim as an underscore-prefixed attribute of the same name.

Subclasses whose constructor arguments aren’t stored under a matching attribute name, or that accept non-serializable arguments (e.g. references to other nodes, as TaskModuleNode does), must override this method.

process(inp: Any) → Any

Method definition that needs to be implemented by subclasses.

• Arguments:

  • inp: object or list of objects being received for processing from parent nodes.

• Returns:

  • the output being consumed by child nodes.

videoflow.core.node.JoinPolicyArg

a policy object, the plain dict it serializes to (how it arrives when a worker reconstructs the node), or None.

Type:

What a node’s join_policy= argument accepts

alias of JoinPolicy | dict | None

class videoflow.core.node.Leaf(*args: Any, **kwargs: Any)

Bases: Node

Node with no children.

class videoflow.core.node.ModuleNode(entry_node: Node, exit_node: Node, *args: Any, **kwargs: Any)

Bases: Node

Module node that wraps a subgraph of computation. Each node of the Module must be a ProcessorNode or a ModuleNode itself. For simplicity, a module node has exaclty one node as entry point, and exactly one node as exit point. If for some reason a ModuleNode has flag one_process set to True:

• Then any module within the subgraph must also be of that type, or an exception will be thrown.

• No process inside the module can be allocated to a gpu, or an exception will be thrown

• Arguments:

  • entry_node (Node): The node that sits at the top of the subgraph

  • exit_node (Node): The node that sits at the top of the subgraph

• Raises:

  • ValueError if:

    • There is at least one node in the sequence that is not instance of ProcessorNode or of ModuleNode

    • There is a cycle in the subgraph

    • The exit_node is not reachable from the entry_node

    • The flag one_process is set to True, and any of the following conditions is true:

      • There is a ModuleNode within the subgraph that does not have that flag set to true too.

      • There is at least one node in the sequence that has device_type GPU

get_params() → NoReturn

Returns a JSON-serializable dict of keyword arguments sufficient to reconstruct an equivalent node via type(node)(**node.get_params()). This is what lets a worker process (running in its own container, with no access to the object that built the graph) recreate the exact node it is responsible for running.

Default implementation: walks every class in this node’s MRO that declares its own __init__, inspects that __init__’s named parameters (skipping *args/**kwargs), and for each parameter name looks up self._<name> and then self.<name>. This matches the prevailing convention in this codebase of storing constructor arguments verbatim as an underscore-prefixed attribute of the same name.

Subclasses whose constructor arguments aren’t stored under a matching attribute name, or that accept non-serializable arguments (e.g. references to other nodes, as TaskModuleNode does), must override this method.

property nodes: List[ProcessorNode]

class videoflow.core.node.Node(name: str | None = None, image: str | None = None)

Bases: object

Represents a computational node in the graph. It is also a callable object. It can be call with the list of parents on which it depends.

• Arguments:

  • name (str): a unique, stable identifier for this node within a flow. Used as the node’s identity everywhere outside the process that built the graph: message broker subject names, Kubernetes resource names, and logging. If not given, one is auto-generated from the class name and a per-class construction counter; uniqueness within a flow is validated when the flow is built (see videoflow.core.graph.GraphEngine), not at construction time, since a collision can only be detected once the whole graph is known.

  • image (str): the container image ref that this node’s worker runs in on Kubernetes, e.g. ghcr.io/me/app:v1. Declare it here when a node intrinsically needs a specific environment (a GPU model image, say). If omitted, the image is taken from the deploy-time default (--image); a deploy-time --image-override beats both. Ignored by the local engine, which runs workers in the current Python environment.

Two class-level declarations (RFC 0006, plan Phase 3) describe how a node’s results may be recovered; they are class attributes rather than constructor parameters so a node’s get_params() — and with it every compiled spec — is unchanged by declaring them:

• deterministic: whether the same input always yields the same output (True by default). A stochastic component sets it False.

• replay_policy: 'recompute' (the default: a redelivered input is processed again, which for a deterministic node yields the same output and the same message id) or 'committed' (a result the worker committed before a crash is re-published byte-for-byte from the runtime ledger, never recomputed — what a nondeterministic component under a reliable profile needs).

Two more (plan Phase 4) describe what a GPU node needs from its grant, checked by the worker before the node is opened (runtime.gpucheck):

• gpu_fallback: 'cpu' (the default: a node granted fewer devices than its gpu_count — none at all on a GPU-less host — still opens, and the worker reports the shortfall and the CPU execution explicitly) or 'none' (a hard requirement: the worker refuses to open the node under a short, empty or unverifiable grant, RUN-044).

• requires_peer_access: a multi-device node whose execution path needs peer access between its devices (NVLink/PCIe P2P). Verified against the delivered devices before readiness; a two-device grant without the property fails the node instead of silently satisfying a count (RUN-043).

Two more (plan Phase 5) declare execution shapes no shipped engine provides; they exist so a declaration is refused at admission — before anything is deployed — rather than silently run as ordinary nodes:

• execution_group: the name of a fused execution group this node belongs to (RUN-035). Members of one group would run in one worker with their internal edges never serialized through the broker; an engine that cannot do that (both shipped engines) rejects the flow with VF_INCOMPATIBLE_PROFILE.

• batching_policy: a dynamic-batching contract the runtime would have to honour — {'max_batch': N, 'max_wait_ms': D, 'fairness': ...} (RUN-036). Distinct from transport fetch batching; rejected the same way.

add_child(child: Node) → None

Adds child to the set of childs that depend on it.

batching_policy: Dict[str, Any] | None = None

property children: Set[Node] | None

Returns a set of the child nodes

close() → None

This method is called by the task running after finishing doing all consuming, processing or producing because of and end signal receival. Should be used to close any resources that were opened by the open() method, such as files, tensorflow sessions, etc.

deterministic: bool = True

execution_group: str | None = None

get_params() → Dict[str, Any]

Returns a JSON-serializable dict of keyword arguments sufficient to reconstruct an equivalent node via type(node)(**node.get_params()). This is what lets a worker process (running in its own container, with no access to the object that built the graph) recreate the exact node it is responsible for running.

Default implementation: walks every class in this node’s MRO that declares its own __init__, inspects that __init__’s named parameters (skipping *args/**kwargs), and for each parameter name looks up self._<name> and then self.<name>. This matches the prevailing convention in this codebase of storing constructor arguments verbatim as an underscore-prefixed attribute of the same name.

Subclasses whose constructor arguments aren’t stored under a matching attribute name, or that accept non-serializable arguments (e.g. references to other nodes, as TaskModuleNode does), must override this method.

gpu_fallback: str = 'cpu'

property image: str | None

The container image ref declared for this node (or None to use the deploy-time default).

property name: str

The stable, unique-within-a-flow string identity of the node. This is the identifier used for broker subjects, Kubernetes resource names, and node reconstruction in worker processes.

open() → None

This method is called by the task runner before doing any consuming, processing or producing. Should be used to open any resources that will be needed during the life of the task, such as opening files, tensorflow sessions, etc.

property parents: List[Node] | None

Returns a list with the parent nodes

remove_child(child: Node) → None

replay_policy: str = 'recompute'

required_assets() → List[AssetRequirement]

The files this node needs on the machine it runs on, with their content identity (plan Phase 4, RUN-033). The worker verifies every one before the node is opened: a missing file or different bytes end the worker with ResourceUnavailable naming the host, rather than processing with a model or input that is not the declared one. Default: none. Override in a node whose constructor takes the paths (they are known in the worker, after get_params() rebuilt the node).

requires_peer_access: bool = False

class videoflow.core.node.OneTaskProcessorNode(device_type: str = 'cpu', name: str | None = None, **kwargs: Any)

Bases: ProcessorNode

Used for processes that keep internal state so they are easily parallelizable. The main use of this class if for processes that can only run one task, such as trackers and aggregators.

class videoflow.core.node.ProcessorNode(nb_tasks: int = 1, device_type: str = 'cpu', name: str | None = None, partition_by: str | None = None, join_policy: JoinPolicy | dict | None = None, gpu_count: int | None = None, gpu_memory_gib: int | float | None = None, delivery: str | None = None, on_error: str | None = None, **kwargs: Any)

Bases: ErrorHandlingMixin, Node

• Arguments:

  • nb_tasks (int): number of parallel replicas to allocate for this processor.

  • device_type (str): videoflow.core.constants.CPU or GPU.

  • partition_by (str): if set (and nb_tasks > 1), replicas partition the input by a key instead of competing for it: each message is handled by exactly one replica chosen by hash(key) % nb_tasks. The key is the message’s trace_id (the special value 'trace_id', which co-locates both halves of a join on the same replica) or a metadata field name. Required for a multi-parent (join) node with nb_tasks > 1.

  • join_policy (JoinPolicy | dict): for multi-parent nodes, how to handle a join group that never completes (timeout + missing policy). Defaults per flow type when unset.

  • gpu_count (int): whole physical GPUs each replica is granted (device_type=GPU only — a positive count on a CPU node is a build error). N > 1 grants N whole devices on one host, visible as cuda:0..N-1 (RFC 0003); locally the engine partitions the host’s devices to match. Unset (None, the default) means one device; whether the 1 was explicit or defaulted is kept in gpu_provenance.

  • gpu_memory_gib (int | float): GPU memory this node needs, in GiB (device_type=GPU only). Under --gpu-mode mix each replica gets an exclusive MIG slice of at least this size, chosen by the layout solver — the card is shared, the slice is not (RFC 0004). Other modes ignore it (the node gets a whole device). Mutually exclusive with gpu_count > 1: a model cannot span MIG slices, so a node either declares a fraction of one device or whole devices, never both.

  • delivery (str): 'at-least-once' or 'best-effort', overriding the flow type’s preset — see ErrorHandlingMixin.

  • on_error (str): disposition for exceptions nothing classifies — see ErrorHandlingMixin.

  • name (str): see Node.

change_device(device_type: str) → None

property device_type: str

Returns the preferred device type to use to run the processor’s code

property gpu_count: int

Whole physical GPUs each replica is granted (device_type=GPU only).

property gpu_memory_gib: int | float | None

GPU memory demand in GiB (drives the mix strategy’s MIG slice choice), or None.

property gpu_provenance: Dict[str, str]

'node' for an explicit argument, 'default' for the built-in default, and 'descriptor' when a remote component’s component.yaml supplied it (see videoflow.core.provenance). Deliberately not a NodeSpec field — the serialized spec is unchanged; videoflow.core.compiler.gpu_provenance collects it per flow.

Type:

Where each GPU requirement value came from, as {field: source} over device_type, gpu_count and gpu_memory_gib

property join_policy: JoinPolicy | None

Returns the JoinPolicy object (or None), reconstructed from the stored dict.

property nb_tasks: int

Returns the number of tasks to allocate to this processor

property partition_by: str | None

partition_key_policy: Dict[str, Any] | None = None

How a partitioned replica treats a record whose partition key is unusable (RFC 0006, RUN-020): None is the default PartitionKeyPolicy (reject: dead-lettered as VF_POISON_PARTITION_KEY); a class may declare {'invalid': 'fallback', 'fallback_partition': 0} instead. A class attribute, so get_params() and the compiled spec are unchanged.

process(inp: Any) → Any

Method definition that needs to be implemented by subclasses.

• Arguments:

  • inp: object or list of objects being received for processing from parent nodes.

• Returns:

  • the output being consumed by child nodes.

class videoflow.core.node.ProducerNode(is_finite: bool = True, name: str | None = None, **kwargs: Any)

Bases: Node

The producer node does not receive input, and produces input. Each time the next() method is called, it produces a new input.

It would have been more natural to implement the ProducerNode as a generator, but a node is shipped to its worker as (class path, get_params()) and rebuilt there with type(node)(**get_params()) — a generator has no such reconstructable form, so it cannot cross into the worker process.

• Arguments:

  • is_finite (bool): True (the default) if a call to next() will eventually raise StopIteration on its own (e.g. reading a fixed video file). Set to False for producers that read from an unbounded/live source (e.g. an RTSP stream) and only stop when told to. The Kubernetes execution engine uses this to decide whether to deploy the producer as a Job (finite) or a Deployment (infinite).

  • name (str): see Node.

Class-level declaration replayable (RFC 0006 MSGID-6, off by default): the source has a stable position of its own (a frame index, a record offset), so a re-run or a restart re-mints the identical message ids and downstream deduplication and sink idempotency engage. A replayable producer implements seek(offset) so a replacement resumes after the last accepted offset the runtime checkpointed; a live source (the default) mints a fresh capture epoch per process instead (MSGID-5).

analysis_version (MSGID-6, replayable sources only): a deliberately new analysis of the same media declares a version and mints {node}:{analysis_version}:{offset}, a namespace of its own, instead of re-minting — and colliding with — the identities of the earlier analysis. None (the default) is the plain {node}:{offset} form.

analysis_version: str | None = None

property is_finite: bool

next() → Any

Returns next produced element.

Raises StopIteration after the last element has been produced and a call to self.next happens.

replayable: bool = False

seek(offset: int) → None

Position the source so the next next() yields item offset + 1 (a replayable producer resuming after a restart; offset is the last accepted item, 0 for none). The default ignores it: a live source has no position to return to.

class videoflow.core.node.TaskModuleNode(entry_node: ProcessorNode, exit_node: ProcessorNode, nb_tasks: int = 1, name: str | None = None, **kwargs: Any)

Bases: ProcessorNode

Processor node that wraps a graph of processor nodes. This has the effect that instead of allocating one task per processor node in the graph, only one task process is allocated for the entire subgraph.

• Arguments:

  • entry_node (Node): The node that sits at the top of the subgraph

  • exit_node (Node): The node that sits at the top of the subgraph

  • nb_tasks (int) The number of parallel tasks to allocate

• Raises:

  • ValueError if:

    • There is at least one node in the subgraph that is not instance of ProcessorNode

    • nb_tasks parameter is greater than one and there is at least one node in the sequence that derives from OneTaskProcessorNode.

    • There is at least one node in the sequence that has device_type GPU

    • The subgraph has less than one node.

    • There is a node of type TaskModuleNode among the nodes of the subgraph.

Note: because it wraps live references to other nodes, TaskModuleNode does not support get_params()-based reconstruction and is not yet supported by the distributed Kubernetes execution path (see videoflow.core.compiler).

get_params() → NoReturn

Returns a JSON-serializable dict of keyword arguments sufficient to reconstruct an equivalent node via type(node)(**node.get_params()). This is what lets a worker process (running in its own container, with no access to the object that built the graph) recreate the exact node it is responsible for running.

Default implementation: walks every class in this node’s MRO that declares its own __init__, inspects that __init__’s named parameters (skipping *args/**kwargs), and for each parameter name looks up self._<name> and then self.<name>. This matches the prevailing convention in this codebase of storing constructor arguments verbatim as an underscore-prefixed attribute of the same name.

Subclasses whose constructor arguments aren’t stored under a matching attribute name, or that accept non-serializable arguments (e.g. references to other nodes, as TaskModuleNode does), must override this method.

process(*inp: Any) → Any

Method definition that needs to be implemented by subclasses.

• Arguments:

  • inp: object or list of objects being received for processing from parent nodes.

• Returns:

  • the output being consumed by child nodes.

videoflow.core.policies module

Policies that tune how a node treats its inputs: how a multi-parent node aligns them (JoinPolicy), and what happens to one when the node fails on it (DeliveryPolicy).

Both are attached to a node and travel with it as a plain dict, so they survive get_params() serialization into a worker that never sees the object graph.

A JoinPolicy decides two things:

• How input groups are formed (mode): by lineage (trace, the default — inputs that descend from the same originating message of a single producer) or by event time (time — inputs whose event_ts fall within a tolerance of each other, which is how streams from independent producers such as multiple cameras and sensors are fused).

• What to do with a group that never completes — a real possibility when one branch drops a message (REALTIME) or stalls: timeout + missing policy, and for time-aligned joins, an optional quorum that lets a late group emit with the parents it has.

A DeliveryPolicy decides what a failure costs. Its defaults come from the flow type, because the correlation is real — a REALTIME flow wants freshest-wins and a BATCH flow wants completeness — but each axis is separately overridable per node, because loss tolerance is really a property of what a node does with a message, not of the flow it happens to live in. The case that forces the split: a REALTIME flow whose frame pipeline wants drop-on-failure but whose final sink writes alerts to a database, where a dropped message is a missed incident.

videoflow.core.policies.ACTION_NAK = 'nak'

What the messenger does with a failed input group. Returned by DeliveryPolicy.action_for; executed by NATSMessenger.fail_inputs.

videoflow.core.policies.AT_LEAST_ONCE = 'at-least-once'

How much a node is willing to lose. at-least-once retries and dead-letters; best-effort drops a failed message so the freshest one wins.

videoflow.core.policies.DEFAULT_DLQ_SAMPLE_PER_MINUTE = 5

Dead-letters admitted per (code, node) per minute under DLQ_SAMPLED.

videoflow.core.policies.DEFAULT_MAX_RETRIES = 3

Default number of times a message is retried after the first delivery attempt, for an at-least-once node. max_deliver = retries + 1.

videoflow.core.policies.DLQ_FULL = 'full'

What reaches the dead-letter queue. sampled exists so best-effort failures stop being invisible: “we drop things” is true of load shedding and false of exceptions, and deleting the evidence of a bug is not a retention policy.

class videoflow.core.policies.DeliveryPolicy(delivery: str = 'at-least-once', max_retries: int = 3, dlq: str = 'full', on_error: str | None = None, sample_per_minute: int = 5)

Bases: object

What a node’s failure costs: whether the message is retried, dropped or dead-lettered, and how many attempts it gets.

The whole point of separating this from the flow type is that the action now depends on the error’s disposition (see videoflow.core.errors), not only on how patient the flow is. A message that will never parse should not burn four attempts and fourteen seconds of backoff on its way to the same dead-letter queue; and a worker whose GPU has wedged should not blame — and dead-letter — every message it touches.

• Arguments:

  • delivery: at-least-once (retry, then dead-letter) or best-effort (drop on failure). Defaults per flow type.

  • max_retries: redelivery attempts after the first, for an at-least-once node. Ignored for best-effort, which never redelivers.

  • dlq: full (every exhausted message), sampled (a bounded number per code per minute — the best-effort default) or off.

  • on_error: the disposition assumed for an exception nothing has classified. None means the framework default (transient).

  • sample_per_minute: cap for dlq = 'sampled'.

action_for(disposition: str, num_delivered: int) → str

The action for a failed input group, given how the error was classified and how many times the broker has delivered it.

The table, which spec/PROTOCOL.md §7.3 states normatively:

disposition	best-effort	at-least-once
poison	sampled DLQ, term	DLQ immediately, term
transient	term	nak until budget, then DLQ
worker_fatal	nak	nak

worker_fatal never dead-letters in either mode: the message is fine, this worker is not, so it goes back for another replica (or this pod’s replacement) while the circuit breaker takes the worker out.

• Arguments:

  • disposition: one of videoflow.core.errors.DISPOSITIONS.

  • num_delivered: the broker’s delivery count for this message, starting at 1 for the first attempt.

• Raises:

  • ValueError: on an unknown disposition (the message names the known ones).

classmethod default_for(flow_type: str) → DeliveryPolicy

BATCH is at-least-once with a full DLQ (completeness matters). REALTIME is best-effort with a sampled DLQ — it drops failed messages, but keeps a bounded specimen of each distinct failure so a bug is still diagnosable.

classmethod from_dict(d: Dict[str, Any] | None) → DeliveryPolicy | None

property max_deliver: int

Broker-side delivery cap for this node’s durable consumers. A best-effort node never redelivers (1); an at-least-once node gets retries + 1. Provisioning and the messenger must agree on this, so both read it here.

classmethod resolve(flow_type: str, override: Dict[str, Any] | None = None, max_retries: int | None = None) → DeliveryPolicy

The effective policy for one node: the flow-type preset, with a node’s own delivery=/on_error= applied on top, and a deployment-level retry count (VF_MAX_RETRIES) applied last.

Overriding delivery alone also moves the DLQ mode and the retry budget with it, because those are what the mode means — an at-least-once sink in a REALTIME flow wants a real DLQ, not a sampled one, and would otherwise silently keep max_retries = 0.

retry_delay(num_delivered: int, jitter: float = 1.0) → float

Backoff before a redelivery. Jittered because a deterministic schedule makes N replicas that failed together retry together — a thundering herd against whatever just recovered.

• Arguments:

  • num_delivered: the broker’s delivery count (1 for the first attempt).

  • jitter: multiplier in [0.5, 1.5]; the caller supplies the random draw so the function stays pure and testable.

to_dict() → Dict[str, Any]

videoflow.core.policies.INVALID_KEY_REJECT = 'reject'

What a partitioned node does with a record whose partition key is unusable — absent, None, empty, or not a string (RUN-020). Never hashed as the string "None" under RFC 0006: that silent fallback made every invalid record an undeclared hot partition.

videoflow.core.policies.JOIN_TRACE = 'trace'

How input groups are formed at a multi-parent node.

class videoflow.core.policies.JoinPolicy(timeout_seconds: float | None = None, missing: str = 'drop', max_pending: int = 256, mode: str = 'trace', tolerance_ms: float | None = None, quorum: int | None = None, collect: dict | None = None)

Bases: object

• Arguments:

  • timeout_seconds: how long to wait for the rest of a join group before applying missing (or emitting a quorum group). None means no timeout (wait forever). For mode='time' this is the lateness bound: the answer to “how long after a window’s first message may a straggler still arrive”.

  • missing: one of drop / wait / error (see constants above).

  • max_pending: hard cap on buffered incomplete groups; the oldest is evicted (as drop) beyond this, protecting against unbounded memory.

  • mode: trace (default) or time. time groups inputs whose event_ts (stamped by the producers) are within tolerance_ms of each other, instead of requiring a shared upstream trace id — required to join branches that descend from different producers.

  • tolerance_ms: (time mode only, required) two messages from different parents belong to the same group when their event times differ by at most this much. Pick it below the fastest parent’s inter-message period (e.g. for 50fps cameras, < 20ms).

  • quorum: (time mode only) minimum number of synchronized parents that must be present for a timed-out group to still be emitted (missing parents are passed to process() as None). None (default) means all parents are required and a timed-out group is handled by missing. With N cameras, quorum=k gives “emit with at least k views”. Requires timeout_seconds.

  • collect: (time mode only) dict {parent_name: window_ms} marking high-rate parents (e.g. a 500Hz sensor vs 50fps cameras) that should not join 1:1: every message of that parent whose event_ts is within window_ms of the group’s time is delivered as a list in that parent’s position. Collect parents never gate completeness and don’t count toward quorum; a group holds for the largest collect window after completing so trailing samples can arrive.

classmethod default_for(flow_type: str) → JoinPolicy

BATCH waits (completeness matters; bounded by max_pending). REALTIME times out and drops (a dropped sibling frame must not stall the join forever).

classmethod from_dict(d: Dict[str, Any] | None) → JoinPolicy | None

to_dict() → Dict[str, Any]

working_set(nb_parents: int) → int

The most deliveries a join may hold un-acked while it waits for groups to complete: max_pending incomplete groups, each holding up to nb_parents - 1 halves, plus the group being assembled. The credit a join’s durables are provisioned with must cover it, or an adversarial parent ordering (every A before any B) fills the credit with halves that can never complete and the join deadlocks (RUN-005).

videoflow.core.policies.MISSING_DROP = 'drop'

What to do with an incomplete join group once it times out.

videoflow.core.policies.ORDER_ARRIVAL = 'arrival'

in the order they arrive, or by their sequence number within a bounded horizon.

Type:

How a stateful node orders the records of one partition (RUN-021)

class videoflow.core.policies.OrderingPolicy(mode: str = 'arrival', horizon: int = 8, late: str = 'drop')

Bases: object

The declared temporal semantics of a keyed stateful node, applied through a ReorderBuffer per partition key so no arrival-order race changes them.

• Arguments:

  • mode: arrival (records are applied as they come; the default and today’s behaviour) or sequence (records are released in sequence order; a gap is waited for up to horizon later records).

  • horizon: (sequence) how many later records may be held back waiting for a gap before it is given up on.

  • late: what happens to a record that arrives after its slot was given up: drop or mark (delivered with late = True).

classmethod from_dict(d: Dict[str, Any] | None) → OrderingPolicy

to_dict() → Dict[str, Any]

class videoflow.core.policies.PartitionKeyPolicy(invalid: str = 'reject', fallback_partition: int = 0)

Bases: object

• Arguments:

  • invalid: reject (the default) or fallback.

  • fallback_partition: the replica index an invalid key routes to under fallback; must be below the node’s nb_tasks, checked when bound.

classmethod from_dict(d: Dict[str, Any] | None) → PartitionKeyPolicy

static is_valid_key(value: Any) → bool

A usable partition key: a non-empty string, or an int/bool — never None, ‘’ or a container.

to_dict() → Dict[str, Any]

class videoflow.core.policies.ReorderBuffer(policy: OrderingPolicy, start: int = 1)

Bases: object

The reference interpreter of an OrderingPolicy for one partition key. offer(seq, record) returns the records that may be applied now, oldest first, each as (seq, record, late): under arrival every record at once; under sequence the next expected record and whatever follows it contiguously, a duplicate never twice, and a gap only once horizon later records have piled up behind it — at which point the missing slots are given up on and a record for one of them is late (dropped, or marked).

property expected: int

flush() → list

End of input: release everything held, in order, marked as it is (nothing more is coming).

offer(seq: int, record: Any) → list

classmethod restore(policy: OrderingPolicy, snapshot: Dict[str, Any]) → ReorderBuffer

A buffer positioned as snapshot recorded it (the inverse of snapshot).

snapshot() → Dict[str, Any]

The buffer’s whole position — next expected, held records, applied sequences, drop count — as a JSON-serializable dict, so a stateful node can checkpoint it with its own state (RUN-021/RUN-022) and a replacement resumes the declared ordering exactly where the crashed process left it. Held records are the node’s own values and must be JSON-serializable too.

videoflow.core.provenance module

Provenance-aware resolution of a node’s GPU requirement (RFC 0003 / RFC 0004).

A node’s GPU requirement — device_type, gpu_count, gpu_memory_gib and, at deploy time, the extended-resource name — can be declared in several places: an explicit argument on the node or on component(...), a remote component’s component.yaml (spec.resources.gpu, and its device list), the deploy CLI’s --gpu-resource-name default, and a GPU strategy’s own resolution (the mix solver stamps a sharer’s MIG profile). Before this module the merge was spread over core/remote.py, core/node.py and deploy/gpu.py with x if y is None else z chains: the precedence was implicit, nobody could tell afterwards where a resolved value had come from, and two declarations that genuinely contradicted each other were silently overridden or reported without naming the loser.

The resolver here is one pure function over tagged declarations. Every candidate value arrives as a Declaration naming its field, its source and whether it is hard (a requirement) or soft (a default another source may override). The rules, which reproduce exactly what the framework did before for every input it accepted:

1. A hard declaration always beats a soft one. Among declarations of equal hardness the source order SOURCE_PRECEDENCE decides: node, strategy, descriptor, cli-default, default. So an explicit gpu_count= beats the descriptor’s spec.resources.gpu.count, which beats the built-in 1; a strategy’s resolved resource name beats --gpu-resource-name, which beats nvidia.com/gpu.

2. Two hard declarations of one field from different sources that disagree are a contradiction (GpuRequirementConflict, a CapabilityError whose remedy names both sources). The descriptor’s device list counts as a hard device_type declaration when it names a single device: a component that only runs on GPU contradicts a node asking for CPU.

3. A GPU-only requirement (gpu_count > 1 or a gpu_memory_gib) on a node whose device_type resolves to CPU is a contradiction when the two come from different sources — with one documented exception: a soft memory demand describes the component’s GPU flavor, so a CPU run drops it (recorded in GpuResolution.dropped) rather than failing (RFC 0004). A descriptor’s count default is deliberately not dropped: the author must say gpu_count=1 to run a multi-GPU component on CPU (RFC 0003).

4. gpu_count > 1 and a memory demand are mutually exclusive (a model cannot span MIG slices); from different sources that is a contradiction naming both.

Contradictions within one source — a node passing device_type='cpu' and gpu_count=2 in the same call — are not this module’s business: the node constructor already rejects them with the messages the API documents.

The output is the resolved values plus {field: source} provenance. That record is what ProcessorNode.gpu_provenance and videoflow.core.compiler.gpu_provenance expose; it deliberately lives beside NodeSpec rather than on it, so that a spec’s serialized form — every specs ConfigMap byte — is unchanged.

class videoflow.core.provenance.Declaration(field: str, value: Any, source: str, hard: bool = True)

Bases: object

One candidate value for one field of a node’s GPU requirement.

• Arguments:

  • field: one of FIELDS.

  • value: the declared value (None is a legitimate declaration for gpu_memory_gib: “no memory demand”).

  • source: one of SOURCE_PRECEDENCE.

  • hard: True for a requirement, False for a default another source may override.

field: str

hard: bool = True

source: str

value: Any

exception videoflow.core.provenance.GpuRequirementConflict(message: str, remedy: str | None = None, **context: Any)

Bases: CapabilityError, ValueError

Two declarations of one node’s GPU requirement contradict each other and neither is a default the other may override. The message names both values and both sources; remedy says which of the two to change.

A CapabilityError (code VF_CAPABILITY, exit code 2): the graph asks a component for something its own declaration rules out. It is also a ValueError because component() and the node constructors promise graph authors a ValueError for every build-time misconfiguration, and the CLI converts that class at its boundaries — the taxonomy class is what the CLI renders, the builtin base is what existing callers catch.

class videoflow.core.provenance.GpuResolution(values: Dict[str, ~typing.Any]=<factory>, provenance: Dict[str, str]=<factory>, dropped: Dict[str, str]=<factory>)

Bases: object

The resolver’s answer: the winning value per field and where it came from.

• Attributes:

  • values: field -> resolved value.

  • provenance: field -> the source of the winning declaration.

  • dropped: field -> the source of a soft GPU-only default that was set aside because the node runs on CPU (rule 3 above). Evidence for the provenance table; empty when nothing was dropped.

dropped: Dict[str, str]

provenance: Dict[str, str]

values: Dict[str, Any]

videoflow.core.provenance.SOURCE_CLI_DEFAULT = 'cli-default'

A deploy-level default from the CLI (--gpu-resource-name).

videoflow.core.provenance.SOURCE_DEFAULT = 'default'

What the framework assumes when nobody says anything.

videoflow.core.provenance.SOURCE_DESCRIPTOR = 'descriptor'

spec.resources.gpu and spec.device.

Type:

A remote component’s component.yaml

videoflow.core.provenance.SOURCE_NODE = 'node'

An explicit argument on the node (ProcessorNode(gpu_count = 2), component(..., device_type = 'gpu')).

videoflow.core.provenance.SOURCE_PRECEDENCE = ('node', 'strategy', 'descriptor', 'cli-default', 'default')

Precedence among declarations of equal hardness, highest first. A hard declaration beats a soft one regardless of this order.

videoflow.core.provenance.SOURCE_STRATEGY = 'strategy'

A GPU strategy’s own resolution at deploy time (mix stamping a MIG profile).

videoflow.core.provenance.builtin_defaults() → List[Declaration]

The soft declarations a node makes by saying nothing: CPU, one device, no memory demand.

videoflow.core.provenance.descriptor_declarations(device: Sequence[str], gpu_count: int | None = None, gpu_memory_gib: int | float | None = None) → List[Declaration]

What a component descriptor declares. spec.resources.gpu.count / memoryGiB are soft — RFC 0003/0004 make them defaults, not floors. A single-entry spec.device list is a hard device_type: a component that only ships a GPU flavor requires a GPU (a two-entry list declares nothing; the node’s choice merely has to be one of them, which component() checks).

videoflow.core.provenance.node_declarations(device_type: str, gpu_count: int | None = None, gpu_memory_gib: int | float | None = None) → List[Declaration]

The hard declarations an explicit node / component() call makes. device_type is always declared (the node API has no “unspecified” device); gpu_count and gpu_memory_gib only when the caller passed them — None means “I did not say”, not “zero”.

videoflow.core.provenance.resolve_gpu_requirements(declarations: Iterable[Declaration], subject: str = 'node') → GpuResolution

Resolve a node’s GPU requirement from every declaration made about it.

• Arguments:

  • declarations: the candidates, in any order. Every field the caller cares about should carry a soft SOURCE_DEFAULT declaration (see builtin_defaults) so its provenance is always recorded.

  • subject: how to name the node in an error ("component 'acme/x'").

• Returns:

  • a GpuResolution: values, {field: source} provenance, and the soft GPU-only defaults dropped because the node runs on CPU.

• Raises:

  • GpuRequirementConflict (a CapabilityError and a ValueError): two sources make incompatible hard declarations, or a source’s hard declaration is incompatible with another source’s GPU-only requirement (see the module docstring for the exact rules). The remedy names both sources.

  • ValueError: a declaration names an unknown field or source.

videoflow.core.remote module

Remote (language-agnostic) components in a Python-authored graph.

A component(ref, params=...) node stands in for a component that runs as its own container image and is authored in any language. It behaves like a normal Producer/Processor/Consumer node for graph-building (wiring, validation, the compiler, manifests), but it carries no Python implementation — its next/process/consume run out-of-process in the vendor image, driven by that image’s own SDK speaking the wire protocol (spec/PROTOCOL.md). The Python process only ever builds and compiles the graph; it never imports the component.

The three concrete kinds subclass the existing node base classes so every isinstance dispatch in graph.py/compiler.py/flow.py keeps working unchanged; RemoteNodeMixin marks them for the compiler, which records a component_ref + descriptor instead of a Python node_class.

class videoflow.core.remote.RemoteConsumer(component_ref: str, descriptor: ComponentDescriptor, params: Dict[str, Any], metadata: bool = False, idempotent: bool = False, join_policy: JoinPolicy | dict | None = None, name: str | None = None, image: str | None = None)

Bases: RemoteNodeMixin, ConsumerNode

consume(*inputs: Any) → None

Method definition that needs to be implemented by subclasses.

• Arguments:

  • item: the item being received as input (or consumed).

class videoflow.core.remote.RemoteNodeMixin

Bases: object

Shared state/identity for the three remote node kinds.

property component_command: List[str] | None

property component_ref: str

property descriptor: ComponentDescriptor

get_params() → Dict[str, Any]

The component’s own params, delivered via VF_NODE_PARAMS_JSON. For a native component these are the only params (routing settings reach it via env); for a Python component, the reconstructing worker calls pythonClass(**params), so node-level settings the class needs (nb_tasks/device_type) are merged in — matching what a normal node’s get_params() would carry. Every videoflow node constructor accepts these via **kwargs.

property local_command: List[str] | None

class videoflow.core.remote.RemoteProcessor(component_ref: str, descriptor: ComponentDescriptor, params: Dict[str, Any], nb_tasks: int = 1, device_type: str = 'cpu', partition_by: str | None = None, join_policy: JoinPolicy | dict | None = None, name: str | None = None, image: str | None = None, gpu_count: int | None = None, gpu_memory_gib: int | float | None = None, gpu_provenance: Dict[str, str] | None = None)

Bases: RemoteNodeMixin, ProcessorNode

process(*inputs: Any) → Any

Method definition that needs to be implemented by subclasses.

• Arguments:

  • inp: object or list of objects being received for processing from parent nodes.

• Returns:

  • the output being consumed by child nodes.

class videoflow.core.remote.RemoteProducer(component_ref: str, descriptor: ComponentDescriptor, params: Dict[str, Any], is_finite: bool = True, name: str | None = None, image: str | None = None)

Bases: RemoteNodeMixin, ProducerNode

next() → Any

Returns next produced element.

Raises StopIteration after the last element has been produced and a call to self.next happens.

videoflow.core.remote.component(ref: str | ComponentDescriptor, params: Dict[str, Any] | None = None, name: str | None = None, nb_tasks: int = 1, device_type: str = 'cpu', partition_by: str | None = None, join_policy: JoinPolicy | dict | None = None, image: str | None = None, is_finite: bool | None = None, metadata: bool = False, idempotent: bool = False, gpu_count: int | None = None, gpu_memory_gib: int | float | None = None) → Node

Create a graph node backed by a language-agnostic component described by ref (a path to a component.yaml or, later, an oci:// ref). Returns a Producer/Processor/Consumer node per the descriptor’s role, so it wires into a Python graph exactly like a native node:

from videoflow.core import Flow, component
tracker = component('components/sort', params={'iou_threshold': 0.3})(detector)

Validation happens here, at graph-build time, so a misconfiguration is a build error in the authoring script rather than a crash inside the vendor container: params are checked against the descriptor’s JSON Schema, device_type against the declared devices, the image against what the descriptor provides, and the join policy against what the component says it accepts.

• Arguments:

  • ref: descriptor reference (local path now; oci:// in Phase 6), or an already-loaded ComponentDescriptor.

  • params: component params dict (validated + defaulted against the descriptor).

  • device_type: ‘cpu’ or ‘gpu’ (must be in the descriptor’s device list).

  • image: explicit image ref override (else the descriptor’s image for the device).

  • is_finite: producers only; defaults to the descriptor’s finite.

  • nb_tasks/partition_by/join_policy/metadata/idempotent: as on the native nodes.

  • gpu_count: processors only; as on ProcessorNode (whole physical devices per replica). Resolution order: explicit argument, then the descriptor’s spec.resources.gpu.count (RFC 0003), then 1. The descriptor value is a default, not a floor — a component with a hard minimum should verify in open(). Passing it for a producer or consumer component is an error, not a silent no-op.

  • gpu_memory_gib: processors only; as on ProcessorNode (declared GPU memory demand in GiB, RFC 0004). Resolution order: explicit argument, then the descriptor’s spec.resources.gpu.memoryGiB, then None. Same non-processor rejection as gpu_count.

The GPU resolution runs through videoflow.core.provenance so the node records where each value came from (node.gpu_provenance), and so a genuine contradiction between two sources — a GPU-only component asked to run on CPU, an explicit gpu_count=2 against a descriptor memory default — is rejected with a GpuRequirementConflict (a CapabilityError and a ValueError) whose remedy names both sources.

videoflow.core.supervision module

When to give up — on a message, on a worker, or on a whole flow.

Three policies live here, sharing one reason to change: they all answer “has this stopped being worth waiting for?”, and they are all pure (no I/O, no optional dependencies) so both engines and the task loop can hold them.

• ConsecutiveFailureBreaker — worker-scoped, and specifically for the failures nothing classified. An error that says it is worker-fatal already ends its worker on the spot (see videoflow.core.task); the breaker is what catches the same illness when it arrives unlabelled. Data failures are sparse and independent, worker failures are dense and correlated, and counting a run of them tells the two apart without needing the taxonomy to be right.

• ProgressDeadline — node-scoped. Distinguishes “slow” from “wedged” by watching acks against pending work rather than the wall clock, because a legitimately slow model and a hung one look identical to a timer.

• SupervisionPolicy — engine-scoped. How a dead worker is restarted. Held here rather than in either engine so the local engine and the Kubernetes engine agree by construction: the same object is honoured by a supervisor thread in one and rendered into a Job backoffLimit in the other. They used to diverge, and the divergence meant the development environment was the one place recovery was never exercised.

The lifecycle events at the bottom are the other half of that parity: both engines emit the same records, so one renderer serves both.

class videoflow.core.supervision.ConsecutiveFailureBreaker(threshold: int = 10, node_name: str = '')

Bases: object

Trips when a worker fails threshold messages in a row — the signature of a sick worker rather than of bad data. Any successful ack resets the count, so poison messages interleaved with successful ones never trip it.

This is the backstop for failures that arrive unclassified. A node that raises an explicitly worker-fatal error is believed immediately; the breaker is what notices the same illness when it shows up as a run of ordinary-looking exceptions instead.

The failure mode it exists for: one pod’s GPU wedges at 03:00 behind a library error nothing recognizes and, by 03:20, ten thousand perfectly good messages are in the DLQ while the flow looks healthy from outside. With the breaker the pod dies after ten, Kubernetes replaces it, and those ten are redelivered.

• Arguments:

  • threshold: consecutive failures that trip it. 0 disables the breaker.

  • node_name: named in the raised error, since that is what an operator reads.

check() → None

• Raises:

  • WorkerUnhealthy: if the breaker has tripped. The caller lets this out of the run loop so the un-acked inputs return to the broker for a healthy replica.

property consecutive_failures: int

record_failure(exc: BaseException) → None

A message failed. Counts toward the trip regardless of disposition — the breaker deliberately does not consult the taxonomy, because its whole value is catching the failures the taxonomy got wrong.

record_success() → None

A message was processed and acked: the worker is demonstrably alive.

property tripped: bool

videoflow.core.supervision.DEFAULT_BREAKER_THRESHOLD = 10

Consecutive failures before a worker declares itself sick. Ten is high enough that a run of unlucky-but-independent bad messages does not trip it, and low enough that a wedged worker is out in seconds rather than after it has dead-lettered a stream.

videoflow.core.supervision.DEFAULT_PROGRESS_TIMEOUT_SECONDS = 300

Seconds a node may ack nothing while work is pending before it is declared stalled. Generous, because the false positive (killing a slow but healthy node) is worse than the false negative (a wedged node found a minute later).

class videoflow.core.supervision.EventLog(events: list = <factory>)

Bases: object

Collects lifecycle events for rendering. Deliberately a plain list rather than a callback interface: there is exactly one consumer (the CLI renderer), and dump_failed_logs/report_failures becoming one function is the point.

emit(event: object) → None

events: list

failed_nodes() → list

Node names that gave up, in first-seen order.

restart_count(node: str | None = None) → int

class videoflow.core.supervision.FlowStalled(detail: str)

Bases: object

The flow can never finish: unschedulable pods, or a node that stopped progressing.

detail: str

class videoflow.core.supervision.NodeExited(node: str, replica: int, code: int, reason: Dict[str, object] | None = None)

Bases: object

A worker exited. code is the process exit status; reason is the structured cause it reported (from its termination log), when it managed to.

property clean: bool

code: int

property disposition: str | None

node: str

reason: Dict[str, object] | None = None

replica: int

class videoflow.core.supervision.NodeGaveUp(node: str, replica: int, attempts: int, reason: Dict[str, object] | None = None)

Bases: object

A worker will not be restarted again — the flow cannot complete on its own from here, so this is what triggers the supervisor’s control-abort.

attempts: int

node: str

reason: Dict[str, object] | None = None

replica: int

class videoflow.core.supervision.NodeRestarted(node: str, replica: int, attempt: int, delay: float)

Bases: object

attempt: int

delay: float

node: str

replica: int

class videoflow.core.supervision.NodeStarted(node: str, replica: int, attempt: int = 0)

Bases: object

A worker was launched. attempt is 0 for the first launch.

attempt: int = 0

node: str

replica: int

class videoflow.core.supervision.ProgressDeadline(timeout_seconds: float = 300, pending_probe: Callable[[], ~typing.Any] | None=None, node_name: str = '', clock: Callable[[], float]=<built-in function monotonic>, unknown_grace_seconds: float | None = None)

Bases: object

Trips when a node has acked nothing for timeout seconds while work was pending — work is available and none is being done.

Pending-aware on purpose. A wall-clock deadline cannot tell a slow model from a hung one, and an idle node (nothing upstream to do) is not stalled at all; checking acks against the broker’s pending count separates all three. This is the only stall detection a BATCH flow has: every BATCH node is a Job, Job pods have their probes stripped, and neither wait_for_completion nor the Job itself carries an overall timeout.

• Arguments:

  • timeout_seconds: allowed silence while work is pending. 0 disables it.

  • pending_probe: returns how many messages are waiting for this node. Injected rather than imported so this stays pure — the messenger passes its own broker query.

  • node_name: named in the raised error.

  • clock: monotonic time source; injected for tests.

  • unknown_grace_seconds: how long the probe may keep answering Unknown (the broker could not be observed) before that is raised as BrokerUnavailable. Unknown neither resets the window (that would hide a stall) nor trips it (that would blame the node for the broker). Default: twice timeout_seconds.

check() → None

• Raises:

  • ProgressStalled: if nothing has been acked for timeout while the broker still reports pending work. The probe is consulted only after the silence threshold is crossed, so the common path costs nothing.

record_progress() → None

Called on every ack (and on every failure — a failure is still work being done).

silent_for() → float

class videoflow.core.supervision.SupervisionPolicy(max_restarts: int = 3, backoff_seconds: Tuple[float, ...] = (10.0, 20.0, 40.0), restart_on: FrozenSet[str] = frozenset({'transient', 'worker_fatal'}))

Bases: object

How an engine responds to a worker exiting non-zero.

The defaults mirror the Kubernetes Job semantics the manifests already encode (backoffLimit: 3), so the local engine and the cluster agree by construction rather than by coincidence. Only the backoff differs — see local().

• Attributes:

  • max_restarts: attempts after the first launch. 0 disables restarts.

  • backoff_seconds: delay before each restart; the last value repeats.

  • restart_on: dispositions worth restarting for. POISON is deliberately absent — a worker that died of a bad message will die of it again, and three more identical crashes help nobody.

backoff_seconds: Tuple[float, ...] = (10.0, 20.0, 40.0)

delay_for(attempt: int) → float

Backoff before restart number attempt (0-based); the last value repeats.

classmethod disabled() → SupervisionPolicy

No restarts — run-local --no-restart, for a tight debug loop.

classmethod local() → SupervisionPolicy

Same restart count as the cluster, compressed backoff. A developer watching a terminal should not wait seventy seconds to learn that a node is genuinely broken, but the recovery path still has to be exercised — that is the entire reason the local supervisor exists.

max_restarts: int = 3

restart_on: FrozenSet[str] = frozenset({'transient', 'worker_fatal'})

should_restart(attempt: int, disposition: str | None = None) → bool

• Arguments:

  • attempt: how many restarts have already been spent (0 on the first failure).

  • disposition: the classified cause, when the worker managed to report one (via its termination log). None — a worker that died without saying why — is restarted: an unexplained death is far more often a crash worth retrying than a poison message.

videoflow.core.supervision.render_event(event: object) → str | None

One human line per event, or None for events not worth printing.

videoflow.core.task module

class videoflow.core.task.ConsumerTask(consumer: ConsumerNode, messenger: Messenger, has_children: bool, parent_names: List[str], ctx: RuntimeContext | None = None, idempotency_store: IdempotencyStore | None = None, breaker: ConsecutiveFailureBreaker | None = None, deadline: ProgressDeadline | None = None, on_error: str = 'transient', watchdog: ProgressWatchdog | None = None)

Bases: NodeTask

It runs forever, blocking until it receives a message from every parent node through the messenger. It consumes the message and does not publish anything back down the pipe — consumers are the leaves of the graph.

class videoflow.core.task.NodeTask(computation_node: Node, messenger: Messenger, has_children: bool, ctx: RuntimeContext | None = None, breaker: ConsecutiveFailureBreaker | None = None, deadline: ProgressDeadline | None = None, on_error: str = 'transient', watchdog: ProgressWatchdog | None = None)

Bases: Task

A NodeTask is a wrapper around a videoflow.core.node.Node that is able to interact with the execution environment through a messenger. Nodes receive input and/or produce output, but tasks are the ones that run in infinite loops, receiving inputs from the environment and passing them to the computation node, and taking outputs from the computation node and passing them to the environment.

• Arguments:

  • computation_node

  • messenger (Messenger): the messenger that will communicate between nodes.

  • has_children (bool): True if this node has at least one downstream child in the graph — used to skip publishing when nothing would ever consume it.

  • breaker: trips when the worker fails too many messages in a row (see videoflow.core.supervision). Constructed by the worker and handed down, like idempotency_store.

  • deadline: trips when the node stops acking while work is pending.

  • on_error (str): disposition for exceptions nothing classifies.

  • watchdog: re-checks deadline on its own thread while the node is inside process()/consume() (see videoflow.runtime.watchdog). Started once open() has returned and stopped when the loop ends; it must hold the same deadline instance, or it measures silence the loop never resets.

property computation_node: Node

Returns the current computation node

run() → None

Starts the task in an infinite loop. If this method is called and the set_messenger() method has not been called yet, an assertion error will happen.

A failure in open() or in the run loop publishes an ABORT marker before propagating, so the graph downstream of this node terminates instead of hanging on an end-of-stream that will never arrive.

class videoflow.core.task.ProcessorTask(processor: ProcessorNode, messenger: Messenger, has_children: bool, parent_names: List[str], ctx: RuntimeContext | None = None, breaker: ConsecutiveFailureBreaker | None = None, deadline: ProgressDeadline | None = None, on_error: str = 'transient', watchdog: ProgressWatchdog | None = None)

Bases: NodeTask

It runs forever, first blocking until it receives a message from every parent node through the messenger. Then it passes the merged inputs to the processor node and, when it gets back the output, uses the messenger to publish it down the flow. If every parent has signaled termination, it passes termination message down the flow and breaks from infinite loop.

change_device(device_type: str) → None

property device_type: str

class videoflow.core.task.ProducerTask(producer: ProducerNode, messenger: Messenger, has_children: bool, ctx: RuntimeContext | None = None, breaker: ConsecutiveFailureBreaker | None = None, deadline: ProgressDeadline | None = None, on_error: str = 'transient', watchdog: ProgressWatchdog | None = None, resume_offset: int = 0)

Bases: NodeTask

It runs forever calling the next() method in the producer node. At each iteration it checks for a termination signal, and if so it sends a termination message to its child task and breaks the infinite loop.

A producer has no inputs, so it has no per-message failure path: an exception from next() is not something that can be naked or dead-lettered. It ends the producer, and NodeTask.run publishes ABORT on the way out so the rest of the graph does not wait forever for an end-of-stream.

class videoflow.core.task.Task

Bases: object

run() → None

Starts the task in an infinite loop.

videoflow.core.task.invoke_node(method: Callable[[...], Any], *args: Any, node: str = '', replica: int = 0, trace_id: str | None = None, seq: int | None = None, ctx: RuntimeContext | None = None, run_async: Callable[[Any], Any] | None = None, on_error: str = 'transient') → Any

The one seam where user code becomes framework-legible.

Calls a node method, passing ctx only if it declares it and awaiting the result if it is a coroutine, then converts whatever it raised into a classified VideoflowRuntimeError carrying the message’s identity. Without this, the task loop’s except Exception catches the error and throws away everything structural about it: the retry ladder cannot tell a bad message from a sick worker, and the DLQ records free text nothing can aggregate.

BaseException — KeyboardInterrupt, SystemExit — and StopIteration deliberately pass through untouched: the first two are not message failures, and the third is a producer’s normal end of stream.

• Arguments:

  • method: the node’s next/process/consume/open/close.

  • node, replica, trace_id, seq: identity stamped into the error’s context.

  • ctx: the RuntimeContext, passed only to methods that declare it.

  • run_async: runs an awaitable to completion. Injected because the event loop belongs to the task, which keeps node coroutines off the messenger’s I/O loop so a node’s async work never blocks broker fetches and acks.

  • on_error: disposition for an exception nothing classifies.

• Raises:

  • VideoflowRuntimeError: whatever the node raised, classified and enriched.

videoflow.core.task.raise_if_aborted(entries: List[dict], messenger: Messenger, has_children: bool) → None

Ends this node when any parent reports that it terminated abnormally.

A clean end-of-stream and an abnormal one are different facts, and only one of them used to exist on the wire: a producer that died mid-run left its children blocking forever on an EOS that was never coming. ABORT is that missing fact, and it walks the graph exactly the way EOS does — this node propagates it to its own children before stopping.

• Raises:

  • UpstreamAborted: naming the parent that aborted and carrying its error.