Meta¶
sakshi.meta
¶
Metacognitive control loop.
The meta-loop ships five primitives:
MetaControllerruns the six-phase cycle.MetaSchedulingPolicydecides whether to run the meta-cycle on this iteration.DeliberationGatedecides between routine and deliberative reasoning paths.CanalizationMetricsreports stuck-loop signals as a typed struct.InterventionExecutorvalidates every control action against recent history and a host-supplied permission policy; theInterventionTypetaxonomy labels the pattern of an intervention orthogonally to its mechanism.
HEALTHY_DEPTH_CEILING = 0.5
module-attribute
¶
PATHOLOGICAL_DEPTH_FLOOR = 0.8
module-attribute
¶
DEFAULT_BUDGET_FLOOR = 0.1
module-attribute
¶
DEFAULT_CONFIDENCE_FLOOR = 0.6
module-attribute
¶
DEFAULT_FAILURE_CEILING = 0.4
module-attribute
¶
DEFAULT_COOLDOWN_SECONDS = 30.0
module-attribute
¶
DEFAULT_HISTORY_SIZE = 100
module-attribute
¶
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
¶
CognitiveAssessor
¶
Default Sakshi assessor for monitoring metrics.
assess(monitoring_data: dict[str, Any]) -> dict[str, Any]
async
¶
Assess cognitive performance from monitoring metrics.
MetaController
¶
Package-native metacognitive controller.
run_meta_cycle(cycle_trace: CycleTrace, *, opacity_level: float = 0.0, plan_failure_goal_ids: Iterable[str] = ()) -> MetaCycleResult
async
¶
Execute MONITOR -> INTERPRET -> EVALUATE -> INTEND -> PLAN -> CONTROL.
MetaCycleResult
dataclass
¶
Result of one metacognitive cycle.
actions: list[ControlAction] = field(default_factory=list)
class-attribute
instance-attribute
¶
assessment: dict[str, Any] = field(default_factory=dict)
class-attribute
instance-attribute
¶
monitoring_data: dict[str, Any] = field(default_factory=dict)
class-attribute
instance-attribute
¶
intervention_records: list[InterventionRecord] = field(default_factory=list)
class-attribute
instance-attribute
¶
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
¶
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
¶
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.
ModificationIntegrityGuard
¶
Bases: Protocol
Validate a proposed plan/goal rewrite against existing constraints.
validate_modification(*, before: GoalConstraint, after: GoalConstraint) -> GuardVerdict
¶
RewardIntegrityGuard
¶
Bases: Protocol
Validate a goal-achievement claim against exogenous evidence.
validate_achievement(*, goal_id: str, evidence_keys: Iterable[str]) -> GuardVerdict
¶
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]
¶
DenyByDefaultPolicy
¶
Safe policy that denies every intervention until a host policy replaces it.
is_permitted(action: ControlAction, history: Iterable[InterventionRecord]) -> tuple[bool, str]
¶
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
¶
EveryCyclePolicy
¶
Run the meta-cycle on every iteration.
decide_to_run(*, anomalies_since_last_run: int, canalization: CanalizationMetrics) -> SchedulingDecision
¶
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
¶
OnAnomalyPolicy
¶
Run only when at least one anomaly fired since the previous run.
decide_to_run(*, anomalies_since_last_run: int, canalization: CanalizationMetrics) -> SchedulingDecision
¶
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
¶
metrics_from_static_cycles(*, static_cycles: int, max_history: int, perturbation_count: int = 0, perturbation_capacity: int = 0, temperature_sensitivity: float = 1.0) -> CanalizationMetrics
¶
Convenience factory wrapping the most common derivation.
perturbation_resistance is computed as
perturbation_count / perturbation_capacity when capacity is
positive; otherwise it defaults to 0.0.
build_world_state_from_trace(cycle_trace: CycleTrace) -> WorldStateSnapshot
¶
Build a world-state snapshot from PERCEIVE-style phase output.
collect_goal_relevant_dimensions(active_goals: Iterable[Goal]) -> set[str]
¶
Collect string dimensions named by active goals.
compute_gfe(efe_values: list[float]) -> float
¶
Compute package-local GFE proxy from per-phase EFE values.
biases_against_exploration(balance: KnowledgeRewardBalance) -> bool
¶
Return whether this preference biases away from exploration.
Convenience predicate hosts use when deciding whether to suppress a curiosity-motivated goal because the agent's budget is exhausted and it should be exploiting.
biases_toward_exploration(balance: KnowledgeRewardBalance) -> bool
¶
Return whether this preference biases toward exploration.
make_audit_record(guard_name: str, target: str, verdict: GuardVerdict, metadata: dict[str, Any] | None = None) -> GuardAuditRecord
¶
Build a frozen audit record from a guard verdict.