Core¶

Core — mvp.core

Cluster: Core Infrastructure | Type: component | MCP Tools: None

Overview¶

core provides the foundation types that every other G6 component builds on, including the generic AIBlock base class, the railway-oriented Result monad, guardrailed SearchTree, resource-bounding primitives, and structural protocols.

When to use:

• Defining custom reasoning blocks by subclassing AIBlock
• Building typed, composable pipelines with PipelineBlock and >> chaining
• Enforcing token, time, and disk resource bounds on any pipeline execution

Example:

from mvp.core import AIBlock, Result

class DoubleBlock(AIBlock[int, int, None]):
    def infer(self, data: int) -> Result[int]:
        return Result.ok(data * 2)

block = DoubleBlock(name="double")
result = block.infer(21)
# result.ok → True; result.value → 42

Reliability envelope guarantee: when a ReliabilityEnvelope carries conflicting state, the two public state fields (completion_state and reliability_label) are reconciled to the weakest honest label, and any explicit degradation signal (degraded=True or a degradation_reason) caps the result at non-verified — so verified can never coexist with a degraded or blocked state. A legitimately verified result may still carry a green/amber warning_card; the card's verified flag is a derived echo of the reconciled state, not an independent downgrade trigger.

Works well with: every other component — core is the universal dependency

Public API¶

AIBlock(Generic[Input, Output, State])¶

The atomic building block of the G6 hyperdistillation system.

Methods:

infer(data: Input) -> Result[Output]¶

Single-shot computation: transform Input → Result[Output].

verify(output: Output, contract: BlockContract | None = None) -> EvidenceBundle¶

Verify output using a registered per-harness verification method.

bias() -> dict[str, 'Any']¶

Structural assumptions about this block's hypothesis space.

contract() -> BlockContract¶

Runtime obligations this block promises to satisfy.

then(other: 'AIBlock[Output, Any, Any]') -> 'PipelineBlock[Input, Any, None]'¶

Chain self >> other into a sequential pipeline.

PipelineBlock(AIBlock[Input, Output, None])¶

Two AIBlocks chained sequentially: left.emit() → right.ingest().

Constructor:

Methods:

infer(data: Any) -> Result[Any]¶

ArtifactSchemaError(ValueError)¶

FailureMode¶

A known way this block can go wrong.

TransitionOutcome¶

A single probability-weighted outcome from a TransitionKernel evaluation.

ResourceTriple¶

Abstract formal resource bounds for a BoundedAgent.

BoundedAgent¶

Pure formal description of a resource-bounded agent.

SafetyMonitor(ABC)¶

Abstract base interface for runtime safety monitors.

Methods:

reset() -> None¶

Reset all accumulated state. Called at the start of each episode.

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

Process one agent step.

is_violated() -> bool¶

Return True if any safety property has been violated so far.

StepRationale¶

Audit trace record for a single agent step.

SafetyDecisionReport¶

Structured verdict returned by a SafetyVerifier or CSF evaluation.

SafetyLevel(str, Enum)¶

Qualitative safety evaluation outcome from TASP-style tiered evaluation.

RiskClassification(str, Enum)¶

Risk classification for governance-aware operations.

OperationMode(str, Enum)¶

Agent autonomy level for governance enforcement.

QualitativeSafetyVerdict¶

Structured verdict from a qualitative safety evaluation.

StrategySpec¶

Specification for a strategic orchestration plan.

ImpedanceBounds¶

Safety envelope for impedance-based motor control.

Methods:

clamp_k(k: float) -> float¶

Clamp stiffness to valid range.

clamp_c(c: float) -> float¶

Clamp damping to valid range.

EnergyModel¶

Tracks computational energy consumption as analogue of physical energy E(t).

Methods:

record_cycle(tokens: int = 0, api_calls: int = 0, wall_clock_s: float = 0.0) -> float¶

Record one cycle's resource consumption. Returns cycle energy.

total_energy() -> float¶

Total cumulative energy spent.

mean_cycle_energy() -> float¶

Mean energy per cycle (0.0 if no cycles recorded).

energy_trend() -> float¶

Energy trend: negative = improving (using less energy per cycle).

G6Error¶

Structured error with actionable guidance.

Methods:

