Reliability¶

mvp.reliability — Component Reliability Framework (CRF)

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

Overview¶

Public API¶

`BenchmarkSpecFactory`¶

Creates BenchmarkSpec objects for components based on their domain.

Methods:

`create_for_component(component: str, domain: str = 'general', sample_size: int = 25, score_threshold: float = 0.8, max_cycles: int = 5) -> BenchmarkSpec`¶

`ComponentFailureClassifier`¶

Generic FailureClassifier for CRF.

Methods:

`classify(result: TaskResult) -> FailureAnalysis`¶

`ExtractionResult`¶

Field	Type	Default
`component`	`str`	required
`source_t3_run_id`	`str`	required
`exemplars_created`	`int`	0
`exemplars_pending`	`int`	0
`fme_created`	`int`	0
`fme_promoted`	`int`	0
`recipes_created`	`int`	0
`profile_updated`	`bool`	False
`executed_at`	`float`	field(default_factory=time.time)

`KnowledgeExtractor`¶

Extracts exemplars and recipe candidates from a completed LearningResult.

Constructor:

Parameter	Type	Default
`store`	`ReliabilityStore \\| None`	None

Methods:

`extract(component: str, learning_result: LearningResult, source_run_id: str = '', quality_threshold: float = _ACCEPT_THRESHOLD) -> ExtractionResult`¶

`OrchestratorConfig`¶

Field	Type	Default
`priority`	`str`	'red_first'
`max_parallel`	`int`	3
`research_before_t3`	`bool`	True
`skip_green`	`bool`	False
`budget_usd_per_component`	`float`	0.0
`memory_limit_mb`	`int`	2000

`OrchestratorResult`¶

Field	Type	Default
`total_components`	`int`	0
`researched`	`int`	0
`trained`	`int`	0
`improved`	`int`	0
`unchanged`	`int`	0
`failed`	`int`	0
`total_cost_usd`	`float`	0.0
`duration_sec`	`float`	0.0

`T3Orchestrator`¶

Batch-runs T3 across all registered components in priority order.

Constructor:

Parameter	Type	Default
`store`	`ReliabilityStore \\| None`	None

Methods:

`priority_order(config: OrchestratorConfig | None = None) -> list[str]`¶

Return component names sorted: red first, then amber, then green.

`run_next(n: int = 1, config: OrchestratorConfig | None = None, **kwargs: Any) -> list[Any]`¶

Run T3 for the next n highest-priority components.

`run_all(config: OrchestratorConfig | None = None, **kwargs: Any) -> OrchestratorResult`¶

Run T3 for all registered components in priority order.

`ReliabilityRegistrar`¶

Writes ComponentProfile records to ReliabilityStore for @reliability_tracked classes.

Constructor:

Parameter	Type	Default
`store`	`ReliabilityStore \\| None`	None

Methods:

`register_class(cls: type) -> ComponentProfile`¶

Persist a @reliability_tracked class as a ComponentProfile (idempotent).

`scan_registry(component_registry: Any) -> list[ComponentProfile]`¶

Scan all classes in a component registry and register tracked ones.

`auto_enroll_production_ready(component_registry: Any) -> list[ComponentProfile]`¶

Register all @reliability_tracked classes in the registry. Alias for scan_registry.

`ReliabilityBridge`¶

Collects ScorerInputs from all upstream systems for a named component.

Constructor:

Parameter	Type	Default
`theory_db_dir`	`str`	''
`_scorecard_store`	`Any`	None
`_reliability_store`	`ReliabilityStore \\| None`	None

Methods:

`collect_inputs(component: str) -> ScorerInputs`¶

`ResearchResult`¶

Field	Type	Default
`component`	`str`	required
`exemplars_created`	`int`	0
`exemplars_pending`	`int`	0
`benchmark_spec`	`BenchmarkSpec \\| None`	None
`expert_exemplars_found`	`int`	0
`domain_principles_created`	`int`	0
`templates_generated`	`list[str]`	field(default_factory=list)
`research_sources`	`list[str]`	field(default_factory=list)
`synthesis_model`	`str`	''
`quality_threshold_used`	`float`	0.9
`executed_at`	`float`	field(default_factory=time.time)

`ComponentResearcher`¶

Autonomous cold-start benchmark pipeline with injectable LLM and search.

Constructor:

Parameter	Type	Default
`store`	`ReliabilityStore \\| None`	None
`benchmark_dir`	`str \\| None`	None
`max_download_mb`	`int`	50
`max_exemplar_payload_kb`	`int`	64
`max_benchmark_json_mb`	`int`	10
`memory_watermark_mb`	`int`	500
`_llm_fn`	`LLMFn \\| None`	None
`_search_fn`	`SearchFn \\| None`	None
`scenarios_root`	`Path \\| str \\| None`	None
`scenario_store`	`Any \\| None`	None

Methods:

`research(component_name: str, max_task_instances: int = 20, quality_threshold: float = 0.9, model: str = 'openrouter/anthropic/claude-opus-4.7', force: bool = False) -> ResearchResult`¶

Run the cold-start pipeline. Idempotent when force=False.

`AgentSpec`¶

