| from __future__ import annotations |
|
|
| from .models import ProcessGroup, ProcessNode, max_risk |
|
|
|
|
| STALE_CANDIDATE_SECONDS = 6 * 3600 |
| STRONG_STALE_SECONDS = 24 * 3600 |
|
|
|
|
| def assess_node(node: ProcessNode) -> None: |
| node.risk_level = "ok" |
| node.risk_reasons.clear() |
| node.stale_state = "healthy" |
| node.stale_reasons.clear() |
|
|
| if node.classification.is_bridge or node.classification.is_mcp: |
| if node.sample.age_seconds >= STRONG_STALE_SECONDS: |
| node.stale_state = "strong_stale_candidate" |
| node.stale_reasons.append("age over 24h") |
| node.risk_level = "warn" |
| node.risk_reasons.append("strong stale candidate") |
| elif node.sample.age_seconds >= STALE_CANDIDATE_SECONDS: |
| node.stale_state = "stale_candidate" |
| node.stale_reasons.append("age over 6h") |
| node.risk_level = "info" |
| node.risk_reasons.append("stale candidate") |
|
|
| if node.sample.state == "Z": |
| node.risk_level = max_risk(node.risk_level, "warn") |
| node.risk_reasons.append("zombie process") |
|
|
|
|
| def assess_group(group: ProcessGroup) -> None: |
| for node in group.nodes: |
| assess_node(node) |
|
|
| reasons: list[str] = [] |
| level = "ok" |
| if group.child_count >= 50: |
| level = "critical" |
| reasons.append("child count >= 50") |
| elif group.child_count >= 20: |
| level = max_risk(level, "warn") |
| reasons.append("child count >= 20") |
|
|
| if group.bridge_count >= 12: |
| level = max_risk(level, "critical") |
| reasons.append("bridge count >= 12") |
| elif group.bridge_count >= 6: |
| level = max_risk(level, "warn") |
| reasons.append("bridge count >= 6") |
| elif group.bridge_count >= 2: |
| level = max_risk(level, "info") |
| reasons.append("bridge count >= 2") |
|
|
| stale = [ |
| node |
| for node in group.nodes |
| if node.stale_state in {"stale_candidate", "strong_stale_candidate"} |
| ] |
| if stale: |
| level = max_risk(level, "info") |
| reasons.append(f"{len(stale)} stale candidate(s)") |
|
|
| for node in group.nodes: |
| if node.risk_level in {"warn", "critical"}: |
| level = max_risk(level, node.risk_level) |
|
|
| group.risk_level = level |
| group.risk_reasons = reasons |
| group.stale_candidates = stale |
|
|
|
|
| def assess_groups(groups: list[ProcessGroup]) -> tuple[dict[str, int], list[str]]: |
| for group in groups: |
| assess_group(group) |
|
|
| bridge_total = sum(group.bridge_count for group in groups) |
| mcp_total = sum(group.mcp_count for group in groups) |
| agent_related_total = sum(len(group.nodes) for group in groups) |
| stale_total = sum(len(group.stale_candidates) for group in groups) |
| max_fanout = max((group.child_count for group in groups), default=0) |
| cpu_hot = sum( |
| 1 for group in groups for node in group.nodes if node.sample.cpu_percent_interval >= 80 |
| ) |
|
|
| warnings: list[str] = [] |
| if mcp_total >= 50 or bridge_total >= 50: |
| warnings.append("critical: global MCP/bridge count >= 50") |
| elif mcp_total >= 20 or bridge_total >= 20: |
| warnings.append("warn: global MCP/bridge count >= 20") |
|
|
| if agent_related_total >= 70: |
| warnings.append("critical: agent-related process count >= 70") |
| elif agent_related_total >= 30: |
| warnings.append("warn: agent-related process count >= 30") |
|
|
| stats = { |
| "agent_roots": len(groups), |
| "mcp_count": mcp_total, |
| "bridge_count": bridge_total, |
| "agent_related_count": agent_related_total, |
| "stale_candidates": stale_total, |
| "max_fanout": max_fanout, |
| "cpu_hot": cpu_hot, |
| } |
| return stats, warnings |
|
|
|
|
