videoflow.backends package

Backend contracts: the responsibility boundaries behind a running flow.

Videoflow builds a graph on one machine and executes it on many. Between the graph and the workers sit a handful of backends — a message transport, a payload store, an accelerator allocator, an execution environment — and a runtime that decides what correct processing means regardless of which concrete backend is underneath. This package holds the contracts for those boundaries, the truthful outcome types they share, and reference in-memory implementations that double as the executable specification every real adapter is tested against.

  • MessagingBackend (messaging): how does an envelope reach its required consumers?

  • PayloadStore (payload): where are the image bytes, and how long must they survive?

  • AcceleratorAllocationBackend (allocation): which accelerators may this workload use, under what guarantees? (Implemented by deploy.allocation_local for run-local, deploy.allocation_kubernetes over the GPU strategies, and render-only deploy.allocation_dra.)

  • FlowRuntime (runtime): what constitutes correct processing and recovery?

  • the composition planner (capabilities): can this graph meet its requested contract?

Two rules run through all of it. Policy lives above adapters: an adapter reports what it can guarantee, and the planner rejects a request it cannot meet instead of quietly weakening it. And observations are truthful: a read that failed is Unknown, never zero, never empty, never complete.

Nothing here imports a broker, a store or a cluster client at module scope, so the contracts can be imported wherever the graph can.

Subpackages

Submodules

videoflow.backends.allocation module

The AcceleratorAllocationBackend contract: which accelerator resources may a workload use, and under what guarantees.

Named for what it decides, not for one mechanism: a Kubernetes DRA driver, the device plugin, videoflow’s managed MIG geometry and a local nvidia-smi walk are all implementations. What every implementation must share:

  • Inventory and plans are advisory. A feasible plan is a snapshot, never a reservation; authoritative allocation must survive concurrent demand.

  • Reservation is a compare-and-swap. Ownership is written under a server enforced precondition (a Kubernetes resourceVersion or a JSON-patch test op), and a stale rollback cannot remove a newer owner.

  • Readiness is observed, not inferred. “Allocated”, “prepared” and “application-ready” are different states; a historical success label is not evidence for a new operation.

  • Reads that failed are Unknown. An occupancy listing the API refused does not prove a device idle.

  • Release is idempotent and generation-fenced, and a retained workload keeps its allocation until an explicit later release.

Memory is expressed in bytes with separate meanings — usable minimum, reserved, hard limit, declared peak — because one gpu_memory_gb with three meanings was how a scheduler reservation got mistaken for isolation.

class videoflow.backends.allocation.AcceleratorAllocationBackend[source]

Bases: ABC

abstractmethod bindings(claim_id: str, workload_id: str) → WorkloadBindings[source]
abstractmethod capabilities(environment: Mapping[str, Any]) → AllocationCapabilities[source]
abstractmethod inventory(scope: Mapping[str, Any]) → Known[InventorySnapshot] | Unknown[source]
abstractmethod observe(claim_id: str) → Known[ClaimObservation] | Unknown[source]
abstractmethod plan(requests: Sequence[WorkloadRequest], snapshot: InventorySnapshot) → FeasiblePlan | Infeasible[source]
abstractmethod reconcile(claim_id: str, desired: str, expected_generation: str) → ClaimObservation[source]
abstractmethod release(claim_id: str, operation_id: str, expected_generation: str, keep_workloads: bool = False) → ReleaseObservation[source]
abstractmethod reserve(plan: FeasiblePlan, operation_id: str, expected_generation: str | None) → ClaimObservation[source]
class videoflow.backends.allocation.ClaimObservation(claim_id: str, owner: str, desired_generation: str, observed_generation: str | None, status: str, grant: tuple[videoflow.backends.allocation.DeviceIdentity, ...] | None, evidence: Mapping[str, Any]=<factory>)[source]

Bases: object

claim_id: str
desired_generation: str
evidence: Mapping[str, Any]
grant: tuple[DeviceIdentity, ...] | None
observed_generation: str | None
owner: str
status: str
class videoflow.backends.allocation.Constraint(key: str, operator: str, values: tuple[str, ...], hard: bool = True)[source]

Bases: object

A hard requirement or a soft preference on device or node attributes.

hard: bool = True
key: str
operator: str
values: tuple[str, ...]
class videoflow.backends.allocation.DeliveredGrant(workload_id: str, devices: tuple[DeviceIdentity, ...], exclusive: bool, requested: int, policy: str, host: str = 'observed')[source]

Bases: object

What a workload actually received, distinct from what it requested.

devices: tuple[DeviceIdentity, ...]
exclusive: bool
static from_dict(d: Mapping[str, Any]) → DeliveredGrant[source]
host: str = 'observed'

observed when the grant reflects a successful host read; unobserved when discovery failed and the launcher went ahead without one — an empty device list then means “could not tell”, never “a zero-GPU machine”.

policy: str
requested: int
to_dict() → dict[str, Any][source]
workload_id: str
class videoflow.backends.allocation.DeviceIdentity(node: str | None, ordinal: int | None, uuid: str | None, mig_uuid: str | None, product: str, memory_bytes: int | None, mig_profile: str | None = None)[source]

Bases: object

One accelerator as three different identifier spaces: a host ordinal, a physical GPU UUID, and (for a slice) a MIG UUID. They are not interchangeable; a worker maps whatever it was granted onto CUDA-local ordinals itself.

memory_bytes: int | None
mig_profile: str | None = None
mig_uuid: str | None
node: str | None
ordinal: int | None
product: str
uuid: str | None
videoflow.backends.allocation.EXCLUSIVE_FEATURE_PAIRS = (('mps', 'dynamic-mig'), ('mps', 'consumable-capacity'))

MPS shares a device between clients while dynamic MIG re-partitions it, and consumable shares already carve it.

Type:

Combinations no backend can serve at once

videoflow.backends.allocation.FEATURE_DYNAMIC_MIG = 'dynamic-mig'

Feature vocabulary a request may name (WorkloadRequest.features); each is admitted only when the backend’s version matrix lists it as available — a request for dynamic MIG on hardware or a driver that cannot partition is rejected by name, never by a blanket “impossible” (ALLOC-023).

class videoflow.backends.allocation.FeasiblePlan(assignments: Mapping[str, tuple[DeviceIdentity, ...]], geometry: Mapping[str, str], snapshot_generation: str, plan_id: str, notes: tuple[str, ...] = ())[source]

Bases: object

Advisory. plan_id and snapshot_generation tie a later reservation to the inventory it was planned on.

assignments: Mapping[str, tuple[DeviceIdentity, ...]]
geometry: Mapping[str, str]
notes: tuple[str, ...] = ()
plan_id: str
snapshot_generation: str
class videoflow.backends.allocation.Infeasible(reasons: tuple[str, ...])[source]

Bases: object

reasons: tuple[str, ...]
class videoflow.backends.allocation.InventorySnapshot(devices: tuple[DeviceIdentity, ...], occupancy: Mapping[str, int], sharing: Mapping[str, str], owners: Mapping[str, str], completeness: str, observed_at: float, generation: str)[source]

Bases: object

  • Arguments:

    • occupancy: device key -> units in use, as far as the reads could see.

    • sharing: node -> classification (physical, mig, time-sliced, mps, unknown).

    • owners: node -> owner recorded on it (videoflow’s own stamp), if any.

    • completeness: complete when every read succeeded, partial when some read failed — a partial snapshot may plan but may not admit.

completeness: str
devices: tuple[DeviceIdentity, ...]
generation: str
observed_at: float
occupancy: Mapping[str, int]
owners: Mapping[str, str]
sharing: Mapping[str, str]
class videoflow.backends.allocation.ReleaseObservation(claim_id: str, status: str, remaining: tuple[str, ...] = (), reason: str = '')[source]

Bases: object

claim_id: str
reason: str = ''
remaining: tuple[str, ...] = ()
status: str
class videoflow.backends.allocation.WorkloadBindings(env: ~typing.Mapping[str, str], pod_fragment: dict, container_fragment: dict, claim_manifests: list[dict] = <factory>, node_constraints: dict = <factory>)[source]

Bases: object

How a granted claim reaches a workload: environment for a process, and Kubernetes fragments (plain dicts — external API schema) for a pod.

claim_manifests: list[dict]
container_fragment: dict
env: Mapping[str, str]
node_constraints: dict
pod_fragment: dict
class videoflow.backends.allocation.WorkloadRequest(flow_id: str, run_id: str, workload_id: str, device_count: int, sharing: str, minimum_usable_memory_bytes: int | None = None, reserved_memory_bytes: int | None = None, hard_memory_limit_bytes: int | None = None, declared_peak_memory_bytes: int | None = None, features: frozenset[str] = frozenset(), constraints: tuple[videoflow.backends.allocation.Constraint, ...]=(), elasticity: str = 'fixed', host_cpu: str | None = None, host_memory: str | None = None, provenance: Mapping[str, str]=<factory>)[source]

Bases: object

constraints: tuple[Constraint, ...] = ()
declared_peak_memory_bytes: int | None = None
device_count: int
elasticity: str = 'fixed'
features: frozenset[str] = frozenset({})
flow_id: str
hard_memory_limit_bytes: int | None = None
host_cpu: str | None = None
host_memory: str | None = None
minimum_usable_memory_bytes: int | None = None
provenance: Mapping[str, str]
reserved_memory_bytes: int | None = None
run_id: str
sharing: str
workload_id: str
videoflow.backends.allocation.allocation_rejections(requests: Sequence[WorkloadRequest], capabilities: AllocationCapabilities) → list[str][source]

