Spaces:
Running
Running
| """Import-safe Hugging Face adapter over the canonical Nuclear Intelligence runtime. | |
| The Space UI uses this module only as a presentation gateway. Research, safety, | |
| evaluation, knowledge graph, ledger and governance decisions remain implemented | |
| once in the repository's canonical runtime. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import time | |
| from collections import deque | |
| from pathlib import Path | |
| from threading import RLock | |
| from typing import Any, Dict, List, Optional | |
| from core.runtime import RuntimeSettings, build_runtime, runtime_public_status | |
| from core.runtime_config import validate_secret | |
| from core.safety_guard import check_answer, check_query, render_safe_block | |
| try: | |
| from loguru import logger | |
| except Exception: # pragma: no cover | |
| import logging | |
| logger = logging.getLogger("core_hf") | |
| class HeadlessHFAdapter: | |
| """Small, presentation-safe facade over the canonical research runtime.""" | |
| def __init__(self) -> None: | |
| self._core = None | |
| self._loop = None | |
| self._ledger = None | |
| self.settings: Optional[RuntimeSettings] = None | |
| self._load_error: Optional[str] = None | |
| self._work_lock = RLock() | |
| self._public_events: deque[float] = deque() | |
| self._load() | |
| def ready(self) -> bool: | |
| return self._core is not None and self._loop is not None and self._ledger is not None | |
| def providers(self) -> List[str]: | |
| if not self._core: | |
| return [] | |
| return list( | |
| getattr(self._core.llm, "_available_providers", None) | |
| or getattr(self._core.llm, "_available", []) | |
| or [] | |
| ) | |
| def nes_supply(self) -> float: | |
| return float(getattr(self._ledger, "nes_supply", 0.0)) if self._ledger else 0.0 | |
| def _load(self) -> None: | |
| try: | |
| self._core, self._loop, self._ledger, self.settings = build_runtime() | |
| logger.info("Canonical HF adapter ready · providers=%s · nes=%s", self.providers, self.nes_supply) | |
| except Exception as exc: # pragma: no cover | |
| self._load_error = str(exc) | |
| logger.exception("Canonical HF adapter initialization failed") | |
| def _unavailable(self) -> Optional[Dict[str, Any]]: | |
| if not self.ready: | |
| return {"error": f"canonical runtime not initialized: {self._load_error or 'unknown error'}"} | |
| return None | |
| def public_status(self) -> Dict[str, Any]: | |
| status = runtime_public_status(self.settings) if self.settings else {"runtime": "canonical"} | |
| status.update({ | |
| "ready": self.ready, | |
| "providers": self.providers, | |
| "nes_supply": self.nes_supply, | |
| # Never expose exception text, filesystem paths or provider errors. | |
| "degraded": bool(self._load_error), | |
| }) | |
| return status | |
| def system_stats(self) -> Dict[str, Any]: | |
| unavailable = self._unavailable() | |
| if unavailable: | |
| return unavailable | |
| return { | |
| "runtime": self.public_status(), | |
| "core": self._core.get_stats(), | |
| "loop": self._loop.get_stats(), | |
| "ledger": self._ledger.get_stats(), | |
| } | |
| def governance(self) -> Dict[str, Any]: | |
| unavailable = self._unavailable() | |
| if unavailable: | |
| return unavailable | |
| return self._loop.controller.governance_snapshot(self._loop.history) | |
| def _allow_public_request(self) -> bool: | |
| """Apply a small in-process sliding-window guard to public callbacks.""" | |
| limit = int(getattr(self.settings, "public_rate_limit_per_minute", 20) or 20) | |
| now = time.monotonic() | |
| with self._work_lock: | |
| while self._public_events and now - self._public_events[0] >= 60: | |
| self._public_events.popleft() | |
| if len(self._public_events) >= limit: | |
| return False | |
| self._public_events.append(now) | |
| return True | |
| def recent_cycles(self, limit: int = 25) -> List[Dict[str, Any]]: | |
| unavailable = self._unavailable() | |
| if unavailable: | |
| return [unavailable] | |
| return self._loop.get_recent_cycles(max(1, min(int(limit), 100)))[::-1] | |
| def knowledge_search(self, query: str, limit: int = 10) -> List[Dict[str, Any]]: | |
| unavailable = self._unavailable() | |
| if unavailable: | |
| return [unavailable] | |
| cleaned = " ".join((query or "").split()) | |
| if not cleaned: | |
| return [] | |
| max_chars = int(getattr(self.settings, "public_max_query_chars", 2000) or 2000) | |
| if len(cleaned) > max_chars: | |
| return [{"error": f"Query exceeds the public limit of {max_chars} characters."}] | |
| if not self._allow_public_request(): | |
| return [{"error": "Public request limit reached; please retry later."}] | |
| return self._core.kg.search(cleaned, max(1, min(int(limit), 50))) | |
| def ledger_status(self) -> Dict[str, Any]: | |
| unavailable = self._unavailable() | |
| if unavailable: | |
| return unavailable | |
| return {"valid": self._ledger.is_chain_valid(), **self._ledger.get_stats()} | |
| def run_cycle(self, dev_mode: Optional[bool] = None, *, public: bool = False) -> Dict[str, Any]: | |
| unavailable = self._unavailable() | |
| if unavailable: | |
| return unavailable | |
| if public and not bool(getattr(self.settings, "public_cycle_enabled", False)): | |
| return {"error": "Public autonomous cycles are disabled; use the read-only research interface."} | |
| if public and not self._allow_public_request(): | |
| return {"error": "Public request limit reached; please retry later."} | |
| with self._work_lock: | |
| try: | |
| use_dev_mode = self.settings.developer_mode if dev_mode is None else bool(dev_mode) | |
| return self._loop.run_cycle(developer_mode=use_dev_mode).to_dict() | |
| except Exception: | |
| logger.exception("Canonical HF run_cycle failed") | |
| return {"error": "The governed cycle failed safely; inspect operator logs."} | |
| def ask_question(self, question: str, developer_mode: bool = False) -> Dict[str, Any]: | |
| unavailable = self._unavailable() | |
| if unavailable: | |
| return unavailable | |
| cleaned = " ".join((question or "").split()) | |
| max_chars = int(getattr(self.settings, "public_max_query_chars", 2000) or 2000) | |
| if len(cleaned) < 5: | |
| return {"error": "Enter a question of at least five characters."} | |
| if len(cleaned) > max_chars: | |
| return {"error": f"Question exceeds the public limit of {max_chars} characters."} | |
| if not self._allow_public_request(): | |
| return {"error": "Public request limit reached; please retry later."} | |
| query_verdict = check_query(cleaned) | |
| if not query_verdict.allowed: | |
| return {"refused": True, "message": render_safe_block(query_verdict), "verdict": query_verdict.to_dict()} | |
| with self._work_lock: | |
| try: | |
| result = self._core.ask_question( | |
| cleaned, | |
| developer_mode=bool(developer_mode), | |
| use_web_search=self.settings.web_search_enabled, | |
| ) | |
| answer_verdict = check_answer(str(result.get("answer", ""))) | |
| if not answer_verdict.allowed: | |
| return {"refused": True, "message": render_safe_block(answer_verdict), "verdict": answer_verdict.to_dict()} | |
| result["safety"] = {"allowed": True} | |
| return result | |
| except Exception: | |
| logger.exception("Canonical HF manual research failed") | |
| return {"error": "The research request failed safely; inspect operator logs."} | |
| def export_state(self, limit: int = 25) -> Dict[str, Any]: | |
| unavailable = self._unavailable() | |
| if unavailable: | |
| return unavailable | |
| return { | |
| "exported_at": __import__("datetime").datetime.now().isoformat(), | |
| "stats": self.system_stats(), | |
| "governance": self.governance(), | |
| "recent_cycles": self.recent_cycles(limit), | |
| } | |
| def sync_to_hf_dataset(self, report: Dict[str, Any]) -> bool: | |
| """Persist a report and invoke the single repository sync implementation.""" | |
| if not validate_secret("HF_TOKEN", prefix="hf_").usable: | |
| logger.warning("HF sync unavailable: HF_TOKEN is not configured") | |
| return False | |
| try: | |
| root = self.settings.root if self.settings else Path.cwd() | |
| reports_dir = root / "reports" | |
| reports_dir.mkdir(parents=True, exist_ok=True) | |
| report_path = reports_dir / f"cycle_{report['cycle_id']}.json" | |
| report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") | |
| from scripts.sync_huggingface import sync_hf_dataset | |
| return sync_hf_dataset(limit=1) == 0 | |
| except Exception as exc: | |
| logger.warning("HF dataset sync failed: %s", exc) | |
| return False | |
| _DEFAULT_ADAPTER: Optional[HeadlessHFAdapter] = None | |
| def get_adapter() -> HeadlessHFAdapter: | |
| global _DEFAULT_ADAPTER | |
| if _DEFAULT_ADAPTER is None: | |
| _DEFAULT_ADAPTER = HeadlessHFAdapter() | |
| return _DEFAULT_ADAPTER | |
| def run_cycle(dev_mode: Optional[bool] = None) -> Dict[str, Any]: | |
| return get_adapter().run_cycle(dev_mode=dev_mode) | |
| def sync_to_hf_dataset(report: Dict[str, Any]) -> bool: | |
| return get_adapter().sync_to_hf_dataset(report) | |