Self Training¶

self_training — mvp.self_training

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

Overview¶

Maturity audit and production readiness pipeline for G6 components. Audits components against domain-grounded rubrics, tracks maturity scorecards across tiers T0 (stub) through T6 (production), manages component promotion and flagging, and orchestrates automated audit-diagnose-improve-reaudit training loops.

Includes a CompetenceOracle for Bayesian confidence assessment, a TrainingStateMachine for multi-cycle improvement, and domain grounding via external knowledge queries with caching.

Pilot reliability aid, not autonomous production repair

self_training is suitable for MVP and design-partner workflows where a person reviews the diagnosis, generated artifacts, and promotion decision. It audits components, explains failures, tracks maturity, and can attempt bounded improvement strategies, but it does not guarantee that arbitrary user workflows will be fixed automatically or become production-safe without review. For launch and customer-facing copy, describe it as a guided reliability and maturity workflow backed by scorecards and validation evidence, not as fully autonomous self-training.

When to use:

Auditing a component's production readiness against domain-specific rubrics
Running automated improvement loops to elevate a component from T0 to higher tiers
Tracking maturity scorecards and promotion decisions across the component registry

Example:

from mvp.self_training import SelfTrainingBlock, SelfTrainingInput
from mvp.core.registry import get_registry
from mvp.experience_loop import ExperienceLoopBlock

registry = get_registry()
experience_loop = ExperienceLoopBlock()

block = SelfTrainingBlock()
block.set_registry(registry, experience_loop)
# Minimal read-only probe: check whether a component exists in the registry.
result = block.infer(SelfTrainingInput(op="list_rubrics"))
# result.value → SelfTrainingOutput with rubrics catalogue / scorecards / promotions

Caveats and known limitations:

Requires set_registry() post-construction to bind registry and experience loop — block is inert without it
RubricRegistry loads rubrics from a directory at init — degrades to built-in defaults if directory is missing
TrainingStateMachine convergence criteria are internal — no user-visible config for loop termination
ScorecardStore uses SQLite — defaults to in-memory if no path is provided (scorecards lost on restart)
Domain grounding queries are cached but cache has no TTL — stale grounding data is not automatically refreshed

Works well with: learning_layer, align_evals, immune_system, invariants

Public API¶

`CompetenceReport`¶

Unified maturity assessment for a component.

Field	Type	Default
`component`	`str`	required
`composite_score`	`float`	required
`confidence`	`float`	required
`tier`	`int`	required
`tier_name`	`str`	required
`signals_used`	`list[str]`	required
`trend`	`str`	required
`ready_for_promotion`	`bool`	required
`timestamp`	`float`	0.0

`CompetenceOracle`¶

Merges maturity signals into a unified competence assessment.

Field	Type	Default
`audit_history`	`dict[str, list[dict]]`	field(default_factory=dict)

Methods:

`record_audit(component: str, tier: int, score: float) -> None`¶

Record an audit result for trend tracking.

`assess(component: str, audit_tier: int = 0, rubric_score: float = 0.0, experience_beta: tuple[float, float] | None = None, autonomy_phase: str | None = None) -> CompetenceReport`¶

Compute unified competence from available signals.

`batch_assess(audit_results: list[dict]) -> list[CompetenceReport]`¶

Assess all components from audit results.

`Diagnosis`¶

Actionable diagnosis of a component failure.

Field	Type	Default
`component`	`str`	required
`failure_class`	`FailureClass`	required
`description`	`str`	required
`suggested_fix`	`str`	required
`confidence`	`float`	required
`blocking`	`bool`	True

`GroundingCache`¶

Persistent JSON cache for domain grounding queries.

Constructor:

Parameter	Type	Default
`cache_dir`	`str \\| None`	None

Methods:

`load(domain: str) -> list[dict[str, Any]]`¶

`append(domain: str, entry: dict[str, Any]) -> None`¶

`has_query(domain: str, query: str) -> bool`¶

`get_cached_response(domain: str, query: str) -> str | None`¶

`get_cached_entry(domain: str, query: str) -> dict[str, Any] | None`¶

`get_cached_metadata(domain: str, query: str, stale_after_days: float = DEFAULT_STALE_CACHE_DAYS) -> dict[str, Any] | None`¶

`call_count(domain: str | None = None) -> int`¶

`all_domains() -> list[str]`¶

