| name | self-healing-executor |
| description | ð§ Self-Healing æ§è¡åšæèœ - èªåšè¯æé误ãéæ©ä¿®å€çç¥ãæ§è¡ä¿®å€ãéªè¯ç»æ
è§Šååºæ¯ïŒ
- æµè¯å€±èŽ¥æ¶èªåšè§Šå
- 诎"èªåšä¿®å€"
- 诎"è¿è¡ Self-Healing"
- 诎"è¯æå¹¶ä¿®å€"
- èªåšå管éäžçé误æ¢å€
åèœïŒ
- é误åç±»ïŒ8 ç±»é误ïŒ
- çç¥éæ©ïŒ7 ç§ä¿®å€çç¥ïŒ
- ä¿®å€æ§è¡
- ç»æéªè¯
- éç¯æ¥å
|
| version | 2.0.0 |
| author | mcp agent |
| trigger_patterns | ["èªåšä¿®å€","è¿è¡ Self-Healing","è¯æå¹¶ä¿®å€","self-healing"] |
| tags | ["self-healing","error-recovery","automation","fix-strategy"] |
ð§ Self-Healing Executor v2.0
èªåšé©Ÿé©¶æ žå¿æèœïŒè¯æ â ä¿®å€ â éªè¯
çæ¬åæŽïŒv2.0 æŽåè¯å®¡ç»è®ºïŒæ°å¢ä¿®å€åå€å®ãäžæéªè¯ãåå±éåå¡ãLLM é级路åŸãå®å
šå
³é®æäœéå¶
å·¥äœæµçš
[é误åç]
â
âââââââââââââââââââââââââââââââââââââââ
â 0. ä¿®å€åå€å®ïŒåçœ®ïŒ â
â å
çŽ çº§ â è¿å
¥ä¿®å€åŸªç¯ â
â äžå¡çº§ â é级æ¥åïŒåæ¢ïŒ â
âââââââââââââââââââââââââââââââââââââââ
â
âââââââââââââââââââââââââââââââââââââââ
â 1. Error Classification â
â 8 ç±»é误åç±» â
âââââââââââââââââââââââââââââââââââââââ
â
âââââââââââââââââââââââââââââââââââââââ
â 2. Strategy Selection â
â 7 ç§ä¿®å€çç¥ â
âââââââââââââââââââââââââââââââââââââââ
â
âââââââââââââââââââââââââââââââââââââââ
â 3. Fix Execution â
â åºçšéäžçä¿®å€çç¥ â
âââââââââââââââââââââââââââââââââââââââ
â
âââââââââââââââââââââââââââââââââââââââ
â 4. Verification â
â åå±éªè¯éåå¡ïŒL1-L4ïŒ â
â äžæç»æïŒCONFIRMED/UNCERTAIN/FAILED â
âââââââââââââââââââââââââââââââââââââââ
â
[ç»ææ¥å]
0. ä¿®å€åå€å®ïŒå眮ïŒ
ä¿®å€åå€å®æ¯ Self-Healing ç第äžéå
³å¡ïŒå³å®æ¯åŠè¿å
¥èªåšä¿®å€åŸªç¯ã
ä¿®å€ååç±»
| å | æ
éç±»å | å€ççç¥ |
|---|
| å
çŽ çº§ | å®äœåšå€±æãUI 屿§ååãå¯è§æ§ç¶æãå
çŽ é®æ¡ | èªåšä¿®å€ïŒæå€ 3 蜮 L1-L3 éçº§ïŒ |
| äžå¡çº§ | é»èŸéè¯¯ãæ°æ®äžäžèŽãæéé®é¢ãç¶ææºåŒåžž | é级æ¥åïŒè®°åœæ¥å¿ãéç¥äººå·¥ã忢èªåšä¿®å€ïŒ |
å€å®é»èŸ
async def run(self, workflow, initial_error, context):
classified = self.error_analyzer.classify(str(initial_error), context)
if self._is_business_level_failure(classified):
logger.warning("Business-level failure, skipping auto-fix")
return LoopResult(
success=False,
attempts=0,
error="Business-level failure requires manual intervention",
state=LoopState.FAILED,
)
return await self._run_fix_loop(...)
å®å
šå
³é®æäœéå¶
ä»¥äžæäœå±äº Critical 级å«ïŒå¿
é¡»çšæ·äºæ¬¡ç¡®è®€ïŒæç»èªåšä¿®å€ïŒ
| æäœ | å±é©ç级 | å€çæ¹åŒ |
|---|
grant_permission | ðŽ Critical | å¿
é¡»çšæ·æŸåŒææïŒçŠæ¢èªåšä¿®å€ |
revoke_permission | ðŽ Critical | å¿
é¡»çšæ·æŸåŒææïŒçŠæ¢èªåšä¿®å€ |
delete_data | ðŽ Critical | å¿
é¡»çšæ·æŸåŒææïŒçŠæ¢èªåšä¿®å€ |
CRITICAL_OPERATIONS = {
"grant_permission",
"revoke_permission",
"delete_data",
}
def is_critical_operation(self, operation_name: str) -> bool:
return operation_name in self.CRITICAL_OPERATIONS
é误åç±»ïŒ8 ç±»ïŒ
| Category | 诎æ | ç€ºäŸ | ä¿®å€å |
|---|
LocatorNotFound | å
çŽ å®äœå€±èŽ¥ | NoSuchElementException | å
çŽ çº§ |
Timeout | æäœè¶
æ¶ | Timeout waiting for element | å
çŽ çº§ |
AssertionFailed | æèšå€±èŽ¥ | Expected true but was false | äžå¡çº§ |
NetworkError | çœç»é误 | Connection refused | å
çŽ çº§ |
PermissionDenied | æéé误 | Access denied to X | äžå¡çº§ |
InvalidState | ç¶æé误 | Element not clickable | å
çŽ çº§ |
CrashDetected | åºçšåŽ©æº | App crashed | å
çŽ çº§ |
UnknownError | æªç¥é误 | fallback | éå€å® |
äžæ VerificationResult
éªè¯ç»æä»äºæïŒboolïŒåçº§äžºäžæïŒæäŸæŽç²Ÿç»çç¶æè¡šèŸŸïŒ
class VerificationStatus(Enum):
CONFIRMED = "confirmed"
UNCERTAIN = "uncertain"
FAILED = "failed"
@dataclass
class VerificationResult:
status: VerificationStatus
message: str
element_exists: bool = False
element_visible: bool = False
suggestion: str | None = None
| ç¶æ | å«ä¹ | äžäžæ¥è¡åš |
|---|
CONFIRMED | éªè¯éè¿ïŒä¿®å€æå | éåºä¿®å€åŸªç¯ |
UNCERTAIN | éªè¯äžç¡®å®ïŒéçº§å°æŽé«å±çº§éªè¯ | è¿å
¥ L3/L4 éªè¯ |
FAILED | éªè¯å€±èŽ¥ïŒä¿®å€æ æ | å°è¯äžäžäžªçç¥ |
åå±éªè¯éåå¡
éªè¯éçšåå±çç¥ïŒä»äœææ¬å°é«ææ¬éå±éªè¯ïŒ
âââââââââââââââ
â L4: äžå¡æµéªè¯ â é«ææ¬ â ä»
åš L3 倱莥æ¶è§Šå
â (åŒæ¥ïŒå®æŽè·¯åŸ) â
ââââââââ¬âââââââ
â UNCERTAIN
ââââââââŒâââââââ
â L3: 蜻éåèœéªè¯ â äžææ¬ â å
çŽ å¯æäœæ§éªè¯
â (åŒæ¥ïŒäº€äºæµè¯) â
ââââââââ¬âââââââ
â UNCERTAIN
ââââââââŒâââââââ
â L2: å
çŽ å¯æäœæ§ â äœææ¬ â enabled/visible/interactive
â (忥ïŒ<100ms) â
ââââââââ¬âââââââ
â UNCERTAIN
ââââââââŒâââââââ
â L1: å
çŽ ååšæ§ â æäœææ¬ â æ¥è¯¢å
çŽ æ¯åŠååš
â (忥ïŒ<100ms) â
âââââââââââââââ
| å±çº§ | éªè¯å
容 | æ§è¡æ¹åŒ | è¶
æ¶ | éçšåºæ¯ |
|---|
| L1 | å
çŽ ååšäº UI æ äž | 忥 | < 100ms | å¿«é确讀å
çŽ ååš |
| L2 | å
çŽ enabled/visible/interactive | 忥 | < 100ms | 确讀å
çŽ å¯äº€äº |
| L3 | 蜻éåèœéªè¯ïŒtap/swipe 亀äºïŒ | åŒæ¥ | 5s | éªè¯å
çŽ çå®å¯çš |
| L4 | 宿Žäžå¡æµéªè¯ïŒå€æ¥éª€æäœïŒ | åŒæ¥ | 30s | L3 倱莥åçæç»ç¡®è®€ |
éªè¯éåœä¿æ€
鲿¢éªè¯éåœå¯ŒèŽçæ é埪ç¯ïŒ
@dataclass
class LoopConfig:
max_retries: int = 3
max_verification_depth: int = 5
verify_timeout: int = 60
async def _verify_with_depth_protection(self, locator, context):
self._verify_depth += 1
if self._verify_depth > self.config.max_verification_depth:
logger.error("Verification depth exceeded, stopping recursive loop")
return VerificationResult(
status=VerificationStatus.FAILED,
message="Max verification depth exceeded"
)
try:
result = await self.fix_verifier.verify(locator, context)
return result
finally:
self._verify_depth -= 1
ä¿®å€çç¥ïŒ7 å±ïŒ
| Level | çç¥ | éçšåºæ¯ | è¶
æ¶ | éçº§è·¯åŸ |
|---|
| L1 | QuickRetry | ç¬æ¶é误 | 1s | â L2 |
| L2 | AlternativeLocator | å®äœç¬Šé®é¢ | 3s | â L3 |
| L3 | ScrollRetry | éæ»åš | 5s | â L4 |
| L4 | AISuggestion | å€æåºæ¯ | 10s | â L5 |
| L5 | VisualMatch | è§è§çžäŒŒ | 15s | â L6 |
| L6 | SemanticFallback | è¯ä¹æšæ | 20s | â L7 |
| L7 | ReportFailure | ææå€±èŽ¥ | - | ç»æ¢ |
LLM 倱莥é级路åŸ
åœ AI ä¿®å€çç¥ïŒCODE_VALIDATEïŒå€±èŽ¥æ¶ïŒèªåšé级å°ç¡®å®æ§çç¥ïŒ
class StrategyFailureType(Enum):
NOT_APPLICABLE = "not_applicable"
EXECUTION_FAILED = "execution_failed"
LLM_UNAVAILABLE = "llm_unavailable"
def apply_strategy(self, strategy, context) -> FixResult:
try:
result = strategy.action(context)
return FixResult(success=True, strategy_name=strategy.name, ...)
except Exception as e:
if strategy.strategy_type == FixStrategyType.CODE_VALIDATE:
return self._fallback_deterministic_fix(strategy, context)
return FixResult(success=False, strategy_name=strategy.name, ...)
def _fallback_deterministic_fix(self, strategy, context):
"""é级å°ç¡®å®æ§ä¿®å€çç¥ïŒCODE_VALIDATE â APP_RESTART"""
restart_strategy = FixStrategy(
name="fallback_restart",
strategy_type=FixStrategyType.APP_RESTART,
action=self._restart_app,
)
try:
restart_strategy.action(context)
return FixResult(success=True, strategy_name="fallback_restart", ...)
except:
return FixResult(
success=False,
strategy_name="fallback_restart",
failure_type=StrategyFailureType.EXECUTION_FAILED
)
é级路åŸïŒ
CODE_VALIDATE (LLM) 倱莥
â
APP_RESTART (ç¡®å®æ§)
â
è¥å倱莥 â ReportFailure (L7)
äœ¿çšæ¹åŒ
æåšè§Šå
python3 .claude/skills/self-healing-executor/scripts/heal.py --error "Element not found" --context "accessibility_id:submit"
Python API
from self_healing_executor import SelfHealingExecutor
executor = SelfHealingExecutor()
result = executor.heal(
error="Element not found",
context={"locator": "accessibility_id:submit"}
)
if result.verification_result.status == VerificationStatus.CONFIRMED:
print(f"Fixed with {result.strategy_used}")
elif result.verification_result.status == VerificationStatus.UNCERTAIN:
print("Verification uncertain, needs higher-level validation")
else:
print(f"Failed after {result.attempts} attempts")
èªåšåéæ
try:
element.click()
except Exception as e:
executor = SelfHealingExecutor()
result = executor.heal(str(e), get_context())
if result.requires_user_confirmation:
wait_for_user_confirmation(result.critical_operations)
èŸåºç€ºäŸ
æåä¿®å€ïŒCONFIRMEDïŒ
ð§ Self-Healing Executing...
âââ ð Domain Check: Element-level failure (proceed to fix)
âââ ð Classify: LocatorNotFound (confidence: 95%)
âââ ð¯ Strategy: L2 Alternative Locator
âââ â¡ Attempt 1: accessibility_id:submit â accessibility_id:submit_button
â âââ â
L1 Verify: CONFIRMED (element exists)
â âââ â
L2 Verify: CONFIRMED (element interactive)
âââ â
Fixed in 2.1s
Strategy: L2 Alternative Locator
Attempts: 1
Duration: 2.1s
Verification: CONFIRMED
é级æ¥åïŒäžå¡çº§æ
éïŒ
ð§ Self-Healing Executing...
âââ ð Domain Check: Business-level failure (escalate)
âââ ð Classify: PermissionDenied (confidence: 88%)
âââ â ïž Stopping auto-fix: business-level failure
Needs human review: permission denied
Reason: Security-critical operation requires user confirmation
LLM é级
ð§ Self-Healing Executing...
âââ ð Classify: AssertionFailed (confidence: 75%)
âââ ð¯ Strategy: L4 AI Suggestion (CODE_VALIDATE)
âââ â¡ LLM validate... FAILED (LLM unavailable)
âââ ð Fallback: APP_RESTART
âââ â¡ Restart app... SUCCESS
âââ â
Fixed via fallback strategy
Strategy: fallback_restart (APP_RESTART)
Attempts: 2
Duration: 8.3s
é
眮
åš scripts/config.py äžä¿®æ¹ïŒ
CONFIG = {
"max_attempts": 3,
"max_verification_depth": 5,
"verify_timeout": 60,
"escalate_on_fail": True,
"use_ai_for_complex": True,
"verify_after_fix": True,
"notify_on_escalation": True,
"critical_operations": [
"grant_permission",
"revoke_permission",
"delete_data",
],
"require_user_confirmation": True,
}
ç¶æç
| ç¶æç | å«ä¹ |
|---|
| 0 | ä¿®å€æåïŒCONFIRMEDïŒ |
| 1 | ä¿®å€å€±èŽ¥ïŒé人工ïŒFAILED after all strategiesïŒ |
| 2 | é
眮é误 |
| 3 | è¶
æ¶ |
| 4 | äžå¡çº§æ
ééçº§ïŒæ éä¿®å€ïŒ |
äžå
¶ä»æèœçåäœ
test-runner 倱莥 â self-healing-executor â æå â ç»§ç»
â
倱莥 â notification-hub â éç¥
self-healing-executor 倱莥 â notification-hub â ç»äžéç¥
project-health-monitor æ£æµå°çæåš OPEN â self-healing-executor å°è¯æ¢å€
è¯å®¡ç»è®ºæ¥æº
æ¬æèœææ¡£ v2.0 æŽåèª Agent Council è¯å®¡ç»è®ºïŒ
- è¯å®¡æä»¶:
.claude/skills/agent-council/history/self_healing_review.md
- åäž Agent: Software Architect, Security Engineer, Code Reviewer
- è¯å®¡æ¥æ: 2026/05/11
æ žå¿æ¹è¿ç¹
| æ¹è¿é¡¹ | æ¥æº | äŒå
级 |
|---|
| ä¿®å€åå€å®å眮 | Software Architect | P0 |
| äžæ VerificationResult | Code Reviewer | P0 |
| åå±éªè¯éåå¡ | Software Architect | P0 |
| LLM 倱莥éçº§è·¯åŸ | Code Reviewer | P0 |
| å®å
šå
³é®æäœéå¶ | Security Engineer | P0 |
| éªè¯éåœä¿æ€ | Code Reviewer | P1 |
| 审计æ¥å¿æºå¶ïŒHMAC-SHA256 é²ç¯¡æ¹ïŒ | Security Engineer | P0 |
审计æ¥å¿æºå¶
Self-Healing æ§è¡è¿çšå
šçšå®¡è®¡è®°åœïŒéçšé²ç¯¡æ¹è®Ÿè®¡ç¡®ä¿æ¥å¿å®æŽæ§ã
æ¥å¿ç»æ
@dataclass
class AuditLogEntry:
timestamp: str
event_type: str
level: str
operation: str
status: str
details: dict
hmac_signature: str
HMAC-SHA256 é²ç¯¡æ¹
æ¯æ¡æ¥å¿å
å«åºäºå
容计ç®ç HMAC çŸåïŒæ£æµä»»äœç¯¡æ¹è¡äžºïŒ
import hmac
import hashlib
class AuditLogger:
def __init__(self, secret_key: str):
self._secret = secret_key.encode()
def _sign(self, entry: AuditLogEntry) -> str:
"""è®¡ç® HMAC-SHA256 çŸå"""
message = f"{entry.timestamp}:{entry.event_type}:{entry.operation}:{entry.status}"
return hmac.new(self._secret, message.encode(), hashlib.sha256).hexdigest()
def verify(self, entry: AuditLogEntry) -> bool:
"""éªè¯æ¥å¿å®æŽæ§"""
expected = self._sign(entry)
return hmac.compare_digest(entry.hmac_signature, expected)
P0/P1 åŒåžžæ£æµ
| çº§å« | è§Šåæ¡ä»¶ | å€çæ¹åŒ |
|---|
| P0 | å®å
šå
³é®æäœè¢«æç»ãä¿®å€åŸªç¯è¶
è¿æå€§æ¬¡æ°ãäžå¡çº§æ
éé级 | ç«å³éç¥ + è®°åœäž¥éäºä»¶ |
| P1 | LLM é级åçãéªè¯æ·±åºŠè¶
è¿éåŒãè¿ç» 3 次修å€å€±èŽ¥ | è®°åœèŠå + 宿¶æ±æ» |
class P0Alert(Exception):
"""P0 级å«åŒåžžïŒéèŠç«å³å€ç"""
pass
class P1Warning(Exception):
"""P1 级å«åŒåžžïŒéèŠå
³æ³šäœäžçާæ¥"""
pass
def _classify_alert_level(event_type: str, attempts: int, domain: str) -> str:
if domain == "business" or attempts > 5:
return "P0"
elif event_type in ("LLM_FALLBACK", "VERIFICATION_DEPTH_EXCEEDED"):
return "P1"
return "P2"
审计æ¥å¿èŸåºç€ºäŸ
2026-05-12T10:30:45.123Z | FIX_ATTEMPT | P1 | ios_button_click | SUCCESS
âââ strategy: L2 Alternative Locator
âââ locator: accessibility_id:submit â accessibility_id:submit_btn
âââ verification: CONFIRMED
âââ signature: a1b2c3d4e5f6...
2026-05-12T10:30:48.456Z | DOMAIN_CHECK | P0 | grant_permission | FAILURE
âââ reason: Critical operation requires user confirmation
âââ action: Skip auto-fix, escalate to human
âââ signature: f6e5d4c3b2a1...
æ¥å¿ååš
审计æ¥å¿ååšåš .claude/logs/self-healing-audit.logïŒææ¥æèœ®èœ¬ïŒ
.claude/logs/
âââ self-healing-audit.log
âââ 2026-05-10.log
âââ 2026-05-11.log
âââ 2026-05-12.log
äž notification-hub éæ
P0 级å«äºä»¶èªåšè§Šåéç¥ïŒ
if alert_level == "P0":
notification_hub.send_urgent(
title="Self-Healing P0 Alert",
message=f"Critical failure in {operation}: {reason}",
channel="security"
)
elif alert_level == "P1":
pending_warnings.append({
"operation": operation,
"reason": reason,
"timestamp": timestamp
})
审计æ¥å¿å®æŽæ§éªè¯
宿éªè¯å岿¥å¿æ¯åŠè¢«ç¯¡æ¹ïŒ
def verify_audit_log(log_file: Path) -> list[VerificationResult]:
"""éªè¯æ¥å¿æä»¶å®æŽæ§ïŒè¿å被篡æ¹çæ¡ç®"""
tampered = []
with open(log_file) as f:
for line in f:
entry = parse_log_line(line)
if not audit_logger.verify(entry):
tampered.append(entry)
return tampered