Csf Cognitive¶

CSF Cognitive — mvp.csf_cognitive

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

Overview¶

Cognitive safety block providing five stateless safety operations — evaluate, classify_risk, analyse_scenario, check_privacy, and status — that delegate to dedicated evaluator, governance, and scenario modules. The evaluate op runs a tiered harness that checks token budgets, blocked patterns, and context before optionally delegating to an LLM for nuanced judgement. No SQLite store is used; each call is independent and side-effect-free.

Pilot safety aid, not compliance-grade assurance

csf_cognitive is suitable for MVP launch, local MCP use, and design-partner pilots where practical pre-action safety review, privacy checks, configurable guardrails, and audit history are useful. Scenario analysis is strongest when the configured safeguard model is available. If gpt-oss-safeguard or another configured safeguard backend is unavailable, scenario analysis degrades to the local classifier and keyword/rule fallbacks. Treat that degraded mode as a useful triage signal, not a compliance-grade safety review. Regulated, high-stakes, or unattended production workflows still require human review, domain-specific validation, audit policy, and independent legal/compliance assessment. The component is beta and held at tier1_review_pending; it is not promoted to a verified safety certifier.

Public API¶

`CSFCognitiveBlock(AIBlock[CognitiveInput, CognitiveOutput, dict])`¶

Base cognitive safety block with 5 ops (no persistence).

Field	Type	Default
`name`	`str`	'csf_cognitive'
`state`	`dict \\| None`	None
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)

Methods:

`infer(data: CognitiveInput) -> Result[CognitiveOutput]`¶

`CognitiveInput(BaseModel)`¶

Field	Type	Default
`op`	`Literal['evaluate', 'classify_risk', 'analyse_scenario', 'check_privacy', 'status']`	required
`action`	`str`	''
`content`	`str`	''
`context`	`str`	''
`operation`	`str`	''
`policy`	`str`	''
`mode`	`str`	'autonomous'
`max_tokens`	`int`	0
`estimated_tokens`	`int`	0
`blocked_patterns`	`list[str]`	Field(default_factory=list)

`CognitiveOutput(BaseModel)`¶

Field	Type	Default
`op`	`str`	required
`value`	`dict[str, Any]`	Field(default_factory=dict)
`error`	`str`	''
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`policy_citations`	`dict[str, Any]`	Field(default_factory=lambda: {'retrieval_backend': 'unknown', 'calibrated': False})
`completion_state`	`CompletionState \\| None`	None
`warning_card`	`dict[str, Any] \\| None`	None
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str \\| None`	None
`task_id`	`str \\| None`	None
`run_id`	`str \\| None`	None

`CSFCognitiveMCPBlock(AIBlock[MCPCognitiveInput, MCPCognitiveOutput, dict])`¶

Full-featured cognitive safety block with SQLite persistence.

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

Methods:

`infer(data: MCPCognitiveInput) -> Result[MCPCognitiveOutput]`¶

`MCPCognitiveInput(BaseModel)`¶

Field	Type	Default
`op`	Literal['eval_programmatic', 'eval_token_budget', 'eval_llm_judge', 'eval_policy_harness', 'eval_list_evaluators', 'gov_classify_risk', 'gov_check_privacy', 'gov_check_operation_mode', 'gov_create_context', 'gov_list_constraints', 'scenario_analyse', 'scenario_compare', 'scenario_mitigate', 'scenario_log', 'scenario_list', 'llm_safety_review', 'llm_explain_risk', 'llm_suggest_guardrails', 'llm_assess_alignment', 'llm_ground_check', 'cognitive_status', 'cognitive_evaluate', 'cognitive_history', 'cognitive_clear', 'cognitive_configure', 'list_patterns']	required
`action`	`str`	''
`content`	`str`	''
`context`	`str`	''
`operation`	`str`	''
`policy`	`str`	''
`mode`	`str`	'autonomous'
`max_tokens`	`int`	0
`estimated_tokens`	`int`	0
`blocked_patterns_json`	`str`	'[]'
`action_a`	`str`	''
`action_b`	`str`	''
`risks_json`	`str`	'[]'
`risk_class`	`str`	''
`claim`	`str`	''
`specs_content`	`str`	''
`config_json`	`str`	'{}'
`limit`	`int`	50
`scenario_id`	`str`	''

`MCPCognitiveOutput(BaseModel)`¶

Field	Type	Default
`op`	`str`	required
`value`	`Any`	None
`error`	`str`	''
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated'] \\| None`	None
`warning_card`	`dict[str, Any] \\| None`	None
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str \\| None`	None
`task_id`	`str \\| None`	None
`run_id`	`str \\| None`	None

`CognitiveStore`¶

Sync SQLite cognitive safety store with 3 tables.

Constructor:

Parameter	Type	Default
`db_path`	`str`	':memory:'

Methods:

`add_evaluation(context_dict: dict[str, Any], verdict_dict: dict[str, Any], evaluator: str, level: str) -> str`¶

`query_evaluations(limit: int = 50) -> list[dict[str, Any]]`¶

`clear_evaluations() -> int`¶

`add_scenario(action: str, risks: list[dict[str, Any]], mitigations: list[dict[str, Any]] | None = None) -> str`¶

`query_scenarios(limit: int = 50) -> list[dict[str, Any]]`¶

`clear_scenarios() -> int`¶

`add_governance_check(operation: str, risk_class: str, privacy_class: str, mode: str) -> str`¶

`query_governance_checks(limit: int = 50) -> list[dict[str, Any]]`¶

`clear_governance_checks() -> int`¶

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

`clear_all() -> dict[str, int]`¶

Clear all tables. Returns count of deleted rows per table.

`set_config(config: dict[str, Any]) -> None`¶

`get_config() -> dict[str, Any]`¶

MCP Tools¶

Operation	Source
`eval_programmatic`	cognitive_mcp
`eval_token_budget`	cognitive_mcp
`eval_llm_judge`	cognitive_mcp
`eval_policy_harness`	cognitive_mcp
`eval_list_evaluators`	cognitive_mcp
`gov_classify_risk`	cognitive_mcp
`gov_check_privacy`	cognitive_mcp
`gov_check_operation_mode`	cognitive_mcp
`gov_create_context`	cognitive_mcp
`gov_list_constraints`	cognitive_mcp
`scenario_analyse`	cognitive_mcp
`scenario_compare`	cognitive_mcp
`scenario_mitigate`	cognitive_mcp
`scenario_log`	cognitive_mcp
`scenario_list`	cognitive_mcp
`llm_safety_review`	cognitive_mcp
`llm_explain_risk`	cognitive_mcp
`llm_suggest_guardrails`	cognitive_mcp
`llm_assess_alignment`	cognitive_mcp
`llm_ground_check`	cognitive_mcp
`cognitive_status`	cognitive_mcp
`cognitive_evaluate`	cognitive_mcp
`cognitive_history`	cognitive_mcp
`cognitive_clear`	cognitive_mcp
`cognitive_configure`	cognitive_mcp
`list_patterns`	cognitive_mcp

Csf Cognitive¶

Overview¶

Public API¶

CSFCognitiveBlock(AIBlock[CognitiveInput, CognitiveOutput, dict])¶

infer(data: CognitiveInput) -> Result[CognitiveOutput]¶

CognitiveInput(BaseModel)¶

CognitiveOutput(BaseModel)¶

CSFCognitiveMCPBlock(AIBlock[MCPCognitiveInput, MCPCognitiveOutput, dict])¶

infer(data: MCPCognitiveInput) -> Result[MCPCognitiveOutput]¶

MCPCognitiveInput(BaseModel)¶

MCPCognitiveOutput(BaseModel)¶

CognitiveStore¶

add_evaluation(context_dict: dict[str, Any], verdict_dict: dict[str, Any], evaluator: str, level: str) -> str¶

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

clear_evaluations() -> int¶

add_scenario(action: str, risks: list[dict[str, Any]], mitigations: list[dict[str, Any]] | None = None) -> str¶

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

clear_scenarios() -> int¶

add_governance_check(operation: str, risk_class: str, privacy_class: str, mode: str) -> str¶

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

clear_governance_checks() -> int¶

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

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

set_config(config: dict[str, Any]) -> None¶

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

MCP Tools¶

`CSFCognitiveBlock(AIBlock[CognitiveInput, CognitiveOutput, dict])`¶

`infer(data: CognitiveInput) -> Result[CognitiveOutput]`¶

`CognitiveInput(BaseModel)`¶

`CognitiveOutput(BaseModel)`¶

`CSFCognitiveMCPBlock(AIBlock[MCPCognitiveInput, MCPCognitiveOutput, dict])`¶

`infer(data: MCPCognitiveInput) -> Result[MCPCognitiveOutput]`¶

`MCPCognitiveInput(BaseModel)`¶

`MCPCognitiveOutput(BaseModel)`¶

`CognitiveStore`¶

`add_evaluation(context_dict: dict[str, Any], verdict_dict: dict[str, Any], evaluator: str, level: str) -> str`¶

`query_evaluations(limit: int = 50) -> list[dict[str, Any]]`¶

`clear_evaluations() -> int`¶

`add_scenario(action: str, risks: list[dict[str, Any]], mitigations: list[dict[str, Any]] | None = None) -> str`¶

`query_scenarios(limit: int = 50) -> list[dict[str, Any]]`¶

`clear_scenarios() -> int`¶

`add_governance_check(operation: str, risk_class: str, privacy_class: str, mode: str) -> str`¶

`query_governance_checks(limit: int = 50) -> list[dict[str, Any]]`¶

`clear_governance_checks() -> int`¶

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

`clear_all() -> dict[str, int]`¶

`set_config(config: dict[str, Any]) -> None`¶

`get_config() -> dict[str, Any]`¶