`ImprovementCycleResult`¶

Field	Type	Default
`enabled`	`bool`	required
`sensed`	`Any \\| None`	required
`patch`	`HarnessPatch \\| None`	required
`eval_report`	`PatchEvalReport \\| None`	required
`decision`	`PromotionDecision \\| None`	required
`applied`	`bool`	required
`rolled_back`	`bool`	required
`remeasured`	`Any \\| None`	required
`reliability_label`	`str`	required
`reasons`	`list[str]`	required
`rollback_method`	`str`	'extension_revert'

Methods:

`to_dict() -> dict`¶

`HarnessPatch`¶

A proposed harness change (spec §G HarnessPatchProposal, decision-relevant fields).

Field	Type	Default
`patch_id`	`str`	''
`agent_id`	`str`	''
`base_harness_version`	`str`	''
`patch_type`	`str`	''
`target_surfaces`	`list[str]`	field(default_factory=list)
`permission_change`	`str`	'none'
`safety_policy_change`	`str`	'none'
`data_sensitivity_change`	`str`	'none'
`external_action_impact`	`str`	'none'
`removes_human_review`	`bool`	False
`reversible`	`bool`	True
`rollback_method`	`str`	'extension_revert'
`treats_untrusted_content_as_instructions`	`bool`	False
`high_impact_domain`	`bool`	False
`prohibited`	`bool`	False

`PatchEvalReport`¶

The independent evaluation of a candidate patch (spec §H PromotionCriteria).

Field	Type	Default
`targeted_failure_fixed`	`bool`	False
`regression_pass_rate`	`float`	0.0
`safety_violations`	`int`	0
`injection_resilience_not_worse`	`bool`	True
`target_metric_improvement`	`float`	0.0
`evaluator_confidence`	`float`	0.0
`independent_eval`	`bool`	False

`PromotionDecision`¶

Field	Type	Default
`decision`	`str`	required
`risk_tier`	`str`	required
`reliability_label`	`str`	required
`requires_human`	`bool`	required
`auto_promotable`	`bool`	required
`reasons`	`list[str]`	field(default_factory=list)

Methods:

`to_dict() -> dict`¶

`ScorecardStore`¶

Append-only JSONL persistence for audit results and scorecard state.

Constructor:

Parameter	Type	Default
`store_dir`	`str \\| None`	None

Methods:

`append_audit(result: AuditResult) -> None`¶

Append an audit result to the log and update the scorecard.

`get_scorecard() -> ComponentScorecard`¶

`get_score(component: str) -> MaturityScore | None`¶

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

`flag(component: str, reason: str) -> None`¶

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

Read audit log, optionally filtered by component.

`RubricRegistry`¶

Registry of component rubrics, loadable from files or registered programmatically.

Methods:

`register(rubric: ComponentRubric) -> None`¶

`get(component: str, op: str | None = None) -> list[ComponentRubric]`¶

`list_components() -> list[str]`¶

`load_from_dir(rubric_dir: Path) -> int`¶

Load rubrics from JSON files in a directory. Returns count loaded.

`to_dict() -> dict[str, list[dict]]`¶

`RubricAssertion(BaseModel)`¶

A single testable assertion within a rubric.

Field	Type	Default
`field`	`str`	required
`check`	`Literal['exists', 'not_empty', 'gt', 'gte', 'lt', 'lte', 'eq', 'contains', 'type_is', 'length_gte', 'in_set']`	'exists'
`expected`	`Any`	None
`weight`	`float`	1.0
`description`	`str`	''

`ComponentRubric(BaseModel)`¶

Domain-specific evaluation rubric for a component.

Field	Type	Default
`component`	`str`	required
`op`	`str`	required
`description`	`str`	''
`assertions`	`list[RubricAssertion]`	Field(default_factory=list)
`domain_cluster`	`str`	''
`grounding_source`	`str`	''
`test_input`	`dict[str, Any]`	Field(default_factory=dict)
`provenance`	`RubricProvenance`	Field(default_factory=RubricProvenance)

`AuditTask(BaseModel)`¶

Input to a single component audit.

Field	Type	Default
`component`	`str`	required
`op`	`str`	'infer'
`params`	`dict[str, Any]`	Field(default_factory=dict)
`rubric`	`ComponentRubric \\| None`	None

