Skip to content

Adapt Healing

Adapt Healing — mvp.adapt_healing

Cluster: Code Intelligence | Type: component | MCP Tools: 32

Overview

Assisted self-healing component with error detection, diagnosis, patching, verification, and constrained backend-native adapter discovery, plus a 32-op MCP sub-package with 7 SQLite tables for drift detection, context capture, and retry strategies.

Production boundary

adapt_healing is designed for assisted repair and reliability workflows, not arbitrary fully autonomous production code healing. It can capture failures, classify common Python errors, suggest or apply bounded repairs, verify against supplied tests, and record what worked. Mechanical patching covers common patterns, LLM patching is optional and best-effort, and any production use should keep human review, test coverage, backups, and deployment rollback in the loop.

When to use:

  • Assisted error recovery and code patching with explicit verification
  • Drift detection in ML pipelines
  • Building reliability workflows with persistent healing history

Example:

from mvp.adapt_healing import AdaptHealingBlock, HealingInput

block = AdaptHealingBlock(name="heal")
result = block.infer(HealingInput(code="def f(): return 1/0", operation="diagnose"))
# result.ok → True; result.value → HealingOutput with diagnosis and patch

Works well with: meta_programming, database, align_csf

Public API

AdaptHealingBlock(AIBlock[HealingInput, HealingOutput, None])

Self-healing Python code block.

Field Type Default
name str 'adapt_healing'
resource_bounds ResourceBounds \| None None
usage ResourceUsage field(default_factory=ResourceUsage)
llm_fn object None

Methods:

infer(data: HealingInput) -> Result[HealingOutput]

TestCase(BaseModel)

A single input/expected-output pair for code verification.

Field Type Default
inputs dict[str, object] Field(default_factory=dict)
expected_output object None
description str ''

HealingInput(BaseModel)

Input to AdaptHealingBlock.

Field Type Default
code str required
error_message str ''
operation Literal['detect', 'diagnose', 'patch', 'verify'] 'detect'
test_cases list[TestCase] Field(default_factory=list)
patched_code str ''

IssueReport(BaseModel)

A detected code issue.

Field Type Default
line int 0
issue_type str required
description str required
severity Literal['error', 'warning', 'info'] 'warning'

TestResult(BaseModel)

Result of running one test case.

Field Type Default
passed bool required
description str ''
actual_output str ''
expected_output str ''
error str ''
degraded bool False
degradation_reason str \| None None

HealingOutput(BaseModel)

Output from AdaptHealingBlock.

Field Type Default
operation str required
issues list[IssueReport] Field(default_factory=list)
diagnosis str ''
diagnosis_category str ''
patched_code str ''
patch_applied bool False
test_results list[TestResult] Field(default_factory=list)
all_tests_passed bool False
n_issues int 0
degraded bool False
degradation_reason str ''

AdaptHealingMCPBlock(AIBlock[MCPHealingInput, MCPHealingOutput, dict])

Full-featured self-healing block with SQLite persistence.

Field Type Default
name str 'adapt_healing_mcp'
state dict \| None None
db_path str ':memory:'
resource_bounds ResourceBounds \| None None
usage ResourceUsage field(default_factory=ResourceUsage)

Methods:

infer(data: MCPHealingInput) -> Result[MCPHealingOutput]

MCPHealingRecord(BaseModel)

Field Type Default
id str required
record_type str required
key str required
value str required
tags list[str] Field(default_factory=list)
timestamp str required
metadata dict[str, Any] Field(default_factory=dict)

MCPHealingInput(BaseModel)

Field Type Default
op Literal['detect', 'diagnose', 'patch', 'verify', 'heal', 'capture_context', 'generate_hint', 'create_backup', 'restore_backup', 'detect_drift', 'hypothesize', 'retry_config', 'record_healing', 'query_healings', 'summarize_healings', 'store_patch', 'retrieve_patches', 'evaluate_patch', 'record_error', 'query_errors', 'get_error_patterns', 'store_rule', 'retrieve_rules', 'apply_rule', 'search', 'info', 'list_patterns', 'backend_info', 'list_capabilities', 'list_backend_ops', 'native_op_schema', 'backend_native_execute'] required
code str ''
error_message str ''
test_cases_json str ''
patched_code str ''
key str ''
value str ''
tags list[str] Field(default_factory=list)
query str ''
limit int 50
top_k int 5
code_id str ''
task_type str ''
success bool True
issues_json str ''
notes str ''
name str ''
description str ''
issue_type str ''
patch_pattern str ''
replacement str ''
domain str ''
worth float 0.5
condition str ''
action str ''
context_json str ''
func_name str ''
args_json str ''
metrics_json str ''
threshold float 0.1
stop_type str ''
stop_value float 3.0
wait_type str ''
wait_min float 1.0
wait_max float 60.0
retry_on list[str] Field(default_factory=list)
max_rounds int 0
max_llm_calls int 0
threshold_window int 0
idempotency_key str ''
trusted_code bool False
sandbox_token str ''
production_mode bool False
allow_unsandboxed_exec bool False
backend str ''
native_operation str ''
native_args_json str '{}'
dry_run bool False
limits_json str '{}'

