Skip to content

Csf Gate

CSF gate component; csf_gate_mcp exposes 20 tools.

Cluster: Uncategorised | Type: component | MCP Tools: 22

Overview

Single callable entry point for the full CSF safety stack. An agent calls safety_gate() before performing any action. It always returns a verdict, never raises, and fails closed if the audit trail or verifier model is unavailable. Combines risk analysis, CsfAuditLog persistence, trust-envelope checks, and file-based HITL task creation in one call.

When to use:

  • Enforcing the CSF safety check at every agent action boundary
  • Replacing ad-hoc if safe: guards with a formally verified gate
  • Getting a structured SgVerdict with allow, block, or human-review status plus audit evidence

Example:

from mvp.csf_audit.audit_log import CsfAuditLog
from mvp.csf_gate import SafetyGateContext, safety_gate

ctx = SafetyGateContext(
    audit_log=CsfAuditLog(),
    epsilon=0.20,
    hitl_workspace="./workspaces",
)
verdict = safety_gate("file_write", ctx)
if verdict.verdict == "APPROVED":
    perform_action()
elif verdict.verdict == "HITL_REQUIRED":
    print(f"Human review queued: {verdict.hitl_task_id}")

Launch-phase caveat:

The built-in HITL queue is file-based. It is suitable for local development, clean-machine pilots, and first-user workflows where the operator controls the workspace. It is not an enterprise approval system: it does not authenticate reviewers, bind an approval cryptographically to a later execution, or provide database-backed multi-user workflow state. For regulated, multi-tenant, or high-scale deployments, use safety_gate() as the decision point but back HITL with authenticated reviewers, durable task storage, and an execution layer that requires the matching approval before release.

Enforcement guarantees (not merely logged):

  • BLOCKED and HITL_REQUIRED verdicts fail closed: they surface as completion_state="blocked-escalated" with a stable G6_E_CSF_GATE_BLOCKED warning card at the top level of every surface (block, MCP block, FastMCP server). A refusal is never reduced to an advisory audit note, and no surface upgrades a block verdict. APPROVED is the only autonomously executable verdict.
  • HITL self-approval is blocked. On the exposed MCP/CLI surfaces reviewer-identity separation is default-on (G6_HITL_REQUIRE_REVIEWER; opt out only for a documented local-only pilot). An anonymous approver and a requester-equals-approver approval are both rejected fail-closed and do not resolve the task; resolution requires a distinct, named reviewer. The frozen library default stays byte-identical for pilot mode.
  • Adversarial short-circuit. A prompt-injection / adversarial risk short-circuits to BLOCKED in safety_gate() before the agent model build or SafetyVerifier runs, so an injection input can never reach an APPROVED or non-blocked envelope.

Maturity caveat: beta. The enforcement floors, audit trail, and Codex review are real, but there is no calibrated adversarial/safety gold corpus yet, and the agentic path degrades to the deterministic floor offline. Treat keyword-free high-risk rewordings with extra caution.

Live backend dependencies: csf_gate is an adapter over a live CSF stack, not a standalone block. Required: csf (formal verifier — failure escalates to HITL), csf_audit (tamper-evident audit log — failure BLOCKS), csf_auto_model (agent-model build — failure escalates to HITL). Optional/degrading: csf_cognitive (risk analysis — absence yields conservative deterministic classification), csf_strategy (strategy-card tightening — absence keeps default epsilon/method, never a relaxation), and the in-component agentic runtime (default-on; offline/unavailable degrades to the deterministic floor and is surfaced as a loud top-level qualified-draft). The file-backed HITL workspace is required only when a verdict is HITL_REQUIRED; an unwritable workspace BLOCKS.

Works well with: csf, csf_audit, csf_auto_model, csf_cognitive, csf_strategy

Public API

CsfGateInput(BaseModel)

Field Type Default
op str required
parameters dict[str, Any] Field(default_factory=dict)
run_mode str 'beta'
reviewer_signature str ''

CsfGateOutput(BaseModel)

Field Type Default
op str ''
result dict[str, Any] Field(default_factory=dict)
message str ''
agentic_evidence dict[str, Any] Field(default_factory=dict)
completion_state str 'qualified-draft'
warning_card dict[str, Any] \| None None
evidence list[dict[str, Any]] Field(default_factory=list)
degraded bool False
degradation_reason str ''
audit_log_writable bool False
verifier_available bool False
risk_analysis_available bool False
trust_envelope_available bool False
hitl_workspace_writable bool False
configured bool False

CsfGateBlock(AIBlock)

AIBlock wrapper for the CSF safety gate (single entry point for action safety evaluation).

Methods:

infer(input: CsfGateInput) -> Result[CsfGateOutput]

SafetyGateContext

Field Type Default
audit_log CsfAuditLog required
epsilon float 0.2
llm Callable[[str], str] \| Any \| None None
llm_backend str ''
agent_id str ''
agent_user str ''
tier str \| None None
strategy_id str ''
stakes str 'medium'
grounding_coverage float 1.0
model_quality float 0.9
chain_accuracy float 1.0
decomposition_depth int 1
hitl_workspace str './workspaces'
create_hitl_task bool True
hitl_timeout_sec float 300.0
requester_attested bool False

SgVerdict

Field Type Default
verdict str required
tier str required
audit_id str required
risks list[dict] required
p_unsafe float required
meets_bound bool required
reason str required
trust_verdict TrustVerdict \| None None
hitl_task_id str ''
pending_until str ''
agentic_evidence dict field(default_factory=dict)

HITLDecision

Field Type Default
task_id str required
approved bool required
comment str ''
reviewer str ''
override_reason str ''
timed_out bool False
elapsed_sec float 0.0

HITLGate

File-based HITL approval gate.

Field Type Default
workspace str './workspaces'
timeout_sec float DEFAULT_TIMEOUT_SEC
poll_interval_sec float POLL_INTERVAL_SEC
require_reviewer bool False
allowed_reviewers frozenset[str] \| None None

Methods:

request_approval(action: str, audit_id: str = '', tier: str = 'IRREVERSIBLE', risks: list[dict] | None = None, metadata: dict[str, Any] | None = None, block: bool = True, escalation_ticket: dict | None = None) -> HITLDecision

Create a HITL task and optionally block until resolved.

get_pending() -> list[dict]

Return all pending HITL tasks (no .done.json yet).

submit_decision(task_id: str, approved: bool, comment: str = '', reviewer: str = '', override_reason: str = '', reviewer_user: str = '') -> bool

Submit a decision for a pending HITL task.

get_history(limit: int = 50) -> list[dict]

Return resolved HITL tasks (newest first).

Functions

safety_gate(action: str, context: SafetyGateContext) -> SgVerdict

Evaluate whether an action is safe to perform.

MCP Tools

Operation Source
ops csf_gate_mcp
help csf_gate_mcp
check csf_gate_mcp
status csf_gate_mcp
get_info csf_gate_mcp
classify_tier csf_gate_mcp
explain_adversarial csf_gate_mcp
explain_escalation csf_gate_mcp
list_strategies csf_gate_mcp
readiness_check csf_gate_mcp
hitl_pending csf_gate_mcp
hitl_history csf_gate_mcp
hitl_detail csf_gate_mcp
hitl_binding_status csf_gate_mcp
hitl_approve csf_gate_mcp
hitl_reject csf_gate_mcp
audit_query csf_gate_mcp
audit_report csf_gate_mcp
audit_summarize csf_gate_mcp
audit_filter csf_gate_mcp
audit_export csf_gate_mcp
capabilities csf_gate_mcp