Csf¶

CSF — mvp.csf (v3.0.0)

Cluster: Safety & Alignment | Type: component | MCP Tools: 52

Overview¶

CrossLinkedSafetyFramework AIBlock that verifies a BoundedAgent against a safety epsilon bound and returns a SafetyDecisionReport. The SafetyVerifier inside supports two verification methods: union_bound (sum of per-hazard failure probabilities) and worst_case (maximum single-hazard probability). The block is used by Phase B+ components to gate LLM calls, file writes, and code execution with a formal safety verdict before proceeding.

When to use:

• Gating any high-risk operation with a formal safety check before execution
• Enforcing system-wide epsilon risk bounds on agents with known hazard profiles
• Integrating a lightweight safety layer into a pipeline without external SMT solvers

Example:

from mvp.csf import CSFBlock, SafetyQuery
from mvp.core import BoundedAgent

agent = BoundedAgent(name="robot_arm", hazard_probs={"collision": 0.02, "drop": 0.01})
block = CSFBlock(name="csf", default_epsilon=0.05)
result = block.infer(SafetyQuery(agent=agent, epsilon=0.05, method="union_bound"))
# result.value.approved → True/False; result.value.reason → explanation

Works well with: align_csf, align_specs, formal_methods

Public API¶

CompositionResult¶

Result of compositional safety verification (Theorem 1).

FormalCheckResult¶

Result of a formal verification check via an external backend.

CSFBlock(AIBlock[SafetyQuery, SafetyDecisionReport, None])¶

CrossLinkedSafetyFramework as an AIBlock.

Methods:

infer(data: SafetyQuery) -> Result[SafetyDecisionReport]¶

LTLAtom¶

Atomic proposition — leaf of the LTL formula tree.

LTLNot¶

Negation: ¬φ

LTLAnd¶

Conjunction: φ ∧ ψ

LTLOr¶

Disjunction: φ ∨ ψ

LTLNext¶

Next: X φ

LTLAlways¶

Always / Globally: □ φ (G φ)

LTLEventually¶

Eventually / Finally: ◇ φ (F φ)

LTLUntil¶

Until: φ U ψ

LTLRelease¶

Release: φ R ψ (dual of Until)

RollbackFeasibility¶

Result of Definition 5 rollback feasibility analysis.

GitRollbackManager¶

Git-backed rollback manager for an isolated workspace.

Methods:

snapshot(message: str = 'checkpoint') -> Result[str]¶

Stage all changes and create a commit.

rollback(commit_sha: str) -> Result[str]¶

Hard-reset the repository to commit_sha.

current_sha() -> Result[str]¶

Return the SHA of the current HEAD commit.

close() -> None¶

Release GitPython resources held by this manager.

check_feasibility(N: int, operations: list[str], max_delay: float = 60.0) -> RollbackFeasibility¶

Check rollback feasibility per Definition 5.

classify_operation(op: str) -> RiskClassification¶

Classify an operation's risk level using IRREVERSIBLE_OPS.

GameState¶

A single state in the safety game trace.

GameResult¶

Result of solving the 2-player safety verification game.

LTLMonitor(SafetyMonitor)¶

Runtime safety monitor that tracks LTL formula satisfaction over a live trace.

Constructor:

Methods:

reset() -> None¶

update(state: Any, inp: Any, hazard: float) -> None¶

is_violated() -> bool¶

trace() -> list[dict[str, bool]]¶

SafetyQuery¶

Input to CSFBlock — describes what to verify.

SafetyVerifier¶

Verifies whether a BoundedAgent meets an epsilon risk bound.

Constructor:

Methods:

verify(agent: BoundedAgent, method: str = 'union_bound', epsilon: float | None = None) -> SafetyDecisionReport¶

CSFFormalMCPBlock(AIBlock[MCPCSFFormalInput, MCPCSFFormalOutput, dict])¶

26-op CSF Formal Methods MCP block.

Methods:

infer(data: MCPCSFFormalInput) -> Result[MCPCSFFormalOutput]¶

MCPCSFFormalInput(BaseModel)¶

MCPCSFFormalOutput(BaseModel)¶

CSFFormalStore¶

Sync SQLite store with 6 tables for CSF formal methods.

Constructor:

Methods:

add_ltl(name: str, formula_str: str) -> str¶

get_ltl(lid: str) -> dict[str, Any]¶

list_ltl(limit: int = 100) -> list[dict[str, Any]]¶

delete_ltl(lid: str) -> bool¶

add_composition(result_dict: dict) -> str¶

list_compositions(limit: int = 100) -> list[dict[str, Any]]¶

add_game_result(result_dict: dict) -> str¶

list_game_results(limit: int = 100) -> list[dict[str, Any]]¶

add_formal_check(result_dict: dict) -> str¶

list_formal_checks(backend: str = '', limit: int = 100) -> list[dict[str, Any]]¶

add_rollback_assessment(result_dict: dict) -> str¶

list_rollback_assessments(limit: int = 100) -> list[dict[str, Any]]¶

set_metadata(key: str, value: Any) -> str¶

get_metadata(key: str) -> Any¶

count_all() -> dict[str, int]¶

CSFMCPBlock(AIBlock[MCPCSFInput, MCPCSFOutput, dict])¶

26-op Computational Safety Framework MCP block.

Methods:

