feat(launchpad): one-document one-task corpus manifest (#626)

launchpad/project-intelligence/corpus/manifest.py

@@ -0,0 +1,172 @@
+"""One-document one-task corpus manifest -- issue #626.
+
+Turns a caller-supplied plan (one dict per planned canonical or generated
+corpus document) into a validated, deterministic `Manifest` -- the input
+`launchpad/project-intelligence/corpus/issue_plan.py` (#627, not yet built)
+will read to create GitHub tasks without manual rewriting.
+
+This module does not itself decide WHICH documents the corpus needs, what
+their titles are, or which template each gets -- that is real product
+knowledge (`launchpad/docs/corpus/AGENTS.md`'s per-type standards and
+templates, most still unmerged per issue #605) that belongs to whoever
+curates the plan, not to a script guessing on their behalf. What this module
+owns is turning that plan into a manifest with the structural guarantees
+issue #626's definition of done requires: one row per document, every
+required field present, no document assigned to two tasks, no task owning
+two documents, and no Feature exceeding GitHub's 100-sub-issue limit --
+enforced, not assumed.
+
+Run as a library -- `build_manifest(plan)` is the entry point tests and
+future callers use. There is no CLI; the plan itself has no fixed source
+yet (see #626's "Out of scope": authoring canonical corpus documents is not
+this task).
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+
+# GitHub's current sub-issue limit per parent, cited directly in #626's
+# definition of done. If GitHub raises this limit, this constant is the one
+# place to change it.
+_MAX_CHILDREN_PER_FEATURE = 100
+
+_REQUIRED_KEYS = frozenset(
+    {
+        "path",
+        "filename",
+        "issue_title",
+        "parent_feature",
+        "priority",
+        "start_date",
+        "target_date",
+        "effort",
+        "blockers",
+        "template",
+        "purpose",
+        "audiences",
+        "source_start_points",
+    }
+)
+
+
+class ManifestValidationError(Exception):
+    """The plan violates one of #626's structural guarantees. Never silently dropped or renamed."""
+
+
+@dataclass(frozen=True)
+class ManifestRow:
+    path: str
+    filename: str
+    issue_title: str
+    parent_feature: str
+    priority: str
+    start_date: str | None
+    target_date: str | None
+    effort: str
+    blockers: tuple[str, ...]
+    template: str
+    purpose: str
+    audiences: tuple[str, ...]
+    source_start_points: tuple[str, ...]
+
+    def to_dict(self) -> dict:
+        return {
+            "path": self.path,
+            "filename": self.filename,
+            "issue_title": self.issue_title,
+            "parent_feature": self.parent_feature,
+            "priority": self.priority,
+            "start_date": self.start_date,
+            "target_date": self.target_date,
+            "effort": self.effort,
+            "blockers": list(self.blockers),
+            "template": self.template,
+            "purpose": self.purpose,
+            "audiences": list(self.audiences),
+            "source_start_points": list(self.source_start_points),
+        }
+
+
+@dataclass
+class Manifest:
+    rows: list[ManifestRow] = field(default_factory=list)
+
+    def to_json(self) -> str:
+        payload = {"rows": [row.to_dict() for row in self.rows]}
+        return json.dumps(payload, indent=2, sort_keys=True) + "\n"
+
+
+def _row_from_plan_entry(entry: dict) -> ManifestRow:
+    missing = _REQUIRED_KEYS - entry.keys()
+    if missing:
+        raise ManifestValidationError(
+            f"plan entry for {entry.get('path', '<no path given>')!r} "
+            f"is missing required field(s): {sorted(missing)}"
+        )
+    return ManifestRow(
+        path=entry["path"],
+        filename=entry["filename"],
+        issue_title=entry["issue_title"],
+        parent_feature=entry["parent_feature"],
+        priority=entry["priority"],
+        start_date=entry["start_date"],
+        target_date=entry["target_date"],
+        effort=entry["effort"],
+        blockers=tuple(entry["blockers"]),
+        template=entry["template"],
+        purpose=entry["purpose"],
+        audiences=tuple(entry["audiences"]),
+        source_start_points=tuple(entry["source_start_points"]),
+    )
+
+
+def _check_no_duplicate_paths(rows: list[ManifestRow]) -> None:
+    seen: dict[str, ManifestRow] = {}
+    for row in rows:
+        if row.path in seen:
+            raise ManifestValidationError(
+                f"document {row.path!r} is assigned to two tasks: "
+                f"{seen[row.path].issue_title!r} and {row.issue_title!r}"
+            )
+        seen[row.path] = row
+
+
+def _check_no_duplicate_issue_titles(rows: list[ManifestRow]) -> None:
+    seen: dict[str, ManifestRow] = {}
+    for row in rows:
+        if row.issue_title in seen:
+            raise ManifestValidationError(
+                f"task {row.issue_title!r} would own two hand-authored canonical "
+                f"documents: {seen[row.issue_title].path!r} and {row.path!r}"
+            )
+        seen[row.issue_title] = row
+
+
+def _check_feature_child_limits(rows: list[ManifestRow]) -> None:
+    counts: dict[str, int] = {}
+    for row in rows:
+        counts[row.parent_feature] = counts.get(row.parent_feature, 0) + 1
+    for feature, count in counts.items():
+        if count > _MAX_CHILDREN_PER_FEATURE:
+            raise ManifestValidationError(
+                f"parent feature {feature!r} would own {count} document tasks, "
+                f"exceeding GitHub's {_MAX_CHILDREN_PER_FEATURE}-sub-issue limit"
+            )
+
+
+def build_manifest(plan: list[dict]) -> Manifest:
+    """Validate `plan` and return a deterministic Manifest sorted by document path.
+
+    Raises ManifestValidationError on any structural violation -- a missing
+    field, a path or issue_title reused across rows, or a Feature exceeding
+    the sub-issue limit -- rather than dropping or silently coercing the bad
+    row. Re-running against the same `plan` list always returns rows in the
+    same order (sorted by path) and therefore the same `to_json()` output.
+    """
+    rows = [_row_from_plan_entry(entry) for entry in plan]
+    _check_no_duplicate_paths(rows)
+    _check_no_duplicate_issue_titles(rows)
+    _check_feature_child_limits(rows)
+    return Manifest(rows=sorted(rows, key=lambda r: r.path))

launchpad/project-intelligence/corpus/tests/test_manifest.py

@@ -0,0 +1,167 @@
+"""Unit tests for the one-document one-task corpus manifest -- issue #626.
+
+Run:  python3 -m unittest discover -s launchpad/project-intelligence/corpus/tests -p "test_*.py"
+
+Every test constructs its own plan list in-memory -- there is no real
+"the corpus's actual plan" fixture to read, because curating that plan is
+explicitly out of scope for this task (see manifest.py's module docstring).
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import sys
+import unittest
+from pathlib import Path
+
+_MANIFEST_PATH = Path(__file__).resolve().parent.parent / "manifest.py"
+_spec = importlib.util.spec_from_file_location("corpus_manifest", _MANIFEST_PATH)
+manifest = importlib.util.module_from_spec(_spec)
+sys.modules["corpus_manifest"] = manifest
+_spec.loader.exec_module(manifest)
+
+
+def _entry(**overrides) -> dict:
+    base = {
+        "path": "launchpad/docs/corpus/capabilities/chat.md",
+        "filename": "chat.md",
+        "issue_title": "task: document capabilities/chat.md",
+        "parent_feature": "#608",
+        "priority": "P2",
+        "start_date": None,
+        "target_date": None,
+        "effort": "M",
+        "blockers": [],
+        "template": "capability",
+        "purpose": "Describe the chat capability's contract.",
+        "audiences": ["agent", "contributor"],
+        "source_start_points": ["desktop_feature:chat"],
+    }
+    base.update(overrides)
+    return base
+
+
+class RequiredFieldsTest(unittest.TestCase):
+    def test_every_required_field_is_carried_through(self) -> None:
+        result = manifest.build_manifest([_entry()])
+
+        row = result.rows[0].to_dict()
+        self.assertEqual(
+            set(row.keys()),
+            {
+                "path",
+                "filename",
+                "issue_title",
+                "parent_feature",
+                "priority",
+                "start_date",
+                "target_date",
+                "effort",
+                "blockers",
+                "template",
+                "purpose",
+                "audiences",
+                "source_start_points",
+            },
+        )
+
+    def test_missing_field_is_rejected_not_defaulted(self) -> None:
+        entry = _entry()
+        del entry["template"]
+
+        with self.assertRaises(manifest.ManifestValidationError) as ctx:
+            manifest.build_manifest([entry])
+        self.assertIn("template", str(ctx.exception))
+
+
+class DuplicatePathTest(unittest.TestCase):
+    def test_the_same_document_assigned_to_two_tasks_is_rejected(self) -> None:
+        first = _entry(issue_title="task: document capabilities/chat.md")
+        second = _entry(issue_title="task: document capabilities/chat.md (duplicate)")
+
+        with self.assertRaises(manifest.ManifestValidationError) as ctx:
+            manifest.build_manifest([first, second])
+        self.assertIn("assigned to two tasks", str(ctx.exception))
+
+
+class DuplicateIssueTitleTest(unittest.TestCase):
+    def test_one_task_owning_two_documents_is_rejected(self) -> None:
+        first = _entry(path="launchpad/docs/corpus/capabilities/chat.md", filename="chat.md")
+        second = _entry(path="launchpad/docs/corpus/capabilities/forum.md", filename="forum.md")
+        # Both entries share the same issue_title -- one task, two documents.
+
+        with self.assertRaises(manifest.ManifestValidationError) as ctx:
+            manifest.build_manifest([first, second])
+        self.assertIn("own two hand-authored canonical documents", str(ctx.exception))
+
+
+class FeatureChildLimitTest(unittest.TestCase):
+    def test_a_feature_at_exactly_the_limit_is_accepted(self) -> None:
+        plan = [
+            _entry(
+                path=f"launchpad/docs/corpus/capabilities/cap-{i}.md",
+                filename=f"cap-{i}.md",
+                issue_title=f"task: document capabilities/cap-{i}.md",
+            )
+            for i in range(100)
+        ]
+
+        result = manifest.build_manifest(plan)
+
+        self.assertEqual(len(result.rows), 100)
+
+    def test_a_feature_over_the_limit_is_rejected(self) -> None:
+        plan = [
+            _entry(
+                path=f"launchpad/docs/corpus/capabilities/cap-{i}.md",
+                filename=f"cap-{i}.md",
+                issue_title=f"task: document capabilities/cap-{i}.md",
+            )
+            for i in range(101)
+        ]
+
+        with self.assertRaises(manifest.ManifestValidationError) as ctx:
+            manifest.build_manifest(plan)
+        self.assertIn("101 document tasks", str(ctx.exception))
+
+    def test_different_features_are_counted_independently(self) -> None:
+        plan = [_entry(parent_feature="#608"), _entry(path="x2.md", filename="x2.md", issue_title="t2", parent_feature="#609")]
+
+        result = manifest.build_manifest(plan)
+
+        self.assertEqual(len(result.rows), 2)
+
+
+class DeterminismTest(unittest.TestCase):
+    def test_rerunning_against_the_same_plan_produces_no_diff(self) -> None:
+        plan = [
+            _entry(path="b.md", filename="b.md", issue_title="t-b"),
+            _entry(path="a.md", filename="a.md", issue_title="t-a"),
+        ]
+
+        first = manifest.build_manifest(plan).to_json()
+        second = manifest.build_manifest(plan).to_json()
+
+        self.assertEqual(first, second)
+
+    def test_rows_are_sorted_by_path_regardless_of_input_order(self) -> None:
+        plan = [
+            _entry(path="z.md", filename="z.md", issue_title="t-z"),
+            _entry(path="a.md", filename="a.md", issue_title="t-a"),
+        ]
+
+        result = manifest.build_manifest(plan)
+
+        self.assertEqual([row.path for row in result.rows], ["a.md", "z.md"])
+
+
+class EmptyPlanTest(unittest.TestCase):
+    def test_an_empty_plan_produces_an_empty_manifest_not_an_error(self) -> None:
+        result = manifest.build_manifest([])
+
+        self.assertEqual(result.rows, [])
+        self.assertEqual(result.to_json(), '{\n  "rows": []\n}\n')
+
+
+if __name__ == "__main__":
+    unittest.main()