vibe.review.parsing.rules.engine

Rule engine for the parsing pipeline.

Rules are declarative specifications that: - Match nodes based on predicates - Apply actions (tag, classify, transform) - Track provenance of decisions

Rules are organized by layer and applied in priority order.

Layer

Valid pipeline layers for rules.

ActionType

Types of rule actions.

RuleAction

An action to perform when a rule matches.

Attributes:
  • action_type (ActionType) –

    The type of action.

  • params (dict[str, Any]) –

    Action-specific parameters.

Rule

A parsing rule.

Attributes:
  • rule_id (str) –

    Unique identifier.

  • name (str) –

    Human-readable name.

  • layer (Layer) –

    Which layer this rule applies to (extraction, layout, structure, semantic).

  • priority (int) –

    Higher priority rules fire first (default 100).

  • predicates (list[Predicate]) –

    Conditions that must all match.

  • actions (list[RuleAction]) –

    Actions to take when rule matches.

  • provenance_note (str) –

    Note for provenance log.

  • enabled (bool) –

    Whether rule is active.

RuleMatch

Result of matching a rule against a node.

Attributes:
  • rule (Rule) –

    The matched rule.

  • node (IRNode) –

    The node that matched.

  • confidence (float) –

    Match confidence (0.0 to 1.0).

RuleResult

Result of applying rules to nodes.

Attributes:
  • matches (list[RuleMatch]) –

    All rule matches found.

  • nodes_modified (int) –

    Count of nodes modified.

  • rules_fired (int) –

    Count of rules that fired.

  • provenance_events (list[ProvenanceEvent]) –

    Generated provenance events.

RuleEngine

Engine for evaluating and applying parsing rules.

Rules are evaluated in priority order. When multiple rules match, the highest priority rule wins (or first in case of tie).

__init__ 

__init__(rules: list[Rule] | None = None, context: dict[str, Any] | None = None, predicate_functions_dir: Path | None = None) -> None

Initialize the rule engine.

Parameters:
  • rules (list[Rule] | None, default: None ) –

    Initial set of rules.

  • context (dict[str, Any] | None, default: None ) –

    Context variables for predicate evaluation.

  • predicate_functions_dir (Path | None, default: None ) –

    Directory containing predicates.py for function loading.

add_rule 

add_rule(rule: Rule) -> None

Add a rule to the engine.

add_rules 

add_rules(rules: list[Rule]) -> None

Add multiple rules.

remove_rule 

remove_rule(rule_id: str) -> bool

Remove a rule by ID. Returns True if found.

get_rules_for_layer 

get_rules_for_layer(layer: str) -> list[Rule]

Get all rules for a specific layer, sorted by priority.

match 

match(node: IRNode, layer: str) -> list[RuleMatch]

Find all rules that match a node.

Parameters:
  • node (IRNode) –

    The node to match against.

  • layer (str) –

    The pipeline layer to filter rules.
Returns:
  • list[RuleMatch] –

    List of RuleMatch objects, sorted by priority.

apply 

apply(nodes: Sequence[IRNode], layer: str, action_handlers: dict[ActionType, Callable[[IRNode, RuleAction], IRNode]] | None = None) -> RuleResult

Apply rules to a list of nodes.

Parameters:
  • nodes (Sequence[IRNode]) –

    Nodes to process.

  • layer (str) –

    Pipeline layer.

  • action_handlers (dict[ActionType, Callable[[IRNode, RuleAction], IRNode]] | None, default: None ) –

    Custom handlers for action types.
Returns:
  • RuleResult –

    RuleResult with matches and modifications.

apply_deferred_actions 

apply_deferred_actions(nodes: Sequence[IRNode]) -> list[IRNode]

Apply deferred merge/split actions to a node list.

Call this after apply() to process any merge/split markers. This method modifies the node list in place and returns it.

Parameters:
  • nodes (Sequence[IRNode]) –

    List of nodes (may include nodes marked for merge/split).
Returns:
  • list[IRNode] –

    Modified node list with merges and splits applied.