MCPHealingOutput(BaseModel)

Field Type Default
ok bool True
op str required
data dict[str, Any] \| None None
error dict[str, Any] \| None None
key str ''
value str ''
found bool False
count int 0
records list[MCPHealingRecord] Field(default_factory=list)
retrieved list[str] Field(default_factory=list)
scores list[float] Field(default_factory=list)
summary str ''
message str ''
patched_code str ''
issues list[dict[str, Any]] Field(default_factory=list)
test_results list[dict[str, Any]] Field(default_factory=list)
all_tests_passed bool False
metadata dict[str, Any] Field(default_factory=dict)
degraded bool False
degradation_reason str ''
warning_card dict[str, Any] \| None None
completion_state Literal['verified', 'qualified-draft', 'blocked-escalated'] 'qualified-draft'
request_id str ''
run_id str ''

HealingStore

Sync SQLite healing store with 7 tables.

Constructor:

Parameter Type Default
db_path str ':memory:'

Methods:

add_healing(code_id: str, original_code: str, patched_code: str, error_message: str, issues_json: str, success: bool, task_type: str, tags: list[str], notes: str, idempotency_key: str = '') -> str

query_healings(code_id: str = '', task_type: str = '', success: bool | None = None, limit: int = 50) -> list[dict[str, Any]]

update_drift_session(session_name: str, value: float) -> None

Update running Welford stats for a named drift session.

get_drift_session(session_name: str) -> dict | None

upsert_patch(name: str, description: str, issue_type: str, patch_pattern: str, replacement: str, domain: str, worth: float, tags: list[str]) -> str

query_patches(query: str = '', issue_type: str = '', domain: str = '', top_k: int = 5) -> list[dict[str, Any]]

update_patch_eval(name: str, success: bool) -> bool

add_error(code_id: str, error_type: str, error_message: str, code_snippet: str, task_type: str, notes: str) -> str

query_errors(error_type: str = '', task_type: str = '', limit: int = 50) -> list[dict[str, Any]]

get_error_patterns(min_frequency: int = 2) -> list[dict[str, Any]]

upsert_rule(name: str, description: str, issue_type: str, condition: str, action: str, domain: str, worth: float, tags: list[str]) -> str

query_rules(query: str = '', issue_type: str = '', domain: str = '', top_k: int = 5) -> list[dict[str, Any]]

add_context(code_id: str, error_type: str, error_message: str, func_name: str, traceback_frames: list[dict], detail: dict, task_type: str) -> str

query_contexts(code_id: str = '', error_type: str = '', limit: int = 20) -> list[dict[str, Any]]

add_backup(code_id: str, code_content: str, task_type: str, notes: str) -> str

list_backups(code_id: str = '', limit: int = 20) -> list[dict[str, Any]]

get_backup(backup_id: str = '', code_id: str = '') -> dict | None

upsert_retry_strategy(name: str, description: str, stop_type: str, stop_value: float, wait_type: str, wait_min: float, wait_max: float, retry_on: list[str]) -> str

get_retry_strategy(name: str) -> dict[str, Any] | None

increment_retry_use(name: str) -> None

text_search(query: str, top_k: int = 5) -> list[dict[str, Any]]

TF-IDF search across healings, patches, errors, rules, contexts, and backups.

count_all() -> dict[str, int]

MCP Tools

Operation Source
detect healing_mcp
diagnose healing_mcp
patch healing_mcp
verify healing_mcp
heal healing_mcp
capture_context healing_mcp
generate_hint healing_mcp
create_backup healing_mcp
restore_backup healing_mcp
detect_drift healing_mcp
hypothesize healing_mcp
retry_config healing_mcp
record_healing healing_mcp
query_healings healing_mcp
summarize_healings healing_mcp
store_patch healing_mcp
retrieve_patches healing_mcp
evaluate_patch healing_mcp
record_error healing_mcp
query_errors healing_mcp
get_error_patterns healing_mcp
store_rule healing_mcp
retrieve_rules healing_mcp
apply_rule healing_mcp
search healing_mcp
info healing_mcp
list_patterns healing_mcp
backend_info healing_mcp
list_capabilities healing_mcp
list_backend_ops healing_mcp
native_op_schema healing_mcp
backend_native_execute healing_mcp