Why the backend cannot serve these requests, before anything is planned or written — the validation-before-write boundary. Empty means admitted. Each reason names the request, the feature/policy/sharing kind it needs and what the backend (adapter/authority) actually offers, so the operator can tell “this driver version lacks the gate” from “this hardware cannot do it”.

videoflow.backends.capabilities module

Capability profiles and the composition planner.

A backend’s name is not a guarantee. “JetStream” can be a single server on an emptyDir or a three-replica cluster on persistent disks; “Redis” can be an evictable cache or an append-only durable store; “GPU” can be a whole device, a MIG slice with hardware isolation, or a time-share that only the scheduler accounts for. So every adapter reports a versioned capabilities record, the flow (or its defaults) states what it requires as profiles, and the planner compares the two before a single stream is provisioned, a single worker starts, or a single label is written. A requirement the composition cannot meet is rejected with the channel, the profile and the missing guarantee named; it is never quietly downgraded to whatever the backend does offer.

Messaging profiles (ARCHITECTURE §2):

  • live_latest: bounded backlog and age, explicit drop policy and scope (latest per key versus global), observable gaps; no offline replay promise.

  • reliable_work: every accepted item stays recoverable for each required logical consumer until committed or durably failed; overflow backpressures, rejects before acceptance, or records a terminal outcome — never evicts.

  • durable_control: reconciled run state and durable terminal records across controller and worker restarts; a transient notification may wake workers but cannot be the sole record of stop, abort or completion.

  • replay_archive: retention independent of working consumers, with payloads and provenance; a replay is a new execution with explicit recipients.

Allocation guarantees name the mechanism behind a promise, because “reserved” means three different things: enforcement says whether a memory bound is enforced by hardware (MIG), by a runtime cap (MPS pinned memory), by scheduler accounting only (a consumable capacity or a whole-device count), or not at all.

FlowRequirements is deliberately not a field of NodeSpec: the spec is serialised with asdict into every specs ConfigMap, so a new field changes bytes for every existing flow. Requirements travel as a separate document beside the specs and as environment variables that are emitted only when set.

videoflow.backends.capabilities.ADMISSION_TIMEOUT_ENV = 'VF_ADMISSION_TIMEOUT_SECONDS'

How long the in-container admission checks (the provision entrypoint before it creates anything, a worker before it opens its node) give the broker and the payload store to answer a read-back, connect included. Unset ⇒ the default; the answer past the deadline is Unknown('timeout'), never a guess.

class videoflow.backends.capabilities.AllocationCapabilities(adapter: str, authority: str, exclusive_device: bool, isolated_mig: bool, cooperative_sharing: bool, memory_enforcement: str, multi_device: bool, topology_verification: bool, elastic: bool, admission_boundary: bool, version_matrix: Mapping[str, Any]=<factory>)[source]

Bases: object

adapter: str
admission_boundary: bool
authority: str
cooperative_sharing: bool
elastic: bool
exclusive_device: bool
isolated_mig: bool
memory_enforcement: str
multi_device: bool
topology_verification: bool
version_matrix: Mapping[str, Any]

version (a string), features (the allocation.FEATURE_* names available) and whatever else the adapter records (feature-gate states, the API group served).

Type:

What the adapter’s version can do, keyed by name

class videoflow.backends.capabilities.CompositionPlan(channel_profiles: Mapping[str, str], channel_retention: Mapping[str, str], restart_safe: bool, notes: tuple[str, ...] = ())[source]

Bases: object

The admitted composition: which profile each channel got, the retention class it compiles to, and the notes a human reads to see why. An instance exists only if nothing was rejected — a rejection is an exception, not a field.

channel_profiles: Mapping[str, str]
channel_retention: Mapping[str, str]
notes: tuple[str, ...] = ()
render() → str[source]
restart_safe: bool
videoflow.backends.capabilities.EFFECT_AT_LEAST_ONCE = 'at_least_once'

What a sink can promise about its external effects (ConsumerNode.effect_guarantee).

class videoflow.backends.capabilities.ExecutionCapabilities(engine: str, restart_supervision: bool, readiness_states: tuple[str, ...], pvc_mounts: bool, autoscaling_controllers: tuple[str, ...] = (), execution_groups: bool = False, dynamic_batching: bool = False)[source]

Bases: object

What an execution engine advertises. execution_groups (fused groups whose internal edges never touch the broker, RUN-035) and dynamic_batching (a runtime batching contract distinct from transport fetch batching, RUN-036) are declared False by both shipped engines: a flow that declares either is refused at admission rather than run as ordinary nodes.

autoscaling_controllers: tuple[str, ...] = ()
dynamic_batching: bool = False
engine: str
execution_groups: bool = False
pvc_mounts: bool
readiness_states: tuple[str, ...]
restart_supervision: bool
class videoflow.backends.capabilities.FlowRequirements(profiles: tuple[~videoflow.backends.capabilities.ProfileRequest, ...]=(), restart_safe: bool = False, exactly_once_effects: tuple[str, ...]=(), resources: Mapping[str, ~typing.Mapping[str, str]]=<factory>, priority_class: str | None = None, rollout_policy: str | None = None, tolerated_failures: int = 0, sink_guarantees: Mapping[str, str]=<factory>, effect_retention_seconds: float | None = None, replay_horizon_seconds: float | None = None, execution_groups: Mapping[str, tuple[str, ...]]=<factory>, batching: Mapping[str, ~typing.Mapping[str, ~typing.Any]]=<factory>)[source]

Bases: object

Everything a flow asks of its backends beyond the graph itself. Serialised beside the specs (compile_to_dict()['requirements']) only when non-empty, so flows that never touch it produce byte-identical documents.

batching: Mapping[str, Mapping[str, Any]]

Dynamic-batching contracts the nodes declare (Node.batching_policy): node -> policy. Admitted only against dynamic_batching (RUN-036).

effect_retention_seconds: float | None = None

a marker that expires inside the replay horizon cannot certify exactly-once.

Type:

How long effect markers are kept, and how far back a replay may reach

exactly_once_effects: tuple[str, ...] = ()
execution_groups: Mapping[str, tuple[str, ...]]

group name -> member nodes. Admitted only against an engine advertising execution_groups (RUN-035); none does today.

Type:

Fused execution groups the nodes declare (Node.execution_group)

static from_dict(d: Mapping[str, Any] | None) → FlowRequirements[source]
is_empty() → bool[source]
priority_class: str | None = None
profiles: tuple[ProfileRequest, ...] = ()
replay_horizon_seconds: float | None = None
resources: Mapping[str, Mapping[str, str]]
restart_safe: bool = False
rollout_policy: str | None = None
sink_guarantees: Mapping[str, str]

What each sink declares about its external effects (ConsumerNode.effect_guarantee, only the non-default idempotent_key declarations): the one thing that can admit exactly_once_effects for it (RUN-017).

to_dict() → dict[str, Any][source]
tolerated_failures: int = 0

Broker pod losses accepted work must survive (the deployment’s declared fault model, MSG-021): f needs persistent storage and 2f + 1 stream copies, read back — never inferred from durable names or pod counts.

videoflow.backends.capabilities.LEDGER_NONE = 'none'

What a backend knows about the fate of a publication after the fact.

videoflow.backends.capabilities.MAX_STREAMS_ENV = 'VF_MAX_STREAMS'

the measured limit a benchmark (MSG-026) established on the deployment’s hardware, so a graph beyond it is refused at admission instead of partially provisioned.

Type:

Operator-declared supported graph size for the JetStream adapter

class videoflow.backends.capabilities.MessagingCapabilities(adapter: str, version: str, retained_backlog: bool, recoverable_delivery: bool, latest_per_key: bool, dedup_window_seconds: int | None, publication_ledger: str, replication_factor: Known[int] | Unknown, persistent_storage: Known[bool] | Unknown, max_payload_bytes: Known[int] | Unknown, credit_resizable: bool, control_shares_data_slot: bool, durable_control: bool = False, archive: bool = False, mixed_retention_per_channel: bool = False, max_streams: Known[int] | Unknown | None = None, max_consumers: Known[int] | Unknown | None = None)[source]

Bases: object

What a messaging adapter, as configured, can guarantee. Observations that depend on the deployed broker (replication, storage, payload limit) are Observation values: an adapter that could not read them says so, and the planner treats that as a reason to reject a profile that depends on them.

adapter: str
archive: bool = False
control_shares_data_slot: bool
credit_resizable: bool
dedup_window_seconds: int | None
durable_control: bool = False
latest_per_key: bool
max_consumers: Known[int] | Unknown | None = None
max_payload_bytes: Known[int] | Unknown
max_streams: Known[int] | Unknown | None = None

The largest graph the adapter supports, as streams and consumers (MSG-026): the broker account’s limits read back (-1 = unlimited) capped by the operator’s declared supported size (VF_MAX_STREAMS / VF_MAX_CONSUMERS, graph_limits_from_env); None when the adapter declares nothing.

mixed_retention_per_channel: bool = False
persistent_storage: Known[bool] | Unknown
publication_ledger: str
recoverable_delivery: bool
replication_factor: Known[int] | Unknown
retained_backlog: bool
version: str
videoflow.backends.capabilities.PARENT_REPLICAS_ENV = 'VF_PARENT_REPLICAS'

The per-parent replica counts (ENV-11), positionally aligned with VF_PARENT_NAMES.

videoflow.backends.capabilities.PROFILE_REQUESTS_ENV = 'VF_PROFILE_REQUESTS_JSON'

The environment variable carrying an operator’s explicit channel-profile requests to a worker (deploy --require-profile); absent when none were made, so the default worker environment is unchanged (D8).

