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