Solver¶
Solver component — 15-step problem solving orchestrator.
Cluster: Goal & Planning | Type: component | MCP Tools: 38
Overview¶
15-step problem-solving orchestrator that accepts a canonical 20-field SolverInput and executes a structured solution pipeline with adaptive depth (4/8/15 steps), per-step LLM backend overrides, resource bounds, breakpoints, guardrails, checkpoints, launch-time step overrides, retry configs, presets, and beta/production run mode. Produces a structured SolverOutput with step-level traces, completion_state (verified, qualified-draft, or blocked-escalated), degradation evidence, integrity verification, and self-healing via solver_healer.
Production mode is fail-closed while the block contract verification method remains tier1_review_pending: run_mode="production" requires a non-blank reviewer_signature, otherwise the run is refused before engine construction. The read-only op="info" path reports maturity, verification method, production gate status, optional subsystem availability, planner suppression state, presets, and completion-state semantics without executing solver steps.
Agentic planner suppression is surfaced honestly. G6_DISABLE_LLM, G6_SOLVER_AGENTIC_RUNTIME, and G6_LLM_BACKEND control whether the grounded planner is constructed; offline, kill-switch, or construction-failure suppression records agentic_evidence.suppression, marks the output degraded, and keeps the top-level state at qualified-draft instead of overclaiming verified.
When to use:
- Solving open-ended analytical or engineering problems through a systematic multi-step reasoning process
- Running a benchmark evaluation harness that requires controlled, reproducible problem-solving steps
- Embedding a structured solver into the hyperdistillation pipeline to generate high-quality training traces
Example:
from mvp.solver import SolverBlock, SolverInput
block = SolverBlock(name="solver")
result = block.infer(SolverInput(
goal="Design a caching strategy for a high-traffic API",
context="Redis available, 10k req/s peak load",
subtasks=["Identify hot paths", "Choose eviction policy", "Estimate memory budget"],
))
# result.ok -> True; result.value -> SolverOutput with steps, answer, completion_state
Works well with: recursive_architect, evoskill, hyperdistillation
Public API¶
AgentPlan¶
Planned substeps for agentic execution.
| Field | Type | Default |
|---|---|---|
substeps | list[dict] | required |
reasoning | str | required |
AgentProtocol¶
Plan -> Execute -> Verify -> Learn loop for agent-mode steps.
Constructor:
| Parameter | Type | Default |
|---|---|---|
voting_pool | VotingPool | required |
red_flag_detector | RedFlagDetector | required |
registry | - | None |
Methods:
execute(context: StepContext) -> Result[dict]¶
Full agentic loop: plan, execute substeps, verify, learn.
PersistenceConfig¶
Backup configuration.
| Field | Type | Default |
|---|---|---|
backup_enabled | bool | True |
backup_destination | str | '' |
max_backup_size_mb | float | 500.0 |
BackupManager¶
Creates timestamped local backups of a project directory.
Constructor:
| Parameter | Type | Default |
|---|---|---|
project_dir | str | required |
config | PersistenceConfig | required |
Methods:
backup_now() -> Result[str]¶
Copy project_dir to backup_destination/{timestamp}/.
check_size() -> Result[dict]¶
Check total backup size. Warn if > max_backup_size_mb.
schedule_backup(interval_minutes: int, task_status = None) -> None¶
Background trio task that backs up on a schedule.
SolverBlock(AIBlock[SolverInput, SolverOutput, dict])¶
15-step problem solving orchestrator. Deep interface: just call infer().
| Field | Type | Default |
|---|---|---|
name | str | 'solver' |
resource_bounds | ResourceBounds \| None | None |
usage | ResourceUsage | field(default_factory=ResourceUsage) |
internal_project_dir | str \| None | None |
Methods:
infer(data: SolverInput) -> Result[SolverOutput]¶
CritiqueSolver¶
Two-pass critique loop for designated solver steps.
Constructor:
| Parameter | Type | Default |
|---|---|---|
prompt_engine | PromptEngine | required |
Methods:
should_critique(step_num: int, complexity: str) -> bool¶
execute_with_critique(step_num: int, executor: StepExecutor, project_state: dict, complexity: str) -> Result[dict]¶
DeepAgentProtocol¶
Real agentic loop: explore -> plan -> execute (CSF-gated) -> verify.
Constructor:
| Parameter | Type | Default |
|---|---|---|
registry | - | None |
prompt_engine | - | None |
exemplar_verifier | - | None |
csf_block | - | None |
Methods:
execute(context: StepContext) -> Result[dict]¶
ExemplarVerificationResult¶
Result of exemplar-based verification.
| Field | Type | Default |
|---|---|---|
has_exemplars | bool | required |
score | float | required |
details | str | required |
exemplar_count | int | 0 |
ExemplarVerifier¶
Verify answers by comparing against similar past successful solutions.
Constructor:
| Parameter | Type | Default |
|---|---|---|
registry | - | None |
case_bank | - | None |
retriever | - | None |
Methods:
available() -> bool¶
verify(goal: str, context: str, current_answer: str, min_similarity: float = 0.5, top_k: int = 3) -> ExemplarVerificationResult¶
FailureCollector¶
Collects solver failure patterns for downstream evolution.
Constructor:
| Parameter | Type | Default |
|---|---|---|
path | str \| None | None |
Methods:
record_failure(goal: str, step_traces: list[dict], failure_reason: str) -> None¶
get_recent_failures(n: int = 20) -> list[dict]¶
get_failure_patterns() -> dict[str, int]¶
Group failures by reason and return counts.
count() -> int¶
GitConfig¶
Git tracking configuration.
| Field | Type | Default |
|---|---|---|
enable_tracking | bool | False |
remote_url | str | '' |
branch_prefix | str | 'solver' |
GitTracker¶
Git-based step tracking for solver projects.
Constructor:
| Parameter | Type | Default |
|---|---|---|
project_dir | str | required |
config | GitConfig | required |
Methods:
enabled() -> bool¶
init_project(project_id: str) -> Result[str]¶
Initialize git repo, create branch, add .gitignore, optionally set remote.
commit_step(step_number: int, step_name: str, status: str) -> Result[str]¶
Commit current state with step metadata in message.
push() -> Result[str]¶
Push to remote if configured.
revert_to_step(step_number: int) -> Result[str]¶
Create a new branch from the step's commit. Never rewrites history.
get_step_sha(step_number: int) -> Result[str]¶
Find the commit SHA for a given step number.
IntegrityReport¶
Result of a full integrity check.
| Field | Type | Default |
|---|---|---|
sqlite_ok | bool | required |
jsonl_ok | bool | required |
git_ok | bool | required |
issues | list[str] | required |
repaired | bool | required |
DataIntegrityManager¶
Checks SQLite, JSONL, and git integrity for a solver project.
Constructor:
| Parameter | Type | Default |
|---|---|---|
project_dir | str | required |
Methods:
check_sqlite() -> Result[list[str]]¶
PRAGMA integrity_check, table existence, JSON validity.
check_jsonl() -> Result[list[str]]¶
Line-by-line parse, required fields, truncation detection.
check_git() -> Result[list[str]]¶
Git fsck via GitPython (optional).
check_all() -> IntegrityReport¶
Run all integrity checks and return combined report.
repair() -> Result[IntegrityReport]¶
Backup first, then rebuild SQLite from JSONL (JSONL is source of truth).
LearningMemoryStore¶
Cross-session learning persistence.
Constructor:
| Parameter | Type | Default |
|---|---|---|
db_path | str \| None | None |
legacy_path | str \| None | None |
Methods:
purge_retained_goal_text() -> int¶
Digest any goal text written before this store started digesting.
record_session(goal: str, success: bool, tokens: int, artifacts_used: int = 0, distilled_steps: int = 0, complexity: str = 'unknown', strategy: str = 'baseline') -> str¶
get_session_count() -> int¶
get_recent_sessions(n: int = 10) -> list[dict]¶
update_mastery(signature: str, task_type: str, success: bool, tokens: int) -> None¶
get_mastery_for_class(signature: str) -> dict | None¶
get_mastered_signatures(min_confidence: float = 0.8) -> list[str]¶
update_strategy_effectiveness(strategy: str, problem_class: str, success: bool, improvement: float = 0.0) -> None¶
get_strategy_stats(strategy: str, problem_class: str) -> dict | None¶
get_all_strategy_stats() -> list[dict]¶
record_calibration(model_id: str, language: str, max_tier_passed: int, recommended_grade: str, tier_results: dict | None = None) -> int¶
get_latest_calibration(model_id: str) -> dict | None¶
get_calibration_history(model_id: str, limit: int = 10) -> list[dict]¶
update_framework_success(model_id: str, framework: str, success: bool, confidence: float = 0.0) -> None¶
get_framework_success(model_id: str, framework: str) -> dict | None¶
get_all_framework_success(model_id: str) -> list[dict]¶
get_cross_session_insights(goal: str, context: str = '') -> dict¶
Aggregate cross-session learning state for pre_solve enrichment.
close() -> None¶
LLMTaskClassifier¶
Classify task complexity using LLM-based virtualization assessment.
Constructor:
| Parameter | Type | Default |
|---|---|---|
registry | - | required |
prompt_engine | PromptEngine \| None | None |
Methods:
classify(goal: str, context: str | None = None) -> TaskComplexity¶
LLMQualityChecker¶
Two-phase output quality check: cheap heuristics first, LLM second.
Constructor:
| Parameter | Type | Default |
|---|---|---|
registry | - | None |
red_flag_detector | RedFlagDetector \| None | None |
Methods:
check(text: str, goal: str, step_name: str, step_number: int = 0) -> dict¶
MetaLearner¶
Thompson-sampling meta-learner over learning strategies.
Constructor:
| Parameter | Type | Default |
|---|---|---|
memory_store | LearningMemoryStore | required |
Methods:
select_strategy(goal: str, context: str, available: list[str], problem_class: str = 'general') -> str¶
Sample from Beta posteriors and pick the strategy with highest draw.
record_outcome(strategy: str, problem_class: str, success: bool, improvement: float = 0.0) -> None¶
Record outcome and update strategy effectiveness.
get_strategy_report() -> dict¶
Return a summary of strategy effectiveness across all classes.
PromptEngine¶
Ports the prototype_2025 G6ProblemSolver reasoning patterns.
Constructor:
| Parameter | Type | Default |
|---|---|---|
registry | - | required |
Methods:
get_constraints(goal: str, context: str) -> str¶
extract_first_principles(goal: str, constraints: str) -> str¶
extract_heuristics(goal: str, constraints: str) -> str¶
virtualize(goal: str, constraints: str, context: str) -> str¶
satisficing_strategy(goal: str, constraints: str, assessment: str) -> str¶
generate_goal_tree_params(goal: str, constraints: str, assessment: str, strategy: str) -> dict¶
build_goal_tree(objective: str, breadth: int, depth: int) -> GoalTree¶
assess(goal: str, context: str) -> dict¶
Run the full 8-step assessment phase. Returns a dict of results.
generate_prompt(task: str, context: str | None = None) -> str¶
generate_prompt_edit(description: str, context: str | None = None) -> str¶
evaluate_solution_critically(problem: str, constraints: str, solution: str) -> str¶
check_research_needed(subgoal: str, context: str) -> tuple[bool | None, str]¶
apply_theory_of_mind(user_input: str) -> str¶
ConstraintSpec(BaseModel)¶
A constraint on the solver's output.
| Field | Type | Default |
|---|---|---|
raw | str | required |
metric_key | str \| None | None |
threshold | float \| None | None |
BreakpointSpec(BaseModel)¶
Pause execution after a specific step.
| Field | Type | Default |
|---|---|---|
after_step | int | required |
description | str | '' |
condition | str \| None | None |
GuardrailSpec(BaseModel)¶
Runtime guardrail evaluated during execution.
| Field | Type | Default |
|---|---|---|
name | str | required |
expression | str | required |
message | str | required |
CheckpointSpec(BaseModel)¶
Checkpoint requiring metric evaluation or human approval.
| Field | Type | Default |
|---|---|---|
name | str | required |
description | str | required |
metric_key | str | required |
threshold | float | 1.0 |
after_step | int \| None | None |
StepOverride(BaseModel)¶
Per-step configuration override.
| Field | Type | Default |
|---|---|---|
step | int | required |
llm_backend | str \| None | None |
llm_model | str \| None | None |
skip | bool | False |
max_parallel | int \| None | None |
StepRetryConfig(BaseModel)¶
Per-step retry configuration (Issue 14).
| Field | Type | Default |
|---|---|---|
step | int | required |
max_retries | int | 3 |
retry_on_red_flag | bool | True |
SolverInput(BaseModel)¶
Canonical input for the 15-step solver.
| Field | Type | Default |
|---|---|---|
op | Literal['solve', 'run', 'infer', 'info'] | 'solve' |
goal | str | required |
context | str \| None | None |
subtasks | list[str] | Field(default_factory=list) |
resource_bounds | ResourceBoundsSchema \| None | None |
constraints | list[ConstraintSpec] | Field(default_factory=list) |
breakpoints | list[BreakpointSpec] | Field(default_factory=list) |
guardrails | list[GuardrailSpec] | Field(default_factory=list) |
checkpoints | list[CheckpointSpec] | Field(default_factory=list) |
workspace_base_dir | str | '' |
operation_mode | str \| None | None |
step_overrides | list[StepOverride] | Field(default_factory=list) |
step_retry_configs | list[StepRetryConfig] | Field(default_factory=list) |
complexity_override | str \| None | None |
include_visualisation | bool | False |
dry_run | bool | False |
explain | bool | False |
preset | Literal['quick', 'thorough', 'formal'] \| None | None |
run_mode | Literal['beta', 'production'] | 'beta' |
reviewer_signature | str \| None | None |
StepResult(BaseModel)¶
Result of a single step execution.
| Field | Type | Default |
|---|---|---|
step_number | int | required |
step_name | str | required |
status | Literal['completed', 'skipped', 'failed', 'distilled'] | required |
output | dict | required |
tokens_used | int | required |
latency_ms | int | required |
llm_backend_used | str | required |
execution_strategy | Literal['scripted', 'hybrid', 'agent', 'self_consistency', 'deep_agent'] | required |
fallback_used | bool | False |
fallback_component | str \| None | None |
heuristic_trace | list[str] | Field(default_factory=list) |
critique_applied | bool | False |
exemplar_score | float \| None | None |
degraded | bool | False |
degradation_reason | str \| None | None |
SolverFailure(BaseModel)¶
Typed degradation record (FS5 typed-failure discrimination).
| Field | Type | Default |
|---|---|---|
step_number | int | required |
step_name | str | required |
failed_dep | str | required |
chosen_fallback | str | required |
confidence_impact | Literal['reduced', 'unknown', 'none'] | 'reduced' |
required_operator_action | str | required |
SolverOutput(BaseModel)¶
Output of a complete solver run.
| Field | Type | Default |
|---|---|---|
project_id | str | required |
status | Literal['completed', 'halted', 'reverted', 'paused', 'cancelled', 'input_error'] | required |
completion_state | CompletionState | 'qualified-draft' |
steps | list[StepResult] | required |
failures | list[SolverFailure] | Field(default_factory=list) |
final_answer | str | '' |
total_tokens | int | required |
total_latency_ms | int | required |
total_cost_usd | float | required |
constraints_satisfied | dict[str, bool] | required |
artifacts_distilled | int | required |
complexity_detected | str | 'complex' |
workspace_path | str \| None | None |
git_branch | str \| None | None |
assessment_data | dict \| None | None |
exemplar_verification | dict \| None | None |
error | str | '' |
resumed_from_step | int \| None | None |
paused_at_step | int \| None | None |
reasoning_traces | dict[int, str] | Field(default_factory=dict) |
learning_loop_state | dict | Field(default_factory=dict) |
degraded | bool | False |
degradation_reason | str \| None | None |
agentic_evidence | dict \| None | None |
optimality | OptimalityStamp | Field(default_factory=OptimalityStamp) |
ProgressEvent(BaseModel)¶
Emitted after each step completes for real-time monitoring.
| Field | Type | Default |
|---|---|---|
project_id | str | required |
step_number | int | required |
step_name | str | required |
status | str | required |
tokens_used | int | required |
latency_ms | int | required |
cumulative_tokens | int | required |
cumulative_cost_usd | float | required |
elapsed_ms | int | required |
steps_remaining | int | required |
EstimatedCost(BaseModel)¶
Pre-execution cost estimate.
| Field | Type | Default |
|---|---|---|
estimated_tokens | int | required |
estimated_cost_usd | float | required |
estimated_duration_ms | int | required |
steps_to_execute | list[int] | required |
steps_to_skip | list[int] | Field(default_factory=list) |
complexity | str | required |
confidence | str | required |
DryRunResult(BaseModel)¶
Preview of execution plan without running.
| Field | Type | Default |
|---|---|---|
complexity | str | required |
steps_to_execute | list[int] | required |
steps_to_skip | list[int] | required |
execution_strategies | dict[int, str] | required |
estimated_cost | EstimatedCost | required |
components_used | dict[int, list[str]] | required |
degraded | bool | False |
degradation_reason | str \| None | None |
RunComparison(BaseModel)¶
Comparison between two solver runs.
| Field | Type | Default |
|---|---|---|
run_a_id | str | required |
run_b_id | str | required |
steps_changed | list[int] | Field(default_factory=list) |
score_delta | float | 0.0 |
token_delta | int | 0 |
latency_delta_ms | int | 0 |
per_step_diffs | list[dict] | Field(default_factory=list) |
HealingTier(str, Enum)¶
SolverHealer¶
Tiered healing bridge for the solver engine.
| Field | Type | Default |
|---|---|---|
error_log | list[dict] | field(default_factory=list) |
Methods:
attempt_heal(step_number: int, error: str, code: str = '', tier: HealingTier = HealingTier.LOCAL_RETRY) -> Result[dict]¶
Attempt healing at the specified tier.
record_error(step_number: int, error: str, context: dict) -> None¶
Record an error for future analysis.
suggest_tier(attempt: int, max_retries: int) -> HealingTier¶
Suggest healing tier based on attempt count.
MCP Tools¶
| Operation | Source |
|---|---|
ops | solver_ops_mcp |
help | solver_ops_mcp |
run_solver | solver_ops_mcp |
get_status | solver_ops_mcp |
cancel_run | solver_ops_mcp |
get_step_result | solver_ops_mcp |
get_project_state | solver_ops_mcp |
list_projects | solver_ops_mcp |
get_project | solver_ops_mcp |
delete_project | solver_ops_mcp |
export_project | solver_ops_mcp |
import_project | solver_ops_mcp |
skip_step | solver_ops_mcp |
override_step | solver_ops_mcp |
retry_step | solver_ops_mcp |
get_step_details | solver_ops_mcp |
list_step_registry | solver_ops_mcp |
backup_project | solver_ops_mcp |
restore_backup | solver_ops_mcp |
check_integrity | solver_ops_mcp |
repair_integrity | solver_ops_mcp |
get_event_log | solver_ops_mcp |
get_config | solver_ops_mcp |
update_config | solver_ops_mcp |
get_guardrails | solver_ops_mcp |
validate_input | solver_ops_mcp |
get_execution_strategy | solver_ops_mcp |
estimate_cost | solver_ops_mcp |
dry_run | solver_ops_mcp |
resume_run | solver_ops_mcp |
compare_runs | solver_ops_mcp |
readiness_check | solver_ops_mcp |
list_strategies | solver_ops_mcp |
plan_steps | solver_ops_mcp |
classify_visual | solver_ops_mcp |
recommend_strategy | solver_ops_mcp |
explain_plan | solver_ops_mcp |
list_patterns | solver_ops_mcp |