feat: integrate personal research memory and external scholar search

README.md

@@ -90,6 +90,50 @@ 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`.
+
+## 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 |

hegi/adapters/__init__.py

@@ -1,6 +1,16 @@
 """Stable host and transport boundaries used by HEGI."""
 
 from .hermes import HermesLLMAdapter, HermesToolDispatcher
+from .research_to_skill import ResearchMemoryBackend, ResearchToSkillAdapter
+from .scholar import ScholarAdapter, ScholarError
 from .telegram import TelegramBotAdapter
 
-__all__ = ["HermesLLMAdapter", "HermesToolDispatcher", "TelegramBotAdapter"]
+__all__ = [
+    "HermesLLMAdapter",
+    "HermesToolDispatcher",
+    "ResearchMemoryBackend",
+    "ResearchToSkillAdapter",
+    "ScholarAdapter",
+    "ScholarError",
+    "TelegramBotAdapter",
+]

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,
+            }