class videoflow.backends.capabilities.PayloadCapabilities(adapter: str, durable: Known[bool] | Unknown, evictable: Known[bool] | Unknown, atomic_multikey: Known[bool] | Unknown, max_object_bytes: int | None, reader_identities: bool, reference_forwarding: bool = False, persistent_storage: Known[bool] | Unknown = <factory>)[source]

Bases: object

durable: an accepted object outlives a restart of the store’s process (persistence on) and is never dropped while a reader holds it (noeviction) — what reliable_work asks. persistent_storage: the data lives on a volume that outlives the store’s pod — what surviving a pod loss (tolerated_failures) additionally asks; a dev Redis writes its append-only file to an emptyDir and is durable without being that. The wire cannot tell the two apart (CONFIG GET dir names a path, not what backs it), so a read-back leaves it Unknown and the profile record on the Service (or the operator) declares it.

reference_forwarding: a stage that republishes an unchanged payload shares the canonical object instead of writing a copy per hop (PAY-017). Neither shipped store declares it; every hop writes its own copy, which the benchmark measures as a copy amplification of one object per frame-bearing stage — declared, never hidden.

adapter: str
atomic_multikey: Known[bool] | Unknown
durable: Known[bool] | Unknown
evictable: Known[bool] | Unknown
max_object_bytes: int | None
persistent_storage: Known[bool] | Unknown
reader_identities: bool
reference_forwarding: bool = False
class videoflow.backends.capabilities.ProfileRequest(channel: str, profile: str, options: Mapping[str, ~typing.Any]=<factory>)[source]

Bases: object

  • Arguments:

    • channel: the producing node whose output channel the profile applies to.

    • profile: one of MESSAGING_PROFILES.

    • options: profile-specific knobs — latest_per_key (bool), horizon_seconds (int, the recovery/replay horizon), key (str).

channel: str
static from_dict(d: Mapping[str, Any]) → ProfileRequest[source]
options: Mapping[str, Any]
profile: str
to_dict() → dict[str, Any][source]
videoflow.backends.capabilities.RETENTION_LIMITS = 'limits'

Retention classes a channel can be compiled to. A channel has exactly one.

videoflow.backends.capabilities.RUNTIME_STORE_ENV = 'VF_RUNTIME_STORE_URL'

the ledger every worker of the run keeps its records in.

Type:

The runtime store’s URL (RFC 0006 ENV-10)

class videoflow.backends.capabilities.RuntimeCapabilities(store: str, durable: videoflow.backends.outcomes.Known[bool] | videoflow.backends.outcomes.Unknown, shared_across_processes: bool, restart_safe_joins: bool, elastic_state: bool)[source]

Bases: object

durable: Known[bool] | Unknown
elastic_state: bool
restart_safe_joins: bool
shared_across_processes: bool
store: str
videoflow.backends.capabilities.admission_timeout_from_env(value: str | None) → float[source]

VF_ADMISSION_TIMEOUT_SECONDS as seconds: the default when unset.

  • Raises:

    • ConfigError: a value that is not a positive number.

videoflow.backends.capabilities.combined_limit(account: Known[int] | Unknown | None, declared: int | None) → Known[int] | Unknown | None[source]

The tighter of an account limit (-1 unlimited) and a declared cap; None when neither says anything.

videoflow.backends.capabilities.default_requirements(flow_type: str, specs: Sequence[NodeSpec]) → FlowRequirements[source]

The requirements a flow makes without saying anything: one profile request per edge, derived from the flow type and each consuming node’s delivery override. Two consumers of one channel may therefore ask for different profiles; whether one channel can serve both is the planner’s decision.

videoflow.backends.capabilities.graph_limits_from_env(environ: Mapping[str, str]) → tuple[int | None, int | None][source]

(max_streams, max_consumers) declared in the environment, or None each; a non-integer is a ConfigError.

videoflow.backends.capabilities.kubernetes_execution_capabilities(autoscaling: bool = False) → ExecutionCapabilities[source]

The Kubernetes engine: kubelet restarts, PVC mounts, KEDA when asked for; no fusion or batching.

videoflow.backends.capabilities.local_execution_capabilities() → ExecutionCapabilities[source]

The local engine: subprocess supervision, no claims, no controllers, no fusion or batching.

videoflow.backends.capabilities.plan_composition(requirements: FlowRequirements, messaging: MessagingCapabilities, payload: PayloadCapabilities | None = None, runtime: RuntimeCapabilities | None = None, allocation: AllocationCapabilities | None = None, execution: ExecutionCapabilities | None = None, payload_refs_in_use: bool = False) → CompositionPlan[source]

Admit the composition or reject it — never downgrade it.

  • Arguments:

    • requirements: what the flow asks for (default_requirements when it says nothing).

    • messaging / payload / runtime / allocation / execution: what the composed backends advertise. None means that backend is not part of the composition; a profile needing it is then rejected.

    • payload_refs_in_use: whether any channel offloads payloads to the store, which makes the store’s durability part of the messaging guarantee.

  • Raises:

    • IncompatibleProfile: a requested guarantee is definitely not provided; the diagnostics list every one.

    • UnobservableState: a requested guarantee depends on a capability the adapter could not observe (replication, persistence, store durability). Unknown is not a pass.

videoflow.backends.capabilities.profile_for_edge(flow_type: str, delivery: Mapping[str, Any] | None) → str[source]

The profile an edge requests under today’s presets: BATCH is reliable work, REALTIME is live-latest, and a node’s delivery override flips its own inputs (an at-least-once sink inside a REALTIME flow asks for reliable work).

videoflow.backends.capabilities.realtime_default_capabilities_note(flow_type: str) → str[source]

One line for videoflow explain: what the flow type’s preset implies.

videoflow.backends.capabilities.requests_env(explicit: Sequence[ProfileRequest]) → dict[str, str][source]

The worker environment entry for explicit requests — empty when there are none.

videoflow.backends.capabilities.requests_from_env(value: str | None) → list[ProfileRequest][source]

The inverse of requests_env on the worker side; [] when unset.

videoflow.backends.faults module

Named fault barriers for deterministic failure injection.

A conformance case is a story with a fault in the middle: “crash after the group is persisted but before its outputs are published”, “drop the acknowledgment of the settlement”, “let a second owner write between the read and the update”. Repeating a happy-path test many times does not reach those interleavings; a barrier does. Adapters and the runtime call barrier(name, **context) at each named point; in production nothing is installed and the call is one attribute read. A test installs a FaultSchedule mapping barrier names to actions, and afterwards asks which barriers actually fired — a scheduled fault that never fired makes the test INVALID_TEST, not PASS.

Barriers reach worker subprocesses too: FaultSchedule.to_env() serialises the schedule into VF_FAULT_SCHEDULE_JSON and names a marker directory in VF_FAULT_MARKER_DIR; videoflow.runtime.worker installs it from the environment, and every hit appends a marker file the parent can count.

Actions:

  • Crash(exit_code): write the marker, then os._exit — no finally, no atexit, exactly the death a kill produces.

  • RaiseError(factory) / RaiseError.typed(code, message, disposition): raise.

  • Delay(seconds): sleep, then continue.

  • DropResponse: continue, but the adapter reports the call’s outcome as Unknown — the lost-receipt model at the adapter boundary.

  • Pause(name): block until FaultSchedule.release(name) (or the marker <dir>/<name>.release appears), so two processes can be interleaved by hand.

  • Nth(n, action): apply action on the n-th hit only.

class videoflow.backends.faults.Crash(exit_code: int = 137)[source]

Bases: object

exit_code: int = 137
class videoflow.backends.faults.Delay(seconds: float)[source]

Bases: object

seconds: float
class videoflow.backends.faults.DropResponse[source]

Bases: object

class videoflow.backends.faults.FaultSchedule(actions: Mapping[str, Crash | RaiseError | Delay | DropResponse | Pause | Nth], marker_dir: str | None = None)[source]

Bases: object

  • Arguments:

    • actions: barrier name -> action.

    • marker_dir: directory for cross-process markers; created on install.

fired() → dict[str, int][source]

Hit counts per barrier: in-process hits plus marker files written by other processes.

classmethod from_env(environ: Mapping[str, str] | None = None) → FaultSchedule | None[source]
hit(name: str, context: Mapping[str, Any]) → Hit[source]
install() → FaultSchedule[source]
release(name: str) → None[source]

Let a Pause(name) continue, in this process and in any process sharing the marker dir.

to_env() → dict[str, str][source]

Environment variables a worker subprocess needs to install the same schedule.

unfired() → list[str][source]
uninstall() → None[source]
class videoflow.backends.faults.Hit(fired: bool, drop_response: bool = False)[source]

Bases: object

What barrier() returns: whether the adapter must report this call’s outcome as unknown.

drop_response: bool = False
fired: bool
class videoflow.backends.faults.Nth(n: int, action: Any)[source]

Bases: object

action: Any
n: int
class videoflow.backends.faults.Pause(name: str, timeout_seconds: float = 60.0)[source]

Bases: object

name: str
timeout_seconds: float = 60.0
class videoflow.backends.faults.RaiseError(factory: Callable[[], BaseException], spec: Mapping[str, str] | None = None)[source]

Bases: object

factory: Callable[[], BaseException]
spec: Mapping[str, str] | None = None

Serialisable form for cross-process schedules (see typed); None when the factory is an in-process callable that cannot travel.

static typed(code: str, message: str, disposition: str = 'transient') → RaiseError[source]

A raise action built from the error taxonomy, serialisable into a worker’s env.

