File size: 3,657 Bytes
06357e7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | 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
|