Goals¶
sakshi.goals
¶
Goal-management core.
This package contains pure goal graph, validation, selection, transform, and outcome-memory logic. Host-specific execution summaries, graph persistence, basin hooks, and legacy goal-service synchronization belong in adapters outside this package.
DEFAULT_PARENT_COUPLING = 0.8
module-attribute
¶
MAX_COUPLING_BASINS = 10
module-attribute
¶
DEFAULT_AUDITOR_WINDOW = 500
module-attribute
¶
RISK_TYPES: frozenset[MotivationType] = frozenset({MotivationType.POWER, MotivationType.SURPRISE})
module-attribute
¶
AnomalyExplainer
¶
Produce structured anomaly explanations from package-native inputs.
Runtime histories and basin profiles enter through constructor args or per-call overrides. No event bus, graph database, or basin service is imported by this class.
explain(anomaly_event: object, *, anomaly_history: Iterable[Mapping[str, Any] | object] | None = None, goal_graph: GoalGraph | None = None, basin_profile_provider: BasinProfileProvider | None = None) -> AnomalyExplanation
async
¶
Return a best-effort explanation, never blocking goal generation.
explain_distribution(anomaly_event: object, *, anomaly_history: Iterable[Mapping[str, Any] | object] | None = None, goal_graph: GoalGraph | None = None, basin_profile_provider: BasinProfileProvider | None = None, top_k: int = 3) -> list[AnomalyExplanation]
async
¶
Return up to top_k ranked competing hypotheses.
Where explain collapses to the single most-shifted key,
this method returns several candidate explanations — one per
meaningfully shifted key, sorted by descending confidence.
The shape lets a meta-cycle hedge across competing diagnoses:
a SWAP_MODULE action on the top-1 hypothesis is sound only
when its confidence exceeds the runner-up by a comfortable
margin; otherwise the host should defer or gather more
evidence.
AnomalyExplanation
¶
Bases: BaseModel
Structured causal hypothesis for a detected anomaly.
anomaly_class: str = Field(description='Most-shifted activation key')
class-attribute
instance-attribute
¶
recurrence_count: int = Field(description='Times this anomaly class appeared in supplied history')
class-attribute
instance-attribute
¶
prior_resolution: str | None = Field(default=None, description='Prior goal resolution: achieved|abandoned|delegated|active|unresolved')
class-attribute
instance-attribute
¶
prior_goal_id: str | None = Field(default=None)
class-attribute
instance-attribute
¶
basin_stability: float = 0.0
class-attribute
instance-attribute
¶
basin_strength: float = 0.0
class-attribute
instance-attribute
¶
is_novel: bool = True
class-attribute
instance-attribute
¶
hypothesis: str = ''
class-attribute
instance-attribute
¶
confidence: float = 0.3
class-attribute
instance-attribute
¶
explained_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)))
class-attribute
instance-attribute
¶
BasinProfile
¶
GoalGenerator
¶
Convert anomaly-shaped inputs into actionable goals.
generate_from_anomaly(anomaly_event: AnomalyEvent | object, *, explanation: AnomalyExplanation | None = None, add_to_graph: bool = False) -> Goal | None
async
¶
Generate a goal from a statistical anomaly event.
generate_from_canalization_event(event_data: Mapping[str, Any], *, add_to_graph: bool = False) -> Goal | None
async
¶
Generate a goal from a canalization-detection payload.
mutate_goal(goal_id: str, anomaly_count: int, *, add_to_graph: bool = False) -> Goal | None
async
¶
Generate a widened investigation goal for persistent anomalies.
GoalEdge
dataclass
¶
GoalGraph
¶
Partial ordering of goals with a current-goal stack.
Root goals have no prerequisites. Child goals are blocked until
their parent has status ACHIEVED. No implicit cycle repair is
attempted: callers must add parents before children.
goal_count: int
property
¶
Total number of goals in the graph.
current_goal_stack_ids: list[str]
property
¶
Current goal stack as a list of goal IDs; top is the last item.
add_goal(goal: Goal, parent_id: str | None = None, coupling_strength: float = DEFAULT_PARENT_COUPLING) -> GoalNode
¶
Add a goal to the graph.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
goal
|
Goal
|
Goal to add. |
required |
parent_id
|
str | None
|
Optional parent goal ID. If provided, the parent must already exist. |
None
|
coupling_strength
|
float
|
Strength for the parent-child coupling edge. |
DEFAULT_PARENT_COUPLING
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If a goal with the same ID already exists. |
KeyError
|
If |
get_node(goal_id: str) -> GoalNode
¶
Return the GoalNode for a goal ID.
mark_achieved(goal_id: str) -> None
¶
Mark a goal as achieved and remove it from continuation state.
mark_abandoned(goal_id: str) -> None
¶
Mark a goal as abandoned and remove it from continuation state.
delegate_goal(goal_id: str, delegate_to: str) -> None
¶
Delegate a goal and remove it from the active frontier.
get_active_frontier() -> list[Goal]
¶
Return active goals whose prerequisites are satisfied.
iter_goals() -> list[Goal]
¶
Return every goal currently in the graph (any status).
Order is insertion-stable. Hosts that need to scan terminal goals
for persistence should prefer :meth:get_goals_by_status instead
of pulling this and filtering — the explicit filter form keeps
intent visible in caller code.
get_goals_by_status(statuses: set[GoalStatus]) -> list[Goal]
¶
Return goals whose status is in statuses.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
statuses
|
set[GoalStatus]
|
Set of :class: |
required |
Returns:
| Type | Description |
|---|---|
list[Goal]
|
Matching goals in insertion order. Empty list if no matches |
list[Goal]
|
or |
to_coupling_matrix() -> dict[tuple[str, str], float]
¶
Export active parent-child basin edges as a coupling matrix.
Only active goals with non-empty basin_name values are included,
capped to MAX_COUPLING_BASINS by descending goal priority.
set_plan(goal_id: str, plan: GoalPlan) -> None
¶
Attach a plan to a goal node.
get_plan(goal_id: str) -> GoalPlan | None
¶
Return the plan attached to a goal, if present.
push_current_goal(goal_id: str) -> None
¶
Push a goal ID onto the current-goal stack.
pop_current_goal() -> str | None
¶
Pop the current-goal stack, returning None if it is empty.
peek_current_goal() -> str | None
¶
Return the top current-goal ID without removing it.
as_dict() -> dict[str, Any]
¶
Return a lightweight serializable view useful for debugging.
GoalNode
dataclass
¶
A node in a hierarchical goal graph.
goal: Goal
instance-attribute
¶
parent: GoalNode | None = None
class-attribute
instance-attribute
¶
children: list[GoalNode] = field(default_factory=list)
class-attribute
instance-attribute
¶
plan: GoalPlan | None = None
class-attribute
instance-attribute
¶
delegate_to: str | None = None
class-attribute
instance-attribute
¶
is_blocked: bool
property
¶
True if this node's parent is not achieved yet.
is_active: bool
property
¶
True if this goal can currently be pursued.
GoalMonitor
¶
Check whether a goal is still meaningful to pursue.
check_validity(goal: Goal, world_state: WorldStateSnapshot) -> GoalMonitorResult
¶
Check whether pursuing this goal is still meaningful.
check_outcome_history(goal: Goal, outcome_memory: GoalOutcomeMemory) -> GoalMonitorResult | None
¶
Return historically-futile result when structured outcomes warrant it.
check_validity_with_store(goal: Goal, world_state: WorldStateSnapshot, store: GoalStateStore, *, valid_at: datetime | None = None) -> GoalMonitorResult
async
¶
Check validity and confirm against host world-state store.
The store receives an opaque query. Hosts decide how to interpret
predicate_name and valid_at. If the store raises, this method
returns the in-memory result so host-store outages do not block
metacognitive flow.
GoalMonitorResult
dataclass
¶
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.
GoalOutcomeClosureService
¶
Derive and apply goal closure from execution outcomes.
close_from_summary(summary: GoalExecutionSummary) -> list[GoalOutcomeRecord]
async
¶
Classify a focused goal's execution outcome and record it.
GoalOutcomeMemory
¶
Small in-memory store for structured goal outcome records.
record(outcome: GoalOutcomeRecord) -> GoalOutcomeRecord
¶
Store and return a goal outcome record.
list_records(*, goal_id: str | None = None, instruction_id: str | None = None, predicate_name: str | None = None, outcome_status: GoalStatus | None = None) -> list[GoalOutcomeRecord]
¶
Return records matching all provided filters.
recent(n: int = 10) -> list[GoalOutcomeRecord]
¶
Return the most recent n records (newest first).
Order is by record insertion; callers that need a different sort key can post-process the result.
find_similar(*, predicate_name: str, outcome_status: GoalStatus | None = None, limit: int = 10) -> list[GoalOutcomeRecord]
¶
Return up to limit recent records sharing a predicate name.
Filters by predicate_name and optionally outcome_status,
returns the most recent matches first. Useful for "have we
tried something like this before" checks before issuing an
identical goal.
hit_rate(*, predicate_name: str | None = None) -> float
¶
Return the fraction of recorded records that achieved their goal.
Optionally narrowed to one predicate name. Returns 0.0 when the matching set is empty.
clear() -> None
¶
Remove all stored records.
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
¶
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
¶
GoalSelector
¶
Rank goals by ModSelectionCriteria score.
select(goals: list[Goal], performance_map: dict[str, float] | None = None, limiting_factor_map: dict[str, float] | None = None) -> list[Goal]
¶
Return goals ordered highest selection score first.
ModSelectionCriteria
¶
Compute a bounded selection score for one goal.
score(goal: Goal, performance: float, limiting_factor: float) -> float
¶
Compute performance / limiting_factor, capped to [0.0, 1.0].
GoalTransformer
¶
Apply predicate transforms to goals.
identity(goal: Goal) -> Goal
¶
Return an equivalent active copy of the goal.
generalize(goal: Goal) -> Goal
¶
Broaden predicate scope by dropping the last argument.
specialize(goal: Goal, constraints: dict[str, Any]) -> Goal
¶
Narrow predicate scope by adding constraints.
abstract(goal: Goal) -> Goal
¶
Remove concrete arguments, leaving the symbolic predicate.
concretize(goal: Goal, world_state: WorldStateSnapshot) -> Goal
¶
Bind a symbolic predicate to args from a world-state snapshot.
apply(transform_type: TransformType, goal: Goal, *, constraints: dict[str, Any] | None = None, world_state: WorldStateSnapshot | None = None) -> Goal
¶
Apply a transform by enum value.
TransformType
¶
Bases: StrEnum
Supported goal-transform operations.
IDENTITY = 'identity'
class-attribute
instance-attribute
¶
GENERALIZE = 'generalize'
class-attribute
instance-attribute
¶
SPECIALIZE = 'specialize'
class-attribute
instance-attribute
¶
ABSTRACT = 'abstract'
class-attribute
instance-attribute
¶
CONCRETIZE = 'concretize'
class-attribute
instance-attribute
¶
DomainRegistry
¶
Registry of valid predicates and required predicate arguments.
predicates: dict[str, list[str]] = {}
instance-attribute
¶
add_predicate(name: str, required_args: list[str] | None = None) -> None
¶
Register a valid predicate name and optional required args.
is_known(predicate_name: str) -> bool
¶
True if the predicate name is registered.
required_args_for(predicate_name: str) -> list[str]
¶
Return required args for a predicate, or an empty list.
GoalValidationResult
dataclass
¶
GoalValidator
¶
Validate goals before insertion into a graph.
This validator checks predicate membership and required argument
presence. It deliberately does not inspect runtime world state;
world-state validity belongs to monitoring logic behind a host's
GoalStateStore.
domain: DomainRegistry
property
¶
The domain registry backing this validator.
validate(goal: Goal, world_state: WorldStateSnapshot | None = None) -> GoalValidationResult
¶
Validate a goal against the domain registry.
world_state is accepted for API symmetry with monitors and
future host adapters, but structural validation does not use it.
find_most_shifted_key(current: Mapping[str, float], baseline: Mapping[str, float]) -> str
¶
Return the key with the largest activation shift from baseline.
severity_to_priority(severity: str) -> int
¶
Map anomaly severity label to goal priority.