diff --git a/src/personal_mcp/registry/contracts.py b/src/personal_mcp/registry/contracts.py new file mode 100644 index 0000000..d53e043 --- /dev/null +++ b/src/personal_mcp/registry/contracts.py @@ -0,0 +1,15 @@ +from .models.registry import DocsRegistry +from .models.registry import PromptRecord +from .models.registry import PromptSummaryRecord +from .models.registry import ReferenceRecord +from .models.registry import SkillRecord +from .models.registry import SkillSummaryRecord + +__all__ = [ + "DocsRegistry", + "PromptRecord", + "PromptSummaryRecord", + "ReferenceRecord", + "SkillRecord", + "SkillSummaryRecord", +] diff --git a/src/personal_mcp/registry/issues.py b/src/personal_mcp/registry/issues.py deleted file mode 100644 index 7eb52bf..0000000 --- a/src/personal_mcp/registry/issues.py +++ /dev/null @@ -1,22 +0,0 @@ -from dataclasses import dataclass - - -@dataclass(frozen=True) -class RegistryIssue: - code: str - message: str - skill_id: str | None - path: str - hint: str - - -class DocsRegistryValidationError(Exception): - def __init__(self, errors: list[RegistryIssue]) -> None: - self.errors = errors - summary = "\n".join( - [ - (f"{issue.code}: {issue.message} (skill={issue.skill_id or 'unknown'}, path={issue.path})") - for issue in errors - ] - ) - super().__init__(summary) diff --git a/src/personal_mcp/registry/load.py b/src/personal_mcp/registry/load.py index 6163b95..8ec2d3b 100644 --- a/src/personal_mcp/registry/load.py +++ b/src/personal_mcp/registry/load.py @@ -1,448 +1,229 @@ -from importlib.resources import files +from __future__ import annotations + +import importlib +from collections import defaultdict +from pathlib import Path from pathlib import PurePosixPath +from typing import Any -from pydantic import ValidationError +import yaml -from ..skills.document_loader import _discover_top_level_references -from ..skills.document_loader import _parse_frontmatter -from ..skills.document_loader import _validate_prompt_frontmatter -from ..skills.document_loader import _validate_skill_frontmatter -from ..skills.document_loader import _walk_markdown -from .contracts import DocsRegistry -from .contracts import PromptRecord -from .contracts import PromptSummaryRecord -from .contracts import ReferenceRecord -from .contracts import SkillRecord -from .contracts import SkillSummaryRecord -from .issues import DocsRegistryValidationError -from .issues import RegistryIssue +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.models.common import ReferenceEntry +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 +from personal_mcp.skills.document_loader import _normalize_docs_path +from personal_mcp.skills.document_loader import _reference_id_from_filename +from personal_mcp.skills.document_loader import _title_from_reference_filename +from personal_mcp.skills.document_loader import _validate_prompt_frontmatter +from personal_mcp.skills.document_loader import _validate_skill_frontmatter -def _ensure_no_cycles( - skills_by_id: dict[str, SkillRecord], -) -> list[tuple[str, str]]: - state: dict[str, int] = {} - stack: list[str] = [] - cycles: set[tuple[str, str]] = set() - - def visit(skill_id: str) -> None: - state[skill_id] = 1 - stack.append(skill_id) - - for dependency in skills_by_id[skill_id].depends_on: - if dependency not in skills_by_id: - continue - dep_state = state.get(dependency, 0) - if dep_state == 0: - visit(dependency) - continue - if dep_state == 1: - cycle_start = stack[stack.index(dependency)] - cycle_path = stack[stack.index(dependency) :] + [dependency] - cycles.add((cycle_start, " -> ".join(cycle_path))) - - stack.pop() - state[skill_id] = 2 - - for skill_id in sorted(skills_by_id): - if state.get(skill_id, 0) == 0: - visit(skill_id) - - return sorted(cycles) +def _parse_frontmatter(raw_frontmatter: str | None, *, path: PurePosixPath) -> dict[str, Any]: + 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 _load_skills( - *, - docs_markdown_by_path: dict[str, str], - skills_root, -) -> tuple[dict[str, SkillRecord], list[SkillSummaryRecord], list[RegistryIssue]]: - skills_by_id: dict[str, SkillRecord] = {} - summaries: list[SkillSummaryRecord] = [] - issues: list[RegistryIssue] = [] - - for skill_dir in sorted(skills_root.iterdir(), key=lambda item: item.name): - if not skill_dir.is_dir(): +def _discover_reference_entries(bundle: SkillFilesBundle) -> dict[str, ReferenceEntry]: + discovered: dict[str, ReferenceEntry] = {} + for reference_doc in bundle.references: + ref_id = _reference_id_from_filename(reference_doc.relpath.name) + if ref_id is None: continue - - skill_dir_name = skill_dir.name - skill_rel_root = PurePosixPath("skills").joinpath(skill_dir_name) - skill_doc_relpath = skill_rel_root.joinpath("SKILL.md").as_posix() - skill_doc_file = skill_dir.joinpath("SKILL.md") - - if not skill_doc_file.is_file(): - issues.append( - RegistryIssue( - code="missing_skill_document", - message="missing required SKILL.md", - skill_id=skill_dir_name, - path=skill_doc_relpath, - hint="add docs/skills//SKILL.md", - ) - ) - continue - - skill_markdown = skill_doc_file.read_text(encoding="utf-8") - try: - raw_frontmatter, _ = _parse_frontmatter( - skill_markdown, - path=skill_doc_relpath, - ) - frontmatter = _validate_skill_frontmatter( - raw_frontmatter, - skill_dir_name=skill_dir_name, - ) - except (ValueError, ValidationError) as exc: - issues.append( - RegistryIssue( - code="invalid_frontmatter", - message=str(exc), - skill_id=skill_dir_name, - path=skill_doc_relpath, - hint="fix SKILL.md YAML frontmatter to match the contract", - ) - ) - continue - - effective_reference_entries = _discover_top_level_references(skill_dir=skill_dir) - effective_reference_entries.update(frontmatter.x_personal_mcp.references) - - references: dict[str, ReferenceRecord] = {} - for ref_id, ref_entry in effective_reference_entries.items(): - ref_relpath = skill_rel_root.joinpath(ref_entry.path).as_posix() - if ref_relpath not in docs_markdown_by_path: - issues.append( - RegistryIssue( - code="missing_reference", - message=f"reference target is missing for ref_id '{ref_id}'", - skill_id=frontmatter.name, - path=ref_relpath, - hint="fix x-personal-mcp.references path or add the referenced markdown file", - ) - ) - continue - - references[ref_id] = ReferenceRecord( - ref_id=ref_id, - uri=f"resource://skills/{frontmatter.name}/references/{ref_id}", - relpath=ref_relpath, - mime_type=ref_entry.mime_type, - title=ref_entry.title, - content=docs_markdown_by_path[ref_relpath], - ) - - skill_id = frontmatter.name - if skill_id in skills_by_id: - issues.append( - RegistryIssue( - code="duplicate_skill_id", - message="duplicate skill id discovered", - skill_id=skill_id, - path=skill_doc_relpath, - hint="ensure each skill directory has a unique id", - ) - ) - continue - - record = SkillRecord( - skill_id=skill_id, - name=frontmatter.name, - description=frontmatter.description, - version=frontmatter.x_personal_mcp.version, - tags=tuple(frontmatter.x_personal_mcp.tags), - capabilities=tuple(frontmatter.x_personal_mcp.capabilities), - depends_on=tuple(frontmatter.x_personal_mcp.depends_on), - document_uri=f"resource://skills/{skill_id}/document", - document_relpath=skill_doc_relpath, - document_content=skill_markdown, - references=references, + discovered[ref_id] = ReferenceEntry( + path=PurePosixPath("references").joinpath(reference_doc.relpath.name).as_posix(), + title=_title_from_reference_filename(reference_doc.relpath.name), ) - skills_by_id[skill_id] = record - summaries.append( - SkillSummaryRecord( - skill_id=record.skill_id, - name=record.name, - description=record.description, - tags=record.tags, - capabilities=record.capabilities, - document_uri=record.document_uri, - version=record.version, - ) + return discovered + + +def _build_skill_record( + *, bundle: SkillFilesBundle, docs_by_relpath: dict[PurePosixPath, MarkdownDocument] +) -> SkillRecord: + frontmatter_raw = _parse_frontmatter(bundle.skill.frontmatter, path=bundle.skill.relpath) + frontmatter = _validate_skill_frontmatter(frontmatter_raw, skill_dir_name=bundle.slug) + metadata = frontmatter.x_personal_mcp + + merged_entries = _discover_reference_entries(bundle) + merged_entries.update(dict(metadata.references)) + + references: dict[str, ReferenceRecord] = {} + for ref_id, entry in sorted(merged_entries.items()): + ref_relpath = PurePosixPath("skills").joinpath(bundle.slug).joinpath(entry.path) + if ref_relpath not in docs_by_relpath: + raise KeyError(f"reference document not found for '{metadata.id}:{ref_id}' at {ref_relpath.as_posix()}") + ref_doc = docs_by_relpath[ref_relpath] + references[ref_id] = ReferenceRecord( + ref_id=ref_id, + uri=f"resource://skills/{metadata.id}/references/{ref_id}", + relpath=ref_relpath.as_posix(), + mime_type=entry.mime_type, + title=entry.title, + content=ref_doc.content, ) - return skills_by_id, summaries, issues - - -def _load_prompts( - *, - prompts_root, - skills_by_id: dict[str, SkillRecord], -) -> tuple[dict[str, PromptRecord], list[PromptSummaryRecord], list[RegistryIssue]]: - prompts_by_id: dict[str, PromptRecord] = {} - prompt_summaries: list[PromptSummaryRecord] = [] - issues: list[RegistryIssue] = [] - discovered_prompt_docs: set[str] = set() - - for prompt_dir in sorted(prompts_root.iterdir(), key=lambda item: item.name): - if not prompt_dir.is_dir(): - continue - - prompt_dir_name = prompt_dir.name - prompt_rel_root = PurePosixPath("prompts").joinpath(prompt_dir_name) - prompt_doc_relpath = prompt_rel_root.joinpath("PROMPT.md").as_posix() - prompt_doc_file = prompt_dir.joinpath("PROMPT.md") - - if not prompt_doc_file.is_file(): - continue - - discovered_prompt_docs.add(prompt_doc_relpath) - - prompt_markdown = prompt_doc_file.read_text(encoding="utf-8") - try: - raw_frontmatter, _ = _parse_frontmatter( - prompt_markdown, - path=prompt_doc_relpath, - ) - frontmatter = _validate_prompt_frontmatter( - raw_frontmatter, - prompt_dir_name=prompt_dir_name, - ) - except (ValueError, ValidationError) as exc: - issues.append( - RegistryIssue( - code="invalid_prompt_frontmatter", - message=str(exc), - skill_id=prompt_dir_name, - path=prompt_doc_relpath, - hint="fix PROMPT.md YAML frontmatter to match the contract", - ) - ) - continue - - prompt_id = frontmatter.name - if prompt_id in prompts_by_id: - issues.append( - RegistryIssue( - code="duplicate_prompt_id", - message="duplicate prompt id discovered", - skill_id=prompt_id, - path=prompt_doc_relpath, - hint="ensure each prompt directory has a unique id", - ) - ) - continue - - if prompt_id in skills_by_id: - issues.append( - RegistryIssue( - code="prompt_skill_id_collision", - message="prompt id collides with an existing skill id", - skill_id=prompt_id, - path=prompt_doc_relpath, - hint="use a unique prompt id that does not match a skill id", - ) - ) - continue - - prompt_record = PromptRecord( - prompt_id=prompt_id, - name=frontmatter.name, - description=frontmatter.description, - version=frontmatter.x_personal_mcp.version, - tags=tuple(frontmatter.x_personal_mcp.tags), - capabilities=tuple(frontmatter.x_personal_mcp.capabilities), - arguments=frontmatter.x_personal_mcp.arguments, - document_uri=f"resource://prompts/{prompt_id}/document", - document_relpath=prompt_doc_relpath, - document_content=prompt_markdown, - ) - prompts_by_id[prompt_id] = prompt_record - prompt_summaries.append( - PromptSummaryRecord( - prompt_id=prompt_record.prompt_id, - name=prompt_record.name, - description=prompt_record.description, - tags=prompt_record.tags, - capabilities=prompt_record.capabilities, - document_uri=prompt_record.document_uri, - version=prompt_record.version, - ) - ) - - return prompts_by_id, prompt_summaries, issues - - -def _collect_relationship_and_uri_issues( - *, - skills_by_id: dict[str, SkillRecord], - prompts_by_id: dict[str, PromptRecord], -) -> list[RegistryIssue]: - issues: list[RegistryIssue] = [] - - for skill_id, record in sorted(skills_by_id.items()): - for dependency in record.depends_on: - if dependency == skill_id: - issues.append( - RegistryIssue( - code="self_dependency", - message="skill must not depend on itself", - skill_id=skill_id, - path=record.document_relpath, - hint="remove the skill id from depends_on", - ) - ) - elif dependency not in skills_by_id: - issues.append( - RegistryIssue( - code="missing_dependency", - message=f"depends_on target '{dependency}' does not exist", - skill_id=skill_id, - path=record.document_relpath, - hint="add the missing skill or remove it from depends_on", - ) - ) - - for cycle_start, cycle in _ensure_no_cycles(skills_by_id): - issues.append( - RegistryIssue( - code="dependency_cycle", - message=f"depends_on cycle detected: {cycle}", - skill_id=cycle_start, - path=skills_by_id[cycle_start].document_relpath, - hint="remove at least one dependency edge in the cycle", - ) - ) - - seen_uris: set[str] = set() - for skill_id, record in sorted(skills_by_id.items()): - uris = [record.document_uri] + [ref.uri for ref in record.references.values()] - for uri in uris: - if uri in seen_uris: - issues.append( - RegistryIssue( - code="duplicate_uri", - message=f"duplicate resource URI generated: {uri}", - skill_id=skill_id, - path=record.document_relpath, - hint="ensure unique skill ids and reference ids", - ) - ) - seen_uris.add(uri) - - for prompt_id, record in sorted(prompts_by_id.items()): - uri = record.document_uri - if uri in seen_uris: - issues.append( - RegistryIssue( - code="duplicate_uri", - message=f"duplicate resource URI generated: {uri}", - skill_id=prompt_id, - path=record.document_relpath, - hint="ensure unique prompt ids", - ) - ) - seen_uris.add(uri) - - return issues - - -def load_docs_registry( - *, - package_anchor: str, - docs_root: str = "docs", -) -> DocsRegistry: - docs_dir = files(package_anchor).joinpath(docs_root) - - if not docs_dir.is_dir(): - raise DocsRegistryValidationError( - [ - RegistryIssue( - code="missing_docs_root", - message="docs root directory does not exist", - skill_id=None, - path=docs_root, - hint="configure docs_root to a valid packaged docs path", - ) - ] - ) - - docs_markdown_by_path = { - relpath: doc_file.read_text(encoding="utf-8") for relpath, doc_file in _walk_markdown(docs_dir) - } - - skills_root = docs_dir.joinpath("skills") - if not skills_root.is_dir(): - raise DocsRegistryValidationError( - [ - RegistryIssue( - code="missing_skills_root", - message="skills directory does not exist under docs root", - skill_id=None, - path=f"{docs_root}/skills", - hint="ensure docs/skills is included in packaged docs", - ) - ] - ) - - skills_by_id, summaries, skill_issues = _load_skills( - docs_markdown_by_path=docs_markdown_by_path, - skills_root=skills_root, + return SkillRecord( + skill_id=metadata.id, + name=frontmatter.name, + description=frontmatter.description, + version=metadata.version, + tags=tuple(metadata.tags), + capabilities=tuple(metadata.capabilities), + depends_on=tuple(metadata.depends_on), + document_uri=f"resource://skills/{metadata.id}/document", + document_relpath=bundle.skill.relpath.as_posix(), + document_content=bundle.skill.content, + references=references, ) + +def _build_prompt_record(*, bundle: PromptFilesBundle) -> PromptRecord: + frontmatter_raw = _parse_frontmatter(bundle.prompt.frontmatter, path=bundle.prompt.relpath) + frontmatter = _validate_prompt_frontmatter(frontmatter_raw, prompt_dir_name=bundle.slug) + metadata = frontmatter.x_personal_mcp + + return PromptRecord( + prompt_id=metadata.id, + name=frontmatter.name, + description=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=bundle.prompt.relpath.as_posix(), + document_content=bundle.prompt.content, + ) + + +def _to_summary(skill: SkillRecord) -> SkillSummaryRecord: + return SkillSummaryRecord( + skill_id=skill.skill_id, + name=skill.name, + description=skill.description, + tags=skill.tags, + capabilities=skill.capabilities, + document_uri=skill.document_uri, + version=skill.version, + ) + + +def _to_prompt_summary(prompt: PromptRecord) -> PromptSummaryRecord: + return PromptSummaryRecord( + prompt_id=prompt.prompt_id, + name=prompt.name, + description=prompt.description, + tags=prompt.tags, + capabilities=prompt.capabilities, + document_uri=prompt.document_uri, + version=prompt.version, + ) + + +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] = {} - prompt_summaries: list[PromptSummaryRecord] = [] - prompt_issues: list[RegistryIssue] = [] + 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) - prompts_root = docs_dir.joinpath("prompts") - if prompts_root.is_dir(): - prompts_by_id, prompt_summaries, prompt_issues = _load_prompts( - prompts_root=prompts_root, - skills_by_id=skills_by_id, - ) + for skill_id, skill in skills_by_id.items(): + for dependency in skill.depends_on: + if dependency not in skills_by_id: + raise ValueError(f"skill '{skill_id}' depends_on unknown skill '{dependency}'") - issues = [ - *skill_issues, - *prompt_issues, - *_collect_relationship_and_uri_issues( - skills_by_id=skills_by_id, - prompts_by_id=prompts_by_id, - ), - ] - - if issues: - raise DocsRegistryValidationError(issues) - - skill_ids = tuple(sorted(skills_by_id)) - summary_by_id = {summary.skill_id: summary for summary in summaries} - ordered_summaries = tuple(summary_by_id[skill_id] for skill_id in skill_ids) - prompt_ids = tuple(sorted(prompts_by_id)) - prompt_summary_by_id = {summary.prompt_id: summary for summary in prompt_summaries} - ordered_prompt_summaries = tuple(prompt_summary_by_id[prompt_id] for prompt_id in prompt_ids) - - tag_index: dict[str, list[str]] = {} - capability_index: dict[str, list[str]] = {} - prompt_tag_index: dict[str, list[str]] = {} - for skill_id in skill_ids: - record = skills_by_id[skill_id] - for tag in record.tags: - tag_index.setdefault(tag, []).append(skill_id) - for capability in record.capabilities: - capability_index.setdefault(capability, []).append(skill_id) - - for prompt_id in prompt_ids: - record = prompts_by_id[prompt_id] - for tag in record.tags: - prompt_tag_index.setdefault(tag, []).append(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=skill_ids, - skills_summary_in_load_order=ordered_summaries, + skills_in_load_order=skills_in_order_tuple, + skills_summary_in_load_order=tuple(_to_summary(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={key: tuple(sorted(values)) for key, values in sorted(tag_index.items())}, - capability_to_skill_ids={key: tuple(sorted(values)) for key, values in sorted(capability_index.items())}, + 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=prompt_ids, - prompts_summary_in_load_order=ordered_prompt_summaries, - tag_to_prompt_ids={key: tuple(sorted(values)) for key, values in sorted(prompt_tag_index.items())}, + prompts_in_load_order=prompts_in_order_tuple, + prompts_summary_in_load_order=tuple( + _to_prompt_summary(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), ) diff --git a/src/personal_mcp/registry/models/document.py b/src/personal_mcp/registry/models/document.py deleted file mode 100644 index cb94dc6..0000000 --- a/src/personal_mcp/registry/models/document.py +++ /dev/null @@ -1,56 +0,0 @@ -from pathlib import PurePosixPath -from typing import Self - -from .common import StrictFrozenModel - - -class MarkdownDocumentModel(StrictFrozenModel): - """Normalized markdown document content with path and frontmatter.""" - - relpath: str - content: str - frontmatter: str | None = None - skill_slug: str | None = None - - -class SkillFilesBundleModel(StrictFrozenModel): - """Grouped skill documents partitioned into skill, references, and other files.""" - - slug: str - skill: MarkdownDocumentModel - references: tuple[MarkdownDocumentModel, ...] = () - other: tuple[MarkdownDocumentModel, ...] = () - - @classmethod - def from_parts( - cls, - *, - slug: str, - skill: MarkdownDocumentModel, - references: tuple[MarkdownDocumentModel, ...] = (), - other: tuple[MarkdownDocumentModel, ...] = (), - ) -> Self: - # Ensure deterministic ordering even when the source collection is unordered. - references_sorted = tuple(sorted(references, key=lambda doc: doc.relpath)) - other_sorted = tuple(sorted(other, key=lambda doc: doc.relpath)) - return cls( - slug=slug, - skill=skill, - references=references_sorted, - other=other_sorted, - ) - - -def to_markdown_document_model( - *, - relpath: PurePosixPath, - content: str, - frontmatter: str | None, - skill_slug: str | None, -) -> MarkdownDocumentModel: - return MarkdownDocumentModel( - relpath=relpath.as_posix(), - content=content, - frontmatter=frontmatter, - skill_slug=skill_slug, - ) diff --git a/src/personal_mcp/registry/models/prompt.py b/src/personal_mcp/registry/models/prompt.py index 1003ce9..7db015e 100644 --- a/src/personal_mcp/registry/models/prompt.py +++ b/src/personal_mcp/registry/models/prompt.py @@ -1,8 +1,10 @@ import re from collections.abc import Mapping +from pathlib import PurePosixPath from typing import Any from typing import Literal +import yaml from pydantic import Field from pydantic import field_validator @@ -100,10 +102,21 @@ class PromptFrontmatter(StrictFrozenModel): return value -class PromptDocumentModel(StrictFrozenModel): - """Structured representation of a prompt markdown document.""" +class StoredPrompt(StrictFrozenModel): + """Normalized prompt document content with path and frontmatter for storage in the registry.""" prompt_id: str - relpath: str + relpath: PurePosixPath content: str - frontmatter: PromptFrontmatter + frontmatter: PromptFrontmatter | None = None + + @field_validator("frontmatter", mode="before") + @classmethod + def parse_frontmatter_yaml(cls, value: str | None): + if value is None: + return None + try: + data = yaml.safe_load(value) + except yaml.YAMLError as e: + raise ValueError(f"invalid YAML in frontmatter: {e}") from e + return PromptFrontmatter.model_validate(data) diff --git a/src/personal_mcp/registry/models/skill.py b/src/personal_mcp/registry/models/skill.py index ec143ea..d3b378f 100644 --- a/src/personal_mcp/registry/models/skill.py +++ b/src/personal_mcp/registry/models/skill.py @@ -1,6 +1,7 @@ from collections.abc import Mapping from pathlib import PurePosixPath +import yaml from pydantic import Field from pydantic import field_validator @@ -107,19 +108,31 @@ class SkillFrontmatter(StrictFrozenModel): return value -class SkillDocumentModel(StrictFrozenModel): - """Structured representation of a skill markdown document.""" - - skill_id: str - relpath: PurePosixPath - content: str - frontmatter: SkillFrontmatter - - -class SkillReferenceDocumentModel(StrictFrozenModel): +class StoredSkillReference(StrictFrozenModel): """Structured representation of a skill reference markdown document.""" ref_id: str relpath: PurePosixPath content: str entry: ReferenceEntry + + +class StoredSkill(StrictFrozenModel): + """Structured representation of a skill markdown document.""" + + skill_id: str + relpath: PurePosixPath + content: str + frontmatter: SkillFrontmatter + references: dict[str, StoredSkillReference] = Field(default_factory=dict) + + @field_validator("frontmatter", mode="before") + @classmethod + def parse_frontmatter_yaml(cls, value: str | None): + if value is None: + return None + try: + data = yaml.safe_load(value) + except yaml.YAMLError as e: + raise ValueError(f"invalid YAML in frontmatter: {e}") from e + return SkillFrontmatter.model_validate(data) diff --git a/src/personal_mcp/registry/prompts.py b/src/personal_mcp/registry/prompts.py deleted file mode 100644 index e69de29..0000000 diff --git a/tests/registry/ingest/test_current_docs.py b/tests/registry/ingest/test_current_docs.py index 82e3bc8..faf0b1b 100644 --- a/tests/registry/ingest/test_current_docs.py +++ b/tests/registry/ingest/test_current_docs.py @@ -7,7 +7,6 @@ import pytest 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.load import load_docs_registry pytestmark = pytest.mark.unit @@ -41,10 +40,3 @@ class TestCurrentDocsIngestion: bundles = PromptFilesBundle.from_docs(docs.values()) assert {bundle.slug for bundle in bundles} == expected_slugs - - def test_loads_current_registry(self) -> None: - """Ensures the current docs registry loads without validation errors.""" - registry = load_docs_registry(package_anchor="personal_mcp", docs_root="../../docs") - - assert set(registry.skills_by_id) == {path.parent.name for path in DOCS_ROOT.glob("skills/*/SKILL.md")} - assert set(registry.prompts_by_id) == {path.parent.name for path in DOCS_ROOT.glob("prompts/*/PROMPT.md")}