Field	Type	Default
`agent_id`	`str`	required
`agent_type`	`str`	required
`backend`	`str`	required
`model`	`str`	required
`capabilities`	`tuple[str, ...]`	required
`cost_tier`	`str`	required
`prompt_usd_per_1k`	`float`	required
`completion_usd_per_1k`	`float`	required
`latency_profile`	`str`	required
`supports_tools`	`bool`	required
`max_context_tokens`	`int \\| None`	required
`disabled`	`bool`	False

`AgentRecommendation`¶

Field	Type	Default
`agent_id`	`str`	required
`confidence`	`str`	required
`expected_pass_rate`	`float`	required
`ci_low`	`float`	required
`ci_high`	`float`	required
`rationale`	`str`	required
`cost_effectiveness_score`	`float`	required
`is_fallback`	`bool`	required

`AgentRegistry`¶

Thin wrapper around agent_registry table with lazy baseline seeding.

Constructor:

Parameter	Type	Default
`store`	`ReliabilityStore`	required

Methods:

`get(agent_id: str) -> AgentSpec | None`¶

`list_active() -> list[AgentSpec]`¶

`AgentRouter`¶

Competence-matrix agent router with Bayesian updates and Wilson CI.

Constructor:

Parameter	Type	Default
`store`	`ReliabilityStore`	required

Methods:

`recommend(component: str, task_type: str = 'general', task_complexity: str = 'medium', budget_usd: float = 0.0) -> AgentRecommendation`¶

`record_outcome(component: str, agent_id: str, task_type: str, passed: bool, task_score: float = 0.0, latency_ms: float = 0.0, cost_usd: float = 0.0, task_id: str = '', failure_code: str = '') -> dict`¶

Bayesian update of competence matrix. Returns updated row.

`get_matrix(component: str) -> list[dict]`¶

`LoadReport`¶

Field	Type	Default
`loaded`	`int`	0
`errors`	`int`	0
`warnings`	`list[str]`	field(default_factory=list)
`error_details`	`list[str]`	field(default_factory=list)

`ScenarioLoader`¶

Filesystem ↔ SQLite sync for scenario .md/.json pairs.

Constructor:

Parameter	Type	Default
`store`	`ScenarioStore`	required
`scenarios_root`	`Path \\| str \\| None`	None
`workspace_components_root`	`Path \\| str \\| None`	None

Methods:

`discover(brick: str | None = None) -> LoadReport`¶

Glob scenario .md files, validate against sibling .json, upsert.

`compute_brick_size(brick: str) -> tuple[str, int, dict[str, int]]`¶

Return (tier, scenario_count, mix={'golden': n, 'adversarial': n}).

`RubricEntry(BaseModel)`¶

One weighted criterion in a scenario rubric.

Field	Type	Default
`criterion`	`str`	Field(min_length=1)
`weight`	`float`	Field(ge=0.0, le=1.0)
`kind`	`RubricKind`	'callable'
`callable_ref`	`str`	required

`InvocationSpec(BaseModel)`¶

How to invoke a brick under test.

Field	Type	Default
`component_class`	`str`	required
`input_class`	`str`	required
`method`	`str`	'infer'

`AdversarialVariant(BaseModel)`¶

A patch over the base inputs to generate a related adversarial input.

Field	Type	Default
`name`	`str`	Field(min_length=1)
`patch`	`dict[str, Any]`	Field(default_factory=dict)

`ScenarioFrontmatter(BaseModel)`¶

YAML frontmatter parsed from a scenario .md file.

Field	Type	Default
`scenario_id`	`str`	Field(min_length=1)
`brick`	`str`	Field(min_length=1)
`brick_kind`	`BrickKind`	required
`domain`	`Domain`	required
`persona`	`str \\| None`	None
`kind`	`ScenarioKind`	required
`difficulty`	`Difficulty`	required
`expected_outcome`	`ExpectedOutcome`	required
`estimated_minutes`	`int`	Field(ge=1)
`failure_modes_targeted`	`list[str]`	Field(default_factory=list)
`exemplar_refs`	`list[dict[str, Any]]`	Field(default_factory=list)
`review_status`	`ReviewStatus`	required
`is_pilot`	`bool`	False

`ScenarioHarness(BaseModel)`¶

Machine-readable JSON sibling of a scenario .md file.

Field	Type	Default
`scenario_id`	`str`	Field(min_length=1)
`brick`	`str`	Field(min_length=1)
`brick_kind`	`BrickKind`	required
`domain`	`Domain`	required
`kind`	`ScenarioKind`	required
`difficulty`	`Difficulty`	required
`expected_outcome`	`ExpectedOutcome`	required
`estimated_minutes`	`int`	Field(ge=1)
`inputs`	`dict[str, Any]`	Field(default_factory=dict)
`invocation`	`InvocationSpec`	required
`rubric`	`list[RubricEntry]`	Field(min_length=1)
`pass_threshold`	`float`	Field(ge=0.0, le=1.0)
`failure_modes_targeted`	`list[str]`	Field(default_factory=list)
`exemplar_refs`	`list[dict[str, Any]]`	Field(default_factory=list)
`adversarial_variants`	`list[AdversarialVariant]`	Field(default_factory=list)

`ScenarioPair(BaseModel)`¶

Composite of frontmatter + harness + filesystem provenance.

Field	Type	Default
`frontmatter`	`ScenarioFrontmatter`	required
`harness`	`ScenarioHarness`	required
`md_body`	`str`	required
`md_hash`	`str`	required
`md_path`	`str`	required
`json_path`	`str`	required