`AssertionResult(BaseModel)`¶

Result of evaluating a single rubric assertion.

Field	Type	Default
`assertion`	`RubricAssertion`	required
`passed`	`bool`	required
`actual_value`	`Any`	None
`error`	`str`	''
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None

`AuditResult(BaseModel)`¶

Result of auditing a single component.

Field	Type	Default
`component`	`str`	required
`op`	`str`	required
`tier`	`MaturityTier`	required
`tier_name`	`str`	''
`passed_threshold`	`bool`	False
`raw_output`	`Any`	None
`assertion_results`	`list[AssertionResult]`	Field(default_factory=list)
`score`	`float`	0.0
`error`	`str`	''
`timestamp`	`float`	0.0
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None

`MaturityScore(BaseModel)`¶

Aggregate maturity score for a component across all audits.

Field	Type	Default
`component`	`str`	required
`current_tier`	`MaturityTier`	required
`tier_name`	`str`	''
`audit_count`	`int`	0
`pass_rate`	`float`	0.0
`avg_score`	`float`	0.0
`last_audit_timestamp`	`float`	0.0
`flagged`	`bool`	False
`flag_reason`	`str`	''

`ComponentScorecard(BaseModel)`¶

Full scorecard across all components.

Field	Type	Default
`scores`	`dict[str, MaturityScore]`	Field(default_factory=dict)
`total_components`	`int`	0
`above_threshold`	`int`	0
`below_threshold`	`int`	0
`not_audited`	`int`	0
`timestamp`	`float`	0.0

`SelfTrainingInput(BaseModel)`¶

Field	Type	Default
`op`	`Literal['audit_component', 'audit_all', 'get_scorecard', 'promote', 'flag', 'get_rubric', 'list_rubrics', 'train_component', 'train_all_below', 'get_learning_plan', 'get_competence', 'get_loop_status', 'run_learning_loop', 'run_unified_loop', 'improvement_cycle', 'get_progress', 'source_health']`	'audit_component'
`component`	`str`	''
`op_name`	`str`	'infer'
`params`	`dict[str, Any]`	Field(default_factory=dict)
`tier_minimum`	`int`	5
`flag_reason`	`str`	''
`new_tier`	`MaturityTier`	5
`max_cycles`	`int`	3
`target_components`	`list[str]`	Field(default_factory=list)
`run_mode`	`str`	'beta'
`reviewer_signature`	`str`	''

`SelfTrainingOutput(BaseModel)`¶

