diff --git a/src/discussion_framework/commands.py b/src/discussion_framework/commands.py index 7db8840..bb3935e 100644 --- a/src/discussion_framework/commands.py +++ b/src/discussion_framework/commands.py @@ -6,7 +6,7 @@ import json import re import shutil -from .model import ALL_ARTIFACT_KINDS, DiscussionError, DiscussionIndex, WORKFLOW_KINDS, artifact_path +from .model import ALL_ARTIFACT_KINDS, DiscussionError, DiscussionIndex, WORKFLOW_KINDS, artifact_path, index_lock from .validation import validate, validate_or_raise @@ -24,11 +24,12 @@ def init(root: Path) -> str: ): (discussion_dir / relative).mkdir(parents=True, exist_ok=True) - index_path = discussion_dir / "index.ndjson" - if index_path.exists(): - return f"already initialized: {index_path}" - index_path.write_text(INITIAL_INDEX, encoding="utf-8") - return f"initialized: {discussion_dir}" + with index_lock(root): + index_path = discussion_dir / "index.ndjson" + if index_path.exists(): + return f"already initialized: {index_path}" + index_path.write_text(INITIAL_INDEX, encoding="utf-8") + return f"initialized: {discussion_dir}" def start_agenda( @@ -47,199 +48,205 @@ def start_agenda( if not title.strip(): raise DiscussionError("title is required") - index = DiscussionIndex.load(root) - normalized_ticket = _slugify(ticket) - if not normalized_ticket: - raise DiscussionError("ticket must contain at least one alphanumeric character") - for discussion in index.discussions: - if discussion.get("ticket") == normalized_ticket: - raise DiscussionError(f"discussion already exists for ticket: {normalized_ticket}") + with index_lock(root): + index = DiscussionIndex.load(root) + normalized_ticket = _slugify(ticket) + if not normalized_ticket: + raise DiscussionError("ticket must contain at least one alphanumeric character") + for discussion in index.discussions: + if discussion.get("ticket") == normalized_ticket: + raise DiscussionError(f"discussion already exists for ticket: {normalized_ticket}") - discussion_id = index.next_id("DSC") - agenda_id = index.next_id("AGD") - slug = _slugify(title) or normalized_ticket - agenda_file = f"{agenda_id}-{slug}.md" - today = date.today().isoformat() - normalized_tags = _normalize_tags(tags) - agenda_path = artifact_path(root, "agendas", agenda_file) - agenda_path.parent.mkdir(parents=True, exist_ok=True) - agenda_path.write_text( - _render_agenda( - agenda_id=agenda_id, - ticket=normalized_ticket, - title=title.strip(), - created=today, - tags=normalized_tags, - domain_owner=domain_owner.strip(), - pain=pain.strip(), - context=context.strip(), - questions=questions, - ), - encoding="utf-8", - ) + discussion_id = index.next_id("DSC") + agenda_id = index.next_id("AGD") + slug = _slugify(title) or normalized_ticket + agenda_file = f"{agenda_id}-{slug}.md" + today = date.today().isoformat() + normalized_tags = _normalize_tags(tags) + agenda_path = artifact_path(root, "agendas", agenda_file) + agenda_path.parent.mkdir(parents=True, exist_ok=True) + agenda_path.write_text( + _render_agenda( + agenda_id=agenda_id, + ticket=normalized_ticket, + title=title.strip(), + created=today, + tags=normalized_tags, + domain_owner=domain_owner.strip(), + pain=pain.strip(), + context=context.strip(), + questions=questions, + ), + encoding="utf-8", + ) - discussion = { - "type": "discussion", - "id": discussion_id, - "status": "open", - "ticket": normalized_ticket, - "title": title.strip(), - "created_at": today, - "updated_at": today, - "tags": normalized_tags, - "agendas": [{"id": agenda_id, "file": agenda_file, "status": "open", "created_at": today, "updated_at": today}], - "decisions": [], - "plans": [], - "lessons": [], - } - index.discussions.insert(0, discussion) - index.ensure_next_id_above("DSC", discussion_id) - index.ensure_next_id_above("AGD", agenda_id) - index.save() - validate_or_raise(root) - return { - "discussion_id": discussion_id, - "agenda_id": agenda_id, - "ticket": normalized_ticket, - "title": title.strip(), - "agenda_file": str(agenda_path), - } + discussion = { + "type": "discussion", + "id": discussion_id, + "status": "open", + "ticket": normalized_ticket, + "title": title.strip(), + "created_at": today, + "updated_at": today, + "tags": normalized_tags, + "agendas": [{"id": agenda_id, "file": agenda_file, "status": "open", "created_at": today, "updated_at": today}], + "decisions": [], + "plans": [], + "lessons": [], + } + index.discussions.insert(0, discussion) + index.ensure_next_id_above("DSC", discussion_id) + index.ensure_next_id_above("AGD", agenda_id) + index.save() + validate_or_raise(root) + return { + "discussion_id": discussion_id, + "agenda_id": agenda_id, + "ticket": normalized_ticket, + "title": title.strip(), + "agenda_file": str(agenda_path), + } def accept_agenda(root: Path, discussion_id: str, agenda_id: str | None) -> dict: - index = DiscussionIndex.load(root) - discussion = index.find_discussion(discussion_id) - agenda = _select_artifact(discussion, "agendas", agenda_id) - if agenda.get("status") not in {"open", "review"}: - raise DiscussionError(f"agenda cannot be accepted from status: {agenda.get('status')}") - today = date.today().isoformat() - _set_artifact_status(root, "agendas", agenda, "accepted") - agenda["status"] = "accepted" - agenda["updated_at"] = today - discussion["updated_at"] = today - index.save() - validate_or_raise(root) - return {"discussion_id": discussion_id, "agenda_id": agenda["id"], "status": "accepted"} + with index_lock(root): + index = DiscussionIndex.load(root) + discussion = index.find_discussion(discussion_id) + agenda = _select_artifact(discussion, "agendas", agenda_id) + if agenda.get("status") not in {"open", "review"}: + raise DiscussionError(f"agenda cannot be accepted from status: {agenda.get('status')}") + today = date.today().isoformat() + _set_artifact_status(root, "agendas", agenda, "accepted") + agenda["status"] = "accepted" + agenda["updated_at"] = today + discussion["updated_at"] = today + index.save() + validate_or_raise(root) + return {"discussion_id": discussion_id, "agenda_id": agenda["id"], "status": "accepted"} def add_decision(root: Path, discussion_id: str, agenda_id: str | None, title: str) -> dict: if not title.strip(): raise DiscussionError("title is required") - index = DiscussionIndex.load(root) - discussion = index.find_discussion(discussion_id) - agenda = _select_artifact(discussion, "agendas", agenda_id) - if agenda.get("status") != "accepted": - raise DiscussionError(f"agenda must be accepted before adding a decision: {agenda.get('id')}") - decision_id = index.next_id("DEC") - slug = _slugify(title) or _slugify(discussion.get("ticket", "")) or decision_id.lower() - decision_file = f"{decision_id}-{slug}.md" - today = date.today().isoformat() - decision_path = artifact_path(root, "decisions", decision_file) - decision_path.parent.mkdir(parents=True, exist_ok=True) - decision_path.write_text( - _render_decision( - decision_id=decision_id, - ticket=discussion.get("ticket", ""), - title=title.strip(), - created=today, - tags=discussion.get("tags", []), - agenda_id=agenda["id"], - ), - encoding="utf-8", - ) - entry = { - "id": decision_id, - "file": decision_file, - "status": "open", - "created_at": today, - "updated_at": today, - "ref_agenda": agenda["id"], - } - discussion.setdefault("decisions", []).append(entry) - discussion["status"] = "in_progress" - discussion["updated_at"] = today - index.ensure_next_id_above("DEC", decision_id) - index.save() - validate_or_raise(root) - return {"discussion_id": discussion_id, "decision_id": decision_id, "decision_file": str(decision_path)} + with index_lock(root): + index = DiscussionIndex.load(root) + discussion = index.find_discussion(discussion_id) + agenda = _select_artifact(discussion, "agendas", agenda_id) + if agenda.get("status") != "accepted": + raise DiscussionError(f"agenda must be accepted before adding a decision: {agenda.get('id')}") + decision_id = index.next_id("DEC") + slug = _slugify(title) or _slugify(discussion.get("ticket", "")) or decision_id.lower() + decision_file = f"{decision_id}-{slug}.md" + today = date.today().isoformat() + decision_path = artifact_path(root, "decisions", decision_file) + decision_path.parent.mkdir(parents=True, exist_ok=True) + decision_path.write_text( + _render_decision( + decision_id=decision_id, + ticket=discussion.get("ticket", ""), + title=title.strip(), + created=today, + tags=discussion.get("tags", []), + agenda_id=agenda["id"], + ), + encoding="utf-8", + ) + entry = { + "id": decision_id, + "file": decision_file, + "status": "open", + "created_at": today, + "updated_at": today, + "ref_agenda": agenda["id"], + } + discussion.setdefault("decisions", []).append(entry) + discussion["status"] = "in_progress" + discussion["updated_at"] = today + index.ensure_next_id_above("DEC", decision_id) + index.save() + validate_or_raise(root) + return {"discussion_id": discussion_id, "decision_id": decision_id, "decision_file": str(decision_path)} def accept_decision(root: Path, discussion_id: str, decision_id: str | None) -> dict: - index = DiscussionIndex.load(root) - discussion = index.find_discussion(discussion_id) - decision = _select_artifact(discussion, "decisions", decision_id) - if decision.get("status") not in {"open", "review"}: - raise DiscussionError(f"decision cannot be accepted from status: {decision.get('status')}") - today = date.today().isoformat() - _set_artifact_status(root, "decisions", decision, "accepted") - decision["status"] = "accepted" - decision["updated_at"] = today - discussion["updated_at"] = today - index.save() - validate_or_raise(root) - return {"discussion_id": discussion_id, "decision_id": decision["id"], "status": "accepted"} + with index_lock(root): + index = DiscussionIndex.load(root) + discussion = index.find_discussion(discussion_id) + decision = _select_artifact(discussion, "decisions", decision_id) + if decision.get("status") not in {"open", "review"}: + raise DiscussionError(f"decision cannot be accepted from status: {decision.get('status')}") + today = date.today().isoformat() + _set_artifact_status(root, "decisions", decision, "accepted") + decision["status"] = "accepted" + decision["updated_at"] = today + discussion["updated_at"] = today + index.save() + validate_or_raise(root) + return {"discussion_id": discussion_id, "decision_id": decision["id"], "status": "accepted"} def add_plan(root: Path, discussion_id: str, decision_id: str | None, title: str) -> dict: if not title.strip(): raise DiscussionError("title is required") - index = DiscussionIndex.load(root) - discussion = index.find_discussion(discussion_id) - decision = _select_artifact(discussion, "decisions", decision_id) - if decision.get("status") != "accepted": - raise DiscussionError(f"decision must be accepted before adding a plan: {decision.get('id')}") - plan_id = index.next_id("PLN") - slug = _slugify(title) or _slugify(discussion.get("ticket", "")) or plan_id.lower() - plan_file = f"{plan_id}-{slug}.md" - today = date.today().isoformat() - plan_path = artifact_path(root, "plans", plan_file) - plan_path.parent.mkdir(parents=True, exist_ok=True) - plan_path.write_text( - _render_plan( - plan_id=plan_id, - ticket=discussion.get("ticket", ""), - title=title.strip(), - created=today, - tags=discussion.get("tags", []), - decision_id=decision["id"], - ), - encoding="utf-8", - ) - entry = { - "id": plan_id, - "file": plan_file, - "status": "open", - "created_at": today, - "updated_at": today, - "ref_decisions": [decision["id"]], - } - discussion.setdefault("plans", []).append(entry) - discussion["status"] = "in_progress" - discussion["updated_at"] = today - index.ensure_next_id_above("PLN", plan_id) - index.save() - validate_or_raise(root) - return {"discussion_id": discussion_id, "plan_id": plan_id, "plan_file": str(plan_path)} + with index_lock(root): + index = DiscussionIndex.load(root) + discussion = index.find_discussion(discussion_id) + decision = _select_artifact(discussion, "decisions", decision_id) + if decision.get("status") != "accepted": + raise DiscussionError(f"decision must be accepted before adding a plan: {decision.get('id')}") + plan_id = index.next_id("PLN") + slug = _slugify(title) or _slugify(discussion.get("ticket", "")) or plan_id.lower() + plan_file = f"{plan_id}-{slug}.md" + today = date.today().isoformat() + plan_path = artifact_path(root, "plans", plan_file) + plan_path.parent.mkdir(parents=True, exist_ok=True) + plan_path.write_text( + _render_plan( + plan_id=plan_id, + ticket=discussion.get("ticket", ""), + title=title.strip(), + created=today, + tags=discussion.get("tags", []), + decision_id=decision["id"], + ), + encoding="utf-8", + ) + entry = { + "id": plan_id, + "file": plan_file, + "status": "open", + "created_at": today, + "updated_at": today, + "ref_decisions": [decision["id"]], + } + discussion.setdefault("plans", []).append(entry) + discussion["status"] = "in_progress" + discussion["updated_at"] = today + index.ensure_next_id_above("PLN", plan_id) + index.save() + validate_or_raise(root) + return {"discussion_id": discussion_id, "plan_id": plan_id, "plan_file": str(plan_path)} def mark_plan(root: Path, plan_id: str, status: str) -> dict: if status not in {"open", "review", "accepted", "in_progress", "done", "abandoned"}: raise DiscussionError(f"unsupported plan status: {status}") - index = DiscussionIndex.load(root) - discussion, plan = _find_artifact(index, "plans", plan_id) - current = plan.get("status") - if not _can_transition(current, status): - raise DiscussionError(f"invalid plan transition: {current} -> {status}") - today = date.today().isoformat() - _set_artifact_status(root, "plans", plan, status) - plan["status"] = status - plan["updated_at"] = today - discussion["updated_at"] = today - if status == "in_progress": - discussion["status"] = "in_progress" - index.save() - validate_or_raise(root) - return {"discussion_id": discussion["id"], "plan_id": plan_id, "status": status} + with index_lock(root): + index = DiscussionIndex.load(root) + discussion, plan = _find_artifact(index, "plans", plan_id) + current = plan.get("status") + if not _can_transition(current, status): + raise DiscussionError(f"invalid plan transition: {current} -> {status}") + today = date.today().isoformat() + _set_artifact_status(root, "plans", plan, status) + plan["status"] = status + plan["updated_at"] = today + discussion["updated_at"] = today + if status == "in_progress": + discussion["status"] = "in_progress" + index.save() + validate_or_raise(root) + return {"discussion_id": discussion["id"], "plan_id": plan_id, "status": status} def status(root: Path) -> str: @@ -424,57 +431,58 @@ def housekeep(root: Path, discussion_id: str, lesson: str | None, no_lesson: boo if bool(lesson) == bool(no_lesson): raise DiscussionError("choose exactly one of --lesson or --no-lesson") - index = DiscussionIndex.load(root) - discussion = index.find_discussion(discussion_id) + with index_lock(root): + index = DiscussionIndex.load(root) + discussion = index.find_discussion(discussion_id) - lesson_entry = None - if lesson: - lesson_path = Path(lesson) - if not lesson_path.is_absolute(): - lesson_path = root / lesson_path - lesson_path = lesson_path.resolve() - if not lesson_path.exists(): - raise DiscussionError(f"lesson file not found: {lesson_path}") - lesson_id = _read_frontmatter_field(lesson_path, "id") - if not lesson_id: - lesson_id = lesson_path.name.split("-", 2)[0] + "-" + lesson_path.name.split("-", 2)[1] - relative = lesson_path.relative_to(root.resolve()).as_posix() - today = date.today().isoformat() - lesson_entry = { - "id": lesson_id, - "file": relative, - "status": "done", - "created_at": today, - "updated_at": today, - } - index.ensure_next_id_above("LSN", lesson_id) + lesson_entry = None + if lesson: + lesson_path = Path(lesson) + if not lesson_path.is_absolute(): + lesson_path = root / lesson_path + lesson_path = lesson_path.resolve() + if not lesson_path.exists(): + raise DiscussionError(f"lesson file not found: {lesson_path}") + lesson_id = _read_frontmatter_field(lesson_path, "id") + if not lesson_id: + lesson_id = lesson_path.name.split("-", 2)[0] + "-" + lesson_path.name.split("-", 2)[1] + relative = lesson_path.relative_to(root.resolve()).as_posix() + today = date.today().isoformat() + lesson_entry = { + "id": lesson_id, + "file": relative, + "status": "done", + "created_at": today, + "updated_at": today, + } + index.ensure_next_id_above("LSN", lesson_id) - removed: list[Path] = [] - for kind in WORKFLOW_KINDS: - for artifact in discussion.get(kind, []): - file_value = artifact.get("file") - if not file_value: - continue - path = artifact_path(root, kind, file_value) - if path.exists(): - path.unlink() - removed.append(path) - discussion[kind] = [] + removed: list[Path] = [] + for kind in WORKFLOW_KINDS: + for artifact in discussion.get(kind, []): + file_value = artifact.get("file") + if not file_value: + continue + path = artifact_path(root, kind, file_value) + if path.exists(): + path.unlink() + removed.append(path) + discussion[kind] = [] - if lesson_entry: - lessons = discussion.setdefault("lessons", []) - lessons[:] = [existing for existing in lessons if existing.get("id") != lesson_entry["id"]] - lessons.append(lesson_entry) + if lesson_entry: + lessons = discussion.setdefault("lessons", []) + lessons[:] = [existing for existing in lessons if existing.get("id") != lesson_entry["id"]] + lessons.append(lesson_entry) - discussion["status"] = "done" - discussion["updated_at"] = date.today().isoformat() - index.save() - validate_or_raise(root) + discussion["status"] = "done" + discussion["updated_at"] = date.today().isoformat() + index.save() + validate_or_raise(root) - lines = [f"housekept {discussion_id}", f"removed workflow artifacts: {len(removed)}"] - if lesson_entry: - lines.append(f"registered lesson: {lesson_entry['id']}") - return "\n".join(lines) + lines = [f"housekept {discussion_id}", f"removed workflow artifacts: {len(removed)}"] + if lesson_entry: + lines.append(f"registered lesson: {lesson_entry['id']}") + return "\n".join(lines) def backup_index(root: Path) -> str: diff --git a/src/discussion_framework/model.py b/src/discussion_framework/model.py index 2349637..9ab61fb 100644 --- a/src/discussion_framework/model.py +++ b/src/discussion_framework/model.py @@ -1,8 +1,11 @@ from __future__ import annotations +from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path +import fcntl import json +import os COUNTERS = ("DSC", "AGD", "DEC", "PLN", "LSN", "CLSN") @@ -14,6 +17,19 @@ class DiscussionError(RuntimeError): pass +@contextmanager +def index_lock(root: Path): + discussion_dir = root / "discussion" + discussion_dir.mkdir(parents=True, exist_ok=True) + lock_path = discussion_dir / ".index.lock" + with lock_path.open("a+", encoding="utf-8") as lock_file: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + + @dataclass class DiscussionIndex: root: Path @@ -55,7 +71,13 @@ class DiscussionIndex: def save(self) -> None: records = [self.meta, *self.discussions] text = "\n".join(json.dumps(record, separators=(",", ":"), ensure_ascii=False) for record in records) - self.path.write_text(text + "\n", encoding="utf-8") + tmp_path = self.path.with_name(f".{self.path.name}.{os.getpid()}.tmp") + try: + tmp_path.write_text(text + "\n", encoding="utf-8") + os.replace(tmp_path, self.path) + finally: + if tmp_path.exists(): + tmp_path.unlink() def find_discussion(self, discussion_id: str) -> dict: for discussion in self.discussions: diff --git a/tests/test_cli.py b/tests/test_cli.py index 5135750..2996976 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -3,7 +3,10 @@ from __future__ import annotations import contextlib import io import json +import os from pathlib import Path +import subprocess +import sys import unittest from discussion_framework.cli import main @@ -145,6 +148,52 @@ class CliTest(unittest.TestCase): self.assertIn("status: accepted", decision.read_text(encoding="utf-8")) self.assertIn("status: done", plan.read_text(encoding="utf-8")) + def test_parallel_add_plan_allocates_unique_ids(self) -> None: + with TempDir() as root: + main(["--root", str(root), "init"]) + self.assertEqual(main(["--root", str(root), "start", "parallel-demo", "Parallel Demo"]), 0) + self.assertEqual(main(["--root", str(root), "accept-agenda", "DSC-0001"]), 0) + self.assertEqual(main(["--root", str(root), "add-decision", "DSC-0001", "--title", "Parallel Decision"]), 0) + self.assertEqual(main(["--root", str(root), "accept-decision", "DSC-0001"]), 0) + + script = "from discussion_framework.cli import main; import sys; raise SystemExit(main(sys.argv[1:]))" + env = os.environ.copy() + src = str(Path(__file__).resolve().parents[1] / "src") + env["PYTHONPATH"] = src + os.pathsep + env["PYTHONPATH"] if env.get("PYTHONPATH") else src + processes = [ + subprocess.Popen( + [ + sys.executable, + "-c", + script, + "--root", + str(root), + "add-plan", + "DSC-0001", + "--title", + f"Parallel Plan {number}", + ], + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + for number in range(1, 4) + ] + + results = [process.communicate(timeout=10) + (process.returncode,) for process in processes] + self.assertEqual([returncode for *_streams, returncode in results], [0, 0, 0], results) + + discussion = root / "discussion" + records = [json.loads(line) for line in (discussion / "index.ndjson").read_text(encoding="utf-8").splitlines()] + meta = records[0] + plans = records[1]["plans"] + + self.assertEqual(meta["next_id"]["PLN"], 4) + self.assertEqual(sorted(plan["id"] for plan in plans), ["PLN-0001", "PLN-0002", "PLN-0003"]) + for plan in plans: + self.assertTrue((discussion / "workflow" / "plans" / plan["file"]).exists()) + def test_housekeep_with_lesson_removes_workflow_artifacts(self) -> None: with TempDir() as root: main(["--root", str(root), "init"])