Methods:

`scenario_id() -> str`¶

`brick() -> str`¶

`ScenarioRecord`¶

A row from the scenarios table, decoded into Python types.

Field	Type	Default
`scenario_id`	`str`	required
`brick`	`str`	required
`brick_kind`	`str`	required
`domain`	`str`	required
`persona`	`str`	required
`kind`	`str`	required
`difficulty`	`str`	required
`expected_outcome`	`str`	required
`estimated_minutes`	`int`	required
`inputs`	`dict[str, Any]`	required
`invocation`	`dict[str, Any]`	required
`rubric`	`list[dict[str, Any]]`	required
`pass_threshold`	`float`	required
`failure_modes_targeted`	`list[str]`	required
`exemplar_refs`	`list[dict[str, Any]]`	required
`adversarial_variants`	`list[dict[str, Any]]`	required
`is_pilot`	`bool`	required
`review_status`	`str`	required
`md_path`	`str`	required
`json_path`	`str`	required
`md_hash`	`str`	required
`created_at`	`float`	required
`updated_at`	`float`	required

`ScenarioStore`¶

CRUD over the scenarios table inside reliability.db.

Constructor:

Parameter	Type	Default
`reliability_store`	`ReliabilityStore`	required

Methods:

`upsert(pair: ScenarioPair) -> None`¶

`mark_stale(scenario_id: str) -> int`¶

Set review_status='stale_pending_review' for the given id.

`get(scenario_id: str) -> ScenarioRecord | None`¶

`list_by_brick(brick: str) -> list[ScenarioRecord]`¶

`list_by_domain(domain: str, kind: str | None = None, review_status: str | None = None) -> list[ScenarioRecord]`¶

`list_all() -> list[ScenarioRecord]`¶

`count(brick: str | None = None) -> int`¶

`ReliabilityScore`¶

Field	Type	Default
`component`	`str`	required
`reliability_score`	`float`	required
`tier`	`ReliabilityTier`	required
`theory_strength`	`float`	required
`infer_health`	`float`	required
`audit_score`	`float`	required
`pass_rate`	`float`	required
`confidence`	`float`	required
`signals_used`	`list[str]`	required
`scored_at`	`float`	required

`ComponentProfile`¶

Field	Type	Default
`component`	`str`	required
`display_name`	`str`	required
`domain`	`str`	required
`registered_at`	`float`	required
`last_scored_at`	`float \\| None`	None
`current_score`	`ReliabilityScore \\| None`	None
`known_failure_modes`	`list[str]`	field(default_factory=list)
`benchmark_spec_id`	`str \\| None`	None
`rubric_ids`	`list[str]`	field(default_factory=list)
`tags`	`list[str]`	field(default_factory=list)
`maturity_tier`	`str`	'T0'

`FailureModeEntry`¶

Field	Type	Default
`entry_id`	`str`	required
`component`	`str`	required
`failure_code`	`str`	required
`title`	`str`	required
`trigger_conditions`	`list[str]`	required
`symptoms`	`list[str]`	required
`root_cause`	`str`	required
`recovery_strategy`	`str`	required
`prevention_tips`	`list[str]`	required
`model_specific_notes`	`dict[str, str]`	required
`related_theory_ids`	`list[str]`	required
`promoted_from_theory`	`str \\| None`	required
`theory_strength_at_promotion`	`float \\| None`	required
`component_hash`	`str`	required
`version`	`int`	required
`is_seed`	`bool`	required
`review_status`	`str`	required
`created_at`	`float`	required
`updated_at`	`float`	required
`conflict_note`	`str`	''

`Exemplar`¶

Field	Type	Default
`exemplar_id`	`str`	required
`component`	`str`	required
`kind`	`str`	required
`input_payload`	`dict`	required
`output_summary`	`str`	required
`score`	`float`	required
`domain_tag`	`str`	required
`difficulty_tag`	`str`	required
`model_used`	`str`	required
`failure_codes`	`list[str]`	required
`failure_explanation`	`str`	required
`source_run_id`	`str`	required
`component_hash`	`str`	required
`is_seed`	`bool`	required
`review_status`	`str`	required
`created_at`	`float`	required
`source_instance_id`	`str`	''
`reviewer_notes`	`str`	''

`ReliabilityFailureSignal`¶

Field	Type	Default
`signal_id`	`str`	required
`component`	`str`	required
`failure_codes`	`list[str]`	required
`failed_gates`	`list[str]`	required
`evidence_refs`	`list[str]`	required
`risk_tier`	`str`	'R0'
`reliability_status`	`str`	'unknown'
`completion_state`	`str`	''
`data_sufficiency`	`str`	'unknown'
`agentic_capability_required`	`bool`	False
`source`	`str`	''
`metadata`	`dict[str, Any]`	field(default_factory=dict)
`created_at`	`float`	field(default_factory=time.time)

`Recipe`¶

Field	Type	Default
`recipe_id`	`str`	required
`component`	`str`	required
`title`	`str`	required
`description`	`str`	required
`steps`	`list[dict]`	required
`preconditions`	`list[str]`	required
`postconditions`	`list[str]`	required
`domain_tags`	`list[str]`	required
`success_rate`	`float`	required
`run_count`	`int`	required
`source_theory_ids`	`list[str]`	required
`component_hash`	`str`	required
`is_seed`	`bool`	required
`review_status`	`str`	required
`created_at`	`float`	required
`updated_at`	`float`	required
`last_validated_at`	`float`	0.0
`agentic_workflow`	`AgenticWorkflowSpec \\| None`	None

