Module

contracts.rules_reactive

Reactive bus contract checks.

Validates DependencyIndex configuration at app.check() time:

  • Block references point to real template blocks
  • Derivation graph is a DAG (no cycles)
  • Declared emitted paths are registered in the dependency index
  • Audience-filtered scopes have connection-aware subscribers
4Functions

Functions

check_reactive_block_existence 2 list[ContractIssue] ▼
Verify every BlockRef in the index references a real template block.
def check_reactive_block_existence(dep_index: DependencyIndex, env: Environment) -> list[ContractIssue]
Parameters
Name Type Description
dep_index DependencyIndex
env Environment
Returns
list[ContractIssue]
check_reactive_derivation_dag 1 list[ContractIssue] ▼
Detect cycles in the derivation graph. Cycles are handled safely at runtime (B…
def check_reactive_derivation_dag(dep_index: DependencyIndex) -> list[ContractIssue]

Detect cycles in the derivation graph.

Cycles are handled safely at runtime (BFS visited set), but always indicate a configuration error. Surfacing at check time is better than silent infinite-expansion prevention at runtime.

Parameters
Name Type Description
dep_index DependencyIndex
Returns
list[ContractIssue]
check_reactive_emitted_paths 2 list[ContractIssue] ▼
Warn when declared ChangeEvent paths are not registered in the index.
def check_reactive_emitted_paths(dep_index: DependencyIndex, emitted_paths: object) -> list[ContractIssue]
Parameters
Name Type Description
dep_index DependencyIndex
emitted_paths object
Returns
list[ContractIssue]
check_reactive_audience_scopes 2 list[ContractIssue] ▼
Warn when audience-filtered events target scopes without ConnectionInfo.
def check_reactive_audience_scopes(audience_scopes: object, connection_scopes: object) -> list[ContractIssue]
Parameters
Name Type Description
audience_scopes object
connection_scopes object
Returns
list[ContractIssue]