to_dict() -> dict¶

to_problem_detail(status: int = 400, request_id: str | None = None, instance: str | None = None) -> dict[str, Any]¶

Claim¶

Methods:

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

TestResult¶

SourceRef¶

Assumption¶

EvidenceBundle¶

Portable evidence returned by per-harness verification.

Methods:

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

ExtensionPoints¶

What a harness DECLARES about its T3/self-programming surface.

FailureModeRecord¶

FailureModeRegistry¶

Singleton-style registry keyed by runtime-safe string codes.

Methods:

seed_defaults() -> None¶

register(record: FailureModeRecord) -> None¶

get(code: str | FailureModeCode) -> FailureModeRecord¶

list_all() -> list[FailureModeRecord]¶

is_registered(code: str | FailureModeCode) -> bool¶

StorableMixin¶

Opt-in: block supports save/load of internal state.

Methods:

save(path: str) -> Result[str]¶

Persist the block's state to path. Returns the written path.

load(path: str) -> Result[bool]¶

Restore the block's state from path.

LearnableMixin¶

Opt-in: block supports online learning from supervised signal.

Methods:

learn(data: Any, labels: Any) -> Result[Any]¶

Update block parameters given supervised signal.

IngestEmitMixin¶

Opt-in: block supports staged ingest/emit workflow.

Methods:

ingest(data: Any) -> Result[Any]¶

Accept raw input, validate, and store internally.

emit() -> Result[Any]¶

Produce output from current internal state.

PipelineStep¶

One step in a pipeline: component name + operation + params.

Ingestible(Protocol)¶

Can accept raw input and return a Result.

Methods:

ingest(data: Any) -> Result[Any]¶

Emittable(Protocol)¶

Can produce output from internal state and return a Result.

Methods:

emit() -> Result[Any]¶

Storable(Protocol)¶

Can persist and restore its state.

Methods:

save(path: str) -> Result[str]¶

load(path: str) -> Result[bool]¶

Inferrable(Protocol)¶

Can run forward inference given an input.

Methods:

infer(data: Any) -> Result[Any]¶

Learnable(Protocol)¶

Can update its parameters given labelled data.

Methods:

learn(data: Any, labels: Any) -> Result[Any]¶

InductiveBias(Protocol)¶

Declares its structural assumptions about the hypothesis space.

Methods:

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

ConfigurableBias(Protocol)¶

Can accept runtime bias parameter updates from the evolution loop.

Methods:

configure_bias(params: dict[str, Any]) -> None¶

ResourceBounds¶

Immutable resource constraint envelope for an AIBlock or pipeline.

ResourceUsage¶

Mutable runtime counter — one instance per pipeline execution.

Methods:

record_tokens(count: int, input_tokens: int = 0, output_tokens: int = 0) -> None¶

Append count tokens to the log and update the running total.

record_cost(amount: float) -> None¶

Append amount USD to the cost log and update the running total.

cost_in_window(window_seconds: float) -> float¶

Return total cost recorded within the last window_seconds.

tokens_in_window(window_seconds: float) -> int¶

Return the total tokens recorded within the last window_seconds.

elapsed_seconds() -> float¶

Seconds since this ResourceUsage instance was created.

ResourceGuardrail(Guardrail)¶

Hard constraint that enforces ResourceBounds against a ResourceUsage counter.

Constructor:

Methods:

check(node_data: Any) -> Result[bool]¶

Evaluate all configured bounds against current usage.

Status(Enum)¶

Result(Generic[Output])¶

Lightweight railway Result — wraps a value or an error string.

Constructor:

Methods:

ok(value: Output) -> 'Result[Output]'¶

fail(error: str | object) -> 'Result[Output]'¶

is_ok() -> bool¶

is_fail() -> bool¶

is_failure() -> bool¶

Property alias for is_fail() — for ergonomic boolean checks.

value() -> Output¶

unwrap() -> Output¶

Alias for .value — raises ValueError if FAIL.

error() -> str¶

map(fn: Callable[[Output], T]) -> 'Result[T]'¶

flat_map(fn: Callable[[Output], 'Result[T]']) -> 'Result[T]'¶

or_else(default: Output) -> Output¶

FailureMode(Enum)¶

