System Doctor¶
system_doctor — Self-healing orchestration component for G6.
Cluster: Code Intelligence | Type: component | MCP Tools: 11
Overview¶
Self-healing orchestration component that diagnoses failures, reports backend capabilities, applies bounded tiered healing strategies, verifies candidate fixes where evidence is available, monitors system health, detects semantic drift, runs canary probes, and manages skill pathway selection. Records healing episodes into a competence model so the doctor adapts its strategy preferences over time -- combining a strategy library, tier cascade, and episode recorder into a single block.
Pilot reliability aid, not autonomous production repair
system_doctor is useful for diagnosis, health checks, simple mechanical Python repairs, syntax-validated patches, learning from healing attempts, and escalation when local repair fails. It should not be presented as a general engine that can fix any broken application or autonomously repair production systems without review.
For launch and design-partner workflows, keep claims bounded to assisted reliability: capture the failure, classify likely causes, attempt low-risk local fixes, verify with supplied tests or syntax checks, record the outcome, and escalate unresolved or high-impact failures to a human. Before using generated patches in production, review the diff, run the relevant test suite, keep rollback available, and avoid relying on this component as regulated compliance evidence.
When to use:
- Diagnosing runtime errors in G6 components and attempting bounded local repairs before escalating
- Discovering backend readiness with
capabilitiesbefore attempting repair or monitoring work - Monitoring component health on a schedule and triggering healing when drift or degradation is detected
- Selecting and progressing through a skill pathway (novice -> expert) that controls which healing tier is attempted first
Example:
from mvp.system_doctor import DoctorBlock, DoctorInput
block = DoctorBlock(name="doctor")
result = block.infer(DoctorInput(
op="diagnose",
component_name="solver",
error_message="KeyError: 'answer' in step 7",
error_type="KeyError",
))
# result.ok -> True; result.value -> DoctorOutput with diagnosis, strategy, proposed_fix
Works well with: self_debug, auto_engineer, experience_loop
Public API¶
SystemDoctorDecisionError(ValueError)¶
The LLM did not produce a usable, validated doctor decision.
DoctorDecision¶
Validated error-categorization + routing-posture verdict.
| Field | Type | Default |
|---|---|---|
category | str | required |
baseline_category | str | required |
posture | str | required |
baseline_posture | str | required |
requires_review | bool | False |
degraded | bool | False |
llm_used | bool | False |
reasons | tuple[str, ...] | () |
rationale | str | '' |
raw_response | str | '' |
Methods:
escalated() -> bool¶
category_refined() -> bool¶
agentic_evidence() -> dict[str, Any]¶
to_dict() -> dict[str, Any]¶
SystemDoctorPlanner¶
Runtime-first facade with deterministic fallback + one-way posture clamp.
| Field | Type | Default |
|---|---|---|
runtime | SystemDoctorRuntime \| None | None |
last_decision | DoctorDecision \| None | field(default=None, init=False) |
last_llm_used | bool | field(default=False, init=False) |
last_fallback_reason | str | field(default='', init=False) |
Methods:
diagnose(error_type: str, error_message: str, traceback: str = '', strategy: Any = None, safety_approved: bool | None = None) -> DoctorDecision¶
SystemDoctorBlock(AIBlock[DoctorInput, DoctorOutput, dict])¶
Self-healing orchestrator for the G6 system.
| Field | Type | Default |
|---|---|---|
name | str | 'system_doctor' |
state | dict | field(default_factory=dict) |
db_path | str | '~/.g6/doctor.db' |
Methods:
infer(data: DoctorInput) -> Result[DoctorOutput]¶
Dispatch on op field.
health() -> dict¶
Return health status for production monitoring.
SystemDoctorPatternRuntime¶
Stateless executable mechanisms for system_doctor's applied patterns.
Methods:
tighten(posture: Any, baseline_posture: Any = '', category: Any = '', requires_review: bool = False, degraded: bool = False, unknown_category: bool = False, low_success_rate: bool = False, safety_not_approved: bool = False) -> SystemDoctorPatternReview¶
One-way routing-posture guard (hook-based-safety-guard-rails).
DoctorInput(BaseModel)¶
| Field | Type | Default |
|---|---|---|
op | DoctorOp | 'diagnose' |
component_name | str | '' |
error_message | str | '' |
error_type | str | '' |
traceback | str | '' |
code | str | '' |
operation | str | '' |
proposed_fix | str | '' |
test_cases_json | str | '[]' |
max_tier | int | 4 |
timeout_seconds | float | 30.0 |
task_description | str | '' |
pathway_id | str | '' |
skill_names | list[str] | Field(default_factory=list) |
swap_reason | str | '' |
metrics_json | str | '{}' |
probe_components | list[str] | Field(default_factory=list) |
DoctorOutput(BaseModel)¶
| Field | Type | Default |
|---|---|---|
healed | bool | False |
escalated | bool | False |
tier_used | int | -1 |
tiers_attempted | list[int] | Field(default_factory=list) |
patched_code | str | '' |
fix_description | str | '' |
diagnosis | str | '' |
error_category | str | '' |
root_cause | str | '' |
verified | bool | False |
safety_approved | bool | False |
safety_vub | float | 0.0 |
health_reports | list[dict] | Field(default_factory=list) |
drift_detected | bool | False |
drift_summary | str | '' |
competence | float | 0.0 |
strategy | dict | Field(default_factory=dict) |
improvement_suggestions | list[str] | Field(default_factory=list) |
pathway | dict | Field(default_factory=dict) |
skill_classification | dict | Field(default_factory=dict) |
suggested_skills | list[str] | Field(default_factory=list) |
swap_result | dict | Field(default_factory=dict) |
stats | dict | Field(default_factory=dict) |
episode | dict | Field(default_factory=dict) |
degraded | bool | False |
degradation_reason | str \| None | None |
routing_posture | str | '' |
baseline_posture | str | '' |
requires_review | bool | False |
category_refined | bool | False |
llm_used | bool | False |
agentic_evidence | dict | Field(default_factory=dict) |
completion_state | CompletionState | 'qualified-draft' |
warning_card | dict | Field(default_factory=dict) |
evidence | dict | Field(default_factory=dict) |
request_id | str | '' |
task_id | str | '' |
run_id | str | '' |
Methods:
data() -> dict¶
Flat registry payload for callers using ComponentRegistry.invoke().
HealingEpisode(BaseModel)¶
| Field | Type | Default |
|---|---|---|
episode_id | str | '' |
component_name | str | '' |
error_type | str | '' |
error_message | str | '' |
tier_used | int | -1 |
tiers_attempted | list[int] | Field(default_factory=list) |
fix_applied | str | '' |
verified | bool | False |
safety_checked | bool | False |
outcome | Literal['success', 'failure', 'partial'] | 'failure' |
duration_seconds | float | 0.0 |
tokens_used | int | 0 |
cost_estimate | float | 0.0 |
timestamp | str | '' |
HealthReport(BaseModel)¶
| Field | Type | Default |
|---|---|---|
component_name | str | '' |
status | Literal['healthy', 'degraded', 'failing', 'unreachable'] | 'healthy' |
latency_ms | float | 0.0 |
drift_detected | bool | False |
drift_severity | float | 0.0 |
last_error | str | '' |
SkillClassification(BaseModel)¶
| Field | Type | Default |
|---|---|---|
skill_name | str | '' |
skill_type | Literal['cognitive', 'agentic'] | 'cognitive' |
capabilities | list[str] | Field(default_factory=list) |
resource_profile | str | 'low' |
quality_tier | Literal['fast', 'balanced', 'thorough'] | 'balanced' |
SkillType(str, Enum)¶
SkillProfile¶
| Field | Type | Default |
|---|---|---|
name | str | required |
skill_type | SkillType | required |
capabilities | list[str] | field(default_factory=list) |
typical_latency_ms | float | 0.0 |
typical_tokens | int | 0 |
quality_tier | str | 'balanced' |
ExecutionPathway¶
| Field | Type | Default |
|---|---|---|
pathway_id | str | required |
name | str | required |
description | str | required |
steps | list[PathwayStep] | field(default_factory=list) |
fallback_map | dict[str, str] | field(default_factory=dict) |
quality_mode | str | 'balanced' |
Functions¶
deterministic_doctor_decision(error_type: str, error_message: str, traceback: str = '', strategy: Any = None, safety_approved: bool | None = None) -> DoctorDecision¶
Demoted-real STRUCTURAL baseline (the zero-LLM floor).
apply_category_floor(baseline_category: Any, candidate: Any) -> str¶
Refine-the-unknown category clamp (the regex category is the floor).
apply_posture_floor(baseline: Any, candidate: Any) -> DoctorDecision¶
One-way routing-posture safety clamp (stricter-of, ESCALATE only).
build_default_planner() -> 'SystemDoctorPlanner | None'¶
Build the on-by-default agentic planner, or None when suppressed.
grounded_diagnose(error_type: str, error_message: str, traceback: str = '', strategy: Any = None, safety_approved: bool | None = None, planner: SystemDoctorPlanner | None = None) -> DoctorDecision¶
The ONE shared grounded chokepoint for EVERY system_doctor diagnose decision.
applied_agentic_patterns() -> list[dict[str, Any]]¶
Return compact metadata for system_doctor-applied vendored patterns.
MCP Tools¶
| Operation | Source |
|---|---|
verified | system_doctor_mcp |
qualified-draft | system_doctor_mcp |
blocked-escalated | system_doctor_mcp |
ops | system_doctor_mcp |
help | system_doctor_mcp |
diagnose | system_doctor_mcp |
get_strategy | system_doctor_mcp |
list_patterns | system_doctor_mcp |
capabilities | system_doctor_mcp |
get_stats | system_doctor_mcp |
get_health_report | system_doctor_mcp |