infer(data: MCPCSFInput) -> Result[MCPCSFOutput]¶

MCPCSFInput(BaseModel)¶

MCPCSFOutput(BaseModel)¶

CSFStore¶

Sync SQLite CSF store with 3 tables: events, policies, incidents.

Constructor:

Methods:

add_event(event_type: str, data: dict, metadata: dict | None = None) -> str¶

list_events(event_type: str = '', limit: int = 100, since_ts: float = 0.0) -> list[dict[str, Any]]¶

clear_events(event_type: str = '') -> int¶

get_event_summary() -> dict[str, Any]¶

add_policy(name: str, action: str, resource: str, effect: str, priority: int) -> str¶

check_policy(action: str, resource: str) -> dict[str, Any]¶

list_policies() -> list[dict[str, Any]]¶

remove_policy(name: str) -> bool¶

report_incident(title: str, severity: str, description: str) -> str¶

list_incidents(status: str = '', severity: str = '', limit: int = 100) -> list[dict[str, Any]]¶

resolve_incident(incident_id: str) -> bool¶

search_incidents(query: str, limit: int = 20) -> list[dict[str, Any]]¶

count_all() -> dict[str, int]¶

Functions¶

verify_composition(agent1: BoundedAgent, agent2: BoundedAgent, epsilon1: float, epsilon2: float, rho: float = 0.0, target_epsilon: float = 0.2, total_resources: ResourceTriple | None = None) -> CompositionResult¶

Verify compositional safety of two agents under correlated failures.

check_interface_compatibility(agent1: BoundedAgent, agent2: BoundedAgent) -> tuple[bool, list[str]]¶

Check interface compatibility: O₁ ⊆ I₂ via state set intersection.

check_resource_additivity(agent1: BoundedAgent, agent2: BoundedAgent, total: ResourceTriple) -> tuple[bool, list[str]]¶

Check resource additivity: T₁+T₂≤T, M₁+M₂≤M, N₁+N₂≤N.

agent_to_smv(agent: BoundedAgent, ltl_props: list[LTLFormula] | None = None) -> str¶

Convert a BoundedAgent to NuSMV SMV source.

agent_to_smt(agent: BoundedAgent, property: LTLFormula | None = None, bound: int = 10) -> str¶

Encode a BoundedAgent as a QF_LRA bounded model-checking formula.

agent_to_lean(agent: BoundedAgent) -> str¶

Generate a Lean 4 safety theorem for a BoundedAgent.

check_via_z3(agent: BoundedAgent, ltl_property: LTLFormula | None = None, bound: int = 10) -> FormalCheckResult¶

Check safety via Z3 using QF_LRA probabilistic encoding.

check_via_nusmv(agent: BoundedAgent, ltl_property: LTLFormula | None = None) -> FormalCheckResult¶

Check safety via NuSMV — uses PyNuSMVMCPBlock which routes to:

check_via_lean(agent: BoundedAgent, proof_hint: str = '') -> FormalCheckResult¶

Check safety via Lean 4 using LeanProverBlock (7 proof strategies).

verify_all_backends(agent: BoundedAgent, ltl_properties: list[LTLFormula] | None = None, bound: int = 10) -> list[FormalCheckResult]¶

Run all available backends, skip unavailable ones gracefully.

parse_ltl(s: str) -> LTLFormula¶

Parse an LTL formula string into an AST.

evaluate_ltl(formula: LTLFormula, trace: list[dict[str, bool]], pos: int = 0) -> bool¶

Evaluate an LTL formula over a finite trace using standard semantics.

ltl_to_str(formula: LTLFormula) -> str¶

Pretty-print an LTL formula back to string form.

ltl_to_smv(formula: LTLFormula) -> str¶

Convert an LTL formula to NuSMV LTLSPEC syntax.

ltl_to_smt(formula: LTLFormula, bound: int = 10) -> str¶

Convert an LTL formula to bounded SMT-LIB2 encoding.

ltl_to_lean(formula: LTLFormula) -> str¶

Convert an LTL formula to a Lean 4 proposition string.

phi_A_to_ltl(phi_A: Any, state_names: list[str] | None = None) -> LTLFormula¶

Wrap an existing safety predicate phi_A as G(safe).

list_patterns() -> dict[str, Any]¶

Return the csf applied-pattern + skill surface.

compute_trust_boundary(states: list, operations: list[str]) -> frozenset[str]¶

Compute the trust boundary: states where all reversible ops

snapshot_schedule(N: int) -> list[int]¶

Compute snapshot step indices: every ceil(N/10) steps.

solve_game(agent: BoundedAgent, properties: list[LTLFormula], epsilon: float, input_alphabet: list[Any] | None = None, budget: int = 10000, seed: int | None = None) -> GameResult¶

Solve the 2-player safety verification game.

formulate_game(agent: BoundedAgent, properties: list[LTLFormula], epsilon: float, input_alphabet: list[Any] | None = None) -> dict¶

Formulate (but don't solve) the safety game — returns game structure info.

synthesize_monitor(properties: list[LTLFormula]) -> LTLMonitor¶

Build a concrete LTLMonitor for the given safety properties.

find_counterexample(agent: BoundedAgent, property: LTLFormula, input_alphabet: list[Any] | None = None, budget: int = 10000, seed: int | None = None) -> list | None¶

Find an adversarial input sequence that violates the property, or None.

MCP Tools¶