Safety Engine¶
mvp.safety_engine -- the agentic safety chain.
Cluster: Uncategorised | Type: component | MCP Tools: None
Overview¶
Public API¶
SafetyEngine¶
Runs a planner under the full safety chain and returns a FinalResult.
Constructor:
| Parameter | Type | Default |
|---|---|---|
policy | RuntimePolicy | required |
guard | Guard \| None | _DEFAULT_GUARD |
validator | Validator \| None | None |
verifier | Callable[[RunState], Any] \| None | None |
tool_runner | Callable[[ActionRequest], Any] \| None | None |
Methods:
run(planner: Planner, user_request: str, run_id: str = 'run') -> FinalResult¶
ExecutionResult¶
| Field | Type | Default |
|---|---|---|
action_id | str | required |
ok | bool | required |
output | Any | None |
error | str | '' |
files_touched | tuple[str, ...] | () |
Executor¶
Runs only permit-bearing actions.
Constructor:
| Parameter | Type | Default |
|---|---|---|
ledger | PermitLedger | required |
tool_runner | Callable[[ActionRequest], ExecutionResult] \| None | None |
Methods:
run(action: ActionRequest, permit: Permit | None) -> ExecutionResult¶
Execute action iff permit is a valid, unconsumed binding for it.
FloorCheck¶
The outcome of checking one observed value against one floor.
| Field | Type | Default |
|---|---|---|
name | str | required |
observed | float | required |
required | float | required |
passed | bool | required |
Methods:
blocked() -> bool¶
allowed_or_reaches_next_gate() -> bool¶
A passing floor does not itself authorize anything — it only clears
Floor¶
A minimum bar fixed from trusted config.
| Field | Type | Default |
|---|---|---|
name | str | required |
required_value | float | required |
Methods:
check(observed: float) -> FloorCheck¶
KillSwitch¶
A one-way, run-scoped halt flag.
Methods:
tripped() -> bool¶
reason() -> str¶
trip(reason: str = '') -> None¶
Trip the switch (idempotent / monotone within a run).
reset(token: object) -> None¶
Clear the switch — trusted runtime only.
PatternGuard¶
A deterministic injection / subversion scanner.
Constructor:
| Parameter | Type | Default |
|---|---|---|
block_markers | tuple[str, ...] | _BLOCK_MARKERS |
Methods:
scan_text(text: str, surface: str = '') -> GuardVerdict¶
scan_user_input(user_request: str) -> GuardVerdict¶
scan_plan(plan: Plan) -> GuardVerdict¶
scan_action(action: ActionRequest) -> GuardVerdict¶
PermitLedger¶
A single-use registry of permits the policy engine has issued.
Methods:
issue(action_id: str, run_id: str) -> Permit¶
Mint and register a permit bound to action_id. Policy-engine only.
is_valid_for(permit: Permit, action_id: str) -> bool¶
True iff permit was issued by us, binds action_id, and is unused.
consume(permit: Permit, action_id: str) -> bool¶
Atomically validate and mark permit consumed. Returns success.
PolicyEngine¶
Authorizes (or refuses) one action at a time against trusted policy.
Constructor:
| Parameter | Type | Default |
|---|---|---|
ledger | PermitLedger | required |
Methods:
authorize_action(action: ActionRequest, state: RunState, policy: RuntimePolicy, guard_verdicts: Sequence[GuardVerdict] = (), validator_votes: Sequence[ValidatorVote] = ()) -> AuthorizationDecision¶
MissingPolicyPermitError(RuntimeError)¶
Raised when the executor is asked to run an action without a valid
RuntimePolicy¶
Trusted, immutable per-run policy (ceilings + quorum config).
| Field | Type | Default |
|---|---|---|
max_steps | int | 25 |
max_tool_calls | int | 50 |
max_shell_commands | int | 10 |
max_files_modified | int | 25 |
max_lines_changed | int | 2000 |
max_retries | int | 3 |
max_runtime_seconds | int | 900 |
max_cost_cents | int | 5000 |
max_privilege_level | int | 1 |
allowed_tools | frozenset[str] | field(default_factory=frozenset) |
forbidden_paths | frozenset[str] | field(default_factory=frozenset) |
require_tests_before_final | bool | True |
require_quorum | bool | False |
quorum_n | int | 1 |
quorum_f | int | 0 |
quorum_required_accepts | int | 1 |
ActionRequest¶
A single tool invocation a plan proposes. Untrusted data.
| Field | Type | Default |
|---|---|---|
tool_name | str | required |
arguments | dict[str, Any] | field(default_factory=dict) |
reason | str | '' |
expected_effect | str | '' |
files_touched | tuple[str, ...] | () |
privilege_level | int | 0 |
is_shell_command | bool | False |
estimated_lines_changed | int | 0 |
id | str | field(default_factory=lambda: str(uuid.uuid4())) |
Plan¶
A planner proposal. Untrusted (SELF_CORRECTION.md line 649).
| Field | Type | Default |
|---|---|---|
goal | str | '' |
steps | tuple[str, ...] | () |
proposed_actions | tuple[ActionRequest, ...] | () |
assumptions | tuple[str, ...] | () |
risk_notes | tuple[str, ...] | () |
planner_model_id | str | '' |
prompt_hash | str | '' |
narration | str | '' |
id | str | field(default_factory=lambda: str(uuid.uuid4())) |
GuardVerdict¶
| Field | Type | Default |
|---|---|---|
decision | GuardDecision | required |
reason | str | '' |
matched_patterns | tuple[str, ...] | () |
risk_score | float | 0.0 |
evidence | tuple[str, ...] | () |
surface | str | '' |
ValidatorVote¶
| Field | Type | Default |
|---|---|---|
validator_id | str | required |
fault_domain | str | required |
decision | VoteDecision | required |
reason | str | '' |
evidence | tuple[str, ...] | () |
Permit¶
| Field | Type | Default |
|---|---|---|
action_id | str | required |
run_id | str | required |
issued_by | str | 'policy_engine' |
permit_id | str | field(default_factory=lambda: str(uuid.uuid4())) |
AuthorizationDecision¶
| Field | Type | Default |
|---|---|---|
decision | AuthDecision | required |
reason | str | '' |
permitted_action | ActionRequest \| None | None |
permit | Permit \| None | None |
policy_checks | tuple[str, ...] | () |
guard_verdicts | tuple[GuardVerdict, ...] | () |
validator_votes | tuple[ValidatorVote, ...] | () |
FinalResult¶
| Field | Type | Default |
|---|---|---|
status | RunStatus | required |
reason | str | '' |
summary | dict[str, Any] | field(default_factory=dict) |
Methods:
success(summary: dict[str, Any] | None = None) -> 'FinalResult'¶
aborted(reason: str, summary: dict[str, Any] | None = None) -> 'FinalResult'¶
escalated(reason: str, summary: dict[str, Any] | None = None) -> 'FinalResult'¶
needs_correction(reason: str, summary: dict[str, Any] | None = None) -> 'FinalResult'¶
RunState¶
| Field | Type | Default |
|---|---|---|
policy | RuntimePolicy | required |
run_id | str | '' |
kill_switch | KillSwitch | field(default_factory=KillSwitch) |
steps | int | 0 |
tool_calls | int | 0 |
shell_commands | int | 0 |
lines_changed | int | 0 |
retries | int | 0 |
cost_cents | int | 0 |
files_modified | set[str] | field(default_factory=set) |
status | RunStatus | 'RUNNING' |
done | bool | False |
Methods:
elapsed_seconds() -> float¶
tool_calls_after() -> int¶
shell_commands_after(action: ActionRequest) -> int¶
files_modified_count_after(action: ActionRequest) -> int¶
lines_changed_after(action: ActionRequest) -> int¶
record_execution(action: ActionRequest) -> None¶
record_retry() -> None¶
StaticFinding¶
| Field | Type | Default |
|---|---|---|
rule | str | required |
line_no | int | required |
line | str | required |
path | str | '' |
ExternalVerification¶
| Field | Type | Default |
|---|---|---|
passed | bool | required |
reason | str | '' |
failed_checks | tuple[str, ...] | () |
TestWeakeningReport¶
| Field | Type | Default |
|---|---|---|
weakened | bool | required |
reasons | tuple[str, ...] | field(default_factory=tuple) |
Methods:
test_files_not_weakened() -> bool¶
assertions_not_removed_without_approval() -> bool¶
Functions¶
exceeds_ceiling(action: ActionRequest, state: RunState, policy: RuntimePolicy) -> tuple[bool, str]¶
Return
(exceeded, reason)for action against the trusted policy.
load_floors(trusted_config: Mapping[str, float]) -> dict[str, Floor]¶
Build floors from a trusted config mapping
name -> required_value.
trusted_reset_token() -> object¶
Return the trusted reset token.
bft_shape_ok(policy: RuntimePolicy) -> bool¶
True iff the policy's quorum parameters satisfy the BFT relations.
count_independent_accepts(votes: Sequence[ValidatorVote]) -> int¶
Count APPROVE votes, de-duplicated by fault domain.
quorum_satisfied(votes: Sequence[ValidatorVote], policy: RuntimePolicy) -> bool¶
Return whether votes clear policy's quorum requirement.
default_runtime_policy(**overrides: Any) -> RuntimePolicy¶
Return the default trusted policy, with optional field overrides.
scan_source(text: str, path: str = '') -> list[StaticFinding]¶
Return all §15 findings in text (one pass, line by line).
scan_paths(paths: list[Path]) -> list[StaticFinding]¶
Scan each .py file in paths and return the aggregated findings.
scan_package() -> list[StaticFinding]¶
Scan the safety_engine package's own source (excluding exempt files).
verify_external(unit_tests_passed: bool, static_analysis_findings: Sequence[str] = ()) -> ExternalVerification¶
Run the deterministic verification gate.
reconcile_consensus(quorum_accepted: bool, external: ExternalVerification) -> tuple[str, str]¶
Combine LLM-validator consensus with deterministic verification.
detect_test_weakening(before_files: Mapping[str, str], after_files: Mapping[str, str]) -> TestWeakeningReport¶
Flag a self-correction patch that deletes or weakens tests.