`BenchmarkSpec`¶

Field	Type	Default
`spec_id`	`str`	required
`component`	`str`	required
`name`	`str`	required
`dataset_builder_fn`	`str \\| None`	None
`failure_classifier_fn`	`str \\| None`	None
`sample_size`	`int`	25
`score_threshold`	`float`	0.8
`max_cycles`	`int`	5
`created_at`	`float`	field(default_factory=time.time)

`ScorerInputs`¶

Field	Type	Default
`component`	`str`	required
`active_theories`	`list[Theory]`	required
`infer_metrics`	`InferMetrics`	required
`latest_audit`	`AuditResult \\| None`	required
`maturity_score`	`MaturityScore \\| None`	required
`t3_pass_rate`	`float \\| None`	required
`t3_run_count`	`int`	required

`T3RunConfig`¶

Field	Type	Default
`component`	`str`	required
`max_cycles`	`int`	5
`stagnation_limit`	`int`	2
`stagnation_delta`	`float`	0.01
`db_path`	`str`	''
`phase`	`str`	'train'

`T3RunResult`¶

Field	Type	Default
`component`	`str`	required
`learning_result`	`LearningResult`	required
`reliability_score`	`ReliabilityScore`	required
`theories_synced`	`int`	required
`converged`	`bool`	required
`convergence_reason`	`str`	required
`executed_at`	`float`	field(default_factory=time.time)

`ReliabilityScorer`¶

Computes a live ReliabilityScore from ScorerInputs.

Methods:

`compute(inputs: ScorerInputs) -> ReliabilityScore`¶

`KnowledgeSeedLoader`¶

Discovers and loads reliability_seed.py files into ReliabilityStore.

Constructor:

Parameter	Type	Default
`store`	`ReliabilityStore \\| None`	None

Methods:

`load_all(workspace_root: str | None = None) -> dict[str, int]`¶

