Skip to content

Align Csf

align_csf -- mvp.align_csf

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

Overview

Computational Safety Framework enforcing G6 hazard bounds via union-bound checks, providing make_g6_csf() factory and csf_guarded() decorator for wrapping hazardous operations with bounded, auditable safety decisions.

MVP safety model, not compliance proof

align_csf is useful as an MVP safety gate: it blocks unknown operations, enforces configured epsilon budgets, records decisions, and makes risky actions visible before execution. Its default hazard probabilities are policy priors, not calibrated evidence for every deployment. Passing a CSF check means "within the configured model and budget"; it does not prove real-world safety, satisfy regulated-industry compliance, or remove the need for human review in high-stakes workflows.

When to use:

  • Enforcing safety bounds on hazardous operations (LLM calls, code execution, file writes)
  • Wrapping functions with the csf_guarded() decorator for automatic safety checks
  • Bounded, auditable safety checks with configurable epsilon budgets
  • Read-only capability discovery via Tier 2 capabilities for registered operations, current epsilon, cognitive-check availability, SQLite health, and budget headroom

Example:

from mvp.align_csf import AlignCSFBlock, AlignCSFInput

block = AlignCSFBlock(name="csf")
result = block.infer(AlignCSFInput(operation="code_execute", n_steps=5))
# result.ok -> True; result.value -> AlignCSFOutput with approved, vub, explanation

Use approved as a pre-execution gate, not as a certification. For production pilots, set conservative epsilon values, log decisions, and validate false-positive and false-negative behaviour on the target workload.

Works well with: formal_methods, goal_engine, cegis, align_evals

Public API

AlignCSFBlock(AIBlock[AlignCSFInput, AlignCSFOutput, None])

G6 safety policy layer.

Field Type Default
name str 'align_csf'
resource_bounds ResourceBounds \| None None
usage ResourceUsage field(default_factory=ResourceUsage)
epsilon_override float \| None None
db_path str \| None None

Methods:

close() -> None

Close the optional SQLite logging connection.

infer(data: AlignCSFInput) -> Result[AlignCSFOutput]

record_outcome(operation: str, unsafe: bool, metadata: dict[str, Any] | None = None) -> str

Record observed production outcome data for empirical hazard calibration.

empirical_hazard(operation: str) -> float | None

Return observed unsafe rate for an operation, or None when no outcomes exist.

AlignCSFInput(BaseModel)

Input to AlignCSFBlock — request a safety check for a G6 operation.

Field Type Default
operation Literal['llm_call', 'file_write', 'code_execute', 'external_api', 'rollback', 'image_generate', 'audio_generate', 'video_generate', 'media_download', 'subprocess_ffmpeg', 'subprocess_blender', 'sensor_read', 'learning_update', 'impedance_adjust', 'context_override', 'motor_execute', 'ros_command', 'autonomy_escalate', 'plc_write', 'embodiment_reconfigure', 'gcode_execute', 'computer_use_click', 'computer_use_type', 'computer_use_key', 'computer_use_bash', 'computer_use_screenshot', 'computer_use_scroll'] required
n_steps int 1
epsilon float 0.2
correlated bool \| None None

AlignCSFOutput(BaseModel)

Output from AlignCSFBlock.

Field Type Default
operation str required
per_step_hazard float required
vub float required
epsilon float required
approved bool required
explanation str required
degraded bool False
degradation_reason str ''
number_provenance list[ProvenancedNumber] Field(default_factory=list)

AlignCSFMCPBlock(AIBlock[MCPCSFInput, MCPCSFOutput, dict])

Field Type Default
name str 'align_csf_mcp'
state dict \| None None
db_path str ':memory:'
resource_bounds ResourceBounds \| None None
usage ResourceUsage field(default_factory=ResourceUsage)

Methods:

close() -> None

Close the SQLite-backed store held by this block.

infer(inp: MCPCSFInput) -> Result[MCPCSFOutput]

MCPCSFInput(BaseModel)

Field Type Default
op str required
operation str \| None None
n_steps int 1
epsilon float 0.2
policy_id str \| None None
policy_name str \| None None
policy str \| None None
policy_content str \| None None
policy_tags list[str] Field(default_factory=list)
decision_id str \| None None
decision str \| None None
decision_approved bool \| None None
decision_context str \| None None
hazard_name str \| None None
hazard_probability float \| None None
hazard_description str \| None None
task_id str \| None None
task_description str \| None None
budget_limit float 1.0
hazard float \| None None
per_step_hazard float \| None None
outcome str \| None None
query str \| None None
limit int 20
actor str \| None None
reason str \| None None
expires_at str \| None None

MCPCSFOutput(BaseModel)

Field Type Default
op str required
success bool required
approved bool False
vub float 0.0
epsilon float 0.2
per_step_hazard float 0.0
explanation str ''
policy_id str \| None None
policy dict[str, Any] \| None None
policies list[dict[str, Any]] Field(default_factory=list)
decision_id str \| None None
decision dict[str, Any] \| None None
decisions list[dict[str, Any]] Field(default_factory=list)
hazard_name str \| None None
hazard_signatures list[dict[str, Any]] Field(default_factory=list)
audit_trail_id str \| None None
audit_trail dict[str, Any] \| None None
budget_remaining float 1.0
budget_used float 0.0
budget_total float 1.0
budget_remaining_pct float 100.0
warning str ''
stats dict[str, Any] Field(default_factory=dict)
count int 0
message str ''
error str ''
value dict[str, Any] Field(default_factory=dict)
degraded bool False
degradation_reason str ''
completion_state str ''
warning_card dict[str, Any] \| str Field(default_factory=dict)
evidence list[str] Field(default_factory=list)
request_id str ''
run_id str ''
forced_approval bool False
actor str \| None None
reason str \| None None
expires_at str \| None None

Functions

make_g6_csf(epsilon: float = G6_EPSILON) -> AlignCSFBlock

Return an AlignCSFBlock configured with the G6 safety policy.

csf_guarded(operation: str, n_steps: int = 1, epsilon: float = G6_EPSILON, correlated: bool | None = None, qualitative: bool = False, risk_class: str = '', scenario_check: bool = False, fail_open_on_cognitive_unavailable: bool = False) -> Callable[[Callable[..., Result[Any]]], Callable[..., Result[Any]]]

Decorator that wraps an AIBlock.infer() (or any Result-returning function)

MCP Tools

Operation Source
check align_csf_mcp
approve align_csf_mcp
reject align_csf_mcp
evaluate align_csf_mcp
report align_csf_mcp
store_policy align_csf_mcp
get_policy align_csf_mcp
list_policies align_csf_mcp
delete_policy align_csf_mcp
record_decision align_csf_mcp
get_decision align_csf_mcp
list_decisions align_csf_mcp
compute_union_bound align_csf_mcp
compute_hazard align_csf_mcp
add_hazard_signature align_csf_mcp
get_hazard_signature align_csf_mcp
list_hazard_signatures align_csf_mcp
create_audit_trail align_csf_mcp
get_audit_trail align_csf_mcp
check_budget align_csf_mcp
reset_budget align_csf_mcp
set_epsilon align_csf_mcp
get_epsilon align_csf_mcp
stats align_csf_mcp
info align_csf_mcp
record_outcome align_csf_mcp
get_empirical_hazard align_csf_mcp
calibration_report align_csf_mcp
list_patterns align_csf_mcp
capabilities align_csf_mcp