fixes on paralelism for plans

src/discussion_framework/validation.py

@@ -7,6 +7,7 @@ from .model import (
     COUNTERS,
     DiscussionError,
     DiscussionIndex,
+    WORKFLOW_KINDS,
     artifact_path,
     numeric_part,
 )
@@ -23,6 +24,7 @@ def validate(root: Path) -> list[str]:
 
     max_ids = {prefix: 0 for prefix in COUNTERS}
     seen: set[str] = set()
+    indexed_workflow_files: dict[str, set[str]] = {kind: set() for kind in WORKFLOW_KINDS}
 
     for counter in COUNTERS:
         value = next_ids.get(counter)
@@ -65,12 +67,16 @@ def validate(root: Path) -> list[str]:
                 path = artifact_path(root, kind, file_value)
                 if not path.exists() and _requires_file(discussion, artifact, kind):
                     errors.append(f"{discussion_id}: {artifact_id} file not found: {path}")
+                if kind in indexed_workflow_files:
+                    indexed_workflow_files[kind].add(file_value)
 
     for prefix, max_id in max_ids.items():
         value = next_ids.get(prefix)
         if isinstance(value, int) and value <= max_id:
             errors.append(f"meta.next_id.{prefix} must be greater than max {prefix} id {max_id}")
 
+    _record_orphan_workflow_files(root, indexed_workflow_files, errors)
+
     return errors
 
 
@@ -96,3 +102,17 @@ def _requires_file(discussion: dict, artifact: dict, kind: str) -> bool:
     if discussion.get("status") == "abandoned":
         return artifact.get("status") != "abandoned"
     return True
+
+
+def _record_orphan_workflow_files(
+    root: Path,
+    indexed_workflow_files: dict[str, set[str]],
+    errors: list[str],
+) -> None:
+    for kind in WORKFLOW_KINDS:
+        directory = root / "discussion" / "workflow" / kind
+        if not directory.exists():
+            continue
+        for path in sorted(directory.glob("*.md")):
+            if path.name not in indexed_workflow_files[kind]:
+                errors.append(f"orphan workflow artifact: {path}")

tests/test_cli.py

@@ -194,6 +194,19 @@ class CliTest(unittest.TestCase):
             for plan in plans:
                 self.assertTrue((discussion / "workflow" / "plans" / plan["file"]).exists())
 
+    def test_validate_rejects_orphan_workflow_artifact(self) -> None:
+        with TempDir() as root:
+            main(["--root", str(root), "init"])
+            self.assertEqual(main(["--root", str(root), "start", "orphan-demo", "Orphan Demo"]), 0)
+            orphan = root / "discussion" / "workflow" / "plans" / "PLN-9999-orphan.md"
+            orphan.write_text("---\nid: PLN-9999\n---\n# Orphan\n", encoding="utf-8")
+
+            stderr = io.StringIO()
+            with contextlib.redirect_stderr(stderr):
+                self.assertEqual(main(["--root", str(root), "validate"]), 2)
+
+            self.assertIn("orphan workflow artifact", stderr.getvalue())
+
     def test_housekeep_with_lesson_removes_workflow_artifacts(self) -> None:
         with TempDir() as root:
             main(["--root", str(root), "init"])