Skip to content

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]

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