"""Hermes plugin for M2 Solution Scout v0. The plugin is intentionally fail-closed: no policy opt-in means no hooks do anything beyond loading local config, and raw-source deny keys refuse startup. """ from __future__ import annotations import datetime as _dt import hashlib import json import os import random import re import shutil import socket import subprocess import threading import time from pathlib import Path from typing import Any from urllib.parse import quote POLICY_PATH = Path("~/.config/m2-market/pull-policy.toml").expanduser() STATE_DIR = Path("~/.local/share/m2-market/scout").expanduser() STATE_PATH = STATE_DIR / "state.json" OUTBOX_DIR = STATE_DIR / "outbox" _SECRET_PATTERNS = [ re.compile(r"sk-[A-Za-z0-9_-]{20,}"), re.compile(r"(?i)\b(api[_ -]?key|secret|token)\s*[:=]\s*['\"]?[^'\"\s]{8,}"), ] _EMAIL_RE = re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b") _WORD_RE = re.compile(r"[A-Za-z][A-Za-z0-9_-]{2,}") class PolicyError(RuntimeError): """Policy explicitly requests an unsafe mode.""" def _utcnow() -> _dt.datetime: return _dt.datetime.now(_dt.timezone.utc) def _iso(ts: _dt.datetime | None = None) -> str: return (ts or _utcnow()).replace(microsecond=0).isoformat().replace("+00:00", "Z") def _parse_scalar(value: str) -> Any: value = value.strip() if not value or value.startswith("#"): return "" if "#" in value and not value.lstrip().startswith(("'", '"')): value = value.split("#", 1)[0].strip() if value in {"true", "false"}: return value == "true" if (value.startswith('"') and value.endswith('"')) or ( value.startswith("'") and value.endswith("'") ): return value[1:-1] if value.startswith("[") and value.endswith("]"): inner = value[1:-1].strip() if not inner: return [] return [_parse_scalar(part.strip()) for part in inner.split(",")] try: if "." in value: return float(value) return int(value) except ValueError: return value def parse_toml(text: str) -> dict[str, Any]: """Tiny TOML subset parser, with optional tomli avoided for Python 3.10 hosts.""" data: dict[str, Any] = {} current = data for raw in text.splitlines(): line = raw.strip() if not line or line.startswith("#"): continue if line.startswith("[") and line.endswith("]"): current = data for part in line[1:-1].split("."): current = current.setdefault(part.strip(), {}) continue if "=" not in line: continue key, value = line.split("=", 1) current[key.strip()] = _parse_scalar(value) return data def _read_toml(path: Path) -> dict[str, Any]: try: import tomli # type: ignore except Exception: tomli = None if not path.exists(): return {} raw = path.read_bytes() if tomli is not None: return tomli.loads(raw.decode("utf-8")) return parse_toml(raw.decode("utf-8")) def _section(data: dict[str, Any], *names: str) -> dict[str, Any]: node: Any = data for name in names: if not isinstance(node, dict): return {} node = node.get(name, {}) return node if isinstance(node, dict) else {} def load_policy(path: Path = POLICY_PATH) -> dict[str, Any]: raw = _read_toml(path) scout = _section(raw, "scout") sources = _section(raw, "scout", "sources") summary = _section(raw, "scout", "summary") catalog = _section(raw, "scout", "catalog") store = _section(raw, "scout", "store") telemetry = _section(raw, "scout", "telemetry") enabled = bool(scout.get("enabled", False)) tenant_id = str(scout.get("tenant_id", "") or "").strip() if sources.get("raw_keystrokes") is True or sources.get("raw_client_data") is True: raise PolicyError("Solution Scout refuses raw_keystrokes/raw_client_data policy") if not enabled or not tenant_id: return {"enabled": False, "reason": "disabled_or_missing_tenant"} return { "enabled": True, "mode": scout.get("mode", "suggest"), "desktop_id": str(scout.get("desktop_id", "unknown-desktop")), "operator_id": str(scout.get("operator_id", "unknown-operator")), "tenant_id": tenant_id, "min_score": float(scout.get("min_score", 0.78)), "min_coverage": float(scout.get("min_coverage", 0.55)), "max_proposals_per_day": int(scout.get("max_proposals_per_day", 5)), "max_proposals_per_session": int(scout.get("max_proposals_per_session", 1)), "dismiss_cooldown_days": int(scout.get("dismiss_cooldown_days", 14)), "proposal_cooldown_minutes": int(scout.get("proposal_cooldown_minutes", 45)), "summary": { "max_source_chars": int(summary.get("max_source_chars", 12000)), "max_summary_chars": int(summary.get("max_summary_chars", 900)), "redact_secrets": bool(summary.get("redact_secrets", True)), "redact_client_identifiers": bool(summary.get("redact_client_identifiers", True)), "fail_closed_on_secret": bool(summary.get("fail_closed_on_secret", True)), }, "catalog": { "memory_api_url": str(catalog.get("memory_api_url", "https://memory.machinemachine.ai")), "agent_id": str(catalog.get("agent_id", "market:catalog")), "limit": int(catalog.get("limit", 5)), "routing_strategy": str(catalog.get("routing_strategy", "standard")), }, "store": { "deeplink_scheme": str(store.get("deeplink_scheme", "m2store")), "fallback_command": str(store.get("fallback_command", "m2-market show --json")), }, "telemetry": {"enabled": bool(telemetry.get("enabled", True))}, } def redact(text: str, *, redact_client_identifiers: bool = True) -> tuple[str, dict[str, bool]]: found_secret = False redacted = text for pattern in _SECRET_PATTERNS: redacted, count = pattern.subn("[REDACTED_SECRET]", redacted) found_secret = found_secret or count > 0 found_client = False if redact_client_identifiers: redacted, count = _EMAIL_RE.subn("[REDACTED_EMAIL]", redacted) found_client = count > 0 return redacted, { "secrets_redacted": found_secret, "client_identifiers_redacted": found_client, "failed_closed": False, } def summarize_turn(user_message: str, assistant_response: str, policy: dict[str, Any]) -> dict[str, Any]: summary_policy = policy["summary"] source = (user_message + "\n" + assistant_response)[: summary_policy["max_source_chars"]] clean, scrub = redact( source, redact_client_identifiers=summary_policy.get("redact_client_identifiers", True), ) sentences = re.split(r"(?<=[.!?])\s+", " ".join(clean.split())) summary = " ".join(sentences[:3]).strip() if len(summary) > summary_policy["max_summary_chars"]: summary = summary[: summary_policy["max_summary_chars"]].rsplit(" ", 1)[0].strip() terms = [] for word in _WORD_RE.findall(summary.lower()): if word in {"the", "and", "with", "this", "that", "from", "have", "will", "into"}: continue if word not in terms: terms.append(word) if len(terms) >= 12: break intent_hash = "sha256:" + hashlib.sha256(summary.encode("utf-8")).hexdigest() return { "schema_version": "m2.scout.intent_summary.v1", "summary_id": _id("sum", policy), "observation_ids": [], "desktop_id": policy["desktop_id"], "operator_id": policy["operator_id"], "tenant_id": policy["tenant_id"], "session_id": "", "summary": summary, "intent_terms": terms, "negative_terms": [], "scrub_status": scrub, "intent_hash": intent_hash, "created_at": _iso(), } def _id(prefix: str, policy: dict[str, Any]) -> str: stamp = _utcnow().strftime("%Y%m%d") desktop = re.sub(r"[^A-Za-z0-9]+", "", policy.get("desktop_id", "desktop"))[:12] or "desktop" return f"{prefix}_{stamp}_{desktop}_{random.randrange(36**4):04x}" def _load_state(path: Path = STATE_PATH) -> dict[str, Any]: try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return {"daily": {}, "sessions": {}, "dismissed": {}, "last_proposal_at": None} def _save_state(state: dict[str, Any], path: Path = STATE_PATH) -> None: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(".tmp") tmp.write_text(json.dumps(state, sort_keys=True, indent=2) + "\n", encoding="utf-8") tmp.replace(path) def gate(policy: dict[str, Any], state: dict[str, Any], *, session_id: str = "") -> tuple[bool, str | None]: if not policy.get("enabled"): return False, "disabled" today = _utcnow().date().isoformat() daily_count = int(state.get("daily", {}).get(today, 0)) if daily_count >= policy["max_proposals_per_day"]: return False, "rate_capped" if session_id: session_count = int(state.get("sessions", {}).get(session_id, 0)) if session_count >= policy["max_proposals_per_session"]: return False, "session_capped" last = state.get("last_proposal_at") if last: try: last_ts = _dt.datetime.fromisoformat(str(last).replace("Z", "+00:00")) cooldown = _dt.timedelta(minutes=policy["proposal_cooldown_minutes"]) if _utcnow() - last_ts < cooldown: return False, "proposal_cooldown" except ValueError: pass return True, None def _memory_api_key() -> str: if os.environ.get("M2_MEMORY_API_KEY"): return os.environ["M2_MEMORY_API_KEY"] config = _read_toml(Path("~/.m2-market/config.toml").expanduser()) memory = _section(config, "memory") for key in ("api_key", "memory_api_key", "token"): value = memory.get(key) or config.get(key) if value: return str(value) return "" def _visible(listing: dict[str, Any], tenant_id: str) -> bool: visibility = listing.get("tenant_visibility") or [] status = listing.get("status", "published") return status == "published" and ("*" in visibility or tenant_id in visibility) def search_catalog(summary: dict[str, Any], policy: dict[str, Any]) -> list[dict[str, Any]]: import httpx catalog = policy["catalog"] headers = {} api_key = _memory_api_key() if api_key: headers["X-API-Key"] = api_key with httpx.Client(base_url=catalog["memory_api_url"], headers=headers, timeout=10.0) as client: response = client.post( "/memory/search", json={ "query": summary["summary"], "agent_id": catalog["agent_id"], "routing_strategy": catalog["routing_strategy"], "limit": max(catalog["limit"] * 3, catalog["limit"]), }, ) response.raise_for_status() items = [] for result in response.json().get("results", []): meta = result.get("metadata") or {} listing = meta.get("listing") or {} listing_id = meta.get("listing_id") or listing.get("listing_id") or listing.get("id") if not listing_id or not _visible(listing, policy["tenant_id"]): continue score = float(result.get("score") or 0.0) coverage = _estimate_coverage(summary, listing) if score < policy["min_score"] or coverage < policy["min_coverage"]: continue items.append( { "listing_id": listing_id, "solution_id": listing.get("solution_id", ""), "score": score, "coverage": coverage, "listing": listing, } ) if len(items) >= catalog["limit"]: break return items def _estimate_coverage(summary: dict[str, Any], listing: dict[str, Any]) -> float: terms = set(summary.get("intent_terms") or []) haystack = " ".join( str(listing.get(k, "")) for k in ("name", "summary", "description", "category") ).lower() keywords = listing.get("keywords") or [] if isinstance(keywords, list): haystack += " " + " ".join(str(k).lower() for k in keywords) if not terms: return 0.0 hits = sum(1 for term in terms if term in haystack) return round(min(1.0, hits / max(3, len(terms))), 2) def build_proposal(match: dict[str, Any], summary: dict[str, Any], policy: dict[str, Any]) -> dict[str, Any]: listing = match["listing"] proposal_id = _id("prp", policy) match_id = _id("mat", policy) listing_id = match["listing_id"] deeplink = ( f"{policy['store']['deeplink_scheme']}://listing/{quote(listing_id)}" f"?proposal_id={quote(proposal_id)}&source=scout" ) now = _utcnow() return { "schema_version": "m2.scout.proposal.v1", "proposal_id": proposal_id, "match_id": match_id, "desktop_id": policy["desktop_id"], "operator_id": policy["operator_id"], "tenant_id": policy["tenant_id"], "listing": { "listing_id": listing_id, "name": listing.get("name", listing_id), "summary": listing.get("summary", ""), "price": listing.get("price", {}), "stats": listing.get("stats", {}), }, "confidence": match["score"], "coverage": match["coverage"], "deeplink": deeplink, "shown_at": _iso(now), "expires_at": _iso(now + _dt.timedelta(minutes=policy["proposal_cooldown_minutes"])), "_telemetry": { "session_id": summary.get("session_id", ""), "intent_hash": summary["intent_hash"], "solution_id": match.get("solution_id", ""), }, } def write_proposal_shown(proposal: dict[str, Any], outbox_dir: Path = OUTBOX_DIR) -> dict[str, Any]: telemetry = proposal.pop("_telemetry", {}) event = { "schema_version": "m2.scout.telemetry.v1", "event_id": _id("evt", proposal), "event_type": "proposal_shown", "proposal_id": proposal["proposal_id"], "listing_id": proposal["listing"]["listing_id"], "solution_id": telemetry.get("solution_id", ""), "operator_id": proposal["operator_id"], "desktop_id": proposal["desktop_id"], "tenant_id": proposal["tenant_id"], "session_id": telemetry.get("session_id", ""), "intent_hash": telemetry.get("intent_hash", ""), "score": proposal["confidence"], "coverage": proposal["coverage"], "reason": None, "created_at": proposal["shown_at"], "proposal": proposal, } outbox_dir.mkdir(parents=True, exist_ok=True) path = outbox_dir / f"{_utcnow().date().isoformat()}.jsonl" with path.open("a", encoding="utf-8") as fh: fh.write(json.dumps(event, sort_keys=True) + "\n") return event def notify(proposal: dict[str, Any]) -> None: listing = proposal["listing"] price = listing.get("price") or {} amount = price.get("amount", "?") currency = price.get("currency", "m2cr") body = ( f"This looks like {listing['name']}: ~{int(proposal['coverage'] * 100)}% coverage, " f"{amount} {currency}. Open in M2 Store?" ) if shutil.which("notify-send"): try: subprocess.Popen( [ "notify-send", "M2 Solution Scout", body, f"--action=open=Open", f"--action=dismiss=Dismiss", ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) return except Exception: pass _herdr_toast(body, proposal["deeplink"]) def _herdr_toast(message: str, deeplink: str) -> None: socket_path = os.environ.get("HERDR_SOCKET_PATH", "").strip() pane_id = os.environ.get("HERDR_PANE_ID", "").strip() if not socket_path or not pane_id: return request = { "id": f"m2-solution-scout:{int(time.time() * 1000)}", "method": "pane.toast", "params": { "pane_id": pane_id, "source": "m2-solution-scout", "title": "M2 Solution Scout", "message": message, "deeplink": deeplink, }, } try: client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) client.settimeout(0.5) client.connect(socket_path) client.sendall((json.dumps(request) + "\n").encode("utf-8")) client.close() except Exception: pass def _bump_state( state: dict[str, Any], policy: dict[str, Any], session_id: str, proposal: dict[str, Any] ) -> None: today = _utcnow().date().isoformat() state.setdefault("daily", {})[today] = int(state.setdefault("daily", {}).get(today, 0)) + 1 if session_id: state.setdefault("sessions", {})[session_id] = int( state.setdefault("sessions", {}).get(session_id, 0) ) + 1 state["last_proposal_at"] = _iso() state["last_proposal"] = { "proposal_id": proposal["proposal_id"], "listing_id": proposal["listing"]["listing_id"], "cooldown_until": proposal["expires_at"], } state.setdefault("proposals", []).append(state["last_proposal"]) _save_state(state) def _extract_turn(kwargs: dict[str, Any]) -> tuple[str, str, str]: user_message = str(kwargs.get("user_message") or kwargs.get("prompt") or "") assistant_response = str(kwargs.get("assistant_response") or kwargs.get("response") or "") if not user_message: messages = kwargs.get("messages") if isinstance(messages, list): user_message = "\n".join( str(m.get("content", "")) for m in messages if isinstance(m, dict) and m.get("role") == "user" ) session_id = str(kwargs.get("session_id") or "") return user_message, assistant_response, session_id class ScoutRuntime: def __init__(self) -> None: self.policy: dict[str, Any] = {"enabled": False} self._lock = threading.Lock() def on_session_start(self, **kwargs: Any) -> None: del kwargs self.policy = load_policy() def post_llm_call(self, **kwargs: Any) -> None: user_message, assistant_response, session_id = _extract_turn(kwargs) if not user_message and not assistant_response: return state = _load_state() ok, _reason = gate(self.policy, state, session_id=session_id) if not ok: return worker = threading.Thread( target=self._process, args=(user_message, assistant_response, session_id, state), daemon=True, ) worker.start() def _process( self, user_message: str, assistant_response: str, session_id: str, state: dict[str, Any] ) -> None: with self._lock: try: summary = summarize_turn(user_message, assistant_response, self.policy) summary["session_id"] = session_id matches = search_catalog(summary, self.policy) if not matches: return proposal = build_proposal(matches[0], summary, self.policy) write_proposal_shown(proposal) notify(proposal) _bump_state(state, self.policy, session_id, proposal) except Exception: return _RUNTIME = ScoutRuntime() def register(ctx: Any) -> None: ctx.register_hook("on_session_start", _RUNTIME.on_session_start) ctx.register_hook("post_llm_call", _RUNTIME.post_llm_call)