exception videoflow.backends.faults.UnknownBarrier[source]

Bases: ValueError

A schedule or a call named a barrier that is not in BARRIERS.

videoflow.backends.faults.barrier(name: str, **context: Any) → Hit[source]

The production-side call. One set lookup when no schedule is installed; otherwise delegates to it. Adapters check .drop_response where the lost-receipt model applies, and let RaiseError/Crash propagate.

videoflow.backends.faults.installed() → FaultSchedule | None[source]

videoflow.backends.identity module

Collision-resistant logical identities over today’s physical names.

Every broker subject, stream, durable and Kubernetes resource is derived from user-chosen strings — flow ids, run ids, node names — through two lossy encodings: topology.sanitize (any run of characters outside [A-Za-z0-9_-] becomes a single _) and manifests.k8s_name (lower-case, non-DNS characters to -, truncated to 63). Lossy means collisions: a.b and a_b share a stream; Node and node share a Deployment; two long names can truncate onto each other. And hyphen-joined tuples collide across positions: stream vf-a-b-c-n is the same string for (flow='a-b', run='c') and (flow='a', run='b-c').

The design package asks for identities that are reversible or tied to exact owner metadata. This module does the second without renaming anything (a rename is an RFC): it enumerates every physical name a compiled flow will use, rejects the flow at compile time when two distinct logical identities map to one physical name, and produces the owner labels that provisioning writes into stream and consumer metadata (RFC 0006) so teardown can match exactly instead of by prefix.

The naming functions themselves stay where they are (messaging/topology.py, deploy/manifests.py): this module only calls them, so there is still one source of truth for every name.

class videoflow.backends.identity.Collision(physical: str, identities: tuple[videoflow.backends.identity.LogicalIdentity, ...])[source]

Bases: object

identities: tuple[LogicalIdentity, ...]
physical: str
render() → str[source]
class videoflow.backends.identity.LogicalIdentity(kind: str, parts: tuple[str, ...])[source]

Bases: object

What a physical name means: its kind and the exact logical parts it encodes.

kind: str
parts: tuple[str, ...]
render() → str[source]
videoflow.backends.identity.collisions(specs: Sequence[NodeSpec], flow_id: str, run_id: str) → list[Collision][source]

Physical names that two distinct logical identities of the same kind map to. Different kinds may legitimately share a string (a subject and a stream never occupy the same namespace), so only same-kind clashes count.

videoflow.backends.identity.collisions_across_runs(specs: Sequence[NodeSpec], flow_id: str, run_ids: Iterable[str]) → list[Collision][source]

Collisions between two runs of one flow (r vs r-x), for teardown safety checks.

videoflow.backends.identity.derived_names(specs: Sequence[NodeSpec], flow_id: str, run_id: str, instance_ids: Mapping[str, Sequence[str]] | None = None) → dict[str, list[LogicalIdentity]][source]

Every physical name a compiled flow will create or bind, mapped to the logical identities that produce it. The exact set — not a prefix — that a teardown may delete, and the corpus the collision check runs over.

  • Arguments:

    • instance_ids: per-node EOS instance ids when known (the per-process uuid-suffixed durables are minted at worker start, so provisioning cannot enumerate them; pass what a test or a running flow knows).

videoflow.backends.identity.flow_labels(flow_id: str, kind: str) → dict[str, str][source]

Ownership metadata for a flow-scoped resource — the dead-letter stream, which outlives every run. It carries no run label on purpose: owns is False for it under every run id, so no run’s teardown can match it.

videoflow.backends.identity.has_owner_labels(metadata: Mapping[str, str] | None) → bool[source]

Whether a resource carries videoflow ownership metadata at all — labelled (RFC 0006) versus legacy.

videoflow.backends.identity.node_name_collisions(names: Sequence[str]) → list[Collision][source]

Node names that encode to one physical name under either encoder — the check the graph validator runs before a flow is ever compiled. Flow and run ids are constant across a flow’s nodes, so within one flow only the node part can collide.

videoflow.backends.identity.owner_labels(flow_id: str, run_id: str, node: str | None = None, kind: str | None = None, generation: str | None = None) → dict[str, str][source]

Exact-ownership metadata for a broker resource. Written as JetStream stream / consumer metadata (RFC 0006) and compared verbatim by teardown, which is what makes r and r-x distinguishable when their names share a prefix.

videoflow.backends.identity.owns(metadata: Mapping[str, str] | None, flow_id: str, run_id: str, generation: str | None = None) → bool[source]

Whether metadata names exactly this run (both labels present and equal) and, when generation is given, this provisioning generation too.

videoflow.backends.identity.render_collisions(found: Sequence[Collision], encoder: Callable[[str], str] | None = None) → str[source]

videoflow.backends.messaging module

The MessagingBackend contract: how an envelope reaches its required consumers.

A messaging backend owns channels, publication outcomes, logical subscriptions, delivery leases, settlement semantics, transport retention and replay capabilities. It does not own joining inputs, GPU reservations, application effects or whole-run completion — those belong to the runtime, which composes a backend with a payload store and a durable ledger.

The vocabulary that matters:

  • A subscription is a logical consumer. Ten replicas serving one consumer compete for its work; two distinct downstream consumers each have their own obligations. Fan-out and competing delivery are never interchangeable defaults.

  • A delivery token names the run, the subscription, the message, the attempt and the ownership generation. A stale attempt cannot decide a newer attempt’s outcome: settle on a superseded token returns SettleStale without a broker call.

  • Terminal settlement requires a reference to a durable record (a dead-letter acceptance, a ledger entry). An adapter must refuse to terminate a message whose only trace would be its own disappearance.

  • Observations distinguish zero from unknown. A subscription observation that could not be made is Unknown; queue counters are metrics, not a completion protocol.

class videoflow.backends.messaging.ChannelId(flow_id: str, run_id: str, node: str)[source]

Bases: object

flow_id: str
node: str
run_id: str
class videoflow.backends.messaging.ChannelObservation(first_seq: int, last_seq: int, messages: int, retention: str, observed_at: float)[source]

Bases: object

What a channel retains right now: the first and last stream sequences it holds and how many messages — the eviction facts a reconciler needs (BLOB-14 step 4: a publication below first_seq was evicted and its readers will never take it).

first_seq: int
last_seq: int
messages: int
observed_at: float
retention: str
class videoflow.backends.messaging.ChannelSpec(id: ~videoflow.backends.messaging.ChannelId, profile: str, retention: str, max_msgs: int, max_bytes: int | None, max_age_seconds: float | None, overflow: str, dedup_window_seconds: int, replicas: int, persistence: bool, required_subscriptions: tuple[~videoflow.backends.messaging.SubscriptionId, ...], owner_labels: ~typing.Mapping[str, str] = <factory>, per_subject_limits: bool = False)[source]

Bases: object

What a channel must provide, verified against the broker before production starts. owner_labels is the exact-ownership metadata teardown matches on.

dedup_window_seconds: int
id: ChannelId
max_age_seconds: float | None
max_bytes: int | None
max_msgs: int
overflow: str
owner_labels: Mapping[str, str]
per_subject_limits: bool = False
persistence: bool
profile: str
replicas: int
required_subscriptions: tuple[SubscriptionId, ...]
retention: str
class videoflow.backends.messaging.Completed[source]

Bases: object

class videoflow.backends.messaging.Delivery(token: videoflow.backends.messaging.DeliveryToken, envelope_bytes: bytes, size: int, received_at: float, headers: Mapping[str, str]=<factory>)[source]

Bases: object

envelope_bytes: bytes
headers: Mapping[str, str]
received_at: float
size: int
token: DeliveryToken
class videoflow.backends.messaging.DeliveryToken(subscription: videoflow.backends.messaging.SubscriptionId, message_id: str, stream_sequence: int | None, attempt: int, generation: str)[source]

Bases: object

attempt: int
generation: str
message_id: str
stream_sequence: int | None
subscription: SubscriptionId
class videoflow.backends.messaging.Envelope(channel: ChannelId, publication_id: str, headers: Mapping[str, str], body: bytes, size: int, event_id: str, partition_key: str | None, event_ts: float | None, source_epoch: str | None, source_offset: int | None, schema_version: int, payload_refs: tuple[ImmutablePayloadRef, ...] = (), kind: str = 'data')[source]

Bases: object

A publication. publication_id is the stable operation identity a retry re-uses (it becomes the broker’s dedup key); event_id is the logical event identity the runtime reasons about. Physical stream sequences are diagnostics, not identity.

body: bytes
channel: ChannelId
event_id: str
event_ts: float | None
headers: Mapping[str, str]
kind: str = 'data'
partition_key: str | None
payload_refs: tuple[ImmutablePayloadRef, ...] = ()
publication_id: str
schema_version: int
size: int
source_epoch: str | None
source_offset: int | None
videoflow.backends.messaging.KIND_DATA = 'data'

what the bytes are, which decides the subject they ride and what a receiver does with them. dlq is a dead-letter publication of raw wire bytes (DELIV-8), addressed by the origin node’s channel.

Type:

Envelope.kind values

class videoflow.backends.messaging.LeaseObservation(token: videoflow.backends.messaging.DeliveryToken, renewed: bool, expires_at: float | None, reason: str = '')[source]

Bases: object

expires_at: float | None
reason: str = ''
renewed: bool
token: DeliveryToken
class videoflow.backends.messaging.MessagingBackend[source]

Bases: ABC

The transport contract. Every method returns an outcome type rather than raising for the expected failure modes; exceptions are for programming errors and for faults the caller could not have anticipated.

