From a56c98e735781d27eb6d344101aed2efbe63b8f4 Mon Sep 17 00:00:00 2001 From: Yong Wook Lee Date: Sat, 8 Aug 2026 05:12:49 +0900 Subject: [PATCH 1/2] feat: add read-only research-to-skill memory integration --- README.md | 21 ++ hegi/adapters/__init__.py | 9 +- hegi/adapters/research_to_skill.py | 296 +++++++++++++++++++ hegi/cli.py | 36 +++ hegi/config.py | 22 ++ hegi/config/default.yaml | 6 + hegi/pipeline.py | 33 +++ tests/unit/test_config_state.py | 28 ++ tests/unit/test_research_to_skill_adapter.py | 169 +++++++++++ 9 files changed, 619 insertions(+), 1 deletion(-) create mode 100644 hegi/adapters/research_to_skill.py create mode 100644 tests/unit/test_research_to_skill_adapter.py diff --git a/README.md b/README.md index 586f6c3..14b9822 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,27 @@ Hermes SQLite databases (read-only) [Architecture details](docs/architecture.md) +## Personal Research Memory (Research-to-Skill) + +HEGI can optionally read a Research-to-Skill project as a second, provenance-separated +memory backend. Memory Forest remains the conversation and meeting memory; Personal +Research Memory supplies concepts, claims, evidence, sources, and concept evolution from +the research corpus. HEGI preserves `author`, `external`, `mixed`, and `uncertain` origins +without reclassifying them. + +```yaml +research_memory: + enabled: true + provider: research-to-skill + project_path: /absolute/path/to/research-project + read_only: true +``` + +`HEGI_RESEARCH_MEMORY_PROJECT` may override the project path. The integration reads +canonical `research.json` and never invokes Research-to-Skill writer commands. A missing, +invalid, or corrupt project is reported as unavailable while HEGI continues with Memory +Forest only. Diagnose the backend with `hegi research-memory status`. + ## Supported environment | Component | Supported range | diff --git a/hegi/adapters/__init__.py b/hegi/adapters/__init__.py index 1343dca..9834598 100644 --- a/hegi/adapters/__init__.py +++ b/hegi/adapters/__init__.py @@ -1,6 +1,13 @@ """Stable host and transport boundaries used by HEGI.""" from .hermes import HermesLLMAdapter, HermesToolDispatcher +from .research_to_skill import ResearchMemoryBackend, ResearchToSkillAdapter from .telegram import TelegramBotAdapter -__all__ = ["HermesLLMAdapter", "HermesToolDispatcher", "TelegramBotAdapter"] +__all__ = [ + "HermesLLMAdapter", + "HermesToolDispatcher", + "ResearchMemoryBackend", + "ResearchToSkillAdapter", + "TelegramBotAdapter", +] diff --git a/hegi/adapters/research_to_skill.py b/hegi/adapters/research_to_skill.py new file mode 100644 index 0000000..136d32e --- /dev/null +++ b/hegi/adapters/research_to_skill.py @@ -0,0 +1,296 @@ +"""Read-only adapter for a Research-to-Skill personal research corpus.""" + +from __future__ import annotations + +import json +import os +import re +import stat +from pathlib import Path +from typing import Any, Protocol + + +class ResearchMemoryBackend(Protocol): + def status(self) -> dict[str, Any]: ... + def search(self, query: str) -> dict[str, Any]: ... + def get_concept(self, identifier: str) -> dict[str, Any] | None: ... + def get_claim(self, claim_id: str) -> dict[str, Any] | None: ... + def get_evidence(self, claim_id: str) -> list[dict[str, Any]]: ... + def get_context(self, query: str) -> dict[str, Any]: ... + + +class ResearchToSkillAdapter: + """Read canonical structured artifacts without invoking writer commands.""" + + memory_type = "personal_research" + max_manifest_bytes = 16 * 1024 * 1024 + + def __init__(self, project_path: str | Path): + self.project_path = Path(project_path).expanduser() + + @property + def manifest_path(self) -> Path: + return self.project_path / "research.json" + + def _read_json(self, path: Path) -> dict[str, Any]: + flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(path, flags) + try: + before = os.fstat(descriptor) + if not stat.S_ISREG(before.st_mode): + raise ValueError(f"not a regular file: {path.name}") + if before.st_size > self.max_manifest_bytes: + raise ValueError(f"JSON artifact is too large: {path.name}") + with os.fdopen(descriptor, encoding="utf-8", closefd=False) as handle: + value = json.load(handle) + after = os.fstat(descriptor) + finally: + os.close(descriptor) + if (before.st_ino, before.st_size, before.st_mtime_ns) != ( + after.st_ino, + after.st_size, + after.st_mtime_ns, + ): + raise RuntimeError(f"file changed during read: {path.name}") + if not isinstance(value, dict): + raise ValueError(f"JSON root is not an object: {path.name}") + return value + + def _manifest(self) -> dict[str, Any]: + manifest = self._read_json(self.manifest_path) + required = {"project", "sources", "concepts", "claims", "relations"} + if not required.issubset(manifest): + raise ValueError("research.json is missing canonical collections") + if not all(isinstance(manifest[key], list) for key in required - {"project"}): + raise ValueError("research.json canonical collections must be arrays") + if not isinstance(manifest["project"], dict): + raise ValueError("research.json project must be an object") + for key in ("sources", "concepts", "claims", "relations"): + if not all(isinstance(item, dict) for item in manifest[key]): + raise ValueError(f"research.json {key} must contain objects") + concept_ids = {str(item.get("id", "")) for item in manifest["concepts"]} + claim_ids = {str(item.get("id", "")) for item in manifest["claims"]} + source_ids = {str(item.get("id", "")) for item in manifest["sources"]} + if "" in concept_ids | claim_ids | source_ids: + raise ValueError("research.json contains an empty canonical id") + if len(concept_ids) != len(manifest["concepts"]): + raise ValueError("research.json contains duplicate concept ids") + if len(claim_ids) != len(manifest["claims"]): + raise ValueError("research.json contains duplicate claim ids") + if len(source_ids) != len(manifest["sources"]): + raise ValueError("research.json contains duplicate source ids") + if concept_ids & claim_ids or concept_ids & source_ids or claim_ids & source_ids: + raise ValueError("research.json contains cross-type duplicate ids") + for concept in manifest["concepts"]: + if not set(map(str, concept.get("source_ids", []))).issubset(source_ids): + raise ValueError(f"concept has dangling source reference: {concept.get('id', '')}") + versions = concept.get("versions", []) + if not isinstance(versions, list) or not all(isinstance(item, dict) for item in versions): + raise ValueError(f"concept has invalid versions: {concept.get('id', '')}") + if not {str(item.get("source_id", "")) for item in versions}.issubset(source_ids): + raise ValueError(f"concept version has dangling source: {concept.get('id', '')}") + for claim in manifest["claims"]: + if claim.get("origin") not in {"author", "external", "mixed", "uncertain"}: + raise ValueError(f"invalid claim origin: {claim.get('id', '')}") + if not set(map(str, claim.get("concept_ids", []))).issubset(concept_ids): + raise ValueError(f"claim has dangling concept reference: {claim.get('id', '')}") + if not set(map(str, claim.get("source_ids", []))).issubset(source_ids): + raise ValueError(f"claim has dangling source reference: {claim.get('id', '')}") + evidence = claim.get("evidence", []) + if not isinstance(evidence, list) or not all(isinstance(item, dict) for item in evidence): + raise ValueError(f"claim has invalid evidence: {claim.get('id', '')}") + evidence_sources = {str(item.get("source_id", "")) for item in evidence} + if not evidence_sources.issubset(set(map(str, claim.get("source_ids", [])))): + raise ValueError(f"claim evidence has invalid provenance: {claim.get('id', '')}") + allowed_relations = { + "defines", "supports", "contradicts", "extends", "revises", + "cites", "derived_from", "related_to", + } + for relation in manifest["relations"]: + if str(relation.get("source", "")) not in concept_ids | claim_ids | source_ids: + raise ValueError("relation has a dangling source") + if str(relation.get("target", "")) not in concept_ids | claim_ids | source_ids: + raise ValueError("relation has a dangling target") + if relation.get("type") not in allowed_relations: + raise ValueError("relation has an invalid type") + return manifest + + @staticmethod + def _terms(value: str) -> set[str]: + terms = { + token.casefold() + for token in re.findall(r"[\w가-힣]+", value, flags=re.UNICODE) + if len(token) >= 2 + } + particles = ("으로", "에서", "에게", "까지", "부터", "은", "는", "이", "가", "을", "를", "의") + terms.update( + token[: -len(particle)] + for token in tuple(terms) + for particle in particles + if token.endswith(particle) and len(token) - len(particle) >= 2 + ) + return terms + + def status(self) -> dict[str, Any]: + try: + manifest = self._manifest() + dirty = sum( + 1 + for source in manifest["sources"] + if not source.get("compiled_hash") + or source.get("compiled_hash") != source.get("sha256") + ) + project = manifest["project"] + return { + "memory_type": self.memory_type, + "provider": "research-to-skill", + "project": project.get("name", ""), + "project_id": project.get("id", ""), + "project_path": str(self.project_path), + "sources": len(manifest["sources"]), + "concepts": len(manifest["concepts"]), + "claims": len(manifest["claims"]), + "dirty": dirty, + "validation": "MANIFEST_PASS", + "mode": "READ-ONLY", + "status": "READY", + "writer_lock": (self.project_path / ".research-to-skill.lock").exists(), + } + except Exception as exc: + return { + "memory_type": self.memory_type, + "provider": "research-to-skill", + "project_path": str(self.project_path), + "mode": "READ-ONLY", + "status": "UNAVAILABLE", + "reason": type(exc).__name__, + } + + def get_concept(self, identifier: str) -> dict[str, Any] | None: + manifest = self._manifest() + needle = identifier.strip().casefold() + for concept in manifest["concepts"]: + names = {str(concept.get("id", "")).casefold(), str(concept.get("name", "")).casefold()} + names.update(part.strip() for name in tuple(names) for part in name.split("/")) + if needle in names or any(needle and needle in name for name in names): + result = dict(concept) + result["supporting_claim_ids"] = [ + claim["id"] + for claim in manifest["claims"] + if concept["id"] in claim.get("concept_ids", []) + ] + return result + return None + + def get_claim(self, claim_id: str) -> dict[str, Any] | None: + for claim in self._manifest()["claims"]: + if claim.get("id") == claim_id: + return dict(claim) + return None + + def get_evidence(self, claim_id: str) -> list[dict[str, Any]]: + claim = self.get_claim(claim_id) + return list(claim.get("evidence", [])) if claim else [] + + @staticmethod + def _bounded(value: Any, limit: int = 2000) -> str: + return str(value or "")[:limit] + + def _search(self, manifest: dict[str, Any], query: str) -> dict[str, Any]: + query_terms = self._terms(query) + concepts: list[dict[str, Any]] = [] + for concept in manifest["concepts"]: + haystack = " ".join( + [str(concept.get("id", "")), str(concept.get("name", ""))] + + [str(version.get("definition", "")) for version in concept.get("versions", [])] + ) + score = len(query_terms & self._terms(haystack)) + if score: + concepts.append({**concept, "_score": score}) + concepts.sort(key=lambda item: (-item["_score"], str(item.get("id", "")))) + concepts = [ + { + "id": item["id"], + "name": self._bounded(item.get("name"), 300), + "source_ids": list(item.get("source_ids", [])), + "versions": [ + { + "source_id": version.get("source_id"), + "date": version.get("date"), + "definition": self._bounded(version.get("definition")), + } + for version in item.get("versions", [])[:12] + ], + } + for item in concepts[:5] + ] + selected_concepts = {item["id"] for item in concepts} + claims: list[dict[str, Any]] = [] + for claim in manifest["claims"]: + overlap = len(query_terms & self._terms(str(claim.get("text", "")))) + linked = bool(selected_concepts & set(claim.get("concept_ids", []))) + if linked or overlap >= 2: + claims.append({**claim, "_score": overlap + (2 if linked else 0)}) + claims.sort(key=lambda item: (-item["_score"], str(item.get("id", "")))) + claims = [ + { + "id": item["id"], + "text": self._bounded(item.get("text")), + "origin": item.get("origin"), + "source_ids": list(item.get("source_ids", [])), + "concept_ids": list(item.get("concept_ids", [])), + "confidence": item.get("confidence"), + "evidence": [ + { + "source_id": evidence.get("source_id"), + "locator": evidence.get("locator"), + "summary": self._bounded(evidence.get("summary"), 1000), + "evidence_type": evidence.get("evidence_type"), + } + for evidence in item.get("evidence", [])[:8] + ], + } + for item in claims[:12] + ] + source_ids = { + source_id for claim in claims for source_id in claim.get("source_ids", []) + } | {source_id for concept in concepts for source_id in concept.get("source_ids", [])} + sources = [ + { + key: source.get(key) + for key in ("id", "title", "authors", "year", "doi", "url") + } + for source in manifest["sources"] + if source.get("id") in source_ids + ] + evidence = [ + {"claim_id": claim["id"], **item} + for claim in claims + for item in claim.get("evidence", []) + ] + return { + "memory_type": self.memory_type, + "query": query, + "concepts": concepts, + "claims": claims, + "evidence": evidence, + "sources": sources, + "found": bool(concepts or claims), + } + + def search(self, query: str) -> dict[str, Any]: + return self._search(self._manifest(), query) + + def get_context(self, query: str) -> dict[str, Any]: + try: + manifest = self._manifest() + return {**self._search(manifest, query), "status": "READY", "mode": "READ-ONLY"} + except Exception as exc: + return { + "memory_type": self.memory_type, + "query": query, + "status": "UNAVAILABLE", + "mode": "READ-ONLY", + "reason": type(exc).__name__, + "found": False, + } diff --git a/hegi/cli.py b/hegi/cli.py index 8785b5a..e0f7585 100644 --- a/hegi/cli.py +++ b/hegi/cli.py @@ -18,6 +18,7 @@ import yaml from ._version import __version__ +from .adapters.research_to_skill import ResearchToSkillAdapter from .bootstrap import _atomic_yaml, _backup, setup from .compat import Check, doctor_checks from .config import ( @@ -49,6 +50,31 @@ def _json(value: Any) -> None: print(json.dumps(value, ensure_ascii=False, indent=2)) +def cmd_research_memory_status(args: argparse.Namespace) -> int: + config = load_config(args.config) + settings = config.section("research_memory") + if settings.get("enabled") is not True: + payload = { + "memory_type": "personal_research", + "provider": settings.get("provider", "research-to-skill"), + "mode": "READ-ONLY", + "status": "DISABLED", + } + else: + if settings.get("read_only") is not True: + payload = { + "memory_type": "personal_research", + "provider": settings.get("provider", ""), + "mode": "READ-ONLY", + "status": "UNAVAILABLE", + "reason": "UnsafeConfiguration", + } + else: + payload = ResearchToSkillAdapter(str(settings.get("project_path", ""))).status() + _json(payload) + return 0 if payload["status"] in {"READY", "DISABLED"} else 1 + + def _mask_identifier(value: str) -> str: text = str(value) if len(text) <= 4: @@ -1094,6 +1120,16 @@ def _configure_parser(parser: argparse.ArgumentParser) -> None: hegi_handler=cmd_status ) + research_memory = subparsers.add_parser( + "research-memory", help="inspect the read-only personal research backend" + ) + research_commands = research_memory.add_subparsers( + dest="research_memory_command", required=True + ) + research_commands.add_parser( + "status", help="show Personal Research Memory status" + ).set_defaults(hegi_handler=cmd_research_memory_status) + retry_failed = subparsers.add_parser( "retry-failed", help="retry one frozen failed episode after operator repair" ) diff --git a/hegi/config.py b/hegi/config.py index fa33ea9..6c4e4bc 100644 --- a/hegi/config.py +++ b/hegi/config.py @@ -81,6 +81,12 @@ "reject": ["기억하지 마"], }, }, + "research_memory": { + "enabled": False, + "provider": "research-to-skill", + "project_path": "", + "read_only": True, + }, "daemon": {"poll_seconds": 60}, "reports": {"telegram": True}, "v3": {"enabled": False}, @@ -175,6 +181,9 @@ def _config_from_mapping( home: Path | None = None, ) -> HegiConfig: raw = _deep_merge(DEFAULT_CONFIG, loaded) + project_override = os.environ.get("HEGI_RESEARCH_MEMORY_PROJECT", "").strip() + if project_override: + raw["research_memory"]["project_path"] = project_override home = home if home is not None else config_path.resolve().parent.parent chat_id = str(raw.get("telegram", {}).get("chat_id", "")).strip() agents: list[AgentSourceConfig] = [] @@ -302,12 +311,15 @@ def validate_config(config: HegiConfig, *, require_runtime: bool = False) -> lis except (TypeError, ValueError): errors.append(f"{label}는 정수여야 합니다.") memory = config.section("memory") + research_memory = config.section("research_memory") archive_config = config.section("archive") boolean_fields = { "memory.enabled": memory.get("enabled"), "memory.auto_commit": memory.get("auto_commit"), "memory.auto_draft": memory.get("auto_draft"), "memory.require_professor_approval": memory.get("require_professor_approval"), + "research_memory.enabled": research_memory.get("enabled"), + "research_memory.read_only": research_memory.get("read_only"), "telegram.enabled": config.section("telegram").get("enabled"), "reports.telegram": config.section("reports").get("telegram"), "archive.sync_when_available": archive_config.get("sync_when_available"), @@ -323,6 +335,16 @@ def validate_config(config: HegiConfig, *, require_runtime: bool = False) -> lis errors.append("memory.auto_draft는 안전 경계상 true일 수 없습니다.") if not memory.get("require_professor_approval", True): errors.append("memory.require_professor_approval은 true여야 합니다.") + if research_memory.get("read_only") is not True: + errors.append("research_memory.read_only는 true여야 합니다.") + if research_memory.get("enabled") is True: + if research_memory.get("provider") != "research-to-skill": + errors.append("research_memory.provider는 research-to-skill이어야 합니다.") + project_path = str(research_memory.get("project_path", "")).strip() + if not project_path: + errors.append("research_memory.project_path가 비어 있습니다.") + elif not Path(project_path).expanduser().is_absolute(): + errors.append("research_memory.project_path는 절대 경로여야 합니다.") for key in ("markdown", "json"): if archive_config.get(key) is not True: errors.append(f"archive.{key}는 v2.1.0에서 true여야 합니다.") diff --git a/hegi/config/default.yaml b/hegi/config/default.yaml index 55e4884..47fdf8e 100644 --- a/hegi/config/default.yaml +++ b/hegi/config/default.yaml @@ -78,6 +78,12 @@ memory: reject: - "기억하지 마" +research_memory: + enabled: false + provider: "research-to-skill" + project_path: "" + read_only: true + daemon: poll_seconds: 60 diff --git a/hegi/pipeline.py b/hegi/pipeline.py index 4d8036b..381f701 100644 --- a/hegi/pipeline.py +++ b/hegi/pipeline.py @@ -11,6 +11,7 @@ from ._version import __version__ from .actions import persist_new_actions +from .adapters.research_to_skill import ResearchMemoryBackend, ResearchToSkillAdapter from .analyzer import build_minutes, minimal_minutes from .archive import ArchiveManager from .collector import HermesSQLiteCollector, dedup_key, deduplicate_messages @@ -115,6 +116,7 @@ def __init__( state: StateStore | None = None, llm_client: HermesLLMClient | None = None, memory_backend: Any | None = None, + research_backend: ResearchMemoryBackend | None = None, telegram_sender: Any | None = None, ): self.config = config @@ -157,6 +159,12 @@ def __init__( draft_server=str(memory.get("draft_server", "")), draft_tool=str(memory.get("draft_tool", "")), ) + research_memory = config.section("research_memory") + self.research_backend = research_backend + if self.research_backend is None and research_memory.get("enabled") is True: + self.research_backend = ResearchToSkillAdapter( + str(research_memory.get("project_path", "")) + ) self.telegram_sender = telegram_sender self.logger = RunLogger(config.state_db.parent / "runs.jsonl") @@ -281,6 +289,7 @@ def process_episode( def analyze_episode(self, episode: MeetingEpisode) -> MeetingMinutes: analysis = self.config.section("analysis") + research_context = self._research_context(episode) try: payload = self.analyzer.analyze_payload(episode) minutes = build_minutes(payload, episode) @@ -297,9 +306,33 @@ def analyze_episode(self, episode: MeetingEpisode) -> MeetingMinutes: "prompt_version": analysis.get("prompt_version", "v2.1.0"), "generated_at": datetime.now(UTC).isoformat(), "hegi_version": __version__, + "research_memory": research_context, } + if research_context and research_context.get("status") == "UNAVAILABLE": + minutes.warnings.append( + "Research memory unavailable; continuing with Memory Forest only" + ) return minutes + def _research_context(self, episode: MeetingEpisode) -> dict[str, Any] | None: + if self.research_backend is None: + return None + query_parts = [episode.topic_hint or ""] + query_parts.extend( + message.content for message in episode.messages if message.role == "user" + ) + query = " ".join(part.strip() for part in query_parts if part.strip())[-4000:] + try: + return self.research_backend.get_context(query) + except Exception as exc: + return { + "memory_type": "personal_research", + "status": "UNAVAILABLE", + "mode": "READ-ONLY", + "reason": type(exc).__name__, + "found": False, + } + def _process_episode(self, episode: MeetingEpisode, *, dry_run: bool) -> dict[str, Any]: checkpoint = self.state.episode_by_id(episode.meeting_id) resume_statuses = {"analyzed", "archived", "reporting"} diff --git a/tests/unit/test_config_state.py b/tests/unit/test_config_state.py index 9dd85e7..65ac77f 100644 --- a/tests/unit/test_config_state.py +++ b/tests/unit/test_config_state.py @@ -38,6 +38,34 @@ def test_config_rejects_automatic_memory_writes(tmp_path, monkeypatch): assert any("allow_autonomous_commit" in error for error in errors) +def test_research_memory_is_read_only_and_supports_project_env_override( + tmp_path, monkeypatch +): + config_path = tmp_path / "hegi.yaml" + configured = tmp_path / "configured" + overridden = tmp_path / "overridden" + monkeypatch.setenv("HEGI_RESEARCH_MEMORY_PROJECT", str(overridden)) + config_path.write_text( + f""" +research_memory: + enabled: true + provider: research-to-skill + project_path: {configured} + read_only: true +""", + encoding="utf-8", + ) + + config = load_config(config_path) + assert config.section("research_memory")["project_path"] == str(overridden) + assert not any("research_memory" in error for error in validate_config(config)) + + config.raw["research_memory"]["read_only"] = False + assert any( + "research_memory.read_only" in error for error in validate_config(config) + ) + + def test_existing_state_database_migrates_approval_workflow_columns(tmp_path): path = tmp_path / "legacy.db" import sqlite3 diff --git a/tests/unit/test_research_to_skill_adapter.py b/tests/unit/test_research_to_skill_adapter.py new file mode 100644 index 0000000..71cf2f2 --- /dev/null +++ b/tests/unit/test_research_to_skill_adapter.py @@ -0,0 +1,169 @@ +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +import pytest + +from hegi.adapters.research_to_skill import ResearchToSkillAdapter +from hegi.cli import main + + +@pytest.fixture +def research_project(tmp_path: Path) -> Path: + project = tmp_path / "research" + project.mkdir() + manifest = { + "schema_version": 1, + "project": {"id": "project-1", "name": "Synthetic Research", "slug": "synthetic"}, + "sources": [ + { + "id": "source-1", + "title": "Artificial Nature Paper", + "sha256": "abc", + "compiled_hash": "abc", + } + ], + "concepts": [ + { + "id": "artificial-nature", + "name": "Artificial Nature / 인공자연", + "source_ids": ["source-1"], + "versions": [ + {"source_id": "source-1", "date": "2025", "definition": "초기 협력 환경"}, + {"source_id": "source-1", "date": "2026", "definition": "구조적 실재"}, + ], + } + ], + "claims": [ + { + "id": "claim-001", + "text": "인공자연은 독자적 인과 깊이를 지닌다.", + "origin": "author", + "source_ids": ["source-1"], + "concept_ids": ["artificial-nature"], + "confidence": 0.98, + "evidence": [ + { + "source_id": "source-1", + "locator": {"page": 10, "section": "2.1"}, + "summary": "본문의 명시적 정의", + } + ], + } + ], + "relations": [ + {"source": "claim-001", "target": "artificial-nature", "type": "supports"} + ], + } + (project / "research.json").write_text( + json.dumps(manifest, ensure_ascii=False), encoding="utf-8" + ) + (project / "topic-index.md").write_text( + "# Topic Index\n\n## Artificial Nature\n- claim-001\n", encoding="utf-8" + ) + return project + + +def digest(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def test_adapter_status_and_read_only_retrieval(research_project: Path): + adapter = ResearchToSkillAdapter(research_project) + before = digest(research_project / "research.json") + + assert adapter.status() == { + "memory_type": "personal_research", + "provider": "research-to-skill", + "project": "Synthetic Research", + "project_id": "project-1", + "project_path": str(research_project), + "sources": 1, + "concepts": 1, + "claims": 1, + "dirty": 0, + "validation": "MANIFEST_PASS", + "mode": "READ-ONLY", + "status": "READY", + "writer_lock": False, + } + concept = adapter.get_concept("Artificial Nature") + assert concept is not None + assert [version["definition"] for version in concept["versions"]] == [ + "초기 협력 환경", + "구조적 실재", + ] + assert concept["supporting_claim_ids"] == ["claim-001"] + assert adapter.get_claim("claim-001")["origin"] == "author" + assert adapter.get_evidence("claim-001")[0]["locator"]["page"] == 10 + assert digest(research_project / "research.json") == before + + +def test_search_preserves_provenance_and_rejects_unknown_memory(research_project: Path): + adapter = ResearchToSkillAdapter(research_project) + result = adapter.get_context("내 Artificial Nature 개념은 무엇인가?") + assert result["status"] == "READY" + assert result["memory_type"] == "personal_research" + assert result["claims"][0]["origin"] == "author" + assert result["evidence"][0]["source_id"] == "source-1" + assert result["sources"][0]["id"] == "source-1" + + absent = adapter.get_context("양자 중력 무지개 이론") + assert absent["found"] is False + assert absent["concepts"] == [] + assert absent["claims"] == [] + + korean = adapter.get_context("인공자연은 어떻게 변화했는가?") + assert korean["concepts"][0]["id"] == "artificial-nature" + + +def test_active_writer_lock_does_not_block_stable_read(research_project: Path): + (research_project / ".research-to-skill.lock").write_text( + '{"operation":"compile"}', encoding="utf-8" + ) + status = ResearchToSkillAdapter(research_project).status() + assert status["status"] == "READY" + assert status["writer_lock"] is True + + +def test_unavailable_and_corrupt_manifest_fall_back(tmp_path: Path, research_project: Path): + missing = ResearchToSkillAdapter(tmp_path / "missing").get_context("question") + assert missing["status"] == "UNAVAILABLE" + assert missing["found"] is False + + (research_project / "research.json").write_text("{broken", encoding="utf-8") + corrupt = ResearchToSkillAdapter(research_project).get_context("question") + assert corrupt["status"] == "UNAVAILABLE" + assert corrupt["reason"] == "JSONDecodeError" + + +def test_dangling_manifest_reference_is_unavailable(research_project: Path): + manifest = json.loads((research_project / "research.json").read_text(encoding="utf-8")) + manifest["claims"][0]["concept_ids"] = ["missing-concept"] + (research_project / "research.json").write_text(json.dumps(manifest), encoding="utf-8") + assert ResearchToSkillAdapter(research_project).status()["status"] == "UNAVAILABLE" + + +def test_non_object_manifest_item_is_unavailable(research_project: Path): + manifest = json.loads((research_project / "research.json").read_text(encoding="utf-8")) + manifest["claims"].append(None) + (research_project / "research.json").write_text(json.dumps(manifest), encoding="utf-8") + assert ResearchToSkillAdapter(research_project).status()["status"] == "UNAVAILABLE" + + +def test_research_memory_status_cli(research_project: Path, tmp_path: Path, capsys): + config = tmp_path / "config.yaml" + config.write_text( + "research_memory:\n" + " enabled: true\n" + " provider: research-to-skill\n" + f" project_path: {research_project}\n" + " read_only: true\n", + encoding="utf-8", + ) + assert main(["--config", str(config), "research-memory", "status"]) == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["status"] == "READY" + assert payload["mode"] == "READ-ONLY" From 82b3109cea88ce5e620b070b673d92ae4c5b1462 Mon Sep 17 00:00:00 2001 From: Yong Wook Lee Date: Sat, 8 Aug 2026 05:30:30 +0900 Subject: [PATCH 2/2] feat: add external scholar search integration --- README.md | 23 ++ hegi/adapters/__init__.py | 3 + hegi/adapters/scholar.py | 327 +++++++++++++++++++++++++++++ hegi/cli.py | 60 ++++++ hegi/config.py | 41 ++++ hegi/config/default.yaml | 8 + hegi/research_router.py | 191 +++++++++++++++++ tests/unit/test_config_state.py | 42 ++++ tests/unit/test_scholar_adapter.py | 308 +++++++++++++++++++++++++++ 9 files changed, 1003 insertions(+) create mode 100644 hegi/adapters/scholar.py create mode 100644 hegi/research_router.py create mode 100644 tests/unit/test_scholar_adapter.py diff --git a/README.md b/README.md index 14b9822..4f8c06e 100644 --- a/README.md +++ b/README.md @@ -111,6 +111,29 @@ canonical `research.json` and never invokes Research-to-Skill writer commands. A invalid, or corrupt project is reported as unavailable while HEGI continues with Memory Forest only. Diagnose the backend with `hegi research-memory status`. +## External Scholar Search + +HEGI can search lawful scholarly metadata through OpenAlex with Crossref fallback. External +records always carry `memory_type: external_scholar`; they are never treated as Personal +Research claims or written to Memory Forest. Google Scholar scraping, paywall bypass, PDF +downloads, and automatic Research-to-Skill ingest are not implemented. + +```yaml +scholar: + enabled: true + provider: openalex + fallback_provider: crossref + default_limit: 10 + timeout_seconds: 15 + cache_ttl: 900 +``` + +Run `hegi scholar search "recent AI ecology research"`. Queries that explicitly combine +personal language (for example, “my research”) with comparison or recent-literature +language are routed to a hybrid result with three separate objects: `personal_research`, +`external_scholar`, and metadata-only `cross_analysis`. The in-process TTL cache stores +provider JSON only in memory and never writes to the Research-to-Skill project. + ## Supported environment | Component | Supported range | diff --git a/hegi/adapters/__init__.py b/hegi/adapters/__init__.py index 9834598..bfdcb00 100644 --- a/hegi/adapters/__init__.py +++ b/hegi/adapters/__init__.py @@ -2,6 +2,7 @@ from .hermes import HermesLLMAdapter, HermesToolDispatcher from .research_to_skill import ResearchMemoryBackend, ResearchToSkillAdapter +from .scholar import ScholarAdapter, ScholarError from .telegram import TelegramBotAdapter __all__ = [ @@ -9,5 +10,7 @@ "HermesToolDispatcher", "ResearchMemoryBackend", "ResearchToSkillAdapter", + "ScholarAdapter", + "ScholarError", "TelegramBotAdapter", ] diff --git a/hegi/adapters/scholar.py b/hegi/adapters/scholar.py new file mode 100644 index 0000000..23b838e --- /dev/null +++ b/hegi/adapters/scholar.py @@ -0,0 +1,327 @@ +"""Official-API scholarly search with normalized external provenance.""" + +from __future__ import annotations + +import html +import json +import re +import time +import urllib.error +import urllib.parse +import urllib.request +from collections.abc import Callable +from contextlib import suppress +from dataclasses import dataclass +from typing import Any + +OPENALEX_API = "https://api.openalex.org" +CROSSREF_API = "https://api.crossref.org" +MAX_RESPONSE_BYTES = 4 * 1024 * 1024 +MAX_TEXT_CHARS = 8_000 +ALLOWED_API_HOSTS = {"api.openalex.org", "api.crossref.org"} + + +class ScholarError(RuntimeError): + """A provider response was unavailable or unsafe to consume.""" + + +Transport = Callable[[str, float, int], dict[str, Any]] + + +def _clean_text(value: Any, limit: int = MAX_TEXT_CHARS) -> str: + text = html.unescape(str(value or "")) + text = re.sub(r"<(script|style)\b[^>]*>.*?", " ", text, flags=re.I | re.S) + text = re.sub(r"<[^>]+>", " ", text) + text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", " ", text) + return re.sub(r"\s+", " ", text).strip()[:limit] + + +def _doi(value: Any) -> str | None: + text = str(value or "").strip().lower() + text = re.sub(r"^https?://(?:dx\.)?doi\.org/", "", text) + text = re.sub(r"^doi:\s*", "", text) + return text if re.fullmatch(r"10\.\d{4,9}/[^\s?#]+", text) else None + + +def _safe_url(value: Any, *, hosts: set[str] | None = None) -> str | None: + text = str(value or "").strip() + if not text: + return None + parsed = urllib.parse.urlparse(text) + if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password: + return None + if hosts is not None and parsed.hostname.casefold() not in hosts: + return None + return text[:2_000] + + +class _AllowlistedRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, request, fp, code, msg, headers, newurl): + parsed = urllib.parse.urlparse(newurl) + if parsed.scheme != "https" or parsed.hostname not in ALLOWED_API_HOSTS: + raise ScholarError("provider redirect is not allowlisted") + return super().redirect_request(request, fp, code, msg, headers, newurl) + + +def _default_transport(url: str, timeout: float, max_bytes: int) -> dict[str, Any]: + parsed = urllib.parse.urlparse(url) + if parsed.scheme != "https" or parsed.hostname not in ALLOWED_API_HOSTS: + raise ScholarError("provider URL is not allowlisted") + request = urllib.request.Request( + url, + headers={"Accept": "application/json", "User-Agent": "hermes-hegi/2.1 scholar-read-only"}, + ) + try: + opener = urllib.request.build_opener(_AllowlistedRedirect()) + deadline = time.monotonic() + timeout + with opener.open(request, timeout=timeout) as response: + final = urllib.parse.urlparse(response.geturl()) + if final.scheme != "https" or final.hostname not in ALLOWED_API_HOSTS: + raise ScholarError("provider response URL is not allowlisted") + content_type = response.headers.get_content_type() + declared = response.headers.get("Content-Length") + if declared and int(declared) > max_bytes: + raise ScholarError("provider response is oversized") + chunks = [] + size = 0 + while size <= max_bytes: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise ScholarError("provider response exceeded total timeout") + with suppress(AttributeError): + response.fp.raw._sock.settimeout(remaining) + chunk = response.read(min(64 * 1024, max_bytes + 1 - size)) + if not chunk: + break + chunks.append(chunk) + size += len(chunk) + body = b"".join(chunks) + except (OSError, ValueError, urllib.error.URLError) as exc: + raise ScholarError(type(exc).__name__) from exc + if content_type != "application/json" or len(body) > max_bytes: + raise ScholarError("provider returned an unsafe response") + try: + payload = json.loads(body) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise ScholarError("provider returned malformed JSON") from exc + if not isinstance(payload, dict): + raise ScholarError("provider JSON root must be an object") + return payload + + +@dataclass(slots=True) +class _CacheEntry: + expires_at: float + value: dict[str, Any] + + +class ScholarAdapter: + """Search OpenAlex, falling back to Crossref, without downloading full text.""" + + def __init__( + self, + *, + provider: str = "openalex", + fallback_provider: str = "crossref", + timeout_seconds: float = 15, + cache_ttl: int = 900, + transport: Transport | None = None, + ): + if provider not in {"openalex", "crossref"}: + raise ValueError("unsupported scholar provider") + if fallback_provider not in {"openalex", "crossref", ""}: + raise ValueError("unsupported scholar fallback provider") + self.provider = provider + self.fallback_provider = fallback_provider + self.timeout_seconds = float(timeout_seconds) + self.cache_ttl = int(cache_ttl) + self.transport = transport or _default_transport + self._cache: dict[str, _CacheEntry] = {} + + def _request(self, url: str) -> dict[str, Any]: + now = time.monotonic() + cached = self._cache.get(url) + if cached and cached.expires_at >= now: + return cached.value + payload = self.transport(url, self.timeout_seconds, MAX_RESPONSE_BYTES) + if not isinstance(payload, dict): + raise ScholarError("transport returned a non-object") + if self.cache_ttl > 0: + if len(self._cache) >= 128: + self._cache.pop(next(iter(self._cache))) + self._cache[url] = _CacheEntry(now + self.cache_ttl, payload) + return payload + + @staticmethod + def _abstract(index: Any) -> str | None: + if not isinstance(index, dict): + return None + positions: list[tuple[int, str]] = [] + for word, offsets in index.items(): + if not isinstance(offsets, list): + continue + positions.extend((position, str(word)) for position in offsets if isinstance(position, int)) + return _clean_text(" ".join(word for _, word in sorted(positions))) or None + + @staticmethod + def _openalex_work(item: dict[str, Any]) -> dict[str, Any]: + authorships = item.get("authorships", []) + if not isinstance(authorships, list): + authorships = [] + authors = [ + _clean_text(entry.get("author", {}).get("display_name"), 300) + for entry in authorships + if isinstance(entry, dict) and isinstance(entry.get("author"), dict) + ] + primary = item.get("primary_location") if isinstance(item.get("primary_location"), dict) else {} + source = primary.get("source") if isinstance(primary.get("source"), dict) else {} + oa = item.get("open_access") if isinstance(item.get("open_access"), dict) else {} + work_id = str(item.get("id", "")).rsplit("/", 1)[-1] + doi = _doi(item.get("doi")) + return { + "memory_type": "external_scholar", + "provider": "openalex", + "work_id": work_id, + "title": _clean_text(item.get("display_name") or item.get("title"), 1_000), + "authors": [author for author in authors if author][:100], + "year": item.get("publication_year") if isinstance(item.get("publication_year"), int) else None, + "venue": _clean_text(source.get("display_name"), 500) or None, + "doi": doi, + "abstract": ScholarAdapter._abstract(item.get("abstract_inverted_index")), + "citation_count": item.get("cited_by_count") if isinstance(item.get("cited_by_count"), int) else None, + "canonical_url": _safe_url(item.get("id"), hosts={"openalex.org"}) or (f"https://doi.org/{doi}" if doi else None), + "open_access": oa.get("is_oa") if isinstance(oa.get("is_oa"), bool) else None, + "open_access_url": _safe_url(oa.get("oa_url")), + } + + @staticmethod + def _crossref_work(item: dict[str, Any]) -> dict[str, Any]: + title_value = item.get("title", []) + title = title_value[0] if isinstance(title_value, list) and title_value else title_value + authors = [] + for author in item.get("author", []) if isinstance(item.get("author"), list) else []: + if isinstance(author, dict): + authors.append(_clean_text(" ".join(filter(None, [author.get("given"), author.get("family")])), 300)) + year = None + for key in ("published-print", "published-online", "issued"): + dates = item.get(key, {}).get("date-parts", []) if isinstance(item.get(key), dict) else [] + if dates and isinstance(dates[0], list) and dates[0] and isinstance(dates[0][0], int): + year = dates[0][0] + break + container = item.get("container-title", []) + venue = container[0] if isinstance(container, list) and container else container + doi = _doi(item.get("DOI")) + return { + "memory_type": "external_scholar", + "provider": "crossref", + "work_id": doi or _clean_text(item.get("URL"), 1_000), + "title": _clean_text(title, 1_000), + "authors": [author for author in authors if author][:100], + "year": year, + "venue": _clean_text(venue, 500) or None, + "doi": doi, + "abstract": _clean_text(item.get("abstract")) or None, + "citation_count": item.get("is-referenced-by-count") if isinstance(item.get("is-referenced-by-count"), int) else None, + "canonical_url": f"https://doi.org/{doi}" if doi else _safe_url(item.get("URL")), + "open_access": None, + "open_access_url": None, + } + + def _search_provider( + self, provider: str, query: str, limit: int, year_from: int | None, + year_to: int | None, sort: str, + ) -> list[dict[str, Any]]: + if provider == "openalex": + params: dict[str, str | int] = {"search": query, "per-page": limit} + filters = [] + if year_from: + filters.append(f"from_publication_date:{year_from}-01-01") + if year_to: + filters.append(f"to_publication_date:{year_to}-12-31") + if filters: + params["filter"] = ",".join(filters) + params["sort"] = {"relevance": "relevance_score:desc", "recency": "publication_date:desc", "citations": "cited_by_count:desc"}[sort] + payload = self._request(f"{OPENALEX_API}/works?{urllib.parse.urlencode(params)}") + rows = payload.get("results") + normalizer = self._openalex_work + else: + params = {"query": query, "rows": limit, "sort": {"relevance": "relevance", "recency": "published", "citations": "is-referenced-by-count"}[sort], "order": "desc"} + filters = [] + if year_from: + filters.append(f"from-pub-date:{year_from}-01-01") + if year_to: + filters.append(f"until-pub-date:{year_to}-12-31") + if filters: + params["filter"] = ",".join(filters) + payload = self._request(f"{CROSSREF_API}/works?{urllib.parse.urlencode(params)}") + message = payload.get("message") + rows = message.get("items") if isinstance(message, dict) else None + normalizer = self._crossref_work + if not isinstance(rows, list) or not all(isinstance(item, dict) for item in rows): + raise ScholarError(f"{provider} returned malformed works") + try: + return [ + work for work in (normalizer(item) for item in rows[:limit]) if work["title"] + ] + except (KeyError, TypeError, ValueError) as exc: + raise ScholarError(f"{provider} returned malformed work metadata") from exc + + @staticmethod + def dedupe(works: list[dict[str, Any]]) -> list[dict[str, Any]]: + seen: set[tuple[Any, ...]] = set() + unique = [] + for work in works: + doi = _doi(work.get("doi")) + title = re.sub(r"[^\w가-힣]+", "", str(work.get("title", "")).casefold()) + authors = work.get("authors", []) + key = ("doi", doi) if doi else ("metadata", title, work.get("year"), str(authors[0]).casefold() if authors else "") + if key not in seen: + seen.add(key) + unique.append(work) + return unique + + def search( + self, query: str, limit: int = 10, year_from: int | None = None, + year_to: int | None = None, sort: str = "relevance", + ) -> dict[str, Any]: + query = _clean_text(query, 500) + if not query: + raise ValueError("scholar query must not be empty") + if not 1 <= int(limit) <= 50 or sort not in {"relevance", "recency", "citations"}: + raise ValueError("invalid scholar search options") + for label, value in (("year_from", year_from), ("year_to", year_to)): + if value is not None and (isinstance(value, bool) or not 1000 <= int(value) <= 2100): + raise ValueError(f"invalid {label}") + if year_from is not None and year_to is not None and int(year_from) > int(year_to): + raise ValueError("year_from must not exceed year_to") + providers = [self.provider] + if self.fallback_provider and self.fallback_provider not in providers: + providers.append(self.fallback_provider) + failures = [] + for index, provider in enumerate(providers): + try: + works = self.dedupe(self._search_provider(provider, query, int(limit), year_from, year_to, sort)) + if not works and index + 1 < len(providers): + failures.append(f"{provider}:empty") + continue + return {"memory_type": "external_scholar", "provider": provider, "query": query, "works": works, "status": "READY", "fallback_used": provider != self.provider, "attempted_providers": providers[: index + 1], "provider_failures": failures, "recommend_ingest": []} + except (ScholarError, TimeoutError) as exc: + failures.append(f"{provider}:{type(exc).__name__}") + return {"memory_type": "external_scholar", "provider": None, "query": query, "works": [], "status": "UNAVAILABLE", "fallback_used": len(providers) > 1, "attempted_providers": providers, "provider_failures": failures, "reason": ";".join(failures), "recommend_ingest": []} + + def get_work(self, identifier: str, provider: str | None = None) -> dict[str, Any] | None: + selected = provider or self.provider + clean = _clean_text(identifier, 500) + if selected == "openalex": + doi = _doi(clean) + if not re.fullmatch(r"W\d+", clean, flags=re.I) and not doi: + return None + target = clean if re.fullmatch(r"W\d+", clean, flags=re.I) else f"https://doi.org/{doi}" + payload = self._request(f"{OPENALEX_API}/works/{urllib.parse.quote(target, safe=':/')}") + return self._openalex_work(payload) + doi = _doi(clean) + if not doi: + return None + payload = self._request(f"{CROSSREF_API}/works/{urllib.parse.quote(doi, safe='')}") + message = payload.get("message") + return self._crossref_work(message) if isinstance(message, dict) else None diff --git a/hegi/cli.py b/hegi/cli.py index e0f7585..d1b227c 100644 --- a/hegi/cli.py +++ b/hegi/cli.py @@ -19,6 +19,7 @@ from ._version import __version__ from .adapters.research_to_skill import ResearchToSkillAdapter +from .adapters.scholar import ScholarAdapter from .bootstrap import _atomic_yaml, _backup, setup from .compat import Check, doctor_checks from .config import ( @@ -27,9 +28,11 @@ load_config, state_path_is_bound, validate_config, + validate_scholar_config, ) from .locking import run_with_process_lock from .pipeline import HegiPipeline +from .research_router import ResearchQueryRouter from .state import SCHEMA, StateStore from .yamlio import load_yaml_mapping @@ -75,6 +78,47 @@ def cmd_research_memory_status(args: argparse.Namespace) -> int: return 0 if payload["status"] in {"READY", "DISABLED"} else 1 +def cmd_scholar_search(args: argparse.Namespace) -> int: + config = load_config(args.config) + scholar = config.section("scholar") + if scholar.get("enabled") is not True: + _json({"memory_type": "external_scholar", "status": "DISABLED", "works": []}) + return 0 + scholar_errors = validate_scholar_config(scholar) + if scholar_errors: + _json({ + "memory_type": "external_scholar", + "status": "UNAVAILABLE", + "reason": "InvalidScholarConfiguration", + "errors": scholar_errors, + "works": [], + }) + return 1 + personal = None + research = config.section("research_memory") + if research.get("enabled") is True and research.get("read_only") is True: + personal = ResearchToSkillAdapter(str(research.get("project_path", ""))) + external = ScholarAdapter( + provider=str(scholar.get("provider", "openalex")), + fallback_provider=str(scholar.get("fallback_provider", "crossref")), + timeout_seconds=float(scholar.get("timeout_seconds", 15)), + cache_ttl=int(scholar.get("cache_ttl", 900)), + ) + result = ResearchQueryRouter( + personal_backend=personal, + scholar_backend=external, + default_limit=int(scholar.get("default_limit", 10)), + ).query( + args.query, + limit=args.limit, + year_from=args.year_from, + year_to=args.year_to, + sort=args.sort, + ) + _json(result) + return {"READY": 0, "PARTIAL": 2}.get(str(result.get("status")), 1) + + def _mask_identifier(value: str) -> str: text = str(value) if len(text) <= 4: @@ -1130,6 +1174,22 @@ def _configure_parser(parser: argparse.ArgumentParser) -> None: "status", help="show Personal Research Memory status" ).set_defaults(hegi_handler=cmd_research_memory_status) + scholar = subparsers.add_parser( + "scholar", help="query official scholarly metadata providers" + ) + scholar_commands = scholar.add_subparsers(dest="scholar_command", required=True) + scholar_search = scholar_commands.add_parser( + "search", help="route a personal, external, or hybrid research query" + ) + scholar_search.add_argument("query") + scholar_search.add_argument("--limit", type=int) + scholar_search.add_argument("--year-from", type=int) + scholar_search.add_argument("--year-to", type=int) + scholar_search.add_argument( + "--sort", choices=("relevance", "recency", "citations"), default="relevance" + ) + scholar_search.set_defaults(hegi_handler=cmd_scholar_search) + retry_failed = subparsers.add_parser( "retry-failed", help="retry one frozen failed episode after operator repair" ) diff --git a/hegi/config.py b/hegi/config.py index 6c4e4bc..9cdb4ed 100644 --- a/hegi/config.py +++ b/hegi/config.py @@ -87,6 +87,14 @@ "project_path": "", "read_only": True, }, + "scholar": { + "enabled": False, + "provider": "openalex", + "fallback_provider": "crossref", + "default_limit": 10, + "timeout_seconds": 15, + "cache_ttl": 900, + }, "daemon": {"poll_seconds": 60}, "reports": {"telegram": True}, "v3": {"enabled": False}, @@ -257,6 +265,36 @@ def state_path_is_bound(config: HegiConfig) -> bool: return stat.S_ISREG(metadata.st_mode) and metadata.st_nlink == 1 +def validate_scholar_config(scholar: dict[str, Any]) -> list[str]: + errors: list[str] = [] + provider = scholar.get("provider") + fallback = scholar.get("fallback_provider") + if provider not in {"openalex", "crossref"}: + errors.append("scholar.provider는 openalex 또는 crossref여야 합니다.") + if fallback not in {"openalex", "crossref", ""}: + errors.append("scholar.fallback_provider가 지원되지 않습니다.") + elif fallback and fallback == provider: + errors.append("scholar.fallback_provider는 primary provider와 달라야 합니다.") + limit = scholar.get("default_limit") + if isinstance(limit, bool) or not isinstance(limit, int) or not 1 <= limit <= 50: + errors.append("scholar.default_limit는 1~50 사이 정수여야 합니다.") + timeout = scholar.get("timeout_seconds") + if isinstance(timeout, bool): + errors.append("scholar.timeout_seconds는 양수여야 합니다.") + else: + try: + if float(timeout) <= 0: + errors.append("scholar.timeout_seconds는 양수여야 합니다.") + except (TypeError, ValueError): + errors.append("scholar.timeout_seconds는 숫자여야 합니다.") + cache_ttl = scholar.get("cache_ttl") + if isinstance(cache_ttl, bool) or not isinstance(cache_ttl, int): + errors.append("scholar.cache_ttl은 0 이상의 정수여야 합니다.") + elif cache_ttl < 0: + errors.append("scholar.cache_ttl은 0 이상이어야 합니다.") + return errors + + def validate_config(config: HegiConfig, *, require_runtime: bool = False) -> list[str]: errors: list[str] = [] if not isinstance(config.raw.get("enabled"), bool): @@ -312,6 +350,7 @@ def validate_config(config: HegiConfig, *, require_runtime: bool = False) -> lis errors.append(f"{label}는 정수여야 합니다.") memory = config.section("memory") research_memory = config.section("research_memory") + scholar = config.section("scholar") archive_config = config.section("archive") boolean_fields = { "memory.enabled": memory.get("enabled"), @@ -320,6 +359,7 @@ def validate_config(config: HegiConfig, *, require_runtime: bool = False) -> lis "memory.require_professor_approval": memory.get("require_professor_approval"), "research_memory.enabled": research_memory.get("enabled"), "research_memory.read_only": research_memory.get("read_only"), + "scholar.enabled": scholar.get("enabled"), "telegram.enabled": config.section("telegram").get("enabled"), "reports.telegram": config.section("reports").get("telegram"), "archive.sync_when_available": archive_config.get("sync_when_available"), @@ -345,6 +385,7 @@ def validate_config(config: HegiConfig, *, require_runtime: bool = False) -> lis errors.append("research_memory.project_path가 비어 있습니다.") elif not Path(project_path).expanduser().is_absolute(): errors.append("research_memory.project_path는 절대 경로여야 합니다.") + errors.extend(validate_scholar_config(scholar)) for key in ("markdown", "json"): if archive_config.get(key) is not True: errors.append(f"archive.{key}는 v2.1.0에서 true여야 합니다.") diff --git a/hegi/config/default.yaml b/hegi/config/default.yaml index 47fdf8e..861cee1 100644 --- a/hegi/config/default.yaml +++ b/hegi/config/default.yaml @@ -84,6 +84,14 @@ research_memory: project_path: "" read_only: true +scholar: + enabled: false + provider: "openalex" + fallback_provider: "crossref" + default_limit: 10 + timeout_seconds: 15 + cache_ttl: 900 + daemon: poll_seconds: 60 diff --git a/hegi/research_router.py b/hegi/research_router.py new file mode 100644 index 0000000..2f2bcf5 --- /dev/null +++ b/hegi/research_router.py @@ -0,0 +1,191 @@ +"""Route personal and external research queries without merging provenance.""" + +from __future__ import annotations + +import re +from typing import Any, Protocol + +from .adapters.research_to_skill import ResearchMemoryBackend +from .adapters.scholar import ScholarAdapter + + +class ScholarBackend(Protocol): + def search(self, query: str, limit: int = 10, year_from: int | None = None, + year_to: int | None = None, sort: str = "relevance") -> dict[str, Any]: ... + + +PERSONAL_MARKERS = ( + "내 ", "나의", "내가", "기존 연구", "내 연구", "우리 연구", + "my research", "my theory", "my work", "our research", +) +EXTERNAL_MARKERS = ( + "최근", "최신", "문헌", "학술", "찾아", "외부 연구", "scholar", + "recent literature", "recent research", "external research", "find papers", +) +HYBRID_MARKERS = ("비교", "대조", "차이", "공통점", "연결", "compare", "contrast") + + +def route_research_query(query: str) -> str: + text = query.casefold() + personal = any(marker in text for marker in PERSONAL_MARKERS) + external = any(marker in text for marker in EXTERNAL_MARKERS) + hybrid = any(marker in text for marker in HYBRID_MARKERS) + if personal and (external or hybrid): + return "hybrid" + if external: + return "external" + return "personal" + + +def _terms(value: str) -> set[str]: + stop = {"research", "study", "최근", "연구", "논문", "비교", "대한", "에서", "the", "and", "with"} + return { + token.casefold() + for token in re.findall(r"[\w가-힣]+", value) + if len(token) >= 3 and token.casefold() not in stop + } + + +def _scholar_query(query: str, personal: dict[str, Any] | None) -> str: + terms = re.findall(r"[A-Za-z][A-Za-z0-9-]*", query) + expansions = { + "생태비평": ("ecological", "criticism"), + "생태": ("ecology",), + "알고리즘 권력": ("algorithmic", "power"), + "인공자연": ("artificial", "nature"), + "매체미학": ("media", "aesthetics"), + } + for marker, values in expansions.items(): + if marker in query: + terms.extend(values) + query_folded = query.casefold() + for concept in (personal or {}).get("concepts", [])[:5]: + aliases = re.findall(r"[A-Za-z][A-Za-z0-9-]*", str(concept.get("name", ""))) + concept_id = str(concept.get("id", "")).replace("-", " ").casefold() + if concept_id in query_folded or any(alias.casefold() in query_folded for alias in aliases): + terms.extend(aliases) + unique = [] + for term in terms: + if term.casefold() not in {item.casefold() for item in unique}: + unique.append(term) + return " ".join(unique[:16]) or query + + +def _cross_analysis(personal: dict[str, Any], external: dict[str, Any]) -> dict[str, Any]: + personal_ready = personal.get("status") == "READY" + external_ready = external.get("status") == "READY" + concepts = personal.get("concepts", []) if isinstance(personal, dict) else [] + works = external.get("works", []) if isinstance(external, dict) else [] + similarities = [] + for concept in concepts: + concept_terms = _terms(f"{concept.get('id', '')} {concept.get('name', '')}") + for work in works: + overlap = sorted(concept_terms & _terms(str(work.get("title", "")))) + if overlap: + work_id = str(work.get("work_id", "")) + similarities.append({ + "personal_concept_id": concept.get("id"), + "external_work_id": work_id, + "basis": "shared_title_terms", + "terms": overlap, + }) + personal_dois = { + str(source.get("doi", "")).casefold() + for source in personal.get("sources", []) + if source.get("doi") + } + novelty = [ + { + "external_work_id": work.get("work_id"), + "doi": work.get("doi"), + "basis": "not_present_in_personal_source_metadata", + } + for work in works + if personal_ready + and work.get("doi") + and str(work.get("doi")).casefold() not in personal_dois + ] + differences = [ + { + "personal_scope": [concept.get("id") for concept in concepts], + "external_work_id": work.get("work_id"), + "basis": "separate_provenance; substantive difference requires full-text review", + } + for work in works[:5] + if personal_ready and external_ready + ] + gaps = [] + if personal_ready and external_ready and concepts and works and not similarities: + gaps.append({ + "personal_concept_ids": [concept.get("id") for concept in concepts], + "external_work_ids": [work.get("work_id") for work in works[:5]], + "basis": "no controlled title-term overlap; abstract/full-text review needed", + }) + return { + "memory_type": "synthesis", + "similarities": similarities, + "differences": differences, + "novelty": novelty, + "research_gaps": gaps, + "counterarguments": [], + "limitations": [ + "Metadata-only comparison; no claim-level external inference.", + "Empty counterarguments mean evidence was insufficient, not that none exist.", + *( + ["Personal Research was unavailable; novelty and differences were not assessed."] + if not personal_ready else [] + ), + ], + } + + +class ResearchQueryRouter: + def __init__( + self, + *, + personal_backend: ResearchMemoryBackend | None = None, + scholar_backend: ScholarBackend | None = None, + default_limit: int = 10, + ): + self.personal_backend = personal_backend + self.scholar_backend = scholar_backend + self.default_limit = default_limit + + def query( + self, query: str, *, limit: int | None = None, year_from: int | None = None, + year_to: int | None = None, sort: str = "relevance", + ) -> dict[str, Any]: + route = route_research_query(query) + personal: dict[str, Any] | None = None + external: dict[str, Any] | None = None + if route in {"personal", "hybrid"}: + if self.personal_backend is None: + personal = {"memory_type": "personal_research", "status": "UNAVAILABLE", "found": False} + else: + personal = self.personal_backend.get_context(query) + if route in {"external", "hybrid"}: + if self.scholar_backend is None: + external = {"memory_type": "external_scholar", "status": "UNAVAILABLE", "works": []} + else: + scholar_query = _scholar_query(query, personal) + external = self.scholar_backend.search( + scholar_query, + limit=self.default_limit if limit is None else limit, + year_from=year_from, + year_to=year_to, sort=sort, + ) + external["routed_query"] = scholar_query + result: dict[str, Any] = {"route": route} + if personal is not None: + result["personal_research"] = personal + if external is not None: + result["external_scholar"] = external + if route == "hybrid": + result["cross_analysis"] = _cross_analysis(personal or {}, external or {}) + required = [value for value in (personal, external) if value is not None] + ready = sum(item.get("status") == "READY" for item in required) + result["status"] = "READY" if ready == len(required) else ("PARTIAL" if ready else "UNAVAILABLE") + return result + + +__all__ = ["ResearchQueryRouter", "ScholarAdapter", "route_research_query"] diff --git a/tests/unit/test_config_state.py b/tests/unit/test_config_state.py index 65ac77f..b7fea73 100644 --- a/tests/unit/test_config_state.py +++ b/tests/unit/test_config_state.py @@ -66,6 +66,48 @@ def test_research_memory_is_read_only_and_supports_project_env_override( ) +def test_scholar_config_rejects_unsafe_provider_and_limits(tmp_path): + config_path = tmp_path / "hegi.yaml" + config_path.write_text( + """ +scholar: + enabled: true + provider: google-scholar + fallback_provider: scraper + default_limit: 0 + timeout_seconds: -1 + cache_ttl: -1 +""", + encoding="utf-8", + ) + errors = validate_config(load_config(config_path)) + for label in ( + "scholar.provider", + "scholar.fallback_provider", + "scholar.default_limit", + "scholar.timeout_seconds", + "scholar.cache_ttl", + ): + assert any(label in error for error in errors), label + + +def test_scholar_config_rejects_identical_fallback_and_fractional_limit(tmp_path): + config_path = tmp_path / "hegi.yaml" + config_path.write_text( + """ +scholar: + enabled: true + provider: openalex + fallback_provider: openalex + default_limit: 1.5 +""", + encoding="utf-8", + ) + errors = validate_config(load_config(config_path)) + assert any("fallback_provider" in error and "달라야" in error for error in errors) + assert any("default_limit" in error and "정수" in error for error in errors) + + def test_existing_state_database_migrates_approval_workflow_columns(tmp_path): path = tmp_path / "legacy.db" import sqlite3 diff --git a/tests/unit/test_scholar_adapter.py b/tests/unit/test_scholar_adapter.py new file mode 100644 index 0000000..c722da8 --- /dev/null +++ b/tests/unit/test_scholar_adapter.py @@ -0,0 +1,308 @@ +from __future__ import annotations + +import json + +import pytest + +from hegi.adapters.scholar import ( + MAX_RESPONSE_BYTES, + ScholarAdapter, + ScholarError, + _AllowlistedRedirect, + _default_transport, +) +from hegi.cli import main +from hegi.research_router import ResearchQueryRouter, route_research_query + + +def openalex_payload(): + return { + "results": [ + { + "id": "https://openalex.org/W123", + "display_name": "Mechanocene and AI Ecology", + "publication_year": 2025, + "authorships": [{"author": {"display_name": "Ada Researcher"}}], + "primary_location": {"source": {"display_name": "Digital Ecology"}}, + "doi": "https://doi.org/10.1000/TEST", + "abstract_inverted_index": {"Ecological": [0], "analysis": [1]}, + "cited_by_count": 7, + "open_access": {"is_oa": True, "oa_url": "https://example.org/paper"}, + } + ] + } + + +def crossref_payload(): + return { + "message": { + "items": [ + { + "DOI": "10.1000/fallback", + "title": ["Artificial Nature and Media Aesthetics"], + "author": [{"given": "Grace", "family": "Scholar"}], + "published-online": {"date-parts": [[2024, 1, 2]]}, + "container-title": ["Media Theory"], + "abstract": "Safe summary", + "is-referenced-by-count": 3, + "URL": "https://doi.org/10.1000/fallback", + } + ] + } + } + + +def test_openalex_search_normalizes_external_provenance_and_cache(): + calls = [] + + def transport(url, timeout, max_bytes): + calls.append((url, timeout, max_bytes)) + return openalex_payload() + + adapter = ScholarAdapter(transport=transport, cache_ttl=60) + first = adapter.search("Mechanocene ecology", year_from=2020, sort="citations") + second = adapter.search("Mechanocene ecology", year_from=2020, sort="citations") + work = first["works"][0] + assert first["provider"] == "openalex" + assert work == { + "memory_type": "external_scholar", + "provider": "openalex", + "work_id": "W123", + "title": "Mechanocene and AI Ecology", + "authors": ["Ada Researcher"], + "year": 2025, + "venue": "Digital Ecology", + "doi": "10.1000/test", + "abstract": "Ecological analysis", + "citation_count": 7, + "canonical_url": "https://openalex.org/W123", + "open_access": True, + "open_access_url": "https://example.org/paper", + } + assert first == second + assert len(calls) == 1 + assert first["recommend_ingest"] == [] + + +def test_crossref_fallback_and_html_contamination_removal(): + def transport(url, _timeout, _max_bytes): + if "openalex" in url: + raise ScholarError("offline") + return crossref_payload() + + result = ScholarAdapter(transport=transport).search("artificial nature") + work = result["works"][0] + assert result["fallback_used"] is True + assert result["provider"] == "crossref" + assert work["memory_type"] == "external_scholar" + assert work["doi"] == "10.1000/fallback" + assert work["abstract"] == "Safe summary" + assert "script" not in work["abstract"] + + +def test_empty_primary_result_uses_fallback(): + def transport(url, _timeout, _max_bytes): + return {"results": []} if "openalex" in url else crossref_payload() + + result = ScholarAdapter(transport=transport).search("specialized term") + assert result["provider"] == "crossref" + assert result["fallback_used"] is True + + +def test_malformed_primary_collection_uses_fallback(): + def transport(url, _timeout, _max_bytes): + return {"results": [None]} if "openalex" in url else crossref_payload() + + result = ScholarAdapter(transport=transport).search("specialized term") + assert result["provider"] == "crossref" + assert result["provider_failures"] == ["openalex:ScholarError"] + + +def test_redirect_handler_rejects_non_provider_destination(): + with pytest.raises(ScholarError, match="redirect"): + _AllowlistedRedirect().redirect_request( + None, None, 302, "Found", {}, "http://127.0.0.1/internal" + ) + + +def test_doi_and_metadata_deduplication(): + works = [ + {"doi": "10.1234/SAME", "title": "One", "year": 2024, "authors": ["A"]}, + {"doi": "https://doi.org/10.1234/same", "title": "Duplicate", "year": 2025, "authors": ["B"]}, + {"doi": None, "title": "No DOI!", "year": 2023, "authors": ["First Author"]}, + {"doi": None, "title": "No DOI", "year": 2023, "authors": ["First Author"]}, + ] + assert len(ScholarAdapter.dedupe(works)) == 2 + + +def test_all_providers_unavailable_falls_back_without_raising(): + def unavailable(_url, _timeout, _max_bytes): + raise TimeoutError + + result = ScholarAdapter(transport=unavailable).search("query") + assert result["status"] == "UNAVAILABLE" + assert result["works"] == [] + + +def test_invalid_search_range_and_limit_are_rejected(): + adapter = ScholarAdapter(transport=lambda *_args: openalex_payload()) + with pytest.raises(ValueError): + adapter.search("query", limit=0) + with pytest.raises(ValueError): + adapter.search("query", year_from=2025, year_to=2024) + + +@pytest.mark.parametrize("body", [b"{broken", b"[]"]) +def test_default_transport_rejects_malformed_json(monkeypatch, body): + class Headers: + def get_content_type(self): + return "application/json" + + def get(self, _key): + return None + + class Response: + headers = Headers() + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def read(self, _limit): + return body + + def geturl(self): + return "https://api.openalex.org/works" + + class Opener: + def open(self, *_args, **_kwargs): + return Response() + + monkeypatch.setattr("urllib.request.build_opener", lambda *_args: Opener()) + with pytest.raises(ScholarError): + _default_transport("https://api.openalex.org/works", 1, MAX_RESPONSE_BYTES) + + +def test_default_transport_rejects_oversized_payload(monkeypatch): + class Headers: + def get_content_type(self): + return "application/json" + + def get(self, key): + return str(MAX_RESPONSE_BYTES + 1) if key == "Content-Length" else None + + class Response: + headers = Headers() + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def geturl(self): + return "https://api.openalex.org/works" + + class Opener: + def open(self, *_args, **_kwargs): + return Response() + + monkeypatch.setattr("urllib.request.build_opener", lambda *_args: Opener()) + with pytest.raises(ScholarError, match="oversized"): + _default_transport("https://api.openalex.org/works", 1, MAX_RESPONSE_BYTES) + + +class Personal: + def get_context(self, query): + return { + "memory_type": "personal_research", + "status": "READY", + "query": query, + "concepts": [{"id": "mechanocene", "name": "Mechanocene"}], + "claims": [{"id": "claim-1", "origin": "author"}], + "sources": [], + "found": True, + } + + +class External: + def search(self, query, limit=10, year_from=None, year_to=None, sort="relevance"): + return { + "memory_type": "external_scholar", + "status": "READY", + "provider": "openalex", + "query": query, + "works": [{ + "memory_type": "external_scholar", + "provider": "openalex", + "work_id": "W1", + "title": "Mechanocene ecology", + "authors": ["External Author"], + "year": 2025, + "doi": "10.1/external", + }], + "recommend_ingest": [], + } + + +def test_routing_and_hybrid_provenance_separation(): + assert route_research_query("내 Mechanocene 정의는?") == "personal" + assert route_research_query("최근 Mechanocene 연구를 찾아줘") == "external" + assert route_research_query("compare my research with recent literature") == "hybrid" + assert route_research_query("내 논문에서 Mechanocene은?") == "personal" + query = "최근 AI 생태비평 연구와 내 Mechanocene 이론을 비교해" + result = ResearchQueryRouter( + personal_backend=Personal(), scholar_backend=External() + ).query(query) + assert result["route"] == "hybrid" + assert result["personal_research"]["memory_type"] == "personal_research" + assert result["external_scholar"]["works"][0]["memory_type"] == "external_scholar" + assert "Mechanocene" in result["external_scholar"]["routed_query"] + synthesis = result["cross_analysis"] + assert synthesis["memory_type"] == "synthesis" + assert synthesis["similarities"][0]["personal_concept_id"] == "mechanocene" + assert synthesis["similarities"][0]["external_work_id"] == "W1" + assert synthesis["counterarguments"] == [] + assert result["external_scholar"]["recommend_ingest"] == [] + assert result["status"] == "READY" + + +def test_degraded_hybrid_does_not_assert_novelty(): + result = ResearchQueryRouter( + personal_backend=None, scholar_backend=External() + ).query("compare my research with recent literature") + assert result["status"] == "PARTIAL" + assert result["cross_analysis"]["novelty"] == [] + assert result["cross_analysis"]["differences"] == [] + + +def test_scholar_cli_reports_ready_and_partial_exit_codes(tmp_path, monkeypatch, capsys): + config = tmp_path / "config.yaml" + config.write_text( + "scholar:\n" + " enabled: true\n" + " provider: openalex\n" + " fallback_provider: crossref\n" + " default_limit: 3\n" + " timeout_seconds: 2\n" + " cache_ttl: 0\n", + encoding="utf-8", + ) + monkeypatch.setattr("hegi.cli.ScholarAdapter", lambda **_kwargs: External()) + assert main(["--config", str(config), "scholar", "search", "recent literature"]) == 0 + assert json.loads(capsys.readouterr().out)["status"] == "READY" + assert main([ + "--config", str(config), "scholar", "search", + "compare my research with recent literature", + ]) == 2 + assert json.loads(capsys.readouterr().out)["status"] == "PARTIAL" + + +def test_unexpected_transport_error_is_not_swallowed(): + def programming_error(_url, _timeout, _max_bytes): + raise AssertionError("fixture bug") + + with pytest.raises(AssertionError): + ScholarAdapter(transport=programming_error).search("query")