Public API¶
sakshi
¶
Sakshi: a metacognitive runtime for Python agents.
SinceAnomalyPruner
¶
Keep every phase from the last phase that recorded an anomaly forward.
A phase is considered to have recorded an anomaly if its output
dict carries any of the keys in anomaly_keys with a truthy
value. The default looks for anomaly, anomalies, and
anomaly_event to match the conventions used elsewhere in the
package.
TracePruner
¶
Bases: Protocol
Reduce a CycleTrace to the slice the meta-cycle should reason over.
prune(trace: CycleTrace) -> CycleTrace
¶
WhereExpectationFiredPruner
¶
Keep phases whose output flagged at least one expectation.
A phase is considered to have fired an expectation if its
output dict contains a non-empty expectations list, an
expectation_violations list, or a truthy expectation_fired
flag. The host can extend the keys checked by passing
expectation_keys.
AnomalyEscalationError
¶
Bases: SakshiError
An anomaly persisted beyond the configured escalation threshold.
GoalValidationError
¶
Bases: SakshiError
A goal failed validation against the current world state or schema.
PhaseTransitionError
¶
Bases: SakshiError
A cognitive cycle phase transition was attempted in an invalid state.
PlanSoundnessError
¶
Bases: SakshiError
A plan failed soundness verification before execution.
SakshiError
¶
Bases: Exception
Base class for all Sakshi-raised exceptions.
WorldStateUnavailableError
¶
Bases: SakshiError
The configured GoalStateStore could not return a world-state snapshot.
AcceptingRebelHook
¶
Default hook that accepts every assigned goal.
Backward-compatible default. Hosts wanting principled refusal
inject their own RebelHook implementation.
on_goal_assignment(goal: Goal, expectations: Iterable[CognitiveExpectation]) -> RebelDecision
¶
MotivationAuditor
¶
Bounded ring-buffer audit log over MotivationEvent records.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
window_size
|
int
|
Maximum number of recent events to retain. |
DEFAULT_AUDITOR_WINDOW
|
events: tuple[MotivationEvent, ...]
property
¶
Snapshot of recorded events, oldest first.
record(event: MotivationEvent) -> MotivationEvent
¶
Append an event to the audit log and return it unchanged.
filter_by_type(motivation_type: MotivationType) -> list[MotivationEvent]
¶
Return events of the given motivation type, oldest first.
acceptance_rate() -> float
¶
Fraction of events with accepted=True.
metrics() -> ComputationalMotivationMetrics
¶
Return the four-scalar metrics view over the current window.
RebelDecision
dataclass
¶
Typed verdict from a RebelHook.on_goal_assignment call.
verdict: RebelVerdict
instance-attribute
¶
goal: Goal | None = None
class-attribute
instance-attribute
¶
reason: str = ''
class-attribute
instance-attribute
¶
decided_at: datetime = field(default_factory=(lambda: datetime.now(UTC)))
class-attribute
instance-attribute
¶
accept(goal: Goal, reason: str = '') -> RebelDecision
classmethod
¶
rewrite(new_goal: Goal, reason: str) -> RebelDecision
classmethod
¶
reject(reason: str) -> RebelDecision
classmethod
¶
RebelHook
¶
Bases: Protocol
Evaluate an assigned goal against agent expectations.
on_goal_assignment(goal: Goal, expectations: Iterable[CognitiveExpectation]) -> RebelDecision
¶
RebelVerdict
¶
CalibrationReport
dataclass
¶
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
|
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.
ConfusionDecision
dataclass
¶
ConfusionWeighter
¶
Bases: Generic[T]
Apply a cost matrix to a probability vector.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
cost_matrix
|
Mapping[T, Mapping[T, float]]
|
|
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.
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 |
DEFAULT_DEPTH_WARN_THRESHOLD
|
depth_drift_threshold
|
int
|
Depth at which the verdict turns to
|
DEFAULT_DEPTH_DRIFT_THRESHOLD
|
widening_transforms
|
tuple[str, ...]
|
Transform names that count as movement
away from the original goal (default: |
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.
LineageVerdict
¶
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.
AlwaysPermitPolicy
¶
Default policy that permits every intervention.
Test- and trivial-host-friendly. Production deployments should provide a policy that consults their own safety state.
is_permitted(action: ControlAction, history: Iterable[InterventionRecord]) -> tuple[bool, str]
¶
CanalizationMetrics
dataclass
¶
Typed snapshot of canalization signals at a point in time.
Attributes:
| Name | Type | Description |
|---|---|---|
depth |
float
|
Normalized [0.0, 1.0] measure of how entrenched the
current policy is. Conventionally
|
dwell_time |
int
|
Number of cycles spent in the current near-static regime. |
perturbation_resistance |
float
|
How much external perturbation (anomaly, expectation violation, intervention) the regime has absorbed without leaving. Higher means harder to break out of. |
temperature_sensitivity |
float
|
Inverse of rigidity. Low values mean the agent's policy is largely deterministic given inputs; high values mean small input changes already produce meaningful policy variation. |
The risk property derives a coarse three-band label from
depth for routing decisions; callers that need more nuance
should read the four numbers directly.
CanalizationRisk
¶
DeliberationDecision
dataclass
¶
DeliberationGate
¶
Decide between routine and deliberative paths.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
confidence_floor
|
float
|
Below this self-reported confidence the gate recommends escalation. Default 0.6. |
DEFAULT_CONFIDENCE_FLOOR
|
failure_ceiling
|
float
|
Above this rolling failure rate the gate recommends escalation. Default 0.4. |
DEFAULT_FAILURE_CEILING
|
budget_floor
|
float
|
When remaining budget falls below this, the gate stays on the routine path even if confidence/failure signals would normally escalate (you can't afford the deliberative path; do the cheap thing). Default 0.1. |
DEFAULT_BUDGET_FLOOR
|
decide(*, confidence: float, recent_failure_rate: float, remaining_budget: float) -> DeliberationDecision
¶
Recommend a path given three small numbers.
DeliberationPath
¶
DenyByDefaultPolicy
¶
Safe policy that denies every intervention until a host policy replaces it.
is_permitted(action: ControlAction, history: Iterable[InterventionRecord]) -> tuple[bool, str]
¶
EveryCyclePolicy
¶
Run the meta-cycle on every iteration.
decide_to_run(*, anomalies_since_last_run: int, canalization: CanalizationMetrics) -> SchedulingDecision
¶
EvidenceRequiringRewardIntegrityGuard
¶
Default: require at least one evidence key to permit achievement.
The host accumulates an evidence set per goal (sensor readings
that confirm the post-condition, completed tool calls, observed
state-change events) and passes the keys to validate_achievement.
The default guard permits when at least min_evidence keys are
present.
Hosts wanting richer logic (signature verification, redundant confirmations, time-window checks) supply a custom implementation of the protocol.
validate_achievement(*, goal_id: str, evidence_keys: Iterable[str]) -> GuardVerdict
¶
GuardAuditRecord
dataclass
¶
One immutable record of a guard decision.
Hosts that want a unified audit log across both guards can
construct these from any GuardVerdict and stream them through
their own event bus or storage.
GuardDecision
¶
GuardVerdict
dataclass
¶
Typed verdict returned by every guard.
IntegrityCriticalModificationGuard
¶
Default: refuse to drop or relax constraints flagged critical.
Specifically:
- If
before.integrity_criticalis True andafter.integrity_criticalbecomes False, deny — the host is trying to demote a critical constraint. - If
before.integrity_criticalis True and any safety constraint frombefore.safety_constraintsis missing fromafter.safety_constraints, deny.
All other modifications are permitted. Hosts wanting tighter rules (require an audit signature, require a quorum, require operator approval) supply their own protocol implementation.
validate_modification(*, before: GoalConstraint, after: GoalConstraint) -> GuardVerdict
¶
InterventionDecision
¶
InterventionExecutor
¶
Gate every meta-cycle intervention through typed validation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
policy
|
InterventionPermissionPolicy | None
|
Host-supplied permission policy. Defaults to
|
None
|
cooldown_seconds
|
float
|
Minimum elapsed seconds between two
interventions of the same |
DEFAULT_COOLDOWN_SECONDS
|
history_size
|
int
|
Bounded ring buffer of past records. |
DEFAULT_HISTORY_SIZE
|
history: tuple[InterventionRecord, ...]
property
¶
Snapshot of recent intervention records (oldest first).
validate(action: ControlAction, *, pattern: InterventionType | None = None, now: datetime | None = None) -> InterventionRecord
¶
Decide whether to permit action and append a record.
pattern is the optional pattern-level label describing what
the meta-cycle was trying to accomplish (independent of the
mechanism-level action.action_type).
record_outcome(record: InterventionRecord, outcome: InterventionOutcome) -> InterventionRecord
¶
Update an existing record with the observed outcome.
The record is replaced in the history deque. Callers should keep the returned value if they cache the record elsewhere.
InterventionOutcome
¶
InterventionPermissionPolicy
¶
Bases: Protocol
Host-supplied policy: is this intervention permissible right now?
is_permitted(action: ControlAction, history: Iterable[InterventionRecord]) -> tuple[bool, str]
¶
InterventionRecord
dataclass
¶
One entry in the executor's audit trail.
action_type: ControlActionType
instance-attribute
¶
target: str
instance-attribute
¶
decision: InterventionDecision
instance-attribute
¶
reason: str = ''
class-attribute
instance-attribute
¶
outcome: InterventionOutcome = InterventionOutcome.UNKNOWN
class-attribute
instance-attribute
¶
pattern: InterventionType | None = None
class-attribute
instance-attribute
¶
occurred_at: datetime = field(default_factory=(lambda: datetime.now(UTC)))
class-attribute
instance-attribute
¶
InterventionType
¶
Bases: StrEnum
Pattern-level label for what kind of intervention is happening.
Independent of the mechanism-level ControlActionType (which
answers "what action did the meta-cycle emit"). The pattern
label answers "what was the meta-cycle trying to accomplish."
A single ADJUST_PRECISION action can implement
DROP_CONFIDENCE or WIDEN_SEARCH depending on the precision
delta sign; the pattern label makes that intent explicit in audit
records.
PAUSE_AND_REEVALUATE = 'pause_and_reevaluate'
class-attribute
instance-attribute
¶
DROP_CONFIDENCE = 'drop_confidence'
class-attribute
instance-attribute
¶
WIDEN_SEARCH = 'widen_search'
class-attribute
instance-attribute
¶
RELAX_GOAL = 'relax_goal'
class-attribute
instance-attribute
¶
FLUSH_MEMORY = 'flush_memory'
class-attribute
instance-attribute
¶
TRIGGER_EXPLORATION = 'trigger_exploration'
class-attribute
instance-attribute
¶
SUSPEND_RECOVERY = 'suspend_recovery'
class-attribute
instance-attribute
¶
ESCALATE_TO_OPERATOR = 'escalate_to_operator'
class-attribute
instance-attribute
¶
KnowledgeRewardBalance
¶
Bases: StrEnum
Host-declared preference between learning and reward.
KNOWLEDGE— prefer information-gathering goals; tolerate lower immediate reward.REWARD— prefer reward-maximizing goals; tolerate lower information gain.HYBRID— accept either; let the host's selector decide.
MetaSchedulingPolicy
¶
Bases: Protocol
Decide whether to run the meta-cycle on this iteration.
Hosts pass the latest CanalizationMetrics (or a default
CanalizationMetrics()) and the count of anomalies recorded
since the previous meta-cycle. Implementations decide.
decide_to_run(*, anomalies_since_last_run: int, canalization: CanalizationMetrics) -> SchedulingDecision
¶
ModificationIntegrityGuard
¶
Bases: Protocol
Validate a proposed plan/goal rewrite against existing constraints.
validate_modification(*, before: GoalConstraint, after: GoalConstraint) -> GuardVerdict
¶
OnAnomalyPolicy
¶
Run only when at least one anomaly fired since the previous run.
decide_to_run(*, anomalies_since_last_run: int, canalization: CanalizationMetrics) -> SchedulingDecision
¶
RewardIntegrityGuard
¶
Bases: Protocol
Validate a goal-achievement claim against exogenous evidence.
validate_achievement(*, goal_id: str, evidence_keys: Iterable[str]) -> GuardVerdict
¶
SchedulingDecision
dataclass
¶
ThrottledByLoadPolicy
¶
Skip the meta-cycle when canalization load is above the threshold.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
max_run_risk
|
CanalizationRisk
|
Highest |
DEEPENING
|
always_run_on_anomaly
|
bool
|
When True, an anomaly overrides the throttle and forces a run. Defaults to True; an anomaly during high load is exactly when a host needs the meta layer. |
True
|
decide_to_run(*, anomalies_since_last_run: int, canalization: CanalizationMetrics) -> SchedulingDecision
¶
AnomalySourceType
¶
Bases: StrEnum
Origin lane of a detected anomaly.
WORLD anomalies are mismatches between expected and observed
environment state and trigger world-model corrections. COGNITIVE
anomalies are mismatches inside the agent's own reasoning trace
(impasse, expectation violation, plan deviation) and trigger
meta-cycle adjustments. COMPOUND anomalies present in both lanes
and are decomposed at detection, not at explanation.
ComputationalMotivationMetrics
dataclass
¶
Summary of motivation quality over a recent window.
All four fields are bounded scalars in [0, 1] so a host can route on them with simple thresholds.
diversity_score: Shannon entropy of the motivation-type distribution divided bylog2(num_types_present).1.0means every type fired equally; lower means a single drive is dominating.stability_score:1.0 - (acceptance-rate variance over a sliding window). Lower means the host is oscillating between accepting and rejecting motivation impulses.risk_assessment: rolling fraction of events taggedPOWERorSURPRISE(the two motivation classes flagged in the literature as risk-correlated). Higher means more risk-seeking motivation.communication_cost: rolling fraction of events with a non-emptyreasonfield. Approximates how much human-facing narrative the motivation system is producing.
CreativityEnvelope
dataclass
¶
Bounds on the host's motivation creativity.
The envelope is host-declared. Sakshi observes a proposed goal-from-motivation against it; goals that fall outside the envelope are rejected and the rejection is logged.
Attributes:
| Name | Type | Description |
|---|---|---|
allowed_predicates |
tuple[str, ...]
|
Predicate names a motivated goal MAY use. Empty tuple means no constraint. |
forbidden_attributes |
tuple[str, ...]
|
Predicate-argument keys the goal MUST
NOT contain. The classic example is |
max_novelty_score |
float
|
Highest acceptable novelty scalar (host- scored, in [0, 1]). Defaults to 1.0 (no cap). |
forbidden_motivation_types |
tuple[MotivationType, ...]
|
Motivation types forbidden from
generating goals at all. |
allowed_predicates: tuple[str, ...] = ()
class-attribute
instance-attribute
¶
forbidden_attributes: tuple[str, ...] = ()
class-attribute
instance-attribute
¶
max_novelty_score: float = 1.0
class-attribute
instance-attribute
¶
forbidden_motivation_types: tuple[MotivationType, ...] = ()
class-attribute
instance-attribute
¶
description: str = ''
class-attribute
instance-attribute
¶
DiscrepancyResolution
¶
Bases: BaseModel
End-to-end record of one symptom-to-plan resolution chain.
The four payload fields mirror the four canonical steps. They are
typed as str rather than richer DTOs because callers wire in
their own world-model, explanation, goal, and plan representations
through the existing GoalStateStore and explanation seams; this
DTO is the join row that ties them together.
cycle_id: str
instance-attribute
¶
level: ResolutionLevel
instance-attribute
¶
source: AnomalySourceType = Field(default=(AnomalySourceType.WORLD), description='Origin lane of the discrepancy. Object-level resolutions default to WORLD; meta-level resolutions should override to COGNITIVE; bridging cases use COMPOUND.')
class-attribute
instance-attribute
¶
symptom: str = Field(..., description='What was observed that diverged from expectation.')
class-attribute
instance-attribute
¶
explanation: str = Field(..., description='Best causal hypothesis linking symptom to a fixable cause.')
class-attribute
instance-attribute
¶
goal_id: str | None = Field(default=None, description='ID of the goal formulated to resolve the cause, if any.')
class-attribute
instance-attribute
¶
plan_id: str | None = Field(default=None, description='ID of the plan generated to achieve the goal, if any.')
class-attribute
instance-attribute
¶
confidence: float = Field(default=0.5, ge=0.0, le=1.0, description='Confidence the host attaches to the explanation step.')
class-attribute
instance-attribute
¶
metadata: dict[str, Any] = Field(default_factory=dict)
class-attribute
instance-attribute
¶
recorded_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)))
class-attribute
instance-attribute
¶
EnvelopeVerdict
dataclass
¶
ExpectationProfile
¶
Bases: BaseModel
Five-property contract a module registers with Sakshi.
The five fields are the typed-monitoring surface: every module that participates in a cycle declares what it is supposed to do, and Sakshi watches the running module against that declaration.
Attributes:
| Name | Type | Description |
|---|---|---|
module_name |
str
|
Identifier for the module the profile guards. |
runtime_bound_seconds |
float
|
Maximum acceptable wall-clock time for
the module's primary entry point. Violations are warnings
unless the module also declares the failure mode
|
output_schema |
dict[str, Any] | str
|
Description of the shape the module is expected to return. Accepts a JSON-schema-like dict, a Python type name (string), or a fully-qualified class path. Sakshi does not enforce the schema — it stores it so observers can validate against it. |
confidence_range |
tuple[float, float]
|
Inclusive [low, high] bounds on the module's self-reported confidence. Used by the calibration tracker to detect drift outside the declared band. |
side_effects_contract |
tuple[str, ...]
|
Tuple of strings naming the modules, blackboard keys, or external services this module is allowed to mutate. Anything outside the contract is a scope violation. |
failure_modes |
tuple[FailureMode, ...]
|
Tuple of |
module_name: str
instance-attribute
¶
runtime_bound_seconds: float = Field(default=2.0, ge=0.0, description='Maximum acceptable wall-clock time for the module call.')
class-attribute
instance-attribute
¶
output_schema: dict[str, Any] | str = Field(default_factory=dict, description='JSON-schema-like dict, type name, or class path.')
class-attribute
instance-attribute
¶
confidence_range: tuple[float, float] = Field(default=(0.0, 1.0), description='Inclusive [low, high] bounds on self-reported confidence.')
class-attribute
instance-attribute
¶
side_effects_contract: tuple[str, ...] = Field(default_factory=tuple, description='Module / blackboard / service names this module may mutate.')
class-attribute
instance-attribute
¶
failure_modes: tuple[FailureMode, ...] = Field(default_factory=tuple, description='Admissible failure shapes the module declares up front.')
class-attribute
instance-attribute
¶
confidence_in_band(confidence: float) -> bool
¶
Return whether confidence falls inside the declared band.
is_declared_failure(name: str) -> bool
¶
Return whether name matches a declared failure mode.
has_side_effect(target: str) -> bool
¶
Return whether target is inside the declared side-effects contract.
FailureMode
¶
GoalEvent
¶
Bases: BaseModel
One transition record in a goal's lifecycle history.
event_type: GoalEventType
instance-attribute
¶
from_mode: GoalMode | None = None
class-attribute
instance-attribute
¶
to_mode: GoalMode | None = None
class-attribute
instance-attribute
¶
cause: str = Field(default='', description='Short human-readable reason for the transition')
class-attribute
instance-attribute
¶
cycle_id: str | None = None
class-attribute
instance-attribute
¶
occurred_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)))
class-attribute
instance-attribute
¶
GoalEventType
¶
Bases: StrEnum
Causes of a transition recorded on a goal's event history.
FORMULATED = 'formulated'
class-attribute
instance-attribute
¶
SELECTED = 'selected'
class-attribute
instance-attribute
¶
DISPATCHED = 'dispatched'
class-attribute
instance-attribute
¶
MONITOR_OK = 'monitor_ok'
class-attribute
instance-attribute
¶
MONITOR_VIOLATION = 'monitor_violation'
class-attribute
instance-attribute
¶
REPAIR_STARTED = 'repair_started'
class-attribute
instance-attribute
¶
REPAIR_COMPLETED = 'repair_completed'
class-attribute
instance-attribute
¶
DEFERRED = 'deferred'
class-attribute
instance-attribute
¶
RESUMED = 'resumed'
class-attribute
instance-attribute
¶
COMPLETED = 'completed'
class-attribute
instance-attribute
¶
GoalMode
¶
Bases: StrEnum
Active-phase axis for a goal in flight.
Mirrors the goal-lifecycle-network state machine: a goal is
formulated, then selected and dispatched into execution, then
monitored, then either repaired or deferred when monitoring finds
trouble. COMPLETED is a terminal mode independent of whether
GoalStatus recorded the close as ACHIEVED or ABANDONED.
FORMULATING = 'formulating'
class-attribute
instance-attribute
¶
SELECTED = 'selected'
class-attribute
instance-attribute
¶
DISPATCHED = 'dispatched'
class-attribute
instance-attribute
¶
MONITORING = 'monitoring'
class-attribute
instance-attribute
¶
REPAIRING = 'repairing'
class-attribute
instance-attribute
¶
DEFERRED = 'deferred'
class-attribute
instance-attribute
¶
COMPLETED = 'completed'
class-attribute
instance-attribute
¶
GoalOperation
¶
Bases: StrEnum
A typed verb describing what just happened to a goal.
The first nine entries form the canonical goal-reasoning lifecycle
(formulate, select, expand, commit, dispatch, monitor, evaluate,
repair, defer). DELEGATE and RESUME cover handoff and
deferred-goal restart, which the canonical lifecycle does not name
explicitly.
FORMULATE = 'formulate'
class-attribute
instance-attribute
¶
SELECT = 'select'
class-attribute
instance-attribute
¶
EXPAND = 'expand'
class-attribute
instance-attribute
¶
COMMIT = 'commit'
class-attribute
instance-attribute
¶
DISPATCH = 'dispatch'
class-attribute
instance-attribute
¶
MONITOR = 'monitor'
class-attribute
instance-attribute
¶
EVALUATE = 'evaluate'
class-attribute
instance-attribute
¶
REPAIR = 'repair'
class-attribute
instance-attribute
¶
DEFER = 'defer'
class-attribute
instance-attribute
¶
DELEGATE = 'delegate'
class-attribute
instance-attribute
¶
RESUME = 'resume'
class-attribute
instance-attribute
¶
GoalOperationEvent
¶
Bases: BaseModel
An emitted record of a single goal operation.
Hosts subscribe to goal.op events on the configured EventBus
to receive these. The package itself never publishes directly; it
constructs the DTO and hands it to whichever bus the host injected.
operation: GoalOperation
instance-attribute
¶
goal_id: str
instance-attribute
¶
cycle_id: str | None = None
class-attribute
instance-attribute
¶
cause: str = Field(default='', description='Short human-readable reason for the operation')
class-attribute
instance-attribute
¶
metadata: dict[str, Any] = Field(default_factory=dict)
class-attribute
instance-attribute
¶
occurred_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)))
class-attribute
instance-attribute
¶
GoalRelevanceFilter
dataclass
¶
Filter a motivated goal against a host-defined value taxonomy.
Hosts declare a tuple of allowed_value_tags. A goal proposed
by the motivation system carries one or more value tags (in
Goal.metadata['value_tags'] by convention); the filter accepts
only goals whose tags overlap with the allowed list.
Attributes:
| Name | Type | Description |
|---|---|---|
allowed_value_tags |
tuple[str, ...]
|
Tags the host considers organizationally valuable. Empty tuple short-circuits to "accept any". |
require_value_tag |
bool
|
When |
MotivationEvent
¶
Bases: BaseModel
One immutable record describing a motivation activation.
Hosts emit one of these every time their motivation system produces (or refuses to produce) a goal. The auditor stores them.
motivation_type: MotivationType
instance-attribute
¶
goal_id: str | None = Field(default=None, description='ID of the goal produced by this motivation. ``None`` when the activation was rejected by the envelope.')
class-attribute
instance-attribute
¶
cycle_id: str | None = None
class-attribute
instance-attribute
¶
accepted: bool = Field(default=True, description='True if the resulting goal was accepted into the goal graph; False if the envelope or relevance filter rejected it.')
class-attribute
instance-attribute
¶
reason: str = ''
class-attribute
instance-attribute
¶
occurred_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)))
class-attribute
instance-attribute
¶
MotivationType
¶
Bases: StrEnum
Coarse taxonomy of motivation sources.
Hosts tag every intrinsic goal with the type that produced it so audit logs and the metrics view show motivation diversity without the host having to reverse-engineer it from goal predicates.
ACHIEVEMENT = 'achievement'
class-attribute
instance-attribute
¶
AFFILIATION = 'affiliation'
class-attribute
instance-attribute
¶
POWER = 'power'
class-attribute
instance-attribute
¶
NOVELTY = 'novelty'
class-attribute
instance-attribute
¶
COMPETENCE = 'competence'
class-attribute
instance-attribute
¶
SURPRISE = 'surprise'
class-attribute
instance-attribute
¶
EXTRINSIC = 'extrinsic'
class-attribute
instance-attribute
¶
UNCLASSIFIED = 'unclassified'
class-attribute
instance-attribute
¶
ProjectionTransparency
¶
Bases: BaseModel
Tier-3 disclosure: forecasts and risk estimates.
cycle_id: str
instance-attribute
¶
horizon_steps: int = Field(default=1, ge=1)
class-attribute
instance-attribute
¶
forecast_states: list[dict[str, Any]] = Field(default_factory=list)
class-attribute
instance-attribute
¶
resource_forecast: dict[str, float] = Field(default_factory=dict)
class-attribute
instance-attribute
¶
risk_estimates: dict[str, float] = Field(default_factory=dict)
class-attribute
instance-attribute
¶
captured_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)))
class-attribute
instance-attribute
¶
ReasoningTransparency
¶
Bases: BaseModel
Tier-2 disclosure: the most recent decision's reasoning chain.
cycle_id: str
instance-attribute
¶
selected_motivator: str = Field(default='', description='What drove the decision (anomaly, intrinsic goal, instruction)')
class-attribute
instance-attribute
¶
candidate_explanations: list[str] = Field(default_factory=list)
class-attribute
instance-attribute
¶
chosen_explanation: str = ''
class-attribute
instance-attribute
¶
choice_rationale: str = ''
class-attribute
instance-attribute
¶
confidence: float = Field(default=0.0, ge=0.0, le=1.0)
class-attribute
instance-attribute
¶
captured_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)))
class-attribute
instance-attribute
¶
ResolutionLevel
¶
Bases: StrEnum
Whether a discrepancy was resolved at the object or meta level.
Object-level discrepancies are mismatches between expected and observed environment state. Meta-level discrepancies are mismatches inside the agent's own reasoning trace (impasse, expectation violation on a phase output, plan deviation).
StatusTransparency
¶
Bases: BaseModel
Tier-1 disclosure: current state and active goals.
cycle_id: str
instance-attribute
¶
current_phase: str = ''
class-attribute
instance-attribute
¶
active_goal_ids: list[str] = Field(default_factory=list)
class-attribute
instance-attribute
¶
pending_plan_ids: list[str] = Field(default_factory=list)
class-attribute
instance-attribute
¶
summary: str = ''
class-attribute
instance-attribute
¶
captured_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)))
class-attribute
instance-attribute
¶
TransparencyLevel
¶
Bases: StrEnum
The three published transparency tiers.
STATUS is the lightest disclosure — current state and active
goals only. REASONING adds the chain that produced the most
recent decision: which motivator fired, which candidate
explanations were considered, why the chosen one won. PROJECTION
adds forward-looking forecasts: expected next states, resource
forecasts, risk estimates.
TrustBifurcation
¶
Bases: BaseModel
Two-axis self-trust scalar attached to an event or report.
competence_confidence reports how reliable the producing
capability is, given recent calibration; integrity_confidence
reports how trustworthy the signal pipeline is. A single integrity
breach can collapse trust faster than competence drift; the
separation gives operators a typed handle.
competence_confidence: float = Field(default=0.5, ge=0.0, le=1.0, description='Reliability of the capability that produced this output.')
class-attribute
instance-attribute
¶
integrity_confidence: float = Field(default=1.0, ge=0.0, le=1.0, description='Trust that the signal itself was not tampered with, censored, or otherwise compromised in transit.')
class-attribute
instance-attribute
¶
aggregate: float
property
¶
Single scalar combining both axes via the geometric mean.
The geometric mean penalizes a weakness on either axis: a competent capability with a compromised pipeline scores low, and vice versa.
TrustRepairAction
¶
Bases: StrEnum
Typed repair move a host may take after a degraded trust report.
GATHER_EVIDENCE = 'gather_evidence'
class-attribute
instance-attribute
¶
WIDEN_HYPOTHESES = 'widen_hypotheses'
class-attribute
instance-attribute
¶
RECALIBRATE_MODULE = 'recalibrate_module'
class-attribute
instance-attribute
¶
VERIFY_INTEGRITY = 'verify_integrity'
class-attribute
instance-attribute
¶
ESCALATE_TO_OPERATOR = 'escalate_to_operator'
class-attribute
instance-attribute
¶
REFRAME_MODEL = 'reframe_model'
class-attribute
instance-attribute
¶
TrustRepairRecommendation
¶
Bases: BaseModel
Typed recommendation for restoring or routing trust.
Sakshi records the recommended repair but does not execute it. Hosts decide whether the action means collecting sensor evidence, adjusting module profiles, escalating to an operator, or trying a different model family.
action: TrustRepairAction
instance-attribute
¶
reason: str = Field(min_length=1, description='Why this repair action is recommended.')
class-attribute
instance-attribute
¶
target: str = Field(default='', description='Optional host label for the affected module, goal, plan, or signal.')
class-attribute
instance-attribute
¶
severity: float = Field(default=0.5, ge=0.0, le=1.0, description='How urgently the host should consider this repair.')
class-attribute
instance-attribute
¶
uncertainty_boundary: UncertaintyBoundary | None = Field(default=None, description='Optional boundary classification that motivated the repair.')
class-attribute
instance-attribute
¶
evidence_needed: tuple[str, ...] = Field(default_factory=tuple, description='Host-specific evidence keys that would support the repair.')
class-attribute
instance-attribute
¶
TrustReport
¶
Bases: BaseModel
Composite host-facing trust DTO.
Bundles the bifurcation, uncertainty type, list of competing
hypothesis labels (when uncertainty_type == AMBIGUITY), a
calibration status string, a coarse trajectory band, and a
machine-readable recommendation. Hosts surface this to humans or
to downstream automation.
cycle_id: str
instance-attribute
¶
subject: str = Field(default='', description="What the report is about: 'goal:gid', 'anomaly:event-7', 'plan:p-3', etc. Free-form for host convenience.")
class-attribute
instance-attribute
¶
trust: TrustBifurcation = Field(default_factory=TrustBifurcation)
class-attribute
instance-attribute
¶
uncertainty_type: UncertaintyType = UncertaintyType.PROBABILITY
class-attribute
instance-attribute
¶
uncertainty_boundary: UncertaintyBoundary = UncertaintyBoundary.STOCHASTIC
class-attribute
instance-attribute
¶
competing_hypothesis_labels: tuple[str, ...] = Field(default_factory=tuple)
class-attribute
instance-attribute
¶
calibration_status: str = Field(default='unknown', description="Free-form label: 'well_calibrated', 'overconfident', 'underconfident', 'unknown'.")
class-attribute
instance-attribute
¶
trajectory: str = Field(default='stable', description="Coarse band: 'improving', 'stable', 'deteriorating', 'at_risk_of_collapse'.")
class-attribute
instance-attribute
¶
recommendation: str = Field(default='', description="Free-form, host-facing: 'safe_to_automate', 'verify_before_action', 'escalate'.")
class-attribute
instance-attribute
¶
repair_recommendations: tuple[TrustRepairRecommendation, ...] = Field(default_factory=tuple, description='Typed trust-repair recommendations hosts may route or surface.')
class-attribute
instance-attribute
¶
notes: str = ''
class-attribute
instance-attribute
¶
created_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)))
class-attribute
instance-attribute
¶
UncertaintyBoundary
¶
Bases: StrEnum
Where the uncertainty appears to live.
UncertaintyType answers what kind of confidence statement the
report is making. UncertaintyBoundary answers what response the
host should consider: gather samples, widen hypotheses, admit there
is no model, or reframe the model itself.
STOCHASTIC = 'stochastic'
class-attribute
instance-attribute
¶
AMBIGUOUS = 'ambiguous'
class-attribute
instance-attribute
¶
IGNORANT = 'ignorant'
class-attribute
instance-attribute
¶
EPISTEMIC = 'epistemic'
class-attribute
instance-attribute
¶
ONTOLOGICAL = 'ontological'
class-attribute
instance-attribute
¶
UncertaintyType
¶
Bases: StrEnum
What kind of uncertainty a confidence value reflects.
PROBABILITY— classical: the output is one of N hypotheses, and confidence reports the modeled probability.AMBIGUITY— multiple competing hypotheses are roughly equiprobable; the host should hedge.IGNORANCE— no model; confidence is a placeholder.
Action
¶
Bases: BaseModel
A single schedulable step produced by a task decomposer.
The shape is intentionally narrow — name, arguments, optional metadata — because hosts already maintain richer action records elsewhere (their own executor, observability stack, etc.). This DTO is the package-internal join row.
AnticipatoryRiskScorer
¶
Three-step risk pipeline for the EVAL phase.
Step 1: risk_model.identify(...) enumerates per-step risks.
Step 2: aggregate_risk_score reduces them plus expected_benefit
into one scalar.
Step 3: classify_band maps the scalar to a coarse label.
Every call returns a frozen PlanRiskAssessment.
assess(*, plan_id: str, plan_steps: Sequence[str], world_state: dict[str, Any] | None = None, expected_benefit: float = 0.5, notes: str = '') -> PlanRiskAssessment
¶
GoalConstraint
dataclass
¶
Bounds on a goal's feasibility.
Attributes:
| Name | Type | Description |
|---|---|---|
initial_state |
tuple[str, ...]
|
Predicates that must hold in the initial state for the goal to be considered well-posed. |
safety_constraints |
tuple[str, ...]
|
Predicates that must remain true throughout execution. Violation should trigger a soundness-check failure or a meta-cycle intervention. |
goal_conditions |
tuple[str, ...]
|
Predicates that, when all satisfied, mean the goal is achieved. |
integrity_critical |
bool
|
When |
initial_state: tuple[str, ...] = ()
class-attribute
instance-attribute
¶
safety_constraints: tuple[str, ...] = ()
class-attribute
instance-attribute
¶
goal_conditions: tuple[str, ...] = ()
class-attribute
instance-attribute
¶
integrity_critical: bool = False
class-attribute
instance-attribute
¶
description: str = ''
class-attribute
instance-attribute
¶
metadata: dict[str, str] = field(default_factory=dict)
class-attribute
instance-attribute
¶
PlanRisk
dataclass
¶
One identified risk on one plan step.
PlanRiskAssessment
dataclass
¶
Aggregated assessment over one plan's identified risks.
plan_id: str
instance-attribute
¶
risks: tuple[PlanRisk, ...] = ()
class-attribute
instance-attribute
¶
expected_benefit: float = 0.5
class-attribute
instance-attribute
¶
risk_score: float = field(default=0.0)
class-attribute
instance-attribute
¶
band: RiskBand = field(default=(RiskBand.LOW))
class-attribute
instance-attribute
¶
notes: str = ''
class-attribute
instance-attribute
¶
RiskBand
¶
RiskModel
¶
Bases: Protocol
Host-supplied identifier for plan-step risks.
Implementations inspect the plan + world state and return a tuple
of PlanRisk records. The package never inspects host state
directly.
identify(plan_id: str, plan_steps: Sequence[str], world_state: dict[str, Any] | None = None) -> tuple[PlanRisk, ...]
¶
TaskDecomposer
¶
Bases: Protocol
Expand a high-level task into a flat list of executable actions.
Implementations may consult world_state to pick a decomposition
appropriate for the current state, or ignore it entirely for
domain-independent expansions. The protocol is sync because most
real planners are CPU-bound; hosts that want async planners can
define their own protocol and adapter.
decompose(task: str, world_state: dict[str, Any] | None = None) -> list[Action]
¶
AlwaysPermitWriteGuard
¶
Default WriteGuard that permits every write.
Use this only in tests or in hosts where Sakshi-originated writes do not require external safety review.
check(source_origin: str, payload: Mapping[str, Any]) -> bool
async
¶
BasinHook
¶
Bases: Protocol
Optional hook for hosts that maintain an attractor-basin field.
Sakshi calls these on goal lifecycle transitions. Hosts that do not
use a basin field supply a no-op implementation; see
NoOpBasinHook for the package default.
Clock
¶
Bases: Protocol
Time source.
Allows tests and deterministic replays to substitute a fake clock.
now() -> datetime
¶
DenyByDefaultWriteGuard
¶
Safe WriteGuard that denies every write until a host policy replaces it.
This is the right placeholder for production scaffolds: it preserves the
protocol shape while failing closed instead of silently permitting writes.
Tests and quickstarts can continue using AlwaysPermitWriteGuard when they
deliberately do not exercise host write policy.
check(source_origin: str, payload: Mapping[str, Any]) -> bool
async
¶
EventBus
¶
Bases: Protocol
Async pub/sub seam for emitting cycle and goal events.
A host wraps its own bus implementation in an adapter that satisfies this protocol.
emit(event_type: str, payload: Mapping[str, Any]) -> None
async
¶
GoalStateStore
¶
Bases: Protocol
Goal-related world-state persistence and retrieval seam.
Hosts that persist world state in a graph database, vector store, or any other backend wrap that backend in an adapter implementing this protocol. Sakshi never imports a specific persistence layer.
NoOpBasinHook
¶
NoOpEventBus
¶
Default EventBus that drops emitted events.
emit(event_type: str, payload: Mapping[str, Any]) -> None
async
¶
WriteGuard
¶
Bases: Protocol
Pre-write safety check seam.
Hosts can route Sakshi-originated writes through their own write- safety policy. Returns True if the write is permitted, False otherwise.
check(source_origin: str, payload: Mapping[str, Any]) -> bool
async
¶
PhaseRegistry
¶
Registry that drives a cognitive cycle through its phases.
Usage:
registry = PhaseRegistry(event_bus=bus)
registry.on_cycle_complete(my_persistence_callback)
await registry.start_cycle(cycle_id="cycle-001")
await registry.record_phase_output("PERCEIVE", perception_result)
await registry.record_phase_output("INTERPRET", interpret_result)
await registry.record_phase_output("EVAL", eval_result)
await registry.record_phase_output(
"INTEND", intend_result, np_snapshot=np_snapshot
)
await registry.record_phase_output("PLAN", plan_result)
await registry.record_phase_output("ACT", act_result)
trace = await registry.finalize_cycle()Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event_bus
|
EventBus
|
An object satisfying |
required |
phase_config
|
dict[str, PhaseSlot] | None
|
Optional override for the default phase mapping. |
None
|
cycle_complete_event_type
|
str
|
String event type the registry emits when a cycle finalizes. Hosts can override to align with their bus's event-name conventions. |
'sakshi.cycle.complete'
|
list_phases() -> list[str]
¶
Return ordered list of registered phase names.
get_phase_config(phase_name: str) -> PhaseConfig
¶
Return PhaseConfig for a phase.
Raises:
| Type | Description |
|---|---|
KeyError
|
If the phase name is not registered. |
on_cycle_complete(callback: CycleCompleteCallback) -> None
¶
Register a callback to fire when finalize_cycle() is called.
start_cycle(cycle_id: str) -> None
async
¶
Begin a new cycle, resetting the current trace.
record_phase_output(phase_name: str, output: dict[str, Any], *, np_snapshot: NPStateSnapshot | None = None) -> None
async
¶
Record the output of a completed phase.
At INTEND, np_snapshot may carry the dominant thought-seed's
NP state and prior type.
finalize_cycle() -> CycleTrace
async
¶
Close the current cycle and fire all registered callbacks.
Raises:
| Type | Description |
|---|---|
PhaseTransitionError
|
If no cycle is active. |
Returns:
| Type | Description |
|---|---|
CycleTrace
|
The finalized |
get_current_trace() -> CycleTrace | None
¶
Return the in-progress trace, or None if no cycle is active.
get_last_completed_trace() -> CycleTrace | None
¶
Return the most recently finalized trace, or None.
classify_failure_mode(failure_mode: FailureMode, *, override: TRAPDimension | None = None) -> TRAPDimension
¶
Classify a FailureMode along the four axes.
Resolution order:
overrideargument, if supplied.- The substring
"trap:<axis>"insidefailure_mode.description(case-insensitive). This is the primary host-supplied signal. - Keyword match against
failure_mode.namethenfailure_mode.description. UNCLASSIFIED.
The function is deterministic and dependency-free.
evaluate_envelope(*, envelope: CreativityEnvelope, motivation_type: MotivationType, predicate_name: str, predicate_args: dict[str, object] | None = None, novelty_score: float = 0.0) -> EnvelopeVerdict
¶
Decide whether a candidate motivation+goal fits the envelope.
The function is pure: same inputs produce the same verdict.
Hosts pass the verdict through MotivationAuditor and to their
own goal-acceptance logic.