| 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)) | |