Standardised failure categories for tree-node diagnostics.

Guardrail¶

A hard constraint on tree expansion.

Methods:

check(node_data: Any) -> Result[bool]¶

Checkpoint¶

A soft assertion that logs a diagnostic without halting expansion.

Methods:

evaluate(node_data: Any) -> Result[bool]¶

Breakpoint¶

An optional debugging hook. When active, invokes callback with the

Methods:

trigger(node_data: Any) -> Result[Any]¶

NodeDiagnostic¶

Per-node structured log entry produced during tree evaluation.

Methods:

passed() -> bool¶

TreeNode(Generic[Node])¶

A single node in the search tree.

Methods:

add_child(child_data: Node) -> 'TreeNode[Node]'¶

is_leaf() -> bool¶

SearchTree(Generic[Node])¶

Guardrailed, diagnostic search tree.

Methods:

new_root(data: Node) -> TreeNode[Node]¶

evaluate_node(node: TreeNode[Node]) -> NodeDiagnostic¶

expand(parent: TreeNode[Node], candidate_nodes: list[Node]) -> Result[list[TreeNode[Node]]]¶

Attempt to add candidate_nodes as children of parent.

best_leaf(key: Callable[[TreeNode[Node]], float] | None = None) -> Result[TreeNode[Node]]¶

Return the leaf node with the highest score (or custom key).

iter_nodes() -> Iterator[TreeNode[Node]]¶

StateSurface¶

ModelProfile¶

Empirical profile for a model family.

StakesConfig¶

Thresholds indexed by stakes level.

SelfCorrectionConfig¶

TrustEnvelope¶

Input envelope for trust evaluation (Codex Sec 4.3).

TrustVerdict¶

Result of trust evaluation against stakes thresholds.

SolvabilityVerdict¶

Result of solvability check (Thm 10: C(M,W,T) >= V(P)).

Functions¶

ensure_schema_version(payload: dict[str, Any], version: str = CURRENT_SCHEMA_VERSION) -> dict[str, Any]¶

normalize_schema_version(payload: dict[str, Any], default: str = '0.0') -> str¶

read_version(payload: dict[str, Any]) -> str¶

is_compatible(payload: dict[str, Any], current: str = CURRENT_SCHEMA_VERSION) -> bool¶

assert_compatible(payload: dict[str, Any], current: str = CURRENT_SCHEMA_VERSION, artifact_type: str = 'artifact') -> None¶

load_versioned_artifact(payload: dict[str, Any], artifact_type: str, current: str = CURRENT_SCHEMA_VERSION, required_fields: tuple[str, ...] = ()) -> dict[str, Any]¶

new_request_id(prefix: str = 'req') -> str¶

get_request_id() -> str | None¶

get_run_id() -> str | None¶

get_task_id() -> str | None¶

bind(request_id: str | None = None, run_id: str | None = None, task_id: str | None = None) -> dict[str, Token[str | None]]¶

reset(tokens: dict[str, Token[str | None]]) -> None¶

bound(request_id: str | None = None, run_id: str | None = None, task_id: str | None = None)¶

load_error_catalog() -> dict[str, dict[str, str]]¶

Load the small local error-code catalog without requiring PyYAML.

error_catalog_record(code: str) -> dict[str, str]¶

Return catalog metadata for code, or an empty record if unknown.

suggested_action_for_code(code: str, default: str = 'Check the request and retry.') -> str¶

Return the user-facing recovery action for an error code.

title_for_code(code: str, default: str = 'Request failed') -> str¶

Return the problem-title text for an error code.

type_for_code(code: str, default_base_url: str = 'https://g6solver.local/problems') -> str¶

Return the RFC 9457 problem type URL for an error code.

status_for_code(code: str, default: int = 400) -> int¶

Return the cataloged HTTP status for an error code when available.

problem_detail_for_code(code: str, detail: str, status: int | None = None, request_id: str | None = None, instance: str | None = None) -> dict[str, Any]¶

Build an RFC 9457-style problem detail from the shared catalog.

to_problem_detail(error: G6Error, status: int = 400, request_id: str | None = None, instance: str | None = None) -> dict[str, Any]¶

unknown_op_error(component: str, op: str, valid_ops: list[str]) -> G6Error¶

