Plans¶
sakshi.plans
¶
Plan validation, deviation tracking, decomposition, and risk.
HIGH_RISK_FLOOR = 0.66
module-attribute
¶
LOW_RISK_CEILING = 0.33
module-attribute
¶
DEFAULT_DEVIATION_THRESHOLD = 0.3
module-attribute
¶
MAX_DEVIATION_THRESHOLD = 0.5
module-attribute
¶
MIN_DEVIATION_THRESHOLD = 0.1
module-attribute
¶
MIN_STEPS_BEFORE_REPLAN = 3
module-attribute
¶
SOUNDNESS_CONFIDENCE_THRESHOLD = 0.5
module-attribute
¶
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
¶
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, ...]
¶
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
¶
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.
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]
¶
DeviationRecord
dataclass
¶
One actual execution step compared against the reference plan.
PlanDeviationTracker
¶
Track deviations between planned and actual execution steps.
dynamic_threshold: float
property
¶
Current adaptive deviation threshold.
records: list[DeviationRecord]
property
¶
Recorded deviations and exact-match observations.
set_reference_plan(steps: list[str], *, reset_threshold: bool = False) -> None
¶
Set the accepted reference plan.
record_action(tool_name: str, args: Mapping[str, Any] | None = None) -> DeviationRecord | None
¶
Record an executed action and return a record only on mismatch.
on_learning_signal(event: Mapping[str, Any] | object) -> None
async
¶
Adjust threshold from a host-wired learning signal.
get_deviation_ratio() -> float
¶
Return proportion of executed steps that deviated.
should_replan() -> bool
¶
True once deviation ratio exceeds threshold after enough steps.
reset() -> None
¶
Clear plan and observation state, preserving adaptive threshold.
PlanConstraint
dataclass
¶
Structured constraint attached to one plan step.
step_id: str
instance-attribute
¶
preconditions: list[str] = field(default_factory=list)
class-attribute
instance-attribute
¶
resource_requirements: dict[str, float] = field(default_factory=dict)
class-attribute
instance-attribute
¶
must_follow: list[str] = field(default_factory=list)
class-attribute
instance-attribute
¶
PlanSoundnessVerifier
¶
Validate plan steps against world state and resource constraints.
verify(plan_steps: list[dict[str, Any]], world_state: dict[str, Any] | None = None, available_resources: dict[str, float] | None = None) -> SoundnessCheckResult
¶
Validate a plan and return a SoundnessCheckResult.
SoundnessCheckResult
dataclass
¶
Result of a plan soundness check.
aggregate_risk_score(risks: Sequence[PlanRisk], *, expected_benefit: float = 0.5) -> float
¶
Combine per-step risks and benefit into a single score in [0, 1].
Strategy: take the maximum severity (so one big risk dominates)
and reduce by (1 - expected_benefit) ** 2 so a high-benefit
plan can absorb more risk than a low-benefit plan can.
classify_band(risk_score: float) -> RiskBand
¶
Map a numeric risk score to a coarse three-band label.