Opt Meta¶

opt_meta — mvp.opt_meta

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

Overview¶

Pareto coordinator for the self-optimisation loop. Combines speed/cost/quality reports, computes a Pareto frontier, selects intervention plans via multi-armed bandit (MAB), and optionally auto-applies interventions through a CSF safety gate. Learning from outcomes feeds back into the MAB for continuous improvement. The opt_meta coordinator itself is deterministic and makes no LLM call; sub-optimizers it coordinates may use model-backed analysis.

Example:

from mvp.opt_meta import OptMetaBlock, OptMetaInput

block = OptMetaBlock()
result = block.infer(OptMetaInput(
    op="coordinate",
    trace={
        "trace_id": "trace-1",
        "workflow_name": "demo_pipeline",
        "timestamp": "2026-05-21T00:00:00Z",
        "steps": [
            {"step_index": 0, "component_name": "llm_router",
             "operation": "infer", "confidence": 0.55, "latency_ms": 4200.0},
        ],
    },
))
# result.value -> OptMetaOutput with Pareto frontier + selected plan

Public API¶

`OptMetaBlock(AIBlock[OptMetaInput, OptMetaOutput, dict])`¶

Meta-optimizer: Pareto coordinator, MAB selection, CSF-gated auto-apply.

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

Methods:

`infer(data: OptMetaInput) -> Result[OptMetaOutput]`¶

`OptMetaInput(BaseModel)`¶

Field	Type	Default
`op`	`Literal['coordinate', 'pareto_analyze', 'select_plan', 'apply_plan', 'record_outcome', 'get_intervention_stats', 'reset_learning', 'list_patterns', 'capabilities']`	'coordinate'
`trace`	`dict \\| None`	None
`reports`	`list[dict] \\| None`	None
`interventions`	`list[dict] \\| None`	None
`outcome`	`dict \\| None`	None
`record_id`	`str`	''
`auto_apply`	`bool`	False
`csf_fail_closed`	`bool`	True
`repo_path`	`str`	''
`request_id`	`str \\| None`	None
`task_id`	`str \\| None`	None
`run_id`	`str \\| None`	None
`weights`	`dict[str, float]`	Field(default_factory=lambda: {'speed': 1 / 3, 'cost': 1 / 3, 'quality': 1 / 3})

`OptMetaOutput(BaseModel)`¶

Field	Type	Default
`op`	`str`	required
`reports`	`list[dict] \\| None`	None
`pareto_frontier`	`list[dict] \\| None`	None
`selected_plan`	`list[dict] \\| None`	None
`auto_applied`	`list[dict] \\| None`	None
`arm_stats`	`dict[str, Any] \\| None`	None
`eigen_behaviors`	`dict[str, Any] \\| None`	None
`mutated_trace`	`WorkflowTrace \\| None`	None
`patterns_payload`	`dict[str, Any] \\| None`	None
`capabilities`	`dict[str, Any] \\| None`	None
`success`	`bool`	True
`error`	`str`	''
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None
`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 \\| None`	None
`task_id`	`str \\| None`	None
`run_id`	`str \\| None`	None
`csf_gate_active`	`bool`	False

Launch caveat¶

Launch positioning

opt_meta is credible as an optimisation recommendation and bounded application component: it can coordinate speed/cost/quality analyzers, choose Pareto-valid interventions, learn from recorded outcomes, and mutate workflow traces behind safety gates. Do not present it as proof of fully autonomous production rewriting or guaranteed end-to-end reliability improvement without a live case study that shows before/after results on a real workflow.

Customer-facing claims should say "G6 recommends and can safely apply optimisation plans, with recorded validation evidence" rather than "the system autonomously improves itself in production."

Production notes¶

Operational caveats

opt_meta recommendations are advisory unless auto_apply=True; most users should review selected interventions before applying them. When auto_apply=True, actions still depend on the CSF gate and rollback support. If safety checks are unavailable and csf_fail_closed=True, actions may be blocked.

Live model pricing is best-effort. If OpenRouter refresh fails, opt_cost falls back to static or caller-supplied pricing and reports the issue in report.pricing.warning. Production API and UI callers should surface that warning, for example through the top-level warnings or pricing_warning fields returned by the MCP optimization tools.

Cost projections are estimates, not invoices. Provider routing, token accounting, and pricing can differ from the projection. Recommendation quality also depends on trace completeness: missing cost, token, quality, dependency, or outcome metadata will reduce confidence.

Learned arm selection depends on accumulated local outcome history. Early runs may choose mostly from Pareto geometry rather than proven workload-specific outcomes.

Operations¶

Op	Description
`coordinate`	Full loop: analyze trace on all 3 axes, Pareto + MAB select, optional auto-apply
`pareto_analyze`	Compute Pareto frontier from speed/cost/quality reports
`select_plan`	MAB-guided selection of the best intervention plan
`apply_plan`	Apply interventions with CSF gate and optional rollback
`record_outcome`	Record intervention outcome for MAB learning
`get_intervention_stats`	Retrieve MAB arm statistics and eigen-behaviors
`reset_learning`	Reset MAB state and learning history
`capabilities`	Offline-safe capability discovery for ops, sub-optimizer reachability, hazards, persistence status, CSF status, maturity, and gate scope

Gateway Tools¶

Gateway Tool	Delegates To
`optimize_pipeline`	`coordinate`
`optimization_stats`	`get_intervention_stats`
`optimization_apply`	`apply_plan` with `auto_apply=True`, enabling the CSF gate on that MCP path

Opt Meta¶

Overview¶

Public API¶

OptMetaBlock(AIBlock[OptMetaInput, OptMetaOutput, dict])¶

infer(data: OptMetaInput) -> Result[OptMetaOutput]¶

OptMetaInput(BaseModel)¶

OptMetaOutput(BaseModel)¶