added prompt ingestion

This commit is contained in:
John Lancaster
2026-06-21 15:51:51 -05:00
parent 4f05f13e45
commit 36032040ae
5 changed files with 231 additions and 48 deletions
+7 -1
View File
@@ -1,7 +1,7 @@
from collections.abc import Generator from collections.abc import Generator
from dataclasses import dataclass from dataclasses import dataclass
from dataclasses import field from dataclasses import field
from importlib.abc import Traversable from importlib.resources.abc import Traversable
from itertools import starmap from itertools import starmap
from pathlib import PurePosixPath from pathlib import PurePosixPath
from typing import Self from typing import Self
@@ -37,6 +37,12 @@ class MarkdownDocument:
if parts[0] == "skills" and len(parts) >= 3: if parts[0] == "skills" and len(parts) >= 3:
return parts[1] return parts[1]
@property
def prompt_slug(self) -> str | None:
parts = self.relpath.parts
if parts[0] == "prompts" and len(parts) >= 3:
return parts[1]
def walk_resources( def walk_resources(
node: Traversable, node: Traversable,
@@ -0,0 +1,48 @@
from collections.abc import Iterable
from dataclasses import dataclass
from importlib.resources.abc import Traversable
from itertools import starmap
from typing import Self
from .document import MarkdownDocument
@dataclass(frozen=True, slots=True)
class PromptFilesBundle:
"""Represents a prompt and all of its associated markdown files."""
slug: str
prompt: MarkdownDocument
other: tuple[MarkdownDocument, ...]
@classmethod
def from_root(cls, root: Traversable) -> list[Self]:
# Should only be used for testing
return list(cls.from_docs(MarkdownDocument.from_root(root).values()))
@classmethod
def from_docs(cls, docs: Iterable[MarkdownDocument]) -> tuple[Self, ...]:
return tuple(starmap(cls.from_paths, group_prompt_paths(docs).items()))
@classmethod
def from_paths(cls, slug: str, paths: set[MarkdownDocument]) -> Self:
prompt = next(iter(p for p in paths if p.relpath.name == "PROMPT.md"))
sorted_paths = tuple(sorted(paths, key=lambda p: p.relpath.as_posix()))
other = tuple(p for p in sorted_paths if p != prompt)
return cls(
slug=slug,
prompt=prompt,
other=other,
)
def group_prompt_paths(docs: Iterable[MarkdownDocument]) -> dict[str, set[MarkdownDocument]]:
"""Group prompts from a list of markdown documents by their prompt slug."""
grouped: dict[str, set[MarkdownDocument]] = {}
for doc in sorted(
filter(lambda d: d.prompt_slug is not None, docs),
key=lambda d: d.relpath.as_posix(),
):
if doc.prompt_slug:
grouped.setdefault(doc.prompt_slug, set()).add(doc)
return grouped
+1 -47
View File
@@ -1,47 +1 @@
from .common import SEMVER_RE """Pydantic models for the document registry."""
from .common import SKILL_ID_RE
from .common import ReferenceEntry
from .common import StrictFrozenModel
from .document import MarkdownDocumentModel
from .document import SkillFilesBundleModel
from .document import to_markdown_document_model
from .prompt import PromptArgumentEntry
from .prompt import PromptDocumentModel
from .prompt import PromptFrontmatter
from .prompt import PromptMetadata
from .registry import DocsRegistry
from .registry import PromptRecord
from .registry import PromptSummaryRecord
from .registry import ReferenceRecord
from .registry import SkillRecord
from .registry import SkillSummaryRecord
from .skill import PersonalMcpMetadata
from .skill import SkillDocumentModel
from .skill import SkillFrontmatter
from .skill import SkillMetadata
from .skill import SkillReferenceDocumentModel
__all__ = [
"SEMVER_RE",
"SKILL_ID_RE",
"DocsRegistry",
"MarkdownDocumentModel",
"PersonalMcpMetadata",
"PromptArgumentEntry",
"PromptDocumentModel",
"PromptFrontmatter",
"PromptMetadata",
"PromptRecord",
"PromptSummaryRecord",
"ReferenceEntry",
"ReferenceRecord",
"SkillDocumentModel",
"SkillFilesBundleModel",
"SkillFrontmatter",
"SkillMetadata",
"SkillRecord",
"SkillReferenceDocumentModel",
"SkillSummaryRecord",
"StrictFrozenModel",
"to_markdown_document_model",
]
+21
View File
@@ -102,6 +102,27 @@ class TestMarkdownDocument:
assert doc.skill_slug is None assert doc.skill_slug is None
class TestPromptSlugProperty:
"""Covers prompt_slug derivation from document relative paths."""
def test_returns_slug(self) -> None:
"""Ensures prompt_slug returns the slug for valid prompt paths."""
doc = MarkdownDocument(relpath=PurePosixPath("prompts/demo/PROMPT.md"), content="#")
assert doc.prompt_slug == "demo"
def test_none_for_non_prompt(self) -> None:
"""Ensures prompt_slug is None for non-prompt paths."""
doc = MarkdownDocument(relpath=PurePosixPath("docs/index.md"), content="#")
assert doc.prompt_slug is None
def test_none_for_incomplete_prompt(self) -> None:
"""Ensures prompt_slug is None for incomplete prompt paths."""
doc = MarkdownDocument(relpath=PurePosixPath("prompts/demo"), content="#")
assert doc.prompt_slug is None
class TestWalkResources: class TestWalkResources:
"""Covers recursive resource walking and markdown filtering behavior.""" """Covers recursive resource walking and markdown filtering behavior."""
+154
View File
@@ -0,0 +1,154 @@
from __future__ import annotations
from collections.abc import Callable
from pathlib import Path
import pytest
from personal_mcp.registry.ingest.document import MarkdownDocument
from personal_mcp.registry.ingest.prompt import PromptFilesBundle
from personal_mcp.registry.ingest.prompt import group_prompt_paths
pytestmark = pytest.mark.unit
MakeDoc = Callable[[str], MarkdownDocument]
class TestPromptFilesBundle:
"""Covers PromptFilesBundle construction and path-based categorization."""
class TestFromRoot:
"""Covers bundle creation from resource roots."""
def test_builds_bundles(self, tmp_path: Path) -> None:
"""Ensures from_root builds bundles from discovered markdown docs."""
alpha = tmp_path / "prompts" / "alpha"
beta = tmp_path / "prompts" / "beta"
alpha.mkdir(parents=True)
beta.mkdir(parents=True)
(alpha / "PROMPT.md").write_text("# alpha\n", encoding="utf-8")
(alpha / "notes.md").write_text("notes\n", encoding="utf-8")
(beta / "PROMPT.md").write_text("# beta\n", encoding="utf-8")
bundles = PromptFilesBundle.from_root(tmp_path)
assert {bundle.slug for bundle in bundles} == {"alpha", "beta"}
def test_delegates_to_from_docs(
self,
tmp_path: Path,
) -> None:
"""Ensures from_root delegates bundle assembly to from_docs."""
alpha = tmp_path / "prompts" / "alpha"
beta = tmp_path / "prompts" / "beta"
alpha.mkdir(parents=True)
beta.mkdir(parents=True)
(alpha / "PROMPT.md").write_text("# alpha\n", encoding="utf-8")
(alpha / "notes.md").write_text("notes\n", encoding="utf-8")
(beta / "PROMPT.md").write_text("# beta\n", encoding="utf-8")
from_root = PromptFilesBundle.from_root(tmp_path)
from_docs = PromptFilesBundle.from_docs(MarkdownDocument.from_root(tmp_path).values())
assert tuple(from_root) == from_docs
class TestFromDocs:
"""Covers bundle creation from preloaded markdown documents."""
def test_groups_by_slug(self, make_doc: MakeDoc) -> None:
"""Ensures from_docs groups documents by prompt slug."""
docs = [
make_doc("prompts/alpha/PROMPT.md"),
make_doc("prompts/alpha/notes.md"),
make_doc("prompts/beta/PROMPT.md"),
]
bundles = PromptFilesBundle.from_docs(docs)
assert {bundle.slug for bundle in bundles} == {"alpha", "beta"}
def test_one_bundle_per_slug(self, make_doc: MakeDoc) -> None:
"""Ensures from_docs produces one PromptFilesBundle per slug."""
docs = [
make_doc("prompts/alpha/PROMPT.md"),
make_doc("prompts/alpha/notes.md"),
make_doc("prompts/alpha/changelog.md"),
]
bundles = PromptFilesBundle.from_docs(docs)
assert len(bundles) == 1
assert bundles[0].slug == "alpha"
class TestFromPaths:
"""Covers classification of prompt and other documents."""
def test_selects_prompt_md(self, make_doc: MakeDoc) -> None:
"""Ensures from_paths selects PROMPT.md as the primary document."""
docs = {
make_doc("prompts/alpha/PROMPT.md"),
make_doc("prompts/alpha/notes.md"),
}
bundle = PromptFilesBundle.from_paths("alpha", docs)
assert bundle.prompt.relpath.name == "PROMPT.md"
def test_collects_other_docs(self, make_doc: MakeDoc) -> None:
"""Ensures from_paths classifies non-prompt docs as other docs."""
docs = {
make_doc("prompts/alpha/PROMPT.md"),
make_doc("prompts/alpha/notes.md"),
make_doc("prompts/alpha/changelog.md"),
}
bundle = PromptFilesBundle.from_paths("alpha", docs)
assert {doc.relpath.as_posix() for doc in bundle.other} == {
"prompts/alpha/changelog.md",
"prompts/alpha/notes.md",
}
class TestGroupPromptPaths:
"""Covers grouping markdown documents by derived prompt slug."""
def test_groups_slugged_docs(self, make_doc: MakeDoc) -> None:
"""Ensures group_prompt_paths groups only documents with a slug."""
docs = [
make_doc("prompts/alpha/PROMPT.md"),
make_doc("prompts/alpha/notes.md"),
make_doc("prompts/beta/PROMPT.md"),
]
grouped = group_prompt_paths(docs)
assert set(grouped) == {"alpha", "beta"}
def test_excludes_unslugged_docs(self, make_doc: MakeDoc) -> None:
"""Ensures group_prompt_paths excludes documents without prompt slugs."""
docs = [
make_doc("docs/index.md"),
make_doc("prompts/legacy.md"),
]
assert group_prompt_paths(docs) == {}
def test_returns_sets(self, make_doc: MakeDoc) -> None:
"""Ensures group_prompt_paths returns sets of docs per slug."""
grouped = group_prompt_paths([make_doc("prompts/alpha/PROMPT.md")])
assert isinstance(grouped["alpha"], set)
def test_stable_grouping(self, make_doc: MakeDoc) -> None:
"""Ensures group_prompt_paths behaves consistently after internal sorting."""
docs = [
make_doc("prompts/beta/PROMPT.md"),
make_doc("prompts/alpha/notes.md"),
make_doc("prompts/alpha/PROMPT.md"),
]
grouped_forward = group_prompt_paths(docs)
grouped_reverse = group_prompt_paths(list(reversed(docs)))
assert grouped_forward == grouped_reverse