Factory for the common 'Unknown op' pattern with fuzzy matching.

missing_param_error(component: str, param: str, op: str | None = None) -> G6Error¶

not_found_error(component: str, resource: str, identifier: str) -> G6Error¶

validation_error(component: str, detail: str) -> G6Error¶

dependency_error(component: str, dependency: str) -> G6Error¶

resource_limit_error(component: str, resource: str, limit: int | float) -> G6Error¶

to_core(rresult: RResult[Any]) -> Result¶

Convert returns.Result → mvp.core.Result.

from_core(result: Result) -> RResult[Any]¶

Convert mvp.core.Result → returns.Result.

safe_invoke(block: Any, data: Any) -> RResult[Any]¶

Call block.infer(data) and return a returns.Result.

flow_blocks(*blocks: Any) -> Callable[[Any], RResult[Any]]¶

Return a function that chains blocks sequentially using bind.

component_maturity(level: Maturity, reason: str = '')¶

Decorator: mark a component class with a maturity level.

get_maturity(cls) -> Maturity | None¶

Return the declared maturity level or None if unmarked.

maturity_reason(cls) -> str¶

component_capabilities(*caps: str)¶

Decorator: declare capabilities a component provides (e.g. 'llm_routing', 'formal_verification').

get_capabilities(cls) -> tuple[str, ...]¶

Return declared capabilities for a component class.

observable(cls)¶

Mark a component as observable — emits lifecycle events for monitoring.

is_observable(cls) -> bool¶

Return whether the class is marked as observable.

optional_import(module_name: str, pip_name: str | None = None)¶

Import an optional dependency, returning a proxy if unavailable.

run_pipeline(steps: list[PipelineStep], initial: dict[str, Any], registry: Any = None, stakes: str = 'medium', cancel_check: Callable[[], bool] | None = None) -> RResult[dict, str]¶

Execute steps sequentially, threading the output of each into

run_parallel(steps: list[PipelineStep], inputs: list[dict[str, Any]], registry: Any = None) -> RResult[list[dict], str]¶

Execute steps concurrently via trio.open_nursery().

run_pipeline_trio(steps: list[PipelineStep], initial: dict[str, Any], registry: Any = None, stakes: str = 'medium', cancel_check: Callable[[], bool] | None = None) -> RResult[dict, str]¶

Synchronous wrapper that calls trio.run() to execute

run_pipeline_traced(steps: list[PipelineStep], initial: dict[str, Any], workflow_name: str = 'unnamed', registry: Any = None) -> tuple[RResult[dict, str], 'WorkflowTrace | None']¶

Synchronous pipeline executor that also returns a WorkflowTrace.

check_compatibility(block_a: Any, block_b: Any) -> RResult[dict, str]¶

Verify output type of block_a matches input type of block_b.

run_parallel_trio(steps: list[PipelineStep], inputs: list[dict[str, Any]], registry: Any = None) -> RResult[list[dict], str]¶

Synchronous wrapper that calls trio.run() to execute

list_production() -> list[str]¶

is_production(harness_id: str) -> bool¶

resolve_model_profile(model_id: str) -> ModelProfile¶

Look up a ModelProfile by substring match against known families.

estimate_chain_reliability(n_steps: int, per_step_accuracy: float = DEFAULT_STEP_CONFIDENCE) -> float¶

Thm 9: A_chain = a^n.

compute_capacity(model_id: str, context_tokens: int, tool_count: int) -> float¶

C(M, W, T) = C_weights + C_context + C_tools.

hitl_threshold(cost_human: float, cost_error: float) -> float¶

tau_trust = 1 - C_human / C_error.

minimum_token_budget(model_id: str, domain: str = 'general', estimated_output_bits: float = 100.0, grounding_depth: int = 1, epsilon: float = 0.05) -> int¶

n_in >= I(y*|x) / R(M) + n_ground(d, epsilon).

evaluate_trust(envelope: TrustEnvelope) -> TrustVerdict¶

Evaluate trust envelope against stakes thresholds and return verdict.

check_solvability(model_id: str, context_tokens: int, tool_count: int, estimated_difficulty: float) -> SolvabilityVerdict¶

Check whether model capacity meets problem variety (Thm 10).