Skip to content

crp.pp

Auto-generated reference for the crp.pp subpackage.

pp

crp.pp

Predictive Positioning & World-Model Induction (SPEC-051).

CausalEdge

Bases: str, Enum

Causal edge kinds.

Rule dataclass

An induced transition rule.

Transition dataclass

One observed action-outcome transition.

SimulationResult dataclass

Outcome of a guarded-dispatch simulation.

WorldModel

Predict outcomes from induced transition rules.

predict(state, action)

Return the best predicted outcome for action in state.

Returns:

Type Description
dict[str, Any] | None

A dict with keys predicted, confidence, support,
dict[str, Any] | None

source when a rule matches; None otherwise.

add_causal_edge(graph, src, dst, kind, textual_conf, interventional_support=0)

Add a causal edge between facts.

Parameters:

Name Type Description Default
graph Any

Fact graph with add_fact and add_edge methods.

 required
src Any

Source fact (or fact id).

 required
dst Any

Target fact (or fact id).

 required
kind CausalEdge

Causal edge kind.

 required
textual_conf float

Confidence from textual extraction (reported causality).

 required
interventional_support int

Number of action-log transitions that support the edge (verified causal evidence).

 0

causal_upstream(graph, node, max_depth=3, kinds=None)

Return causal upstream nodes for node up to max_depth hops.

Parameters:

Name Type Description Default
graph Any

Fact graph with edges_to method.

 required
node Any

Target fact or fact id.

 required
max_depth int

Maximum traversal depth.

 3
kinds set[str] | None

Edge relation types to follow (defaults to causes and enables).

 None

Returns:

Type Description
list[dict[str, Any]]

List of edge dicts: cause, effect, kind, conf.

induce_rules(transitions, min_support=2, min_conf=0.8)

Induce rules from transitions.

Parameters:

Name Type Description Default
transitions list[Transition]

Observed transitions.

 required
min_support int

Minimum number of transitions required for a rule.

 2
min_conf float

Minimum fraction of transitions that must agree on an outcome feature for it to be included in the rule's prediction.

 0.8

Returns:

Type Description
list[Rule]

A list of induced Rule objects.

guarded_dispatch(state, action, risk, world, policy, execute_fn, checkpoint_fn)

Dispatch action after optional simulation for HIGH-risk operations.

Parameters:

Name Type Description Default
state dict[str, Any]

Current state features.

 required
action str

Action to take.

 required
risk str

Risk level string (e.g. LOW, MEDIUM, HIGH).

 required
world WorldModel

World model for outcome prediction.

 required
policy Any

Object with check_predicted_outcome(predicted) and sim_confidence_floor attributes.

 required
execute_fn Callable[[dict[str, Any], str, dict[str, Any] | None], dict[str, Any]]

fn(state, action, prediction) -> result.

 required
checkpoint_fn Callable[[str, dict[str, Any] | None, dict[str, Any]], dict[str, Any]]

fn(reason, prediction, state) -> result.

 required

Returns:

Type Description
dict[str, Any]

The result of execution or checkpoint.

pp.causal_ckf

crp.pp.causal_ckf

Causal edge support for the CKF (SPEC-051 §3.3.3).

Adds causes / enables / prevents edge types to a fact graph and provides counterfactual (causal-upstream) retrieval. The implementation is graph-agnostic: it operates on any object exposing add_fact/add_edge and edges_to methods, such as crp.extraction.types.FactGraph.

CausalEdge

Bases: str, Enum

Causal edge kinds.

add_causal_edge(graph, src, dst, kind, textual_conf, interventional_support=0)

Add a causal edge between facts.

Parameters:

Name Type Description Default
graph Any

Fact graph with add_fact and add_edge methods.

 required
src Any

Source fact (or fact id).

 required
dst Any

Target fact (or fact id).

 required
kind CausalEdge

Causal edge kind.

 required
textual_conf float

Confidence from textual extraction (reported causality).

 required
interventional_support int

Number of action-log transitions that support the edge (verified causal evidence).

 0

causal_upstream(graph, node, max_depth=3, kinds=None)

Return causal upstream nodes for node up to max_depth hops.

Parameters:

Name Type Description Default
graph Any

Fact graph with edges_to method.

 required
node Any

Target fact or fact id.

 required
max_depth int

Maximum traversal depth.

 3
kinds set[str] | None

Edge relation types to follow (defaults to causes and enables).

 None

Returns:

Type Description
list[dict[str, Any]]

List of edge dicts: cause, effect, kind, conf.

pp.induction

crp.pp.induction

World-model rule induction over the Tier-E action log (SPEC-051 §3.3.1).

Induces symbolic transition rules of the form (state_pattern, action) → outcome_pattern from logged (pre_state, action, post_state) transitions. Rules record support count and confidence so callers can decide whether a rule is strong enough to gate an action.

Transition dataclass

One observed action-outcome transition.

Rule dataclass

An induced transition rule.

induce_rules(transitions, min_support=2, min_conf=0.8)

Induce rules from transitions.

Parameters:

Name Type Description Default
transitions list[Transition]

Observed transitions.

 required
min_support int

Minimum number of transitions required for a rule.

 2
min_conf float

Minimum fraction of transitions that must agree on an outcome feature for it to be included in the rule's prediction.

 0.8

Returns:

Type Description
list[Rule]

A list of induced Rule objects.

pp.simulate

crp.pp.simulate

Simulation-before-action gating for HIGH-risk operations (SPEC-051 §3.3.4).

Predicts the outcome of a proposed action using a world model; if the prediction violates policy with sufficient confidence, the action is blocked and a checkpoint is returned instead. This turns reactive oversight into anticipatory oversight.

SimulationResult dataclass

Outcome of a guarded-dispatch simulation.

guarded_dispatch(state, action, risk, world, policy, execute_fn, checkpoint_fn)

Dispatch action after optional simulation for HIGH-risk operations.

Parameters:

Name Type Description Default
state dict[str, Any]

Current state features.

 required
action str

Action to take.

 required
risk str

Risk level string (e.g. LOW, MEDIUM, HIGH).

 required
world WorldModel

World model for outcome prediction.

 required
policy Any

Object with check_predicted_outcome(predicted) and sim_confidence_floor attributes.

 required
execute_fn Callable[[dict[str, Any], str, dict[str, Any] | None], dict[str, Any]]

fn(state, action, prediction) -> result.

 required
checkpoint_fn Callable[[str, dict[str, Any] | None, dict[str, Any]], dict[str, Any]]

fn(reason, prediction, state) -> result.

 required

Returns:

Type Description
dict[str, Any]

The result of execution or checkpoint.

pp.world_model

crp.pp.world_model

Symbolic world model for predictive positioning (SPEC-051 §3.3.2).

Given a proposed action and current state, predicts the outcome by matching against induced rules. The model is intentionally rule-first and gradient-free: rules are inspectable, sample-efficient, and cheap to update as the action log grows.

WorldModel

Predict outcomes from induced transition rules.

predict(state, action)

Return the best predicted outcome for action in state.

Returns:

Type Description
dict[str, Any] | None

A dict with keys predicted, confidence, support,
dict[str, Any] | None

source when a rule matches; None otherwise.