Interpret

sakshi.interpret

Interpretation and anomaly-detection runtime classes.

DEFAULT_DECILES = 10 module-attribute

DEFAULT_MISCALIBRATION_THRESHOLD = 0.05 module-attribute

DEFAULT_WINDOW_SIZE = 200 module-attribute

DEFAULT_DEPTH_DRIFT_THRESHOLD = 5 module-attribute

DEFAULT_DEPTH_WARN_THRESHOLD = 3 module-attribute

ADistanceDetector

Compute A-distance between current activations and a rolling baseline.

The implementation uses normalized L2 distance as a deterministic, dependency-free proxy for minimum classification error:

a_dist ~= 2 * ||current - baseline||_2 / sqrt(d)

baseline_window: int property

baseline_window_size: int property

Alias for compatibility with existing tests.

has_baseline: bool property

True once enough observations exist to form a baseline.

observe(activations: dict[str, float]) -> AnomalyEvent | None async

Process one observation and return an event if anomalous.

get_anomaly_frequency() -> float

Return fraction of observations that triggered an anomaly.

update_baseline(window_size: int | None = None) -> None async

Optionally set a new baseline window size.

get_baseline_snapshot() -> dict[str, float]

Return baseline mean vector using current window size.

AnomalyEvent

Bases: BaseModel

Statistical anomaly detected in an activation stream.

The source field tags the anomaly's origin lane. The A-distance detector defaults to WORLD because it operates on environment activations; meta-cycle internal detectors should construct events with source=COGNITIVE. COMPOUND is reserved for events that legitimately straddle both lanes and must be decomposed before explanation.

a_distance: float = Field(ge=0.0, le=2.0, description='A-distance value in the theoretical 0-2 range') class-attribute instance-attribute

current_activations: dict[str, float] = Field(default_factory=dict) class-attribute instance-attribute

baseline_mean: dict[str, float] = Field(default_factory=dict) class-attribute instance-attribute

detected_at: datetime = Field(default_factory=(lambda: datetime.now(UTC))) class-attribute instance-attribute

threshold_used: float = 0.5 class-attribute instance-attribute

source: AnomalySourceType = Field(default=(AnomalySourceType.WORLD), description='Origin lane: WORLD | COGNITIVE | COMPOUND') class-attribute instance-attribute

severity: str property

PredicateStream

Sliding-window buffer for activation observations.

push(activations: dict[str, float]) -> None

Append a new observation.

get_window(size: int | None = None) -> list[dict[str, float]]

Return the last size observations, or all observations.

mean_vector(keys: list[str] | None = None) -> dict[str, float]

Compute element-wise mean across all stored observations.

AmbiguityReport dataclass

Summary of ambiguity in a belief distribution.

ambiguity_score: float instance-attribute

decision_relevant: bool instance-attribute

blocked_goals: list[str] instance-attribute

epistemic_action: str instance-attribute

confidence: float instance-attribute

hypothesis_count: int instance-attribute

DRFreeAmbiguityDetector

Detect decision-relevant ambiguity from hypothesis probabilities.

AMBIGUITY_THRESHOLD = 0.6 class-attribute instance-attribute

MIN_HYPOTHESES = 2 class-attribute instance-attribute

detect(belief_distribution: dict[str, float], active_goals: list[str], context: dict[str, Any] | None = None) -> AmbiguityReport

AnomalyPersistenceTracker

Track consecutive anomaly counts per goal and compute escalation.

record_anomaly(goal_id: str) -> AnomalyEscalation

Increment anomaly count for a goal and return escalation state.

record_success(goal_id: str) -> None

Reset a goal's anomaly streak after a successful cycle.

get_escalation(goal_id: str) -> AnomalyEscalation

Return current escalation for a goal without changing state.

CalibrationReport dataclass

Aggregated calibration view over the current window.

sample_count: int instance-attribute

self_trust_score: float instance-attribute

warnings: tuple[CalibrationWarning, ...] instance-attribute

CalibrationTracker

Sliding-window tracker for (predicted_confidence, actual) pairs.

Parameters:

Name	Type	Description	Default
window_size	int	Maximum number of recent observations to retain.	DEFAULT_WINDOW_SIZE
deciles	int	Number of equally spaced confidence bins. Default 10.	DEFAULT_DECILES
miscalibration_threshold	float	Absolute delta between predicted mean and empirical hit rate above which a warning is issued for the bin. Default 0.05 (5 percentage points).	DEFAULT_MISCALIBRATION_THRESHOLD

sample_count: int property

record(*, predicted_confidence: float, actually_correct: bool, label: str = '') -> ConfidenceObservation

Record one observation and return its frozen DTO.

report() -> CalibrationReport

Compute the current calibration report.

CalibrationWarning dataclass

One miscalibrated decile.

decile_low: float instance-attribute

decile_high: float instance-attribute

sample_count: int instance-attribute

predicted_mean: float instance-attribute

empirical_hit_rate: float instance-attribute

delta: float instance-attribute

is_overconfident: bool property

is_underconfident: bool property

ConfidenceObservation dataclass

One predicted-confidence / actually-correct pair.

predicted_confidence: float instance-attribute

actually_correct: bool instance-attribute

label: str = '' class-attribute instance-attribute