abstractmethod capabilities() → MessagingCapabilities[source]
abstractmethod close(owned: Sequence[ChannelId], expected_generation: str) → CleanupObservation[source]

Remove exactly the owned channels; report what could not be confirmed removed.

abstractmethod ensure_channel(spec: ChannelSpec, operation_id: str) → VerifiedChannel[source]

Create or reconcile the channel and read back its effective configuration. An immutable mismatch (retention, replicas) raises IncompatibleProfile; “already exists” is never accepted without inspection.

abstractmethod ensure_subscription(spec: SubscriptionSpec, operation_id: str) → VerifiedSubscription[source]
observe_ack_floor(subscription: SubscriptionId) → Known[int] | Unknown[source]

The stream sequence up to which subscription has settled everything (its ack floor), for any subscription of the run — bound by this process or not: a reconciler evaluates other readers’ progress through it. Unknown when the adapter cannot read it, which a reconciler treats as “still required”, never as settled.

observe_channel(channel: ChannelId) → Known[ChannelObservation] | Unknown[source]

The channel’s retained range (ChannelObservation); Unknown when the adapter cannot read it.

abstractmethod observe_publication(envelope: Envelope) → Accepted | Rejected | PublicationUnknown | PublicationUnresolvable[source]

What the backend can say later about envelope.publication_id; Unresolvable when it keeps no ledger.

abstractmethod observe_subscription(subscription: SubscriptionId) → Known[SubscriptionObservation] | Unknown[source]
prefetched(subscription: SubscriptionId) → int[source]

Deliveries fetched from the broker and parked locally for subscription (0 when the backend keeps none).

abstractmethod publish(envelope: Envelope, deadline: float) → Accepted | Rejected | PublicationUnknown | PublicationUnresolvable[source]

Publish with a monotonic deadline; on expiry the outcome is PublicationUnknown, never a guess.

abstractmethod receive(subscription: SubscriptionId, item_credit: int, byte_credit: int, deadline: float) → list[Delivery][source]
receive_any(subscriptions: Sequence[SubscriptionId], item_credit: int, byte_credit: int, deadline: float) → list[tuple[SubscriptionId, Delivery]][source]

Deliveries from whichever of subscriptions has any, waiting until deadline (monotonic) for at least one. The default polls each subscription in turn; an adapter that can wait on all of them at once overrides it. Returns [] at the deadline, so a caller can re-check its own termination conditions rather than block forever.

abstractmethod renew(token: DeliveryToken) → LeaseObservation[source]
set_admission(subscription: SubscriptionId, admit: Callable[[Delivery], bool], on_skip: Callable[[Delivery], None] | None = None) → bool[source]

Ask the backend to run admit on every delivery of subscription before it is parked for receive: a delivery it refuses is settled Completed by the backend at once — so a message this consumer will never process does not occupy its ack window — and, once that settlement is confirmed, handed to on_skip off the backend’s own threads. admit runs on the backend’s thread and must be cheap and must not block. Returns False when the backend keeps no such filter; the caller then decides on the receiving side.

abstractmethod settle(token: DeliveryToken, outcome: Completed | Retry | Terminal, settlement_id: str) → SettleConfirmed | SettleUnknown | SettleStale[source]
shutdown() → None[source]

Release connections and threads; the channels stay (close removes those).

subscribe_control(callback: Callable[[], None]) → None[source]

Invoke callback when the run’s flow-wide stop is published; a backend without a control channel ignores it.

supersede(token: DeliveryToken) → bool[source]

Retire a delivery token locally — the receiver now holds a newer attempt of the same logical message — without any broker settlement. Returns whether the token was live. A later settle of the retired token is SettleStale. The default is a no-op for adapters whose tokens are already fenced by attempt and generation.

class videoflow.backends.messaging.Retry(delay_seconds: float | None = None)[source]

Bases: object

delay_seconds: float | None = None
class videoflow.backends.messaging.SubscriptionId(channel: ChannelId, consumer_node: str, partition: int | None, kind: str = 'data', instance: str | None = None)[source]

Bases: object

  • Arguments:

    • partition: the replica index of a partitioned consumer (every replica has its own logical subscription); None for a competing subscription shared by all replicas.

    • kind: data or eos.

    • instance: a per-process observer instance — end-of-stream markers are observed by every replica through its own subscription, named by the process that created it. None for the logical work subscription.

channel: ChannelId
consumer_node: str
instance: str | None = None
kind: str = 'data'
partition: int | None
class videoflow.backends.messaging.SubscriptionObservation(available: int, leased: int, unresolved: int, dropped: int, rejected_publications: int, observed_at: float, generation: str | None = None)[source]

Bases: object

Counts are for this observation only; unresolved is work the broker holds but can no longer deliver.

available: int
dropped: int
generation: str | None = None
leased: int
observed_at: float
rejected_publications: int
unresolved: int
class videoflow.backends.messaging.SubscriptionSpec(id: SubscriptionId, competing: bool, ack_wait_seconds: float, max_deliver: int, item_credit: int, byte_credit: int, owner_labels: Mapping[str, str]=<factory>)[source]

Bases: object

max_deliver == -1 means the broker never strands a message; the runtime’s ledger budgets attempts.

ack_wait_seconds: float
byte_credit: int
competing: bool
id: SubscriptionId
item_credit: int
max_deliver: int
owner_labels: Mapping[str, str]
class videoflow.backends.messaging.Terminal(record_ref: str)[source]

Bases: object

record_ref names the durable record (DLQ acceptance, ledger id) that outlives the message.

record_ref: str
class videoflow.backends.messaging.VerifiedChannel(spec: videoflow.backends.messaging.ChannelSpec, effective: Mapping[str, Any], mismatches: tuple[str, ...] = ())[source]

Bases: object

effective: Mapping[str, Any]
mismatches: tuple[str, ...] = ()
spec: ChannelSpec
class videoflow.backends.messaging.VerifiedSubscription(spec: videoflow.backends.messaging.SubscriptionSpec, effective: Mapping[str, Any], mismatches: tuple[str, ...] = ())[source]

Bases: object

effective: Mapping[str, Any]
mismatches: tuple[str, ...] = ()
spec: SubscriptionSpec

videoflow.backends.observation module

An append-only observation log — the independent evidence a conformance oracle reads instead of trusting a provider’s own counters.

The design package asks every adapter and the runtime to record, for each logical operation, the identities that let a failure be attributed: the operation id, the source epoch and offset, the event or group id, the logical consumer, the delivery attempt, the ownership generation, the payload digest or obligation, the claim and pod UIDs, the device ids, and a monotonic timestamp. Image bytes are never logged. This module is deliberately tiny: a list of frozen records, a lock, and JSONL in/out, so it can be attached to a worker subprocess through a file path and merged back in the test that spawned it.

class videoflow.backends.observation.Event(kind: str, monotonic: float, wall: float, fields: Mapping[str, ~typing.Any]=<factory>)[source]

Bases: object

One observation. kind names what happened ('publish', 'settle', 'claim' …).

as_dict() → dict[str, Any][source]
fields: Mapping[str, Any]
kind: str
monotonic: float
wall: float
videoflow.backends.observation.FIELDS = ('op_id', 'source_epoch', 'source_offset', 'event_id', 'group_id', 'consumer', 'attempt', 'generation', 'payload_digest', 'obligation_id', 'claim_uid', 'pod_uid', 'device_ids', 'status', 'reason')

Field names every record may carry; adapters use these keys so logs from different backends line up in one oracle.

class videoflow.backends.observation.ObservationLog(path: str | None = None, clock: Any = <built-in function monotonic>)[source]

Bases: object

Thread-safe, append-only, replayable.

  • Arguments:

    • path: when given, every event is also appended as one JSON line, so a subprocess can keep writing while its parent still reads.

    • clock: monotonic clock, injectable for deterministic tests.

emit(kind: str, **fields: Any) → Event[source]
events(kind: str | None = None, **match: Any) → list[Event][source]

Events of kind (any kind when None) whose fields equal every match item.

extend(events: Iterable[Event]) → None[source]

Merge events recorded elsewhere (a subprocess log), keeping monotonic order per source.

static load(path: str) → list[Event][source]

videoflow.backends.outcomes module

Truthful outcome types shared by every backend contract.

The failure that motivates this module is quiet: a broker query that raised was reported as “zero pending”, a lost acknowledgment as “acknowledged”, a node list the API refused as “no GPUs in use”. Each of those turned an unobservable state into a confident answer, and the code above it then made a destructive decision (declared a drain complete, reclaimed a payload, repartitioned a card). The types here make the distinction unavoidable in the type system:

  • Known[T] / Unknown: an observation either carries a value with the time and generation it was observed at, or says why it could not be made. Nothing coerces Unknown to a default.

  • Accepted / Rejected / PublicationUnknown / PublicationUnresolvable: what happened to a publication. Unknown means the side effect may have happened (a lost receipt); Unresolvable means the backend has no ledger to ask, so the runtime must reconcile through its own durable intent.

  • SettleConfirmed / SettleUnknown / SettleStale: what happened to a settlement. A stale settlement is one issued from a delivery attempt that a newer attempt has superseded; it is refused before it reaches the broker.

  • CleanupObservation: what a teardown actually removed, and what it could not — never a silent return on a failed listing.

class videoflow.backends.outcomes.Accepted(publication_id: str, sequence: int | None, duplicate: bool, durability_boundary: str)[source]

Bases: object

The backend durably took the envelope. duplicate is True when it deduplicated a retry.

duplicate: bool
durability_boundary: str
publication_id: str
sequence: int | None
class videoflow.backends.outcomes.CleanupObservation(complete: bool, removed: tuple[str, ...], remaining: tuple[str, ...], reason: str = '')[source]

