nearxos
6185b9b85a
Self-hosted, Docker-based agentic troubleshooting platform: FastAPI backend + LangGraph agent, Next.js UI, tiered LLM routing (local Ollama -> Gemini -> DeepSeek -> OpenRouter), MCP server manager, encrypted device credentials, RBAC, audit log, project-memory + Obsidian integrations, and editable troubleshooting decision rules tuned for the GeneseasX vessel stack. Co-authored-by: Cursor <cursoragent@cursor.com>
298 lines
9.0 KiB
Python
298 lines
9.0 KiB
Python
"""Editable troubleshooting decision rules (YAML on disk).
|
|
|
|
Rules standardize which devices to check and in what order, plus hints for the LLM.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass
|
|
|
|
import yaml
|
|
|
|
from app.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_rules_cache: dict | None = None
|
|
_rules_mtime: float = -1.0
|
|
|
|
|
|
@dataclass
|
|
class MatchedRule:
|
|
id: str
|
|
name: str
|
|
severity: str
|
|
devices: list[str]
|
|
order: list[str]
|
|
hints: list[str]
|
|
steps: list[str]
|
|
broad: bool = False
|
|
|
|
def as_dict(self) -> dict:
|
|
return {
|
|
"id": self.id,
|
|
"name": self.name,
|
|
"severity": self.severity,
|
|
"devices": self.devices,
|
|
"order": self.order,
|
|
"hints": self.hints,
|
|
"steps": self.steps,
|
|
"broad": self.broad,
|
|
}
|
|
|
|
|
|
def rules_path() -> str:
|
|
return os.environ.get("TROUBLESHOOTING_RULES_PATH", settings.troubleshooting_rules_path)
|
|
|
|
|
|
def _normalize(data: dict) -> dict:
|
|
data.setdefault("version", 1)
|
|
data.setdefault("broad_triggers", {"keywords": []})
|
|
data.setdefault("device_keywords", {})
|
|
data.setdefault("rules", [])
|
|
data.setdefault("default", {"severity": "medium", "devices": ["all"], "hints": []})
|
|
return data
|
|
|
|
|
|
def invalidate_rules_cache() -> None:
|
|
global _rules_cache, _rules_mtime
|
|
_rules_cache = None
|
|
_rules_mtime = -1.0
|
|
|
|
|
|
def load_rules(*, force: bool = False) -> dict:
|
|
global _rules_cache, _rules_mtime
|
|
path = rules_path()
|
|
try:
|
|
mtime = os.path.getmtime(path)
|
|
except OSError:
|
|
logger.warning("troubleshooting rules file missing: %s", path)
|
|
return _normalize({})
|
|
|
|
if not force and _rules_cache is not None and mtime == _rules_mtime:
|
|
return _rules_cache
|
|
|
|
try:
|
|
with open(path, encoding="utf-8") as fh:
|
|
data = _normalize(yaml.safe_load(fh) or {})
|
|
except yaml.YAMLError as exc:
|
|
logger.warning("invalid rules YAML %s: %s", path, exc)
|
|
data = _normalize({})
|
|
|
|
_rules_cache = data
|
|
_rules_mtime = mtime
|
|
return data
|
|
|
|
|
|
def read_rules_raw() -> str:
|
|
path = rules_path()
|
|
try:
|
|
with open(path, encoding="utf-8") as fh:
|
|
return fh.read()
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
def save_rules_raw(content: str) -> dict:
|
|
parsed = yaml.safe_load(content)
|
|
if not isinstance(parsed, dict):
|
|
raise ValueError("Rules file must be a YAML mapping at the top level")
|
|
normalized = _normalize(parsed)
|
|
path = rules_path()
|
|
parent = os.path.dirname(path)
|
|
if parent:
|
|
os.makedirs(parent, exist_ok=True)
|
|
with open(path, "w", encoding="utf-8") as fh:
|
|
fh.write(content if content.endswith("\n") else content + "\n")
|
|
invalidate_rules_cache()
|
|
return load_rules(force=True)
|
|
|
|
|
|
def _text_has_term(text: str, term: str) -> bool:
|
|
term = term.lower().strip()
|
|
if not term:
|
|
return False
|
|
if " " in term:
|
|
return term in text
|
|
if len(term) <= 2:
|
|
return term in text.split()
|
|
return bool(re.search(rf"\b{re.escape(term)}\b", text))
|
|
|
|
|
|
def is_broad_issue(title: str, issue: str, data: dict | None = None) -> bool:
|
|
data = data or load_rules()
|
|
text = f"{title} {issue}".lower()
|
|
keywords = data.get("broad_triggers", {}).get("keywords") or []
|
|
return any(kw in text for kw in keywords)
|
|
|
|
|
|
def _rule_match_score(text: str, rule: dict) -> int:
|
|
"""Higher score = more specific match. Zero means no match."""
|
|
match = rule.get("match") or {}
|
|
keywords = match.get("keywords") or []
|
|
all_of = match.get("all_of") or []
|
|
|
|
if all_of and not all(_text_has_term(text, t) for t in all_of):
|
|
return 0
|
|
|
|
score = 0
|
|
for kw in keywords:
|
|
if _text_has_term(text, kw):
|
|
# Multi-word phrases are more specific than single tokens like "sip".
|
|
score += 2 if " " in kw.strip() else 1
|
|
|
|
if all_of:
|
|
score += len(all_of) * 2
|
|
|
|
if score == 0 and not all_of:
|
|
return 0
|
|
return score
|
|
|
|
|
|
def match_rule(title: str, issue: str, data: dict | None = None) -> MatchedRule | None:
|
|
data = data or load_rules()
|
|
text = f"{title} {issue}".lower()
|
|
|
|
if is_broad_issue(title, issue, data):
|
|
default = data.get("default", {})
|
|
return MatchedRule(
|
|
id="broad-health",
|
|
name="Full stack health check",
|
|
severity=default.get("severity", "medium"),
|
|
devices=default.get("devices", ["all"]),
|
|
order=[],
|
|
hints=list(default.get("hints") or ["Check all onboard devices layer by layer"]),
|
|
steps=[],
|
|
broad=True,
|
|
)
|
|
|
|
rules = data.get("rules") or []
|
|
best: dict | None = None
|
|
best_score = 0
|
|
best_priority = -1
|
|
|
|
for rule in rules:
|
|
score = _rule_match_score(text, rule)
|
|
if score <= 0:
|
|
continue
|
|
priority = rule.get("priority") or 0
|
|
if score > best_score or (score == best_score and priority > best_priority):
|
|
best = rule
|
|
best_score = score
|
|
best_priority = priority
|
|
|
|
if not best:
|
|
return None
|
|
|
|
return MatchedRule(
|
|
id=str(best.get("id") or "unnamed"),
|
|
name=str(best.get("name") or best.get("id") or "Rule"),
|
|
severity=str(best.get("severity") or "medium"),
|
|
devices=list(best.get("devices") or ["all"]),
|
|
order=list(best.get("order") or []),
|
|
hints=list(best.get("hints") or []),
|
|
steps=list(best.get("steps") or []),
|
|
)
|
|
|
|
|
|
def device_keyword_map(data: dict | None = None) -> dict[str, list[str]]:
|
|
data = data or load_rules()
|
|
return {k: list(v) for k, v in (data.get("device_keywords") or {}).items()}
|
|
|
|
|
|
def devices_mentioned_in_text(text: str, devices: list[dict], data: dict | None = None) -> list[dict]:
|
|
"""Return onboard devices whose catalog keywords appear in *text*."""
|
|
data = data or load_rules()
|
|
kw_map = device_keyword_map(data)
|
|
matched_devices: list[dict] = []
|
|
for device in devices:
|
|
key = (device.get("catalog_key") or device.get("device_type") or "").lower()
|
|
keywords = kw_map.get(key, [])
|
|
terms = set(keywords + [key, device.get("device_type", ""), device.get("name", "")])
|
|
terms = {t.lower() for t in terms if t}
|
|
if any(_text_has_term(text, term) for term in terms if len(term) > 2):
|
|
matched_devices.append(device)
|
|
return matched_devices
|
|
|
|
|
|
def select_devices_for_issue(
|
|
devices: list[dict],
|
|
issue: str,
|
|
*,
|
|
title: str = "",
|
|
) -> tuple[list[dict], MatchedRule | None]:
|
|
if not devices:
|
|
return [], None
|
|
|
|
data = load_rules()
|
|
matched = match_rule(title, issue, data)
|
|
|
|
text = f"{title} {issue}".lower()
|
|
|
|
if matched:
|
|
if "all" in matched.devices:
|
|
selected = list(devices)
|
|
else:
|
|
want = {d.lower() for d in matched.devices}
|
|
selected = [d for d in devices if (d.get("catalog_key") or "").lower() in want]
|
|
# Union devices explicitly named in the issue (e.g. "proxmox and pfsense").
|
|
if not matched.broad:
|
|
seen = {(d.get("catalog_key") or "").lower() for d in selected}
|
|
for device in devices_mentioned_in_text(text, devices, data):
|
|
key = (device.get("catalog_key") or "").lower()
|
|
if key not in seen:
|
|
selected.append(device)
|
|
seen.add(key)
|
|
|
|
if matched.order:
|
|
order_map = {k.lower(): i for i, k in enumerate(matched.order)}
|
|
selected.sort(
|
|
key=lambda d: order_map.get((d.get("catalog_key") or "").lower(), 999)
|
|
)
|
|
|
|
if selected:
|
|
return selected, matched
|
|
|
|
if is_broad_issue(title, issue, data):
|
|
return devices, matched
|
|
|
|
matched_devices = devices_mentioned_in_text(text, devices, data)
|
|
|
|
# Never fall back to all devices — only run what the issue mentions.
|
|
return matched_devices, matched
|
|
|
|
|
|
def rule_guidance_block(matched: MatchedRule | dict | None) -> str:
|
|
if not matched:
|
|
return ""
|
|
if isinstance(matched, dict):
|
|
name = matched.get("name", "")
|
|
rid = matched.get("id", "")
|
|
severity = matched.get("severity", "medium")
|
|
devices = matched.get("devices") or []
|
|
order = matched.get("order") or []
|
|
hints = matched.get("hints") or []
|
|
steps = matched.get("steps") or []
|
|
else:
|
|
name, rid, severity = matched.name, matched.id, matched.severity
|
|
devices, order, hints, steps = matched.devices, matched.order, matched.hints, matched.steps
|
|
|
|
lines = [
|
|
"\n\n--- Troubleshooting rule (standardized) ---",
|
|
f"Rule: {name} ({rid})",
|
|
f"Severity hint: {severity}",
|
|
]
|
|
if devices:
|
|
lines.append(f"Target devices: {', '.join(devices)}")
|
|
if order:
|
|
lines.append(f"Diagnose order: {' → '.join(order)}")
|
|
if hints:
|
|
lines.append("Hints:")
|
|
lines.extend(f"- {h}" for h in hints)
|
|
if steps:
|
|
lines.append("Suggested steps:")
|
|
lines.extend(f"- {s}" for s in steps)
|
|
return "\n".join(lines)
|