from __future__ import annotations import importlib from collections import defaultdict from pathlib import Path from pathlib import PurePosixPath import yaml from personal_mcp.registry.ingest.document import MarkdownDocument from personal_mcp.registry.ingest.prompt import PromptFilesBundle from personal_mcp.registry.ingest.skill import SkillFilesBundle from personal_mcp.registry.ingest.skill import build_stored_skill from personal_mcp.registry.models.common import _normalize_docs_path from personal_mcp.registry.models.prompt import StoredPrompt from personal_mcp.registry.models.registry import DocsRegistry from personal_mcp.registry.models.registry import PromptRecord from personal_mcp.registry.models.registry import PromptSummaryRecord from personal_mcp.registry.models.registry import ReferenceRecord from personal_mcp.registry.models.registry import SkillRecord from personal_mcp.registry.models.registry import SkillSummaryRecord def _parse_frontmatter(raw_frontmatter: str | None, *, path: PurePosixPath) -> dict[str, object]: """Parse frontmatter YAML into a mapping for downstream validation. This helper is retained for compatibility with model-validation tests that exercise gate behavior directly at parse boundaries. """ if raw_frontmatter is None: raise ValueError(f"missing YAML frontmatter: {path.as_posix()}") parsed = yaml.safe_load(raw_frontmatter) if not isinstance(parsed, dict): raise TypeError(f"frontmatter must parse to an object: {path.as_posix()}") return parsed def _build_skill_record( *, bundle: SkillFilesBundle, docs_by_relpath: dict[PurePosixPath, MarkdownDocument] ) -> SkillRecord: stored = build_stored_skill(bundle=bundle, docs_by_relpath=docs_by_relpath) metadata = stored.frontmatter.x_personal_mcp references: dict[str, ReferenceRecord] = {} for ref_id, ref in sorted(stored.references.items()): references[ref_id] = ReferenceRecord( ref_id=ref_id, uri=f"resource://skills/{metadata.id}/references/{ref_id}", relpath=ref.relpath.as_posix(), mime_type=ref.entry.mime_type, title=ref.entry.title, content=ref.content, ) return SkillRecord( skill_id=metadata.id, name=stored.frontmatter.name, description=stored.frontmatter.description, version=metadata.version, tags=tuple(metadata.tags), capabilities=tuple(metadata.capabilities), document_uri=f"resource://skills/{metadata.id}/document", document_relpath=stored.relpath.as_posix(), document_content=stored.content, references=references, ) def _build_prompt_record(*, bundle: PromptFilesBundle) -> PromptRecord: stored = StoredPrompt.from_bundle(bundle) metadata = stored.frontmatter.x_personal_mcp return PromptRecord( prompt_id=metadata.id, name=stored.frontmatter.name, description=stored.frontmatter.description, version=metadata.version, tags=tuple(metadata.tags), capabilities=tuple(metadata.capabilities), arguments=dict(metadata.arguments), document_uri=f"resource://prompts/{metadata.id}/document", document_relpath=stored.relpath.as_posix(), document_content=stored.content, ) def _build_tag_index_skills( skills_in_order: tuple[str, ...], skills_by_id: dict[str, SkillRecord] ) -> dict[str, tuple[str, ...]]: tag_index: defaultdict[str, list[str]] = defaultdict(list) for skill_id in skills_in_order: for tag in skills_by_id[skill_id].tags: tag_index[tag].append(skill_id) return {tag: tuple(ids) for tag, ids in sorted(tag_index.items())} def _build_capability_index( skills_in_order: tuple[str, ...], skills_by_id: dict[str, SkillRecord] ) -> dict[str, tuple[str, ...]]: capability_index: defaultdict[str, list[str]] = defaultdict(list) for skill_id in skills_in_order: for capability in skills_by_id[skill_id].capabilities: capability_index[capability].append(skill_id) return {capability: tuple(ids) for capability, ids in sorted(capability_index.items())} def _build_tag_index_prompts( prompts_in_order: tuple[str, ...], prompts_by_id: dict[str, PromptRecord], ) -> dict[str, tuple[str, ...]]: tag_index: defaultdict[str, list[str]] = defaultdict(list) for prompt_id in prompts_in_order: for tag in prompts_by_id[prompt_id].tags: tag_index[tag].append(prompt_id) return {tag: tuple(ids) for tag, ids in sorted(tag_index.items())} def _resolve_docs_root(*, package_anchor: str, docs_root: str) -> Path: package = importlib.import_module(package_anchor) package_file = getattr(package, "__file__", None) if package_file is None: raise ValueError(f"package anchor '{package_anchor}' has no file location") resolved = Path(package_file).resolve().parent.joinpath(docs_root).resolve() if not resolved.exists() or not resolved.is_dir(): raise FileNotFoundError(f"docs root does not exist or is not a directory: {resolved}") return resolved def load_docs_registry(*, package_anchor: str, docs_root: str = "docs") -> DocsRegistry: docs_path = _resolve_docs_root(package_anchor=package_anchor, docs_root=docs_root) docs = MarkdownDocument.from_root(docs_path) docs_markdown_by_path = {_normalize_docs_path(relpath.as_posix()): doc.content for relpath, doc in docs.items()} skill_bundles = SkillFilesBundle.from_docs(docs.values()) prompt_bundles = PromptFilesBundle.from_docs(docs.values()) docs_by_relpath = {doc.relpath: doc for doc in docs.values()} skills_by_id: dict[str, SkillRecord] = {} skills_in_load_order: list[str] = [] for bundle in skill_bundles: record = _build_skill_record(bundle=bundle, docs_by_relpath=docs_by_relpath) if record.skill_id in skills_by_id: raise ValueError(f"duplicate skill_id detected: {record.skill_id}") skills_by_id[record.skill_id] = record skills_in_load_order.append(record.skill_id) prompts_by_id: dict[str, PromptRecord] = {} prompts_in_load_order: list[str] = [] for bundle in prompt_bundles: record = _build_prompt_record(bundle=bundle) if record.prompt_id in prompts_by_id: raise ValueError(f"duplicate prompt_id detected: {record.prompt_id}") if record.prompt_id in skills_by_id: raise ValueError(f"prompt_id collides with existing skill_id: {record.prompt_id}") prompts_by_id[record.prompt_id] = record prompts_in_load_order.append(record.prompt_id) skills_in_order_tuple = tuple(skills_in_load_order) prompts_in_order_tuple = tuple(prompts_in_load_order) return DocsRegistry( skills_by_id=skills_by_id, skills_in_load_order=skills_in_order_tuple, skills_summary_in_load_order=tuple( SkillSummaryRecord.from_record(skills_by_id[skill_id]) for skill_id in skills_in_order_tuple ), docs_markdown_by_path=docs_markdown_by_path, docs_markdown_path_index=tuple(sorted(docs_markdown_by_path)), tag_to_skill_ids=_build_tag_index_skills(skills_in_order_tuple, skills_by_id), capability_to_skill_ids=_build_capability_index(skills_in_order_tuple, skills_by_id), prompts_by_id=prompts_by_id, prompts_in_load_order=prompts_in_order_tuple, prompts_summary_in_load_order=tuple( PromptSummaryRecord.from_record(prompts_by_id[prompt_id]) for prompt_id in prompts_in_order_tuple ), tag_to_prompt_ids=_build_tag_index_prompts(prompts_in_order_tuple, prompts_by_id), )