File size: 1,662 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 | from __future__ import annotations
import os
from .models import DryRunCandidate, DryRunReport, ProcessGroup
def build_dry_run(groups: list[ProcessGroup], *, current_pid: int | None = None) -> DryRunReport:
current_pid = os.getpid() if current_pid is None else current_pid
candidates: list[DryRunCandidate] = []
exclusions: list[str] = []
for group in groups:
for node in group.stale_candidates:
if node.sample.pid == current_pid:
exclusions.append(f"PID {node.sample.pid}: dashboard self-process")
continue
if node.sample.uid not in {-1, os.getuid()}:
exclusions.append(f"PID {node.sample.pid}: not current UID")
continue
if not (node.classification.is_bridge or node.classification.is_mcp):
exclusions.append(f"PID {node.sample.pid}: not MCP/bridge")
continue
reasons = tuple(node.stale_reasons + node.risk_reasons + group.risk_reasons)
candidates.append(
DryRunCandidate(
pid=node.sample.pid,
process_key=node.sample.process_key,
known_class=node.classification.known_class,
age_seconds=node.sample.age_seconds,
rss_bytes=node.sample.rss_bytes,
command=node.sample.argv_display,
reasons=reasons or ("stale MCP/bridge candidate",),
)
)
candidates.sort(key=lambda item: (-item.age_seconds, item.pid))
return DryRunReport(candidates=tuple(candidates), exclusions=tuple(exclusions))
|