Bases: object

What a teardown did. complete is False when anything owned could not be confirmed removed — including when the inventory itself could not be read, which is not “nothing to remove”.

complete: bool
reason: str = ''
remaining: tuple[str, ...]
removed: tuple[str, ...]
videoflow.backends.outcomes.DURABILITY_MEMORY = 'memory'

Durability boundaries an Accepted publication can report having crossed.

class videoflow.backends.outcomes.Known(value: T, observed_at: float, generation: str | None = None)[source]

Bases: Generic[T]

A successful observation.

  • Arguments:

    • value: what was observed.

    • observed_at: monotonic time of the observation, so two observations can be ordered without trusting wall clocks.

    • generation: the provider’s version of the observed state when it has one (a Kubernetes resourceVersion, a consumer info sequence, a ledger version). None when the provider offers nothing comparable.

generation: str | None = None
observed_at: float
value: T
class videoflow.backends.outcomes.PublicationUnknown(publication_id: str, reason: str)[source]

Bases: object

A response was lost or a deadline expired after the send; the envelope may be stored.

publication_id: str
reason: str
class videoflow.backends.outcomes.PublicationUnresolvable(publication_id: str, reason: str)[source]

Bases: object

The backend keeps no queryable publication history for this id; only the runtime’s own intent can reconcile it.

publication_id: str
reason: str
class videoflow.backends.outcomes.Rejected(publication_id: str, reason: str, retryable: bool)[source]

Bases: object

The backend definitely did not take the envelope. retryable says whether waiting can help.

publication_id: str
reason: str
retryable: bool
class videoflow.backends.outcomes.SettleConfirmed(token: 'DeliveryToken', settlement_id: str)[source]

Bases: object

settlement_id: str
token: DeliveryToken
class videoflow.backends.outcomes.SettleStale(token: DeliveryToken, current_attempt: int, current_generation: str)[source]

Bases: object

Refused before reaching the broker: a newer attempt owns this logical message.

current_attempt: int
current_generation: str
token: DeliveryToken
class videoflow.backends.outcomes.SettleUnknown(token: DeliveryToken, reason: str)[source]

Bases: object

The settlement was sent but not confirmed; the delivery may still be leased or may redeliver.

reason: str
token: DeliveryToken
class videoflow.backends.outcomes.Unknown(reason: str, observed_at: float, detail: str = '')[source]

Bases: object

An observation that could not be made. reason is a short machine-readable category ('timeout', 'auth', 'malformed', 'unreachable', 'unsupported'); detail is the human-readable why.

detail: str = ''
observed_at: float
reason: str
videoflow.backends.outcomes.is_known(observation: Known[T] | Unknown) → TypeGuard[Known[T]][source]
videoflow.backends.outcomes.known(value: T, generation: str | None = None) → Known[T][source]
videoflow.backends.outcomes.unknown(reason: str, detail: str = '') → Unknown[source]
videoflow.backends.outcomes.value_or(observation: Known[T] | Unknown, default: T) → T[source]

The observed value, or default when unknown. Only for diagnostics and display: a decision that would act on default must branch on is_known instead, which is the whole point of the type.

videoflow.backends.payload module

The PayloadStore contract: where the image bytes are, and how long they must survive.

The reviewed implementation kept one decrement-only counter per blob and deleted the blob when it reached zero. Two deliveries of the same message to the same reader decremented twice; a counter that expired between EXISTS and DECR deleted a blob another reader still needed; a crash after the acknowledgment but before the release leaked the blob until its TTL. The contract here replaces the counter with obligations: named, idempotent, generation-fenced claims by logical readers. A payload is reclaimable only when no obligation names it; a release is idempotent by (obligation_id, generation); a TTL is an expiry policy or a final safety net, never proof that every consumer finished.

Read outcomes are typed: transient failures retry, a missing object is a durable data-loss outcome, a digest mismatch is corruption. None of them is a silent None.

class videoflow.backends.payload.Corrupt(ref: videoflow.backends.payload.ImmutablePayloadRef, expected_digest: str, actual_digest: str)[source]

Bases: object

actual_digest: str
expected_digest: str
ref: ImmutablePayloadRef
class videoflow.backends.payload.DurableReceipt(ref: videoflow.backends.payload.ImmutablePayloadRef, obligation_id: str, deadline: float)[source]

Bases: object

deadline: float
obligation_id: str
ref: ImmutablePayloadRef
class videoflow.backends.payload.ImmutablePayloadRef(store: str, key: str, size: int, digest: str, generation: str, content_id: str)[source]

Bases: object

  • Arguments:

    • store: the store identity the ref belongs to ('redis', 'memory').

    • key: the store-local key (the wire BlobRef.ref).

    • size: byte length of the object.

    • digest: hex SHA-256 of the object, verified on read when the store keeps it.

    • generation: the ownership generation minted at put; a release from an older generation is stale and cannot delete a newer object at the same key.

    • content_id: the logical content identity (event id + producer), so an unchanged frame forwarded through metadata stages can share one object.

content_id: str
digest: str
generation: str
key: str
size: int
store: str
class videoflow.backends.payload.Missing(ref: videoflow.backends.payload.ImmutablePayloadRef)[source]

Bases: object

ref: ImmutablePayloadRef
class videoflow.backends.payload.ObligationLedger[source]

Bases: ABC

What the runtime knows about obligations, for reconciliation: which readers are still required for which refs. A store reconciles its objects against it.