Load all seed files under workspace_root/components/mvp/*/reliability_seed.py.

`ReliabilityStore`¶

SQLite-backed persistence for CRF artefacts (WAL mode, thread-safe).

Constructor:

Parameter	Type	Default
`db_path`	`str`	''

Methods:

`close() -> None`¶

`save_profile(profile: ComponentProfile) -> None`¶

`get_profile(component: str) -> ComponentProfile | None`¶

`list_profiles() -> list[ComponentProfile]`¶

`get_maturity_tier(component: str) -> str | None`¶

Return the stored maturity tier for a component, or None if absent.

`set_maturity_tier(component: str, tier: str) -> None`¶

Set the maturity tier for a component.

`save_fme(fme: FailureModeEntry) -> None`¶

`get_fme(entry_id: str) -> FailureModeEntry | None`¶

`list_fmes(component: str, review_status: str | None = None) -> list[FailureModeEntry]`¶

`get_stale_entries(component: str, current_hash: str) -> dict[str, list]`¶

`save_exemplar(ex: Exemplar) -> None`¶

`get_exemplar(exemplar_id: str) -> Exemplar | None`¶

`list_exemplars(component: str, kind: str | None = None, review_status: str | None = None) -> list[Exemplar]`¶

`soft_delete_exemplar(exemplar_id: str) -> None`¶

`save_recipe(recipe: Recipe) -> None`¶

`get_recipe(recipe_id: str) -> Recipe | None`¶

`list_recipes(component: str, review_status: str | None = None) -> list[Recipe]`¶

`save_recipe_run(run: AgenticWorkflowRun) -> None`¶

`get_recipe_run(run_id: str) -> AgenticWorkflowRun | None`¶

`save_harness_theory(theory: HarnessTheory) -> None`¶

`get_harness_theory(theory_id: str) -> HarnessTheory | None`¶

`list_harness_theories(component: str, status: str | None = 'active') -> list[HarnessTheory]`¶

`save_failure_signal(signal: ReliabilityFailureSignal) -> None`¶

`get_failure_signal(signal_id: str) -> ReliabilityFailureSignal | None`¶

`list_failure_signals(component: str, limit: int = 200) -> list[ReliabilityFailureSignal]`¶

Return a component's failure signals, newest first.

`list_unresolved_failures(component: str, limit: int = 50) -> list[ReliabilityFailureSignal]`¶

Return the component's failure signals not superseded by a later success.

`save_remediation_decision(decision: RemediationDecision) -> None`¶

`get_remediation_decision(decision_id: str) -> RemediationDecision | None`¶

`save_remediation_trace(trace: RemediationTrace) -> None`¶

`list_remediation_traces(component: str, limit: int = 50) -> list[RemediationTrace]`¶

`remediation_summary(component: str = '', limit: int = 50) -> dict[str, Any]`¶

Summarize recent metacognitive remediation traces.

`save_t3_candidate(candidate: T3Candidate) -> None`¶

`record_t3_candidate_event(candidate_id: str, component: str, action: str, from_status: str, to_status: str, actor: str = '', notes: str = '', run_id: str = '', pass_rate: float = 0.0, evidence_refs: list[str] | None = None, metadata: dict[str, Any] | None = None, error: str = '') -> T3CandidateEvent`¶

`list_t3_candidate_events(candidate_id: str, limit: int = 100) -> list[T3CandidateEvent]`¶

`save_t3_execution_artifact(candidate_id: str, component: str, run_id: str, artifact_type: str, title: str, payload: dict[str, Any] | None = None, path: str = '') -> T3ExecutionArtifact`¶

`list_t3_execution_artifacts(candidate_id: str = '', run_id: str = '', limit: int = 100) -> list[T3ExecutionArtifact]`¶

`create_t3_review_item(candidate_id: str, component: str, run_id: str, summary: str, artifact_ids: list[str], risk_tier: str = 'medium', domain: str = 'software', policy: dict[str, Any] | None = None) -> T3ReviewItem`¶

`list_t3_review_items(status: str = '', candidate_id: str = '', limit: int = 100) -> list[T3ReviewItem]`¶

`get_t3_review_item(review_id: str) -> T3ReviewItem | None`¶

`get_t3_review_detail(review_id: str) -> dict[str, Any] | None`¶

`export_t3_review_bundle(review_id: str) -> dict[str, Any] | None`¶

`save_t3_apply_result(result: T3ApplyResult, payload: dict[str, Any] | None = None) -> T3ExecutionArtifact`¶

`save_t3_pr_handoff(handoff: T3PRHandoff) -> T3ExecutionArtifact`¶

`update_t3_review_item(review_id: str, status: str, reviewer_id: str = '', second_reviewer_id: str = '', decision_notes: str = '') -> T3ReviewItem | None`¶

`get_t3_candidate(candidate_id: str) -> T3Candidate | None`¶

`list_t3_candidates(component: str = '', status: str | None = None, limit: int = 50) -> list[T3Candidate]`¶

`update_t3_candidate_status(candidate_id: str, status: str, approved_by: str = '', notes: str = '', run_id: str = '', pass_rate: float | None = None, evidence_refs: list[str] | None = None, error: str = '', actor: str = '') -> T3Candidate | None`¶

`enqueue_t3_candidate_from_summary(component: str) -> T3Candidate | None`¶

`enqueue_t3_candidate_from_theories(component: str, theory_ids: list[str], reason: str = 'benchmark_seed_theories') -> T3Candidate | None`¶

`start_t3_candidate(candidate_id: str, notes: str = '') -> T3Candidate | None`¶

`complete_t3_candidate(candidate_id: str, run_id: str, pass_rate: float, evidence_refs: list[str] | None = None, notes: str = '') -> T3Candidate | None`¶

`fail_t3_candidate(candidate_id: str, error: str, evidence_refs: list[str] | None = None, notes: str = '') -> T3Candidate | None`¶

`save_benchmark_spec(spec: BenchmarkSpec) -> None`¶

`get_benchmark_spec(component: str) -> BenchmarkSpec | None`¶

`record_t3_run(component: str, pass_rate: float, run_id: str) -> None`¶

`get_latest_t3_pass_rate(component: str) -> float | None`¶

`get_t3_run_count(component: str) -> int`¶

`save_extraction_run(run_id: str, component: str, source_t3_run_id: str, exemplars_created: int, exemplars_pending: int, fme_created: int, fme_promoted: int, recipes_created: int, profile_updated: bool) -> None`¶

`update_profile_stats(component: str) -> None`¶

Recompute exemplar/recipe/fme counts from live tables and write back.

`save_domain_principle(component: str, domain: str, category: str, principle: str, exemplar_source: str = '', example: str = '', applicability: str = '') -> str`¶

`list_domain_principles(component: str) -> list[dict]`¶

`list_profiles_with_scores() -> list[tuple[str, float | None, int, float | None]]`¶

Return (component_name, reliability_score, exemplar_count, last_t3_run_at)

`get_component_stats(component: str) -> dict | None`¶

Return all component_profiles columns as a plain dict (thread-safe).

`update_reliability_score(component: str, score: float) -> None`¶

Persist a freshly computed reliability_score (thread-safe).

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

Return t3_run records for a component, newest first.

`list_t3_runs_since(component: str, since_ts: float) -> list[dict]`¶

Return t3_run records since a UNIX timestamp, oldest first.

`upsert_competence(component: str, agent_id: str, task_type: str, passed: bool, cost_usd: float = 0.0, latency_ms: float = 0.0, failure_code: str = '') -> dict`¶

Bayesian update of competence matrix row. Returns updated row.

`get_competence_rows(component: str, task_type: str | None = None) -> list[dict]`¶

Return all competence_matrix rows for a component.

`seed_agents(agents: list[dict]) -> int`¶

INSERT OR IGNORE baseline agents. Returns count inserted.

`rename_agent(old_id: str, new_id: str) -> bool`¶

Move a registry row AND its competence history to a new agent_id.

`sync_baseline_agents(agents: list[dict]) -> list[str]`¶

Correct baseline rows that are ALREADY in the store.

`get_agent(agent_id: str) -> dict | None`¶

Return agent_registry row as dict, or None.

`list_agents(disabled: bool | None = None) -> list[dict]`¶

Return agent_registry rows. disabled=None returns all.

`ComponentT3Runner`¶

Orchestrates a T3 OODA cycle for a single component.

Constructor:

Parameter	Type	Default
`store`	`ReliabilityStore \\| None`	None
`bridge`	`ReliabilityBridge \\| None`	None
`scorer`	`ReliabilityScorer \\| None`	None
`_runner_factory`	`Callable[[LearningConfig], LearningRunner] \\| None`	None

Methods:

`run(config: T3RunConfig, dataset: Dataset, agent_fn: AgentFn, classifier: ComponentFailureClassifier | None = None) -> T3RunResult`¶

Run a full T3 OODA cycle and persist results.

`run_with_holdout(config: T3RunConfig, train_dataset: Dataset, holdout_dataset: Dataset, agent_fn: AgentFn, classifier: ComponentFailureClassifier | None = None, generalization_gap_threshold: float = _DEFAULT_GENERALIZATION_GAP_THRESHOLD) -> HeldOutEvalResult`¶

Held-out re-evaluation: train on A, then re-evaluate the AGENT on an unseen held-out B.

`run_from_spec(component: str, agent_fn: AgentFn, override_config: T3RunConfig | None = None) -> T3RunResult`¶

Run T3 using stored BenchmarkSpec for this component.

`CriterionScore`¶

Field	Type	Default
`criterion`	`str`	required
`weight`	`float`	required
`score`	`float`	required
`callable_ref`	`str`	required
`error`	`str \\| None`	None

`UatResult`¶

Field	Type	Default
`scenario_id`	`str`	required
`brick`	`str`	required
`composite_score`	`float`	required
`pass_threshold`	`float`	required
`passed`	`bool`	required
`criterion_scores`	`list[CriterionScore]`	required
`latency_ms`	`float`	required
`exception`	`str \\| None`	None
`raw_output_repr`	`str`	''

`UatGrader`¶

Run scenarios through their target brick and grade the output.

Constructor:

Parameter	Type	Default
`store`	`ScenarioStore`	required

Methods:

`run(scenario_id: str) -> UatResult | None`¶

`run_all(brick: str | None = None, domain: str | None = None, kind: str | None = None) -> list[UatResult]`¶

`pass_rate(brick: str | None = None, domain: str | None = None, kind: str = 'golden_path', review_status: str = 'accepted') -> float | None`¶

`ReliabilityVaultWriter`¶

Writes accepted CRF entries to an Obsidian-compatible vault directory.

Constructor:

Parameter	Type	Default
`vault_path`	`str \\| None`	None
`store`	`ReliabilityStore \\| None`	None

Methods:

`sync(component: str) -> dict[str, int]`¶

Sync all accepted entries for a component. Returns write counts.

`ReliabilityMCPBlock(AIBlock[MCPReliabilityInput, MCPReliabilityOutput, dict])`¶

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

Methods:

`infer(inp: MCPReliabilityInput) -> Result[MCPReliabilityOutput]`¶

`MCPReliabilityInput(BaseModel)`¶

Field	Type	Default
`op`	`str`	required
`component_name`	`str \\| None`	None
`request_id`	`str \\| None`	None
`query`	`str \\| None`	None
`problem`	`str \\| None`	None
`domain`	`str \\| None`	None
`limit`	`int`	20
`top_k`	`int`	5
`tags`	`list[str]`	Field(default_factory=list)
`score_type`	`str \\| None`	None
`since_iso`	`str \\| None`	None
`until_iso`	`str \\| None`	None
`exemplar_id`	`str \\| None`	None
`exemplar_content`	`dict[str, Any]`	Field(default_factory=dict)
`exemplar_type`	`str \\| None`	None
`recipe_id`	`str \\| None`	None
`recipe_input`	`dict[str, Any]`	Field(default_factory=dict)
`execution_mode`	`str`	''
`run_id`	`str \\| None`	None
`dry_run`	`bool`	False
`rating`	`float \\| None`	None
`rating_notes`	`str \\| None`	None
`observed_failure`	`str \\| None`	None
`theory_id`	`str \\| None`	None
`failure_code`	`str \\| None`	None
`failure_category`	`str \\| None`	None
`severity`	`str \\| None`	None
`task_type`	`str \\| None`	None
`task_complexity`	`str \\| None`	None
`budget_usd`	`float`	0.0
`agent_id`	`str \\| None`	None
`passed`	`bool \\| None`	None
`task_score`	`float \\| None`	None
`latency_ms`	`float \\| None`	None
`cost_usd`	`float`	0.0
`task_id`	`str \\| None`	None
`audit_depth`	`str \\| None`	None
`trend_window_days`	`int`	7
`knowledge_text`	`str \\| None`	None
`knowledge_source`	`str \\| None`	None
`export_format`	`str \\| None`	None
`scenario_id`	`str \\| None`	None
`scenario_kind`	`str \\| None`	None
`scenario_status`	`str \\| None`	None
`incident_id`	`str \\| None`	None
`group_by`	`str \\| None`	None
`since`	`str \\| None`	None
`reliability_status`	`str \\| None`	None
`risk_tier`	`str \\| None`	None
`rationale`	`str \\| None`	None
`policy_key`	`str \\| None`	None
`policy_value`	`Any`	None
`confirm`	`bool`	False
`remediation_payload`	`dict[str, Any]`	Field(default_factory=dict)
`signal_id`	`str \\| None`	None
`decision_id`	`str \\| None`	None
`outcome`	`str \\| None`	None
`seed_output_path`	`str \\| None`	None
`apply_import`	`bool`	False
`candidate_id`	`str \\| None`	None
`candidate_status`	`str \\| None`	None
`review_id`	`str \\| None`	None
`review_status`	`str \\| None`	None
`second_reviewer_id`	`str \\| None`	None
`workspace_root`	`str \\| None`	None
`confirm_text`	`str \\| None`	None
`pass_rate`	`float \\| None`	None
`evidence_refs`	`list[str]`	Field(default_factory=list)

`MCPReliabilityOutput(BaseModel)`¶

Field	Type	Default
`op`	`str`	''
`ok`	`bool`	False
`error`	`str`	''
`records`	`list[dict[str, Any]]`	Field(default_factory=list)
`data`	`dict[str, Any]`	Field(default_factory=dict)
`text`	`str`	''
`count`	`int`	0
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`str`	COMPLETION_STATE_QUALIFIED_DRAFT
`warning_card`	`dict[str, Any]`	Field(default_factory=dict)
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

Functions¶

`reliability_tracked(component_name: str | None = None, domain: str = 'general', tags: list[str] | None = None, benchmark_spec: BenchmarkSpec | None = None)`¶

Class decorator — sets __g6_reliability_* attributes at import time.

`unresolved_failures(component: str, db_path: str = '') -> list[ReliabilityFailureSignal]`¶

Authoritative per-component "unresolved gate failures" query.

Reliability¶

Overview¶

Public API¶

BenchmarkSpecFactory¶

create_for_component(component: str, domain: str = 'general', sample_size: int = 25, score_threshold: float = 0.8, max_cycles: int = 5) -> BenchmarkSpec¶

ComponentFailureClassifier¶

classify(result: TaskResult) -> FailureAnalysis¶

ExtractionResult¶

KnowledgeExtractor¶

extract(component: str, learning_result: LearningResult, source_run_id: str = '', quality_threshold: float = _ACCEPT_THRESHOLD) -> ExtractionResult¶

OrchestratorConfig¶

OrchestratorResult¶

T3Orchestrator¶

priority_order(config: OrchestratorConfig | None = None) -> list[str]¶

run_next(n: int = 1, config: OrchestratorConfig | None = None, **kwargs: Any) -> list[Any]¶

run_all(config: OrchestratorConfig | None = None, **kwargs: Any) -> OrchestratorResult¶

ReliabilityRegistrar¶

register_class(cls: type) -> ComponentProfile¶

scan_registry(component_registry: Any) -> list[ComponentProfile]¶

auto_enroll_production_ready(component_registry: Any) -> list[ComponentProfile]¶

ReliabilityBridge¶

collect_inputs(component: str) -> ScorerInputs¶

ResearchResult¶

ComponentResearcher¶

research(component_name: str, max_task_instances: int = 20, quality_threshold: float = 0.9, model: str = 'openrouter/anthropic/claude-opus-4.7', force: bool = False) -> ResearchResult¶

AgentSpec¶

AgentRecommendation¶

AgentRegistry¶

get(agent_id: str) -> AgentSpec | None¶

list_active() -> list[AgentSpec]¶

AgentRouter¶

recommend(component: str, task_type: str = 'general', task_complexity: str = 'medium', budget_usd: float = 0.0) -> AgentRecommendation¶

record_outcome(component: str, agent_id: str, task_type: str, passed: bool, task_score: float = 0.0, latency_ms: float = 0.0, cost_usd: float = 0.0, task_id: str = '', failure_code: str = '') -> dict¶

get_matrix(component: str) -> list[dict]¶

LoadReport¶

ScenarioLoader¶

discover(brick: str | None = None) -> LoadReport¶

compute_brick_size(brick: str) -> tuple[str, int, dict[str, int]]¶

RubricEntry(BaseModel)¶

InvocationSpec(BaseModel)¶

AdversarialVariant(BaseModel)¶

ScenarioFrontmatter(BaseModel)¶

ScenarioHarness(BaseModel)¶

ScenarioPair(BaseModel)¶

scenario_id() -> str¶

brick() -> str¶

ScenarioRecord¶

ScenarioStore¶

upsert(pair: ScenarioPair) -> None¶

mark_stale(scenario_id: str) -> int¶

get(scenario_id: str) -> ScenarioRecord | None¶

list_by_brick(brick: str) -> list[ScenarioRecord]¶

list_by_domain(domain: str, kind: str | None = None, review_status: str | None = None) -> list[ScenarioRecord]¶

list_all() -> list[ScenarioRecord]¶

count(brick: str | None = None) -> int¶

ReliabilityScore¶

ComponentProfile¶

FailureModeEntry¶

Exemplar¶

ReliabilityFailureSignal¶

Recipe¶

BenchmarkSpec¶

ScorerInputs¶

T3RunConfig¶

T3RunResult¶

ReliabilityScorer¶

compute(inputs: ScorerInputs) -> ReliabilityScore¶

KnowledgeSeedLoader¶

load_all(workspace_root: str | None = None) -> dict[str, int]¶

ReliabilityStore¶

close() -> None¶

save_profile(profile: ComponentProfile) -> None¶

get_profile(component: str) -> ComponentProfile | None¶

list_profiles() -> list[ComponentProfile]¶

get_maturity_tier(component: str) -> str | None¶

set_maturity_tier(component: str, tier: str) -> None¶

save_fme(fme: FailureModeEntry) -> None¶

get_fme(entry_id: str) -> FailureModeEntry | None¶

list_fmes(component: str, review_status: str | None = None) -> list[FailureModeEntry]¶

get_stale_entries(component: str, current_hash: str) -> dict[str, list]¶

`BenchmarkSpecFactory`¶

`create_for_component(component: str, domain: str = 'general', sample_size: int = 25, score_threshold: float = 0.8, max_cycles: int = 5) -> BenchmarkSpec`¶

`ComponentFailureClassifier`¶

`classify(result: TaskResult) -> FailureAnalysis`¶

`ExtractionResult`¶

`KnowledgeExtractor`¶

`extract(component: str, learning_result: LearningResult, source_run_id: str = '', quality_threshold: float = _ACCEPT_THRESHOLD) -> ExtractionResult`¶

`OrchestratorConfig`¶

`OrchestratorResult`¶

`T3Orchestrator`¶

`priority_order(config: OrchestratorConfig | None = None) -> list[str]`¶

`run_next(n: int = 1, config: OrchestratorConfig | None = None, **kwargs: Any) -> list[Any]`¶

`run_all(config: OrchestratorConfig | None = None, **kwargs: Any) -> OrchestratorResult`¶

`ReliabilityRegistrar`¶

`register_class(cls: type) -> ComponentProfile`¶

`scan_registry(component_registry: Any) -> list[ComponentProfile]`¶

`auto_enroll_production_ready(component_registry: Any) -> list[ComponentProfile]`¶

`ReliabilityBridge`¶

`collect_inputs(component: str) -> ScorerInputs`¶

`ResearchResult`¶

`ComponentResearcher`¶

`research(component_name: str, max_task_instances: int = 20, quality_threshold: float = 0.9, model: str = 'openrouter/anthropic/claude-opus-4.7', force: bool = False) -> ResearchResult`¶

`AgentSpec`¶

`AgentRecommendation`¶

`AgentRegistry`¶

`get(agent_id: str) -> AgentSpec | None`¶

`list_active() -> list[AgentSpec]`¶

`AgentRouter`¶

`recommend(component: str, task_type: str = 'general', task_complexity: str = 'medium', budget_usd: float = 0.0) -> AgentRecommendation`¶

`record_outcome(component: str, agent_id: str, task_type: str, passed: bool, task_score: float = 0.0, latency_ms: float = 0.0, cost_usd: float = 0.0, task_id: str = '', failure_code: str = '') -> dict`¶

`get_matrix(component: str) -> list[dict]`¶

`LoadReport`¶

`ScenarioLoader`¶

`discover(brick: str | None = None) -> LoadReport`¶

`compute_brick_size(brick: str) -> tuple[str, int, dict[str, int]]`¶

`RubricEntry(BaseModel)`¶

`InvocationSpec(BaseModel)`¶

`AdversarialVariant(BaseModel)`¶

`ScenarioFrontmatter(BaseModel)`¶

`ScenarioHarness(BaseModel)`¶

`ScenarioPair(BaseModel)`¶

`scenario_id() -> str`¶

`brick() -> str`¶

`ScenarioRecord`¶

`ScenarioStore`¶

`upsert(pair: ScenarioPair) -> None`¶

`mark_stale(scenario_id: str) -> int`¶

`get(scenario_id: str) -> ScenarioRecord | None`¶

`list_by_brick(brick: str) -> list[ScenarioRecord]`¶

`list_by_domain(domain: str, kind: str | None = None, review_status: str | None = None) -> list[ScenarioRecord]`¶

`list_all() -> list[ScenarioRecord]`¶

`count(brick: str | None = None) -> int`¶

`ReliabilityScore`¶

`ComponentProfile`¶

`FailureModeEntry`¶

`Exemplar`¶

`ReliabilityFailureSignal`¶

`Recipe`¶

`BenchmarkSpec`¶

`ScorerInputs`¶

`T3RunConfig`¶

`T3RunResult`¶

`ReliabilityScorer`¶

`compute(inputs: ScorerInputs) -> ReliabilityScore`¶

`KnowledgeSeedLoader`¶

`load_all(workspace_root: str | None = None) -> dict[str, int]`¶

`ReliabilityStore`¶

`close() -> None`¶

`save_profile(profile: ComponentProfile) -> None`¶

`get_profile(component: str) -> ComponentProfile | None`¶

`list_profiles() -> list[ComponentProfile]`¶

`get_maturity_tier(component: str) -> str | None`¶

`set_maturity_tier(component: str, tier: str) -> None`¶

`save_fme(fme: FailureModeEntry) -> None`¶

`get_fme(entry_id: str) -> FailureModeEntry | None`¶

`list_fmes(component: str, review_status: str | None = None) -> list[FailureModeEntry]`¶

`get_stale_entries(component: str, current_hash: str) -> dict[str, list]`¶

`save_exemplar(ex: Exemplar) -> None`¶

`get_exemplar(exemplar_id: str) -> Exemplar | None`¶

`list_exemplars(component: str, kind: str | None = None, review_status: str | None = None) -> list[Exemplar]`¶