Self Model¶
self_model — mvp.self_model
Cluster: Uncategorised | Type: component | MCP Tools: 27
Overview¶
Symbolic self-model for the G6 system. Represents each component as a category object, extracts behavioural contracts (input/output types, preconditions, operations) via AST analysis, and runs layered verification checks across available backends. When optional prover integrations such as Z3, Lean, Prolog, TLA+, or related MCP blocks are installed and configured, the self-model records their results; otherwise it falls back to static analysis, generated specifications, and conservative runtime checks. Treat its proof registry as an auditable verification trace, not as a guarantee that every property was externally machine-proved. Supports BDD generation, FSM modelling, composition checking, and modification guards.
Verification scope
Self-model proof records can include external prover results, static-analysis results, generated formal specifications, and fallback checks. For production-critical, regulated, or safety-critical workflows, require the specific prover backend you intend to rely on, inspect each proof status and artifact, and pair self-model output with tests, staging validation, audit logging, and human review.
When to use:
- Verifying that a component implementation matches its declared contract before deployment
- Generating BDD-style behavioural specifications from component source code
- Detecting unsafe self-modification attempts via the
ModificationGuard
Example:
from mvp.self_model import SelfModelBlock, SelfModelInput
block = SelfModelBlock(name="self_model")
result = block.infer(SelfModelInput(
op="introspect",
component_names=["goal_engine"],
))
# result.ok → True; result.value → SelfModelOutput with extracted contracts
Works well with: csf, formal_methods, immune_system
Public API¶
SelfModelDecisionError(ValueError)¶
The LLM did not produce a usable, validated modification-safety decision.
ModificationDecision¶
Validated self-modification-safety verdict.
| Field | Type | Default |
|---|---|---|
target | str | required |
approved | bool | required |
decision | str | required |
baseline_decision | str | required |
requires_review | bool | False |
requires_human_review | bool | False |
degraded | bool | False |
llm_used | bool | False |
rationale | str | '' |
raw_response | str | '' |
floor | dict[str, Any] \| None | None |
SelfModelRuntime(Protocol)¶
Methods:
decide(target_component: str, proposed_source: str, registry: ProofRegistry | None = None, baseline: ModificationDecision | None = None) -> ModificationDecision¶
LLMSelfModelRuntime¶
Provider-neutral self-modification-risk runtime backed by G6's LLM caller.
Constructor:
| Parameter | Type | Default |
|---|---|---|
llm | LLMCaller \| None | None |
Methods:
decide(target_component: str, proposed_source: str, registry: ProofRegistry | None = None, baseline: ModificationDecision | None = None) -> ModificationDecision¶
SelfModelPlanner¶
Runtime-first facade with deterministic fallback + one-way decision clamp.
Constructor:
| Parameter | Type | Default |
|---|---|---|
runtime | SelfModelRuntime \| None | None |
Methods:
decide(target_component: str, proposed_source: str, registry: ProofRegistry | None = None, baseline: ModificationDecision | None = None) -> ModificationDecision¶
BatchResult¶
Result of a batch verification run.
| Field | Type | Default |
|---|---|---|
results | dict[str, list[ProofRecord]] | field(default_factory=dict) |
contracts | dict[str, ComponentContract] | field(default_factory=dict) |
skipped | list[tuple[str, str]] | field(default_factory=list) |
ProofCache¶
SQLite-backed cache mapping (component, source_hash) to proof results.
Constructor:
| Parameter | Type | Default |
|---|---|---|
db_path | str | '~/.g6/proof_cache.db' |
max_age_seconds | float \| None | None |
Methods:
close() -> None¶
Close the underlying SQLite connection.
get(component: str, source_hash: str) -> list[ProofRecord] | None¶
Return cached proofs or None on miss.
put(component: str, source_hash: str, proofs: list[ProofRecord]) -> None¶
Store proof results in the cache.
invalidate(component: str) -> None¶
Remove all cached entries for a component.
expire_older_than(days: int = 30) -> int¶
Delete cache entries older than days days. Returns count deleted.
stats() -> dict¶
Return cache statistics.
ComponentFSM¶
FSM behavioral model for a G6 component.
| Field | Type | Default |
|---|---|---|
component_name | str | required |
states | list[str] | field(default_factory=lambda: list(AIBLOCK_STATES)) |
transitions_spec | list[dict] | field(default_factory=lambda: list(AIBLOCK_TRANSITIONS)) |
contract | ComponentContract \| None | None |
Methods:
current_state() -> str¶
trigger(event: str) -> bool¶
Fire a trigger event. Returns True if transition succeeded.
get_available_triggers() -> list[str]¶
Return triggers valid from the current state.
reset_to_idle() -> None¶
Force reset to idle state.
to_dot() -> str¶
Export FSM as DOT graph.
SelfModelPatternReview¶
Structured one-way (escalate-only) review for a modification verdict.
| Field | Type | Default |
|---|---|---|
tightened | bool | required |
reason | str | required |
target | str | required |
decision | str | required |
baseline_decision | str | required |
approved | bool | True |
requires_review | bool | False |
requires_human_review | bool | False |
warnings | tuple[str, ...] | () |
reflection | str | '' |
Methods:
to_metadata() -> dict[str, Any]¶
SelfModelPatternRuntime¶
Stateless executable mechanisms for self_model's applied patterns.
Methods:
tighten(target: str, decision: str, baseline_decision: str, approved: bool = True, changes: Any = (), requires_review: bool = False, requires_human_review: bool = False, degraded: bool = False) -> SelfModelPatternReview¶
One-way self-modification guard (hook-based-safety-guard-rails).
ProofRecord¶
A single formal proof or verification result.
| Field | Type | Default |
|---|---|---|
proof_id | str | required |
component | str | required |
property_name | str | required |
property_formal | str | required |
formal_method | str | required |
status | Literal['proved', 'disproved', 'timeout', 'pending', 'monadic_guard', 'skipped'] | required |
proof_artifact | str | required |
counterexample | str | '' |
source_hash | str | '' |
timestamp | str | '' |
layer | Literal['L1_contract', 'L2_composition', 'L3_modification'] | 'L1_contract' |
dependencies | list[str] | field(default_factory=list) |
evidence_level | ProofEvidenceLevel | 'fallback_check' |
degradation_reason | str \| None | None |
SystemCharacterisation¶
A formal characterisation of a component or pipeline.
| Field | Type | Default |
|---|---|---|
char_id | str | required |
scope | str | required |
target | str | required |
method | str | required |
characterisation_type | str | required |
artifact | str | required |
source_hash | str | '' |
timestamp | str | '' |
ProofRegistry¶
SQLite-backed registry for formal proofs and characterisations.
Constructor:
| Parameter | Type | Default |
|---|---|---|
db_path | str | ':memory:' |
Methods:
close() -> None¶
Close the underlying SQLite connection.
register(record: ProofRecord) -> str¶
Store a proof record. Returns the proof_id.
get_proofs(component: str) -> list[ProofRecord]¶
All proofs for a component.
get_proofs_by_method(method: str) -> list[ProofRecord]¶
All proofs using a specific prover.
get_proofs_by_layer(layer: str) -> list[ProofRecord]¶
All proofs at a specific layer.
check_staleness(component: str, current_hash: str) -> list[ProofRecord]¶
Return proofs invalidated by a source hash change.
summary() -> dict¶
Aggregate counts by status, method, layer, component.
export_audit_log() -> str¶
Full human-readable documentation of all proofs.
register_characterisation(char: SystemCharacterisation) -> str¶
Store a characterisation. Returns the char_id.
get_characterisations(target: str) -> list[SystemCharacterisation]¶
All characterisations for a target.
ComponentContract¶
Behavioral contract extracted from a component's source code.
| Field | Type | Default |
|---|---|---|
component_name | str | required |
input_type | str | required |
output_type | str | required |
state_type | str | required |
operations | list[str] | required |
preconditions | dict[str, list[str]] | required |
postconditions | dict[str, list[str]] | required |
invariants | list[str] | required |
purity | dict[str, bool] | required |
error_prefix | str | required |
source_hash | str | required |
ComponentDomain(BaseModel)¶
A component modelled as a category (DomainSpec) + its contract.
| Field | Type | Default |
|---|---|---|
domain | Any | required |
contract | Any | required |
verified | bool | False |
verification_results | dict[str, Any] | Field(default_factory=dict) |
SelfModelInput(BaseModel)¶
Input for SelfModelBlock.
| Field | Type | Default |
|---|---|---|
op | OP_TYPES | required |
component_names | list[str] | Field(default_factory=list) |
pipeline | list[str] | Field(default_factory=list) |
proposed_diff | str | '' |
target_component | str | '' |
max_workers | int | 4 |
use_cache | bool | True |
run_mode | RUN_MODES | 'standard' |
SelfModelOutput(BaseModel)¶
Output from SelfModelBlock.
| Field | Type | Default |
|---|---|---|
op | str | required |
success | bool | required |
components_processed | list[str] | Field(default_factory=list) |
contracts | dict[str, Any] | Field(default_factory=dict) |
verification_results | dict[str, Any] | Field(default_factory=dict) |
sheaf_obstructions | list[str] | Field(default_factory=list) |
csf_decision | str | '' |
explanation | str | '' |
proof_summary | dict[str, Any] | Field(default_factory=dict) |
errors | list[str] | Field(default_factory=list) |
system_map | dict[str, Any] \| None | None |
cache_stats | dict[str, Any] \| None | None |
degraded | bool | False |
degradation_reason | str \| None | None |
completion_state | COMPLETION_STATES | 'qualified-draft' |
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 |
promotion_witness | dict[str, Any] \| None | None |
reliability_envelope | dict[str, Any] \| None | None |
SelfModelBlock(AIBlock[SelfModelInput, SelfModelOutput, dict])¶
Symbolic self-model for the G6 system.
| Field | Type | Default |
|---|---|---|
name | str | 'self_model' |
state | dict \| None | None |
critical_components | list[str] | field(default_factory=lambda: list(_DEFAULT_CRITICAL_COMPONENTS)) |
Methods:
infer(data: SelfModelInput) -> Result[SelfModelOutput]¶
health() -> dict¶
Return health status for production monitoring.
SelfModelSkill¶
| Field | Type | Default |
|---|---|---|
name | str | required |
pattern_slug | str | required |
description | str | required |
executable | bool | required |
mechanism | str | required |
capabilities | tuple[str, ...] | required |
triggers | tuple[str, ...] | required |
risk_notes | tuple[str, ...] | required |
Methods:
compact() -> dict[str, Any]¶
SelfModelSkillCatalog¶
Maps each applied pattern slug to a self-modification-safety skill record.
Methods:
list_skills() -> list[SelfModelSkill]¶
executable_skills() -> list[SelfModelSkill]¶
get(slug: str) -> SelfModelSkill | None¶
SystemCoverage¶
Summary statistics for system-wide verification.
| Field | Type | Default |
|---|---|---|
total_discovered | int | required |
total_verifiable | int | required |
total_verified | int | required |
total_proofs | int | required |
proofs_by_status | dict[str, int] | required |
proofs_by_method | dict[str, int] | required |
proofs_by_layer | dict[str, int] | required |
SystemMap¶
Complete self-understanding map of G6.
| Field | Type | Default |
|---|---|---|
components | dict[str, ComponentContract] | required |
proofs | dict[str, list[ProofRecord]] | required |
dependency_graph | dict[str, list[str]] | required |
type_matrix | dict[str, dict[str, bool]] | required |
coverage | SystemCoverage | required |
skipped | list[tuple[str, str]] | required |
Functions¶
summarize_self_model_agentic_evidence(decisions: list[dict[str, Any]]) -> dict[str, Any]¶
Summarise runtime-vs-fallback self-modification decisions with path redaction.
agentic_planner_enabled(default_enabled: bool = True) -> bool¶
Decide whether the agentic self-modification planner should be used.
validate_modification_decision(decision: ModificationDecision) -> None¶
Assert a FINAL modification verdict is canonical and never relaxed.
deterministic_modification_decision(target_component: str, proposed_source: str, registry: ProofRegistry | None = None) -> ModificationDecision¶
The FIXED, HARD deterministic formal floor for a proposed self-modification.
apply_modification_floor(baseline: ModificationDecision, candidate: ModificationDecision) -> ModificationDecision¶
One-way advisory-tier decision clamp (output-verification-loop).
self_model_pattern_guard(decision: ModificationDecision) -> tuple[ModificationDecision, Any]¶
Apply the always-on one-way structural guard to a decision.
grounded_guard_modification(target_component: str, proposed_source: str, registry: ProofRegistry | None = None, planner: 'SelfModelPlanner | None' = None) -> tuple[ModificationDecision, Any, bool, str]¶
The ONE shared grounded chokepoint for EVERY self-modification decision.
discover_verifiable_components() -> list[str]¶
Use ComponentRegistry to discover all components, then filter to
verify_all(component_names: list[str] | None = None, registry: ProofRegistry | None = None, cache: ProofCache | None = None, max_workers: int = 1, use_cache: bool = True, progress_callback: Callable[[str, int, int], None] | None = None) -> BatchResult¶
Batch verify all (or specified) components with caching and parallelism.
generate_feature(contract: ComponentContract) -> str¶
Generate a Gherkin .feature file from a ComponentContract.
generate_step_definitions(contract: ComponentContract) -> str¶
Generate pytest-bdd step definition file for a component.
write_bdd_files(contract: ComponentContract, features_dir: str | pathlib.Path | None = None, tests_dir: str | pathlib.Path | None = None) -> dict[str, str]¶
Write .feature and step definition files for a component.
write_all_bdd_files(contracts: dict[str, ComponentContract], features_dir: str | pathlib.Path | None = None, tests_dir: str | pathlib.Path | None = None) -> dict[str, str]¶
Write BDD files for all contracts.
check_composition(contracts: list[ComponentContract], registry: ProofRegistry | None = None, default_epsilon: float = 0.05, rho: float = 0.0, target_epsilon: float = 0.2) -> Result[dict]¶
Layer 2 verification: check that a pipeline of components composes safely.
verify_contract(contract: ComponentContract, registry: ProofRegistry | None = None) -> list[ProofRecord]¶
Run all Layer 1 verification tasks for a component contract.
verify_contracts(contracts: dict[str, ComponentContract], registry: ProofRegistry | None = None) -> dict[str, list[ProofRecord]]¶
Verify contracts for multiple components.
generate_fsm(contract: ComponentContract) -> ComponentFSM¶
Auto-generate an FSM from a ComponentContract.
generate_pipeline_fsm(contracts: list[ComponentContract], pipeline_name: str = 'pipeline') -> ComponentFSM¶
Generate a hierarchical FSM for a pipeline of components.
introspect_component(component_name: str) -> Result[ComponentContract]¶
Extract a ComponentContract from a component's source files.
introspect_components(names: list[str]) -> dict[str, Result[ComponentContract]]¶
Introspect multiple components, returning results keyed by name.
compute_source_hash(component_name: str) -> Result[str]¶
Compute current source hash for a component (for staleness detection).
contract_to_domain(contract: ComponentContract) -> Any¶
Convert a ComponentContract to a DomainSpec (category).
guard_modification(target_component: str, proposed_source: str, registry: ProofRegistry | None = None) -> Result[dict]¶
Layer 3: Evaluate whether a proposed modification is safe.
applied_agentic_patterns() -> list[dict[str, Any]]¶
Return compact metadata for self_model-applied vendored patterns.
default_catalogue_probe(capability: str) -> list[dict[str, Any]]¶
Lexical coverage probe over the catalogue vocabulary.
[]=> a genuine gap.
assess_compose_vs_create(required_capabilities: Sequence[str], coverage_probe: Callable[[str], list[dict[str, Any]]] | None = None, max_harnesses_per_capability: int = 8) -> Result[dict]¶
Ashby requisite-variety decision: do catalogued harnesses cover the required capabilities?
get_skill_catalog() -> SelfModelSkillCatalog¶
build_system_map(results: dict[str, list[ProofRecord]], contracts: dict[str, ComponentContract], all_discovered: list[str], skipped: list[tuple[str, str]] | None = None) -> SystemMap¶
Build complete system map from batch verification results.
render_summary(system_map: SystemMap) -> str¶
Human-readable summary report.
export_dot(system_map: SystemMap) -> str¶
Export dependency graph as DOT for Graphviz.
MCP Tools¶
| Operation | Source |
|---|---|
introspect | self_model_mcp |
verify_contract | self_model_mcp |
check_composition | self_model_mcp |
guard_modification | self_model_mcp |
full_verify | self_model_mcp |
verify_all | self_model_mcp |
system_map | self_model_mcp |
capabilities | self_model_mcp |
refresh | self_model_mcp |
status | self_model_mcp |
generate_fsm | self_model_mcp |
generate_pipeline_fsm | self_model_mcp |
export_fsm_dot | self_model_mcp |
generate_bdd_feature | self_model_mcp |
generate_bdd_steps | self_model_mcp |
write_bdd_files | self_model_mcp |
store_proof | self_model_mcp |
query_proofs | self_model_mcp |
audit_log | self_model_mcp |
proof_summary | self_model_mcp |
list_components | self_model_mcp |
component_health | self_model_mcp |
dependency_graph | self_model_mcp |
type_compatibility | self_model_mcp |
search | self_model_mcp |
info | self_model_mcp |
list_patterns | self_model_mcp |