A ledger speaks for the obligation families it is authoritative for and no other: the runtime ledger knows its flow’s readers and publishers’ intents, not an archive’s or a dead-letter pin’s (BLOB-13 families archive/*, dlq/*), so a store cancels an obligation the ledger does not require only when the ledger is authoritative for it. The default — everything — is the whole-truth ledger a flow-wide reconciler or a test builds.

authoritative(obligation_id: str) → bool[source]

Whether this ledger may cancel obligation_id when it does not require it.

abstractmethod required_obligations() → Mapping[str, tuple[str, ...]][source]

ref key -> obligation ids still required.

class videoflow.backends.payload.PayloadBytes(ref: videoflow.backends.payload.ImmutablePayloadRef, data: bytes, digest_verified: bool)[source]

Bases: object

data: bytes
digest_verified: bool
ref: ImmutablePayloadRef
class videoflow.backends.payload.PayloadStore[source]

Bases: ABC

abstractmethod acquire_obligation(ref: ImmutablePayloadRef, obligation_id: str, deadline: float) → DurableReceipt[source]
abstractmethod capabilities() → PayloadCapabilities[source]
abstractmethod inventory() → Known[tuple[ImmutablePayloadRef, ...]] | Unknown[source]
abstractmethod put(data: bytes, content_id: str, contract: RetentionContract) → ImmutablePayloadRef[source]
abstractmethod read(ref: ImmutablePayloadRef) → PayloadBytes | TransientFailure | Missing | Corrupt[source]
abstractmethod reconcile(ledger: ObligationLedger, operation_id: str) → ReclamationObservation[source]
ref_for_key(key: str) → ImmutablePayloadRef[source]

The ref for a wire key (BlobRef.ref), completed from the store’s own metadata when it keeps any: size, digest and generation as recorded at put, so a reader that only has the key can still verify what it reads and release against the right generation. A store without metadata for the key (an RFC 0002 publisher, or metadata that expired) returns a ref with empty digest and generation — readable, unverifiable, and released against nothing.

abstractmethod release_obligation(ref: ImmutablePayloadRef, obligation_id: str, completion_receipt: str) → ReleaseReceipt[source]
abstractmethod renew_obligation(ref: ImmutablePayloadRef, obligation_id: str, deadline: float) → DurableReceipt[source]
class videoflow.backends.payload.ReclamationObservation(reclaimed: tuple[str, ...], retained: tuple[str, ...], unknown: tuple[str, ...])[source]

Bases: object

reclaimed: tuple[str, ...]
retained: tuple[str, ...]
unknown: tuple[str, ...]
class videoflow.backends.payload.ReleaseReceipt(ref: ImmutablePayloadRef, obligation_id: str, remaining: int | None, reclaimed: bool, stale: bool, unknown: bool = False)[source]

Bases: object

stale means the release named an older generation and touched nothing; unknown means the store applied (or may have applied) the release but the response was lost — the caller retries, which is safe because a release is idempotent by (obligation_id, generation).

obligation_id: str
reclaimed: bool
ref: ImmutablePayloadRef
remaining: int | None
stale: bool
unknown: bool = False
class videoflow.backends.payload.RetentionContract(ttl_seconds: int, horizon_seconds: int, durable_required: bool, obligations: tuple[str, ...] = ())[source]

Bases: object

  • Arguments:

    • ttl_seconds: the store-side expiry backstop.

    • horizon_seconds: the recovery/replay horizon the flow promised; a TTL shorter than it is rejected at admission unless obligations pin the object for the whole horizon.

    • durable_required: whether the profile needs the object to survive the store’s own failure model (reliable work) or tolerates loss (live).

    • obligations: the logical readers that will claim the object up front.

durable_required: bool
horizon_seconds: int
obligations: tuple[str, ...] = ()
ttl_seconds: int
class videoflow.backends.payload.TransientFailure(ref: ImmutablePayloadRef, reason: str)[source]

Bases: object

The store was unreachable or slow; the bytes may still exist. Retry. (An outcome, not the core error.)

reason: str
ref: ImmutablePayloadRef
videoflow.backends.payload.admit_retention(contract: RetentionContract) → None[source]

Admit a retention contract before a byte is written (BLOB-14 step 5, PAY-009): a durable_required contract whose TTL is shorter than the recovery horizon it promises is rejected unless obligations pin the object for the whole horizon. Both reference stores extend a pinned object’s life to the horizon, so with obligations the TTL is a backstop rather than the bound; without them it is the bound, and a horizon it cannot cover is a promise nobody keeps.

  • Raises:

    • IncompatibleProfile: the contract promises a horizon its TTL cannot cover and no obligation pins the object.

videoflow.backends.payload_bridge module

A BlobStore façade over a PayloadStore, so the wire codec’s offload path (encode_envelope(blob_store = ...) / hydrate_message(decoded, blob_store)) can drive an obligation-keeping store without the codec learning about obligations, contracts or typed read outcomes.

The codec sees the RFC 0002 interface it always has: put_with_readers returns the wire key, get returns bytes or raises. Behind it the bridge translates:

  • a put becomes PayloadStore.put under a RetentionContract naming the reader obligations the publisher acquires up front (BLOB-14 step 1) plus the publisher’s own intent/<publication_id> for the send in flight; the ref of the last put is kept so the messenger can release that intent once the publication’s outcome is known. The reader set is the bridge’s (VF_BLOB_READER_IDS), whichever of the codec’s two put paths is taken: the RFC 0002 reader count chooses between them and means nothing here. With no reader ids at all the object is TTL-only — no obligations, not even the intent — because a lone intent released on the PubAck would reclaim an object its (unidentified) readers have yet to fetch (ENV-12: ids unset ⇒ count semantics, else TTL-only);

  • a read’s typed outcome becomes the error the messenger’s ladder already understands: TransientFailure (retry — never malformed bytes, BLOB-15), and DecodeError for Missing/Corrupt (terminal for that delivery, dead-lettered with a VF-Error naming the ref and the reason);

  • release is a no-op — obligations are released by reader id through the store, after a confirmed settlement, which is the messenger’s decision.

class videoflow.backends.payload_bridge.PayloadStoreBlobBridge(store: ~videoflow.backends.payload.PayloadStore, obligations: ~typing.Sequence[str], horizon_seconds: int, durable_required: bool, content_id: ~typing.Callable[[], str], intent: ~typing.Callable[[], str | None] = <function PayloadStoreBlobBridge.<lambda>>)[source]

Bases: BlobStore

  • Arguments:

    • store: the obligation-keeping store.

    • obligations: the reader obligation ids every put acquires up front

      (VF_BLOB_READER_IDS: <child> per competing child, <child>/p<i> per partitioned replica).

    • horizon_seconds: the recovery horizon the flow promised; a shorter

      TTL is the admission’s problem, recorded on the contract here.

    • durable_required: whether the channel’s profile needs the bytes to

      survive the store’s own failure model.

    • content_id: the logical content identity of the message being

      published, evaluated at put time.

    • intent: the intent/<publication_id> obligation for the send in

      flight, evaluated at put time; None when there is none.

get(ref: str) → bytes[source]

Resolves a reference previously returned by put() back into bytes.

last_ref: ImmutablePayloadRef | None

The ref of the most recent put — what the publisher releases its intent on.

put(data: bytes, ttl_seconds: int = 3600) → str[source]

Stores data and returns an opaque reference string that get() can resolve later.

put_with_readers(data: bytes, readers: int, ttl_seconds: int = 3600) → str[source]

Stores data expecting exactly readers downstream reads, enabling the store to reclaim the blob once every reader has released it (BLOB-5).

Default implementation ignores readers and delegates to put() — a store with no reclamation support degrades to plain TTL expiry.

release(ref: str) → None[source]

One downstream reader is finished with ref (its message was acked, BLOB-6). Default no-op.

property store: PayloadStore

videoflow.backends.runtime module

Runtime contracts: the durable state that makes processing correct across restarts, independent of the transport underneath.

RuntimeStore is a small versioned key-value store with compare-and-swap; three implementations exist (memory for tests, a file directory for the local engine, Redis for clusters) and the planner reads their capabilities to decide whether a restart-safe profile can be admitted at all. The FlowRuntime built on it (Phase 3 of the backend plan) owns the group ledger, ownership epochs, completion barriers, the publication outbox and source epochs.

This module also carries the one identity rule that must be pure and stable across languages: group_identity — the id of a time-aligned input group is derived from its members, not from a rounded timestamp, so two distinct groups whose event times round to the same microsecond never collide, and a replay of the same members yields the same id.

class videoflow.backends.runtime.CommitReceipt(group_id: str, state_version: str, output_intents: tuple[str, ...], effect_intents: tuple[str, ...])[source]

Bases: object

effect_intents: tuple[str, ...]
group_id: str
output_intents: tuple[str, ...]
state_version: str
class videoflow.backends.runtime.CompletionReceipt(partition_id: str, epoch: int, final_sequence: int, version: str)[source]

Bases: object

epoch: int
final_sequence: int
partition_id: str
version: str
videoflow.backends.runtime.DEFAULT_PARTITION_LEASE_SECONDS = 10.0

How many times a compare-and-swap is retried against a concurrent writer before the caller is told the store would not take the update. How long a partition lease lasts without renewal (VF_PARTITION_LEASE_SECONDS): a crashed holder’s replacement waits at most this long; a live holder renews every third of it from its own thread. Long enough that a scheduler hiccup does not usurp a paused process, short enough that a crash is not a long outage.

videoflow.backends.runtime.DEFAULT_RECONCILE_INTERVAL_SECONDS = 60.0

How often a running worker reconciles payload obligations (BLOB-14 step 4); 0 disables the periodic pass (the start pass always runs).

class videoflow.backends.runtime.FlowRuntime(store: ~videoflow.backends.runtime.RuntimeStore, flow_id: str, run_id: str, node: str, replica_id: int = 0, nb_tasks: int = 1, partition_by: str | None = None, parent_replicas: ~typing.Mapping[str, int] | None = None, clock: ~typing.Callable[[], float] = <built-in function time>, lease_seconds: float = 10.0, sleep: ~typing.Callable[[float], None] = <built-in function sleep>)[source]

Bases: object

The per-node runtime ledger of RFC 0006 CTRL-4, over any RuntimeStore: terminators and received sets (the EOS-7 completion barrier), ownership epochs with fencing, the publication outbox, attempt counts, the terminal log, pending dead-letter handoffs, open join groups, checkpoints and sink-effect markers. Every mutation is a compare-and-swap on one key; every read is what the store holds, so a replacement process starts from the same facts.

What it never does: decide policy. The messenger asks it questions (completion_state, attempts_for) and records facts; the answers hold only as far as the store’s own capabilities() — a memory store makes every record process-local, and the callers know it (durable_shared()).

  • Arguments:

    • store: the RuntimeStore (VF_RUNTIME_STORE_URL).

    • flow_id / run_id / node: the ledger’s namespace, vf/{flow}/{run}/{node}/.

    • replica_id / nb_tasks / partition_by: this replica’s identity within the node.

    • parent_replicas: {parent: nb_tasks} from VF_PARENT_REPLICAS (ENV-11); a parent absent from it has no completion barrier.

    • clock: epoch seconds, injectable for tests.

completion_state takes the durable’s (num_pending, num_ack_pending) observation as the messenger reads it (_consumer_pending).

aborted_parents() → dict[str, Mapping[str, Any]][source]

Parents with a recorded ABORT, with the error each carried — an ABORT outranks a clean EOS (ABORT-3).

acquire_partition(partition_id: str | None = None, expected_epoch: int | None = None, wait: bool = True) → OwnershipToken[source]

Take (or retake) ownership of a partition: the epoch is CAS-incremented and a fresh fencing token minted. expected_epoch refuses the acquisition when someone else has moved the epoch since the caller last saw it.

A partition is leased: the record names its holder and a lease_until the holder renews (renew_partition) and clears on a graceful close (release_partition). Another process asking for a partition whose lease is live waits for it to lapse — at most one lease length — and is refused the moment the holder renews it (a renewal proves the holder alive): a singleton scaled out by adding replicas fails explicitly at bind time instead of two owners fencing each other in turns (RUN-018, RUN-019), while the replacement for a crashed holder takes over as soon as its lease lapses, and at once after a graceful close. wait = False refuses a live lease at once instead of waiting for it to lapse (claim_replica_slot probes the free slots that way first).

  • Raises:

    • StaleAuthority: expected_epoch no longer matches (another owner took over), or the store would not take the update.

    • OwnershipConflict: another live process holds the partition’s lease.

attempts_for(message_id: str) → int[source]

Failed attempts of message_id that count against its budget (never worker_fatal).

capabilities() → RuntimeCapabilities[source]
check_authority(token: OwnershipToken) → None[source]

Refuse a commit under a superseded epoch: the store’s epoch and token must be the caller’s.

checkpoint(state: bytes, replay_position: Mapping[str, Any]) → str[source]

One CAS write: state and position describe the same committed prefix. replay_position is the node’s own vocabulary ({'offset': 42} for a replayable source, {'group': <input key>} for a stateful node, plus the output committed with it). Returns the new version.

clear_pending_handoff(record_id: str) → None[source]
commit_completion(parent: str, token: OwnershipToken | None = None) → CompletionReceipt[source]

Record that parent is complete for this replica’s partition; refused under a superseded epoch of that partition (StaleAuthority). Partitions are independent authorities — a sibling replica’s epoch says nothing about this one’s — so the record is keyed by partition id.

completed_parents(partition_id: str | None = None) → set[str][source]

Parents this replica’s partition (or partition_id) has committed complete.

completion_state(parent: str, durable: str, observation: Known[tuple[int, int]] | Unknown, pending_halves: bool = False) → str[source]

Where parent stands for this node (EOS-7): open until every expected replica’s terminator is recorded; aborted once an ABORT is recorded and the parent is drained; draining while the received count is short of the terminators’ final counts, a join still holds a half, or the broker still reports deliveries; unknown while the broker cannot be observed — never complete on evidence that was never obtained.

current_epoch(partition_id: str | None = None) → int[source]
durable_shared() → bool[source]

Whether records here outlive this process and are seen by sibling replicas: the EOS-7 and D11 gate.

effect_seen(key: str) → bool[source]
property flow_id: str
group_members(group_id: str) → dict[str, tuple[str, str, int]][source]
held_partition(partition_id: str | None = None) → OwnershipToken | None[source]

The token this process holds for the partition, if it acquired it already (claim_replica_slot).

intend_publication(publication_id: str, digest: str, payload_refs: Sequence[str] = (), kind: str = 'data', readers: Sequence[str] = ()) → str[source]

Record the intent to publish before the send; returns the key’s version (an existing intent is kept). readers are the reader obligations the put acquired (VF_BLOB_READER_IDS), kept with the refs so a reconciler can tell which readers still owe a release (RuntimeObligationLedger).

key(*parts: str) → str[source]
property lease_seconds: float
mark_effect(key: str, retention_seconds: float) → bool[source]

Mark an effect as applied; False when it already was (within its retention).

property node: str
open_groups() → list[RecoveryRecord][source]
outbox_entry(publication_id: str) → OutboxEntry | None[source]
outbox_of(node: str) → list[dict[str, Any]][source]

Every outbox document of node in this run (a parent’s, for a reader reconciling the obligations it and its siblings owe): the raw records — refs, readers, outcome, seq when accepted.

property parent_replicas: Mapping[str, int]
partition_id() → str[source]
pending_handoffs() → list[PendingHandoff][source]
persist_group(group_id: str, members: Mapping[str, tuple[str, str, int]], tokens: Mapping[str, str]) → RecoveryRecord[source]

Record a group’s members by logical id as they arrive (idempotent; a redelivered member replaces its token).

published_count() → int[source]

Distinct DATA ids this replica published (accepted or deduplicated): a terminator’s seq (EOS-7).

received_ids(parent: str, durable: str) → set[str][source]

The union over this node’s replicas of the distinct ids delivered from parent on durable.

record_attempt(message_id: str, disposition: str) → int[source]

One failed attempt of message_id under disposition; worker_fatal never counts. Returns the budgeted count.

record_pending_handoff(record_id: str, headers: Mapping[str, str], raw: bytes) → None[source]

A dead letter the broker did not accept: kept so a later attempt (or a restart) re-publishes it under the same id.

record_received(parent: str, durable: str, message_id: str) → bool[source]

Note a distinct DATA id delivered from parent on durable to this replica; False when already noted.

record_terminal(record: Mapping[str, Any]) → str[source]

Append a terminal record (a delivery ended without a dead letter) and return its reference.

record_terminator(parent: str, replica_id: int, kind: str, seq: int, error: Mapping[str, Any] | None = None) → bool[source]

Record a parent replica’s terminator before it is acked. Idempotent by (parent, replica_id, kind): returns False when that fact was already recorded (a duplicate marker), True when it is new.

release_partition(token: OwnershipToken) → None[source]

Clear the lease on a graceful close so a replacement takes over at once; a superseded token releases nothing.

renew_partition(token: OwnershipToken) → float[source]

Extend the lease of a partition this process holds; returns the new lease_until. A record that no longer carries the token belongs to another owner: StaleAuthority.

property replica_id: int
resolve_publication(publication_id: str, outcome: Accepted | Rejected | PublicationUnknown | PublicationUnresolvable) → None[source]

Record the send’s outcome; an Accepted (duplicate included) DATA id counts once for EOS-7.

restore_checkpoint() → tuple[bytes | None, dict[str, Any]][source]
property run_id: str
settle_group(group_id: str) → None[source]
source_epoch() → str[source]
property store: RuntimeStore
terminal_entries() → list[dict[str, Any]][source]
terminators(parent: str) → list[TerminatorRecord][source]
unresolved_publications() → list[OutboxEntry][source]

Intents whose send was never confirmed (intent or unknown): what a restart must reconcile first.

videoflow.backends.runtime.LEASE_POLL_SECONDS = 0.5

How often a would-be owner re-reads a held lease while waiting for it to lapse.

class videoflow.backends.runtime.OutboxEntry(publication_id: str, digest: str, payload_refs: tuple[str, ...], outcome: str, kind: str, replica_id: int)[source]

Bases: object

digest: str
kind: str
outcome: str
payload_refs: tuple[str, ...]
publication_id: str
replica_id: int
class videoflow.backends.runtime.OwnershipToken(partition_id: str, epoch: int, fencing_token: str)[source]

Bases: object

A fencing token: commits carry it, and a commit under a superseded epoch is refused.

epoch: int
fencing_token: str
partition_id: str
class videoflow.backends.runtime.PendingHandoff(record_id: str, headers: Mapping[str, str], raw: bytes, attempts: int)[source]

Bases: object

attempts: int
headers: Mapping[str, str]
raw: bytes
record_id: str
class videoflow.backends.runtime.RecoveryRecord(group_id: str, members: Mapping[str, str], version: str)[source]

Bases: object

group_id: str
members: Mapping[str, str]
version: str
class videoflow.backends.runtime.RuntimeStore[source]

Bases: ABC

Versioned KV. cas succeeds only when the key’s current version equals expected_version (None = must not exist). Versions are opaque strings the store mints; callers never fabricate one.

abstractmethod append(log: str, record: bytes) → int[source]
abstractmethod capabilities() → RuntimeCapabilities[source]
abstractmethod cas(key: str, expected_version: str | None, value: bytes) → bool[source]
abstractmethod delete(key: str, expected_version: str | None) → bool[source]
abstractmethod get(key: str) → tuple[bytes | None, str | None][source]
abstractmethod log_entries(log: str) → list[bytes][source]

Every record appended to log, oldest first (empty for an unknown log).

abstractmethod scan(prefix: str) → list[tuple[str, bytes, str]][source]
class videoflow.backends.runtime.TerminatorRecord(parent: str, replica_id: int, kind: str, seq: int, error: Mapping[str, Any] | None)[source]

Bases: object

error: Mapping[str, Any] | None
kind: str
parent: str
replica_id: int
seq: int
videoflow.backends.runtime.claim_replica_slot(store: ~videoflow.backends.runtime.RuntimeStore, flow_id: str, run_id: str, node: str, nb_tasks: int, partition_by: str | None = None, parent_replicas: ~typing.Mapping[str, int] | None = None, slots: int | None = None, clock: ~typing.Callable[[], float] = <built-in function time>, lease_seconds: float = 10.0, sleep: ~typing.Callable[[float], None] = <built-in function sleep>) → FlowRuntime[source]

A replica id for a process that was given none (ENV-5 step 3): the lowest slot 0..slots-1 whose partition lease is free or lapsed, claimed through the ledger, and the FlowRuntime that holds it. A Kubernetes Deployment pod has no ordinal, so its competing replicas would all report as replica 0 — one terminator where the barrier expects N, one lease fought over by N pods; claiming a slot gives the live pods distinct, stable identities, and the replacement for a crashed pod resumes that pod’s slot (and its ledger: unresolved intents, terminators) once its lease lapses.

Free and lapsed slots are taken at once; when none is, each live slot is waited on in turn for at most one lease length (a renewal proves its holder alive). A process that finds every slot held is an extra replica, refused the way a scaled-out singleton is (RUN-018).

  • Arguments:

    • nb_tasks: the node’s declared replica count (what the runtime is built with).

    • slots: how many identities may be claimed — nb_tasks unless a scaler may run more pods (VF_REPLICA_SLOTS, its ceiling), which are wanted replicas rather than extras.

  • Raises:

    • OwnershipConflict: every slot is held under a live lease.

videoflow.backends.runtime.group_identity(members: Mapping[str, tuple[str, str, int]], window_id: str | None, rounded_micros: int | None = None) → str[source]

Canonical identity of an input group.

  • Arguments:

    • members: parent name -> (producer, trace_id, seq) of the member it contributed.

    • window_id: the join window or namespace the group formed in (None for trace joins).

    • rounded_micros: the legacy tw-<µs> prefix component, kept so ids stay sortable by time; distinctness comes from the member hash.

videoflow.backends.runtime.members_signature(members: Sequence[str]) → str[source]
videoflow.backends.runtime.partition_lease_from_env(value: str | None) → float[source]

VF_PARTITION_LEASE_SECONDS parsed: absent means the default; a non-positive or non-numeric value fails fast.

videoflow.backends.runtime.reconcile_interval_from_env(value: str | None) → float[source]

VF_RECONCILE_INTERVAL_SECONDS parsed: absent means the default; negative or non-numeric fails fast.

videoflow.backends.runtime.replayable_trace_id(node: str, offset: int, analysis_version: str | None = None) → str[source]

Trace id of a replayable source’s frame at offset; a new analysis version is a new namespace.

videoflow.backends.runtime.source_epoch_trace_id(node: str, epoch: str, sequence: int) → str[source]

Trace id of a live source’s sequence-th capture within capture epoch epoch.

videoflow.backends.runtime.store_shared_and_durable(store: RuntimeStore) → bool[source]

Whether a store’s records outlive one process and are seen by sibling replicas (what makes a ledger a ledger).