File size: 2,982 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 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
RISK_ORDER = {"ok": 0, "info": 1, "warn": 2, "critical": 3}
def max_risk(*levels: str) -> str:
return max(levels or ("ok",), key=lambda level: RISK_ORDER.get(level, 0))
@dataclass(frozen=True)
class ProcessSample:
sample_ts: float
boot_id: str
pid: int
ppid: int
pgid: int
sid: int
uid: int
user: str
comm: str
state: str
start_time_ticks: int
process_key: str
age_seconds: float
rss_bytes: int
vms_bytes: int
utime_ticks: int
stime_ticks: int
cpu_percent_interval: float
num_threads: int
argv: tuple[str, ...]
argv_display: str
argv_masked_count: int
exe_path: str = ""
cwd: str = ""
cgroup: str = ""
read_errors: tuple[str, ...] = ()
@property
def cpu_ticks(self) -> int:
return self.utime_ticks + self.stime_ticks
@property
def rss_mb(self) -> float:
return self.rss_bytes / 1024 / 1024
@dataclass(frozen=True)
class Classification:
known_class: str
confidence: float
reasons: tuple[str, ...] = ()
is_agent_related: bool = False
is_bridge: bool = False
is_mcp: bool = False
@property
def is_agent(self) -> bool:
return self.known_class.startswith("agent.")
@dataclass
class ProcessNode:
sample: ProcessSample
classification: Classification
parent_key: str | None = None
children: list["ProcessNode"] = field(default_factory=list)
inherited_agent: bool = False
stale_state: str = "healthy"
stale_reasons: list[str] = field(default_factory=list)
risk_level: str = "ok"
risk_reasons: list[str] = field(default_factory=list)
@property
def process_key(self) -> str:
return self.sample.process_key
@property
def pid(self) -> int:
return self.sample.pid
@dataclass
class ProcessGroup:
group_id: str
root: ProcessNode
nodes: list[ProcessNode]
label: str
child_count: int
bridge_count: int
mcp_count: int
total_rss_bytes: int
max_age_seconds: float
risk_level: str = "ok"
risk_reasons: list[str] = field(default_factory=list)
stale_candidates: list[ProcessNode] = field(default_factory=list)
@dataclass(frozen=True)
class DryRunCandidate:
pid: int
process_key: str
known_class: str
age_seconds: float
rss_bytes: int
command: str
reasons: tuple[str, ...]
@dataclass(frozen=True)
class DryRunReport:
candidates: tuple[DryRunCandidate, ...]
exclusions: tuple[str, ...]
note: str = "Dry-run only. No signal or executable cleanup command is generated."
@dataclass
class DashboardSnapshot:
sample_ts: float
processes: list[ProcessSample]
classifications: dict[str, Classification]
nodes: dict[str, ProcessNode]
groups: list[ProcessGroup]
dry_run: DryRunReport
stats: dict[str, Any]
warnings: list[str]
|