Field	Type	Default
`op`	`str`	required
`success`	`bool`	True
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'qualified-draft'
`warning_card`	`dict[str, Any] \\| None`	None
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str`	''
`run_id`	`str`	''
`audit_result`	`AuditResult \\| None`	None
`scorecard`	`ComponentScorecard \\| None`	None
`rubric`	`ComponentRubric \\| None`	None
`rubrics`	`list[str]`	Field(default_factory=list)
`learning_plan`	`list[dict[str, Any]]`	Field(default_factory=list)
`competence`	`dict[str, Any]`	Field(default_factory=dict)
`loop_status`	`dict[str, Any]`	Field(default_factory=dict)
`progress`	`dict[str, Any]`	Field(default_factory=dict)
`training_summary`	`list[dict[str, Any]]`	Field(default_factory=list)
`data`	`dict[str, Any]`	Field(default_factory=dict)
`message`	`str`	''
`error`	`str`	''
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None
`agentic_evidence`	`dict \\| None`	None

`SelfTrainingBlock(AIBlock[SelfTrainingInput, SelfTrainingOutput, None])`¶

Audit components against maturity rubrics, track scores,

Field	Type	Default
`name`	`str`	'self_training'
`store_dir`	`str \\| None`	None
`rubric_dir`	`str \\| None`	None
`reliability_db_path`	`str \\| None`	None

Methods:

`set_registry(registry: Any, experience_loop: Any = None) -> None`¶

Wire the improve_fn into the training state machine.

`infer(data: SelfTrainingInput) -> Result[SelfTrainingOutput]`¶

`run_improvement_cycle_op(sense_fn: Any = None, propose_fn: Any = None, evaluate_fn: Any = None, apply_fn: Any = None, remeasure_fn: Any = None, regressed_fn: Any = None, snapshot_fn: Any = None, enabled: bool | None = None, apply_enabled: bool | None = None, sense_context: SelfTrainingInput | None = None) -> Any`¶

Run one self-improvement cycle (SEED-G7).

`Phase(str, Enum)`¶

`TrainingStateMachine`¶

Orchestrates the audit -> diagnose -> improve -> re-audit loop.

Field	Type	Default
`state`	`LoopState`	field(default_factory=LoopState)
`improve_fn`	`Callable[[str, str, str], bool] \\| None`	None

Methods:

`start(targets: list[str] | None = None, max_iterations: int = 10) -> dict[str, Any]`¶

Initialize the training loop.

`step(audit_results: list[dict]) -> dict[str, Any]`¶

Execute one full iteration of the loop.

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

Return current loop status.

`get_learning_plan(audit_results: list[dict]) -> list[dict[str, Any]]`¶

Recommend which components to train next (worst-first).

`UnifiedCycleRecord`¶

Record of one complete 7-step cycle.

Field	Type	Default
`cycle`	`int`	required
`phase_log`	`list[str]`	field(default_factory=list)
`audit_results`	`list[dict]`	field(default_factory=list)
`diagnoses`	`list[dict]`	field(default_factory=list)
`improvements`	`list[dict]`	field(default_factory=list)
`observations`	`dict`	field(default_factory=dict)
`artifacts_produced`	`int`	0
`components_improved`	`int`	0
`duration_sec`	`float`	0.0

`UnifiedLoopResult`¶

Result of the full unified training loop.

Field	Type	Default
`cycles`	`list[UnifiedCycleRecord]`	field(default_factory=list)
`total_cycles`	`int`	0
`total_improved`	`int`	0
`convergence_reason`	`str`	''
`final_status`	`str`	'idle'

`UnifiedTrainingLoop`¶

Orchestrates the 7-step self-training loop across all subsystems.

Constructor:

Parameter	Type	Default
`state_machine`	`TrainingStateMachine`	required
`audit_fn`	`Any \\| None`	None
`learning_runner`	`Any \\| None`	None
`experience_loop`	`Any \\| None`	None
`registry`	`Any \\| None`	None

Methods:

`run(targets: list[str], max_cycles: int = 5, max_stagnant: int = 2) -> Result[UnifiedLoopResult]`¶

Execute the unified 7-step loop until convergence or max_cycles.

Functions¶

`diagnose(component: str, error: str, tier: int, score: float) -> Diagnosis`¶

Classify a component's failure and suggest remediation.

`diagnose_batch(results: list[dict]) -> list[Diagnosis]`¶

Diagnose all audit results at once.

`ground_domain_query(query: str, domain: str, cache: GroundingCache | None = None, model: str = _PERPLEXITY_MODEL, max_tokens: int = 4000) -> tuple[bool, str]`¶

Query Perplexity for domain-specific factual grounding.

`batch_ground_domain(queries: list[str], domain: str, cache: GroundingCache | None = None, model: str = _PERPLEXITY_MODEL) -> list[tuple[bool, str]]`¶

Run multiple grounding queries for a domain, skipping cached ones.

`run_improvement_cycle(sense_fn: SenseFn, propose_fn: ProposeFn, evaluate_fn: EvaluateFn, apply_fn: ApplyFn | None = None, remeasure_fn: RemeasureFn | None = None, regressed_fn: RegressedFn | None = None, snapshot_fn: SnapshotFn | None = None, enabled: bool | None = None, apply_enabled: bool | None = None, autonomous_low_risk: bool = True) -> ImprovementCycleResult`¶

`make_codex_harness_proposer(caller: Any | None = None, agent_id: str = '', base_harness_version: str = '') -> ProposeFn`¶

`make_no_eval_report() -> PatchEvalReport`¶

`classify_patch_risk(patch: HarnessPatch) -> str`¶

§I risk tiering: final = max(surface base tiers, effect-flag floors). Fail-closed:

`govern_patch_promotion(patch: HarnessPatch, eval_report: PatchEvalReport, autonomous_low_risk: bool = True) -> PromotionDecision`¶

Decide the promotion action for patch. Fail-closed, DOWN-only: hard safety rules and a

`evaluate_assertion(assertion: RubricAssertion, output: Any) -> AssertionResult`¶

Evaluate a single assertion against component output.

`evaluate_rubric(rubric: ComponentRubric, output: Any) -> tuple[list[AssertionResult], float]`¶

Evaluate all assertions in a rubric. Returns (results, weighted_score 0-1).

Self Training¶

Overview¶

Public API¶

CompetenceReport¶

CompetenceOracle¶

record_audit(component: str, tier: int, score: float) -> None¶

assess(component: str, audit_tier: int = 0, rubric_score: float = 0.0, experience_beta: tuple[float, float] | None = None, autonomy_phase: str | None = None) -> CompetenceReport¶

batch_assess(audit_results: list[dict]) -> list[CompetenceReport]¶

Diagnosis¶

GroundingCache¶

load(domain: str) -> list[dict[str, Any]]¶

append(domain: str, entry: dict[str, Any]) -> None¶

has_query(domain: str, query: str) -> bool¶

get_cached_response(domain: str, query: str) -> str | None¶

get_cached_entry(domain: str, query: str) -> dict[str, Any] | None¶

get_cached_metadata(domain: str, query: str, stale_after_days: float = DEFAULT_STALE_CACHE_DAYS) -> dict[str, Any] | None¶

call_count(domain: str | None = None) -> int¶

all_domains() -> list[str]¶

ImprovementCycleResult¶

to_dict() -> dict¶

HarnessPatch¶

PatchEvalReport¶

PromotionDecision¶

to_dict() -> dict¶

ScorecardStore¶

append_audit(result: AuditResult) -> None¶

get_scorecard() -> ComponentScorecard¶

get_score(component: str) -> MaturityScore | None¶

promote(component: str, tier: int) -> None¶

flag(component: str, reason: str) -> None¶

audit_history(component: str | None = None) -> list[dict[str, Any]]¶

RubricRegistry¶

register(rubric: ComponentRubric) -> None¶

get(component: str, op: str | None = None) -> list[ComponentRubric]¶

list_components() -> list[str]¶

load_from_dir(rubric_dir: Path) -> int¶

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

RubricAssertion(BaseModel)¶

ComponentRubric(BaseModel)¶

AuditTask(BaseModel)¶

AssertionResult(BaseModel)¶

AuditResult(BaseModel)¶

MaturityScore(BaseModel)¶

ComponentScorecard(BaseModel)¶

SelfTrainingInput(BaseModel)¶

SelfTrainingOutput(BaseModel)¶

SelfTrainingBlock(AIBlock[SelfTrainingInput, SelfTrainingOutput, None])¶

set_registry(registry: Any, experience_loop: Any = None) -> None¶

infer(data: SelfTrainingInput) -> Result[SelfTrainingOutput]¶

Phase(str, Enum)¶

TrainingStateMachine¶

start(targets: list[str] | None = None, max_iterations: int = 10) -> dict[str, Any]¶

step(audit_results: list[dict]) -> dict[str, Any]¶

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

get_learning_plan(audit_results: list[dict]) -> list[dict[str, Any]]¶

UnifiedCycleRecord¶

UnifiedLoopResult¶

UnifiedTrainingLoop¶

run(targets: list[str], max_cycles: int = 5, max_stagnant: int = 2) -> Result[UnifiedLoopResult]¶

Functions¶

diagnose(component: str, error: str, tier: int, score: float) -> Diagnosis¶

diagnose_batch(results: list[dict]) -> list[Diagnosis]¶

ground_domain_query(query: str, domain: str, cache: GroundingCache | None = None, model: str = _PERPLEXITY_MODEL, max_tokens: int = 4000) -> tuple[bool, str]¶

batch_ground_domain(queries: list[str], domain: str, cache: GroundingCache | None = None, model: str = _PERPLEXITY_MODEL) -> list[tuple[bool, str]]¶

make_codex_harness_proposer(caller: Any | None = None, agent_id: str = '', base_harness_version: str = '') -> ProposeFn¶

make_no_eval_report() -> PatchEvalReport¶

classify_patch_risk(patch: HarnessPatch) -> str¶

govern_patch_promotion(patch: HarnessPatch, eval_report: PatchEvalReport, autonomous_low_risk: bool = True) -> PromotionDecision¶

evaluate_assertion(assertion: RubricAssertion, output: Any) -> AssertionResult¶

evaluate_rubric(rubric: ComponentRubric, output: Any) -> tuple[list[AssertionResult], float]¶

`CompetenceReport`¶

`CompetenceOracle`¶

`record_audit(component: str, tier: int, score: float) -> None`¶

`assess(component: str, audit_tier: int = 0, rubric_score: float = 0.0, experience_beta: tuple[float, float] | None = None, autonomy_phase: str | None = None) -> CompetenceReport`¶

`batch_assess(audit_results: list[dict]) -> list[CompetenceReport]`¶

`Diagnosis`¶

`GroundingCache`¶

`load(domain: str) -> list[dict[str, Any]]`¶

`append(domain: str, entry: dict[str, Any]) -> None`¶

`has_query(domain: str, query: str) -> bool`¶

`get_cached_response(domain: str, query: str) -> str | None`¶

`get_cached_entry(domain: str, query: str) -> dict[str, Any] | None`¶

`get_cached_metadata(domain: str, query: str, stale_after_days: float = DEFAULT_STALE_CACHE_DAYS) -> dict[str, Any] | None`¶

`call_count(domain: str | None = None) -> int`¶

`all_domains() -> list[str]`¶

`ImprovementCycleResult`¶

`to_dict() -> dict`¶

`HarnessPatch`¶

`PatchEvalReport`¶

`PromotionDecision`¶

`to_dict() -> dict`¶

`ScorecardStore`¶

`append_audit(result: AuditResult) -> None`¶

`get_scorecard() -> ComponentScorecard`¶

`get_score(component: str) -> MaturityScore | None`¶

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

`flag(component: str, reason: str) -> None`¶

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

`RubricRegistry`¶

`register(rubric: ComponentRubric) -> None`¶

`get(component: str, op: str | None = None) -> list[ComponentRubric]`¶

`list_components() -> list[str]`¶

`load_from_dir(rubric_dir: Path) -> int`¶

`to_dict() -> dict[str, list[dict]]`¶

`RubricAssertion(BaseModel)`¶

`ComponentRubric(BaseModel)`¶

`AuditTask(BaseModel)`¶

`AssertionResult(BaseModel)`¶

`AuditResult(BaseModel)`¶

`MaturityScore(BaseModel)`¶

`ComponentScorecard(BaseModel)`¶

`SelfTrainingInput(BaseModel)`¶

`SelfTrainingOutput(BaseModel)`¶

`SelfTrainingBlock(AIBlock[SelfTrainingInput, SelfTrainingOutput, None])`¶

`set_registry(registry: Any, experience_loop: Any = None) -> None`¶

`infer(data: SelfTrainingInput) -> Result[SelfTrainingOutput]`¶

`Phase(str, Enum)`¶

`TrainingStateMachine`¶

`start(targets: list[str] | None = None, max_iterations: int = 10) -> dict[str, Any]`¶

`step(audit_results: list[dict]) -> dict[str, Any]`¶

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

`get_learning_plan(audit_results: list[dict]) -> list[dict[str, Any]]`¶

`UnifiedCycleRecord`¶

`UnifiedLoopResult`¶

`UnifiedTrainingLoop`¶

`run(targets: list[str], max_cycles: int = 5, max_stagnant: int = 2) -> Result[UnifiedLoopResult]`¶

`diagnose(component: str, error: str, tier: int, score: float) -> Diagnosis`¶

`diagnose_batch(results: list[dict]) -> list[Diagnosis]`¶

`ground_domain_query(query: str, domain: str, cache: GroundingCache | None = None, model: str = _PERPLEXITY_MODEL, max_tokens: int = 4000) -> tuple[bool, str]`¶

`batch_ground_domain(queries: list[str], domain: str, cache: GroundingCache | None = None, model: str = _PERPLEXITY_MODEL) -> list[tuple[bool, str]]`¶

`make_codex_harness_proposer(caller: Any | None = None, agent_id: str = '', base_harness_version: str = '') -> ProposeFn`¶

`make_no_eval_report() -> PatchEvalReport`¶

`classify_patch_risk(patch: HarnessPatch) -> str`¶

`govern_patch_promotion(patch: HarnessPatch, eval_report: PatchEvalReport, autonomous_low_risk: bool = True) -> PromotionDecision`¶

`evaluate_assertion(assertion: RubricAssertion, output: Any) -> AssertionResult`¶

`evaluate_rubric(rubric: ComponentRubric, output: Any) -> tuple[list[AssertionResult], float]`¶