diff --git a/README.md b/README.md
index 586f6c3..4f8c06e 100644
--- a/README.md
+++ b/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 |
diff --git a/hegi/adapters/__init__.py b/hegi/adapters/__init__.py
index 1343dca..bfdcb00 100644
--- a/hegi/adapters/__init__.py
+++ b/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",
+]
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/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 8785b5a..d1b227c 100644
--- a/hegi/cli.py
+++ b/hegi/cli.py
@@ -18,6 +18,8 @@
import yaml
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 (
@@ -26,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
@@ -49,6 +53,72 @@ 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 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:
@@ -1094,6 +1164,32 @@ 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)
+
+ 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 fa33ea9..9cdb4ed 100644
--- a/hegi/config.py
+++ b/hegi/config.py
@@ -81,6 +81,20 @@
"reject": ["기억하지 마"],
},
},
+ "research_memory": {
+ "enabled": False,
+ "provider": "research-to-skill",
+ "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},
@@ -175,6 +189,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] = []
@@ -248,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):
@@ -302,12 +349,17 @@ 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")
+ scholar = config.section("scholar")
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"),
+ "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"),
@@ -323,6 +375,17 @@ 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는 절대 경로여야 합니다.")
+ 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 55e4884..861cee1 100644
--- a/hegi/config/default.yaml
+++ b/hegi/config/default.yaml
@@ -78,6 +78,20 @@ memory:
reject:
- "기억하지 마"
+research_memory:
+ enabled: false
+ provider: "research-to-skill"
+ 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/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/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 9dd85e7..b7fea73 100644
--- a/tests/unit/test_config_state.py
+++ b/tests/unit/test_config_state.py
@@ -38,6 +38,76 @@ 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_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_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"
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")