"""Per-episode board-review comment store for the workspace.""" from __future__ import annotations import json import re from copy import deepcopy from datetime import datetime, timezone from pathlib import Path from uuid import uuid4 from recoil.core.atomic_write import atomic_write_json from recoil.core.paths import ProjectPaths from recoil.workspace.board import normalize_episode SCHEMA_VERSION = 1 VALID_TARGET_TYPES = frozenset({"board", "panel"}) VALID_TAGS = frozenset({"expand", "compress", "missing_panel", "flow", "note"}) _COMMENT_ID_RE = re.compile(r"^cmt_[0-9a-f]{32}$") _COMMENT_REQUIRED_KEYS = frozenset( { "id", "target_type", "batch_id", "segment_id", "body", "tag", "author", "created_at", "resolved", "resolved_at", } ) class BoardCommentsCorruptError(RuntimeError): """Raised when an existing board_comments.json cannot be trusted.""" def _comments_path(project: str, episode_id: str) -> Path: ids = normalize_episode(episode_id) return ( ProjectPaths.for_project(project).project_root / "prep" / ids.prep_token / "storyboards" / "board_comments.json" ) def load_comments(project: str, episode_id: str) -> dict: ids = normalize_episode(episode_id) path = _comments_path(project, episode_id) if not path.is_file(): return {"schema_version": SCHEMA_VERSION, "episode_id": ids.coverage_id, "comments": []} try: payload = json.loads(path.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError) as exc: raise BoardCommentsCorruptError(f"corrupt board comments file: {path}") from exc _validate_payload(payload, expected_episode_id=ids.coverage_id, path=path) return payload def add_comment( project: str, episode_id: str, *, target_type: str, batch_id: str, segment_id: str | None = None, body: str, tag: str, author: str = "JT", ) -> dict: _validate_new_comment( target_type=target_type, batch_id=batch_id, segment_id=segment_id, body=body, tag=tag, ) if not isinstance(author, str): raise ValueError("author must be a string") payload = load_comments(project, episode_id) comment = { "id": f"cmt_{uuid4().hex}", "target_type": target_type, "batch_id": batch_id, "segment_id": segment_id if target_type == "panel" else None, "body": body, "tag": tag, "author": author, "created_at": _utc_iso(), "resolved": False, "resolved_at": None, } payload["comments"].append(comment) _write_comments(project, episode_id, payload) return deepcopy(comment) def resolve_comment( project: str, episode_id: str, comment_id: str, *, resolved: bool = True, ) -> dict | None: if not isinstance(resolved, bool): raise ValueError("resolved must be a bool") payload = load_comments(project, episode_id) for comment in payload["comments"]: if comment["id"] != comment_id: continue comment["resolved"] = resolved comment["resolved_at"] = _utc_iso() if resolved else None _write_comments(project, episode_id, payload) return deepcopy(comment) return None def delete_comment( project: str, episode_id: str, comment_id: str, ) -> dict | None: payload = load_comments(project, episode_id) for index, comment in enumerate(payload["comments"]): if comment["id"] != comment_id: continue removed = payload["comments"].pop(index) _write_comments(project, episode_id, payload) return deepcopy(removed) return None def _validate_new_comment( *, target_type: str, batch_id: str, segment_id: str | None, body: str, tag: str, ) -> None: if target_type not in VALID_TARGET_TYPES: raise ValueError("target_type must be 'board' or 'panel'") if tag not in VALID_TAGS: raise ValueError("invalid comment tag") if tag == "missing_panel" and target_type == "panel": raise ValueError("missing_panel is a board-level tag") if not isinstance(batch_id, str) or not batch_id: raise ValueError("batch_id must be a non-empty string") if not isinstance(body, str) or not body: raise ValueError("body must be a non-empty string") if target_type == "panel": if not isinstance(segment_id, str) or not segment_id: raise ValueError("segment_id is required for panel comments") elif segment_id is not None: raise ValueError("segment_id must be None for board comments") def _validate_payload(payload: object, *, expected_episode_id: str, path: Path) -> None: if not isinstance(payload, dict): raise BoardCommentsCorruptError(f"{path}: top-level object must be a dict") schema_version = payload.get("schema_version") if ( not isinstance(schema_version, int) or isinstance(schema_version, bool) or schema_version != SCHEMA_VERSION ): raise BoardCommentsCorruptError(f"{path}: unsupported or missing schema_version") episode_id = payload.get("episode_id") if not isinstance(episode_id, str) or episode_id != expected_episode_id: raise BoardCommentsCorruptError(f"{path}: episode_id does not match {expected_episode_id}") comments = payload.get("comments") if not isinstance(comments, list): raise BoardCommentsCorruptError(f"{path}: comments must be a list") for index, comment in enumerate(comments): _validate_comment_record(comment, path=path, index=index) def _validate_comment_record(comment: object, *, path: Path, index: int) -> None: prefix = f"{path}: comments[{index}]" if not isinstance(comment, dict): raise BoardCommentsCorruptError(f"{prefix} must be a dict") missing = _COMMENT_REQUIRED_KEYS - comment.keys() if missing: raise BoardCommentsCorruptError(f"{prefix} missing required fields: {sorted(missing)}") comment_id = comment.get("id") if not isinstance(comment_id, str) or not _COMMENT_ID_RE.match(comment_id): raise BoardCommentsCorruptError(f"{prefix}.id is malformed") target_type = comment.get("target_type") if target_type not in VALID_TARGET_TYPES: raise BoardCommentsCorruptError(f"{prefix}.target_type is malformed") batch_id = comment.get("batch_id") if not isinstance(batch_id, str) or not batch_id: raise BoardCommentsCorruptError(f"{prefix}.batch_id is malformed") body = comment.get("body") if not isinstance(body, str): raise BoardCommentsCorruptError(f"{prefix}.body is malformed") tag = comment.get("tag") if tag not in VALID_TAGS: raise BoardCommentsCorruptError(f"{prefix}.tag is malformed") if tag == "missing_panel" and target_type == "panel": raise BoardCommentsCorruptError(f"{prefix}.tag is invalid for panel comments") author = comment.get("author") if not isinstance(author, str): raise BoardCommentsCorruptError(f"{prefix}.author is malformed") created_at = comment.get("created_at") if not isinstance(created_at, str): raise BoardCommentsCorruptError(f"{prefix}.created_at is malformed") resolved = comment.get("resolved") if not isinstance(resolved, bool): raise BoardCommentsCorruptError(f"{prefix}.resolved is malformed") resolved_at = comment.get("resolved_at") if resolved: if not isinstance(resolved_at, str): raise BoardCommentsCorruptError(f"{prefix}.resolved_at is malformed") elif resolved_at is not None: raise BoardCommentsCorruptError(f"{prefix}.resolved_at must be None") segment_id = comment.get("segment_id") if target_type == "panel": if not isinstance(segment_id, str) or not segment_id: raise BoardCommentsCorruptError(f"{prefix}.segment_id is malformed") elif segment_id is not None: raise BoardCommentsCorruptError(f"{prefix}.segment_id must be None") def _write_comments(project: str, episode_id: str, payload: dict) -> None: path = _comments_path(project, episode_id) path.parent.mkdir(parents=True, exist_ok=True) atomic_write_json(path, payload) def _utc_iso() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")