occurred_at: datetime = field(default_factory=(lambda: datetime.now(UTC))) class-attribute instance-attribute

ConfusionDecision dataclass

Bases: Generic[T]

Result of one cost-weighted classification call.

chosen_class: T instance-attribute

expected_cost: float instance-attribute

naive_argmax: T instance-attribute

naive_max_probability: float instance-attribute

cost_matrix_label: str = '' class-attribute instance-attribute

ConfusionWeighter

Bases: Generic[T]

Apply a cost matrix to a probability vector.

Parameters:

Name	Type	Description	Default
cost_matrix	Mapping[T, Mapping[T, float]]	{predicted_class: {true_class: cost}}.	required
label	str	Optional human-readable label for the matrix; carried into emitted records for audit clarity.	''

label: str property

decide(probabilities: Mapping[T, float]) -> ConfusionDecision[T]

Return the cost-weighted choice plus the naive baseline.

from_uniform(classes: Sequence[T], *, false_positive_cost: float = 1.0, false_negative_cost: float = 1.0, label: str = 'uniform') -> ConfusionWeighter[T] classmethod

Build a uniform off-diagonal cost matrix.

Diagonal entries are 0; off-diagonal entries split between false_positive_cost (predicted-positive when truth is negative — that is, when predicted == classes[0] and true != classes[0]) and false_negative_cost for the complementary case. Useful as a starting point hosts can further specialize.

ExpectationEvaluator

Evaluate registered expectations against phase pre/post state.

register(expectation: CognitiveExpectation) -> None

Register an expectation for a phase.

get_expectations_for_phase(phase_name: str) -> list[CognitiveExpectation]

Return expectations registered for a phase.

evaluate_phase(phase_name: str, cycle_id: str, pre_state: dict[str, Any], post_state: dict[str, Any]) -> list[ExpectationViolation] async

Evaluate all expectations for a phase.

GoalLineageAuditor

Walks the transform chain on a single goal.

Parameters:

Name	Type	Description	Default
depth_warn_threshold	int	Lineage depth at which the verdict turns from ALIGNED to WARN.	DEFAULT_DEPTH_WARN_THRESHOLD
depth_drift_threshold	int	Depth at which the verdict turns to DRIFTED. Must be greater than depth_warn_threshold.	DEFAULT_DEPTH_DRIFT_THRESHOLD
widening_transforms	tuple[str, ...]	Transform names that count as movement away from the original goal (default: generalize and abstract). specialize and concretize are considered re-tightenings and do not count.	WIDENING_TRANSFORMS

audit(goal: Goal, goals_by_id: Mapping[str, Goal]) -> LineageReport

Walk the lineage and produce a typed report.

goals_by_id is the host's lookup over the goal graph (typically GoalGraph.iter_goals() indexed by id). The auditor never holds a reference to the graph itself.

LineageReport dataclass

What the auditor says about one goal's chain back to its origin.

goal_id: str instance-attribute

origin_goal_id: str instance-attribute

depth: int instance-attribute

widening_steps: int instance-attribute

transform_chain: tuple[str, ...] instance-attribute

predicate_changed: bool instance-attribute

verdict: LineageVerdict instance-attribute

LineageVerdict

Bases: StrEnum

Three-band drift classification.

ALIGNED = 'aligned' class-attribute instance-attribute

WARN = 'warn' class-attribute instance-attribute

DRIFTED = 'drifted' class-attribute instance-attribute

TRAPDimension

Bases: StrEnum

Four-axis failure taxonomy.

TRANSPARENCY = 'transparency' class-attribute instance-attribute

REASONING = 'reasoning' class-attribute instance-attribute

ADAPTATION = 'adaptation' class-attribute instance-attribute

PERCEPTION = 'perception' class-attribute instance-attribute

UNCLASSIFIED = 'unclassified' class-attribute instance-attribute

TRAPRouter

Map a classified TRAPDimension to a recommended control action.

The default routing is opinionated; hosts that disagree pass a custom routing mapping at construction.

recommend(dimension: TRAPDimension) -> ControlActionType | None

Return the recommended action, or None if unclassified.

compute_a_distance(current: dict[str, float], baseline_mean: dict[str, float]) -> float

Compute approximate A-distance between current and baseline means.

cost_weighted_argmin(probabilities: Mapping[T, float], cost_matrix: Mapping[T, Mapping[T, float]]) -> tuple[T, float]

Pick the predicted class minimizing expected cost.

cost_matrix[i][j] reads "cost incurred if we predict i when truth is j." Implementations may set cost_matrix[i][i] to zero (or to a small negative value to encode a benefit for correct classification).

expected_cost(probabilities: Mapping[T, float], cost_row: Mapping[T, float]) -> float

Sum of prob[j] * cost[j] over the class space.

classify_failure_mode(failure_mode: FailureMode, *, override: TRAPDimension | None = None) -> TRAPDimension

Classify a FailureMode along the four axes.

Resolution order:

1. override argument, if supplied.
2. The substring "trap:<axis>" inside failure_mode.description (case-insensitive). This is the primary host-supplied signal.
3. Keyword match against failure_mode.name then failure_mode.description.
4. UNCLASSIFIED.

The function is deterministic and dependency-free.