299 lines
10 KiB
Python
299 lines
10 KiB
Python
"""
|
|
F14 Proactive Diagnostics - Runner
|
|
|
|
Orchestrates diagnostic checks with caching, timeouts, and graceful degradation.
|
|
"""
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import time
|
|
import uuid
|
|
from typing import Dict, Any, List, Optional, Callable, Awaitable
|
|
|
|
from ..time_utils import utc_isoformat
|
|
from functools import lru_cache
|
|
|
|
from .models import (
|
|
HealthReport,
|
|
HealthIssue,
|
|
HealthCheckRequest,
|
|
DiagnosticsScope,
|
|
IssueCounts,
|
|
IntentSignature,
|
|
)
|
|
|
|
logger = logging.getLogger("comfyui-doctor.diagnostics")
|
|
|
|
|
|
class DiagnosticsRunner:
|
|
"""
|
|
Orchestrates diagnostic checks on workflow snapshots.
|
|
|
|
Responsibilities:
|
|
- Accept workflow snapshot + scope + options
|
|
- Canonicalize workflow and compute hash
|
|
- Execute registered checks with timeouts
|
|
- Aggregate issues and compute health score
|
|
- Optionally compute intent signature
|
|
- Cache results for short TTL
|
|
"""
|
|
|
|
# Configuration
|
|
DEFAULT_CHECK_TIMEOUT_MS = 2000 # 2s per check
|
|
DEFAULT_GLOBAL_TIMEOUT_MS = 5000 # 5s total
|
|
CACHE_TTL_SECONDS = 60 # Cache results for 60s
|
|
|
|
def __init__(self):
|
|
self._checks: List[Callable[[Dict[str, Any], HealthCheckRequest], Awaitable[List[HealthIssue]]]] = []
|
|
self._check_names: List[str] = []
|
|
self._cache: Dict[str, tuple[HealthReport, float]] = {} # (report, timestamp)
|
|
self._last_report: Optional[HealthReport] = None
|
|
self._intent_scorer: Optional[Any] = None # Set by services/intent
|
|
|
|
def register_check(
|
|
self,
|
|
name: str,
|
|
check_fn: Callable[[Dict[str, Any], HealthCheckRequest], Awaitable[List[HealthIssue]]],
|
|
):
|
|
"""
|
|
Register a diagnostic check function.
|
|
|
|
Args:
|
|
name: Human-readable check name for logging
|
|
check_fn: Async function that takes (workflow, request) and returns List[HealthIssue]
|
|
"""
|
|
if name in self._check_names:
|
|
logger.warning(f"Diagnostic check '{name}' already registered, skipping duplicate.")
|
|
return
|
|
|
|
self._checks.append(check_fn)
|
|
self._check_names.append(name)
|
|
logger.info(f"Registered diagnostic check: {name}")
|
|
|
|
def set_intent_scorer(self, scorer: Any):
|
|
"""Set the intent signature scorer (from services/intent)."""
|
|
self._intent_scorer = scorer
|
|
|
|
def canonicalize_workflow(self, workflow: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Canonicalize workflow for deterministic hashing.
|
|
|
|
Normalizes:
|
|
- Sort keys
|
|
- Strip UI-only fields (position, size, color, etc.)
|
|
- Normalize node IDs to strings
|
|
- Sort nodes by ID for stable ordering
|
|
- Exclude viewport/UI state (extra.ds)
|
|
"""
|
|
if not workflow:
|
|
return {}
|
|
|
|
# Fields to strip (UI-only, no semantic meaning)
|
|
strip_fields = {"pos", "size", "color", "bgcolor", "shape", "flags", "order"}
|
|
|
|
def normalize_node(node: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Normalize a single node, converting id to string."""
|
|
normalized = {}
|
|
for k, v in sorted(node.items()):
|
|
if k in strip_fields:
|
|
continue
|
|
# Normalize node id to string for consistency
|
|
if k == "id" and v is not None:
|
|
normalized[k] = str(v)
|
|
else:
|
|
normalized[k] = v
|
|
return normalized
|
|
|
|
nodes = workflow.get("nodes", [])
|
|
links = workflow.get("links", [])
|
|
|
|
# Normalize nodes and sort by id for deterministic ordering
|
|
normalized_nodes = []
|
|
if isinstance(nodes, list):
|
|
for n in nodes:
|
|
if isinstance(n, dict):
|
|
normalized_nodes.append(normalize_node(n))
|
|
# Sort by id (now string) for stable ordering
|
|
normalized_nodes.sort(key=lambda x: x.get("id", ""))
|
|
|
|
canonical = {
|
|
"nodes": normalized_nodes,
|
|
"links": sorted(links, key=lambda x: x[0] if isinstance(x, list) and x else 0) if isinstance(links, list) else [],
|
|
}
|
|
|
|
# Include extra graph properties if present
|
|
# NOTE: Exclude 'ds' (viewport/zoom state) as it's UI-only
|
|
if "extra" in workflow and isinstance(workflow["extra"], dict):
|
|
extra = workflow["extra"]
|
|
# Only include workflow-relevant extra fields (not ds)
|
|
workflow_extra = {k: v for k, v in extra.items() if k not in ("ds",)}
|
|
if workflow_extra:
|
|
canonical["extra"] = workflow_extra
|
|
|
|
return canonical
|
|
|
|
def compute_workflow_hash(self, workflow: Dict[str, Any]) -> str:
|
|
"""Compute deterministic hash of canonicalized workflow."""
|
|
canonical = self.canonicalize_workflow(workflow)
|
|
json_str = json.dumps(canonical, sort_keys=True, ensure_ascii=False)
|
|
return hashlib.sha256(json_str.encode()).hexdigest()[:32]
|
|
|
|
def _get_cache_key(self, workflow_hash: str, request: HealthCheckRequest) -> str:
|
|
"""Generate cache key from workflow hash and options."""
|
|
options_str = f"{request.include_intent}:{request.max_paths}"
|
|
return f"{workflow_hash}:{options_str}"
|
|
|
|
def _get_cached_report(self, cache_key: str) -> Optional[HealthReport]:
|
|
"""Get cached report if still valid."""
|
|
if cache_key in self._cache:
|
|
report, timestamp = self._cache[cache_key]
|
|
if time.time() - timestamp < self.CACHE_TTL_SECONDS:
|
|
logger.debug(f"Cache hit for {cache_key[:16]}...")
|
|
return report
|
|
else:
|
|
del self._cache[cache_key]
|
|
return None
|
|
|
|
def _cache_report(self, cache_key: str, report: HealthReport):
|
|
"""Cache a report with current timestamp."""
|
|
self._cache[cache_key] = (report, time.time())
|
|
# Limit cache size
|
|
if len(self._cache) > 100:
|
|
# Remove oldest entries
|
|
oldest = sorted(self._cache.items(), key=lambda x: x[1][1])[:50]
|
|
for key, _ in oldest:
|
|
del self._cache[key]
|
|
|
|
async def _run_check_with_timeout(
|
|
self,
|
|
name: str,
|
|
check_fn: Callable[[Dict[str, Any], HealthCheckRequest], Awaitable[List[HealthIssue]]],
|
|
workflow: Dict[str, Any],
|
|
request: HealthCheckRequest,
|
|
timeout_ms: int,
|
|
) -> List[HealthIssue]:
|
|
"""Run a single check with timeout and error handling."""
|
|
try:
|
|
result = await asyncio.wait_for(
|
|
check_fn(workflow, request),
|
|
timeout=timeout_ms / 1000.0,
|
|
)
|
|
return result if isinstance(result, list) else []
|
|
except asyncio.TimeoutError:
|
|
logger.warning(f"Check '{name}' timed out after {timeout_ms}ms")
|
|
return []
|
|
except Exception as e:
|
|
logger.error(f"Check '{name}' failed: {e}")
|
|
return []
|
|
|
|
async def run(self, request: HealthCheckRequest) -> HealthReport:
|
|
"""
|
|
Run diagnostics on a workflow.
|
|
|
|
Args:
|
|
request: Health check request with workflow and options
|
|
|
|
Returns:
|
|
HealthReport with issues, score, and optional intent signature
|
|
"""
|
|
start_time = time.time()
|
|
workflow = request.workflow or {}
|
|
workflow_hash = self.compute_workflow_hash(workflow)
|
|
|
|
# Check cache
|
|
cache_key = self._get_cache_key(workflow_hash, request)
|
|
cached = self._get_cached_report(cache_key)
|
|
if cached:
|
|
return cached
|
|
|
|
# Run all checks concurrently with global timeout
|
|
all_issues: List[HealthIssue] = []
|
|
|
|
if self._checks:
|
|
try:
|
|
check_tasks = [
|
|
self._run_check_with_timeout(
|
|
name,
|
|
check_fn,
|
|
workflow,
|
|
request,
|
|
self.DEFAULT_CHECK_TIMEOUT_MS,
|
|
)
|
|
for name, check_fn in zip(self._check_names, self._checks)
|
|
]
|
|
|
|
results = await asyncio.wait_for(
|
|
asyncio.gather(*check_tasks, return_exceptions=True),
|
|
timeout=self.DEFAULT_GLOBAL_TIMEOUT_MS / 1000.0,
|
|
)
|
|
|
|
for result in results:
|
|
if isinstance(result, list):
|
|
all_issues.extend(result)
|
|
|
|
except asyncio.TimeoutError:
|
|
logger.warning("Global diagnostics timeout reached")
|
|
except Exception as e:
|
|
logger.error(f"Diagnostics runner error: {e}")
|
|
|
|
# Compute intent signature if requested
|
|
intent_signature = None
|
|
if request.include_intent and self._intent_scorer:
|
|
try:
|
|
intent_signature = await self._intent_scorer.compute(
|
|
workflow, workflow_hash
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Intent signature computation failed: {e}")
|
|
|
|
# Compute score and counts
|
|
health_score = HealthReport.compute_health_score(all_issues)
|
|
counts = HealthReport.count_issues(all_issues)
|
|
|
|
# Build report
|
|
duration_ms = int((time.time() - start_time) * 1000)
|
|
report = HealthReport(
|
|
report_id=str(uuid.uuid4()),
|
|
timestamp=utc_isoformat(),
|
|
duration_ms=duration_ms,
|
|
scope=request.scope,
|
|
workflow_hash=workflow_hash,
|
|
health_score=health_score,
|
|
counts=counts,
|
|
issues=all_issues,
|
|
intent_signature=intent_signature,
|
|
)
|
|
|
|
# Cache and store
|
|
self._cache_report(cache_key, report)
|
|
self._last_report = report
|
|
|
|
logger.info(
|
|
f"Diagnostics completed: score={health_score}, "
|
|
f"issues={len(all_issues)}, duration={duration_ms}ms"
|
|
)
|
|
|
|
return report
|
|
|
|
def get_last_report(self) -> Optional[HealthReport]:
|
|
"""Get the last computed report (cached)."""
|
|
return self._last_report
|
|
|
|
def clear_cache(self):
|
|
"""Clear the report cache."""
|
|
self._cache.clear()
|
|
|
|
|
|
# Global instance
|
|
_runner: Optional[DiagnosticsRunner] = None
|
|
|
|
|
|
def get_diagnostics_runner() -> DiagnosticsRunner:
|
|
"""Get or create the global DiagnosticsRunner instance."""
|
|
global _runner
|
|
if _runner is None:
|
|
_runner = DiagnosticsRunner()
|
|
return _runner
|