"""Governed workplan create, update, and archive commands.""" from __future__ import annotations import json import re import uuid from dataclasses import dataclass from datetime import date from pathlib import Path from typing import Any from repo_manager import dual_run, idempotency from repo_manager.gitops import GitError, commit_paths, head_sha, push_ff from repo_manager.identifiers import derive_work_record_uuid, load_fleet_namespace from repo_manager.index_store import append_event, default_index_path, save_index from repo_manager.observe import observe_repository from repo_manager.record_identity import classify_record_id from repo_manager.time import utc_today VALID_WORKPLAN_STATUSES = frozenset( {"proposed", "ready", "active", "blocked", "backlog", "finished", "archived"} ) _FRONTMATTER_RE = re.compile(r"\A---\r?\n(?P.*?)\r?\n---(?P\r?\n.*)?\Z", re.DOTALL) @dataclass class CommandResult: command: str status: str # applied | rejected | failed evidence: dict[str, Any] error: dict[str, Any] | None = None correlation_id: str = "" def to_dict(self) -> dict[str, Any]: out: dict[str, Any] = { "command": self.command, "status": self.status, "correlation_id": self.correlation_id, "evidence": self.evidence, } if self.error: out["error"] = self.error return out def _result_from_prior(command: str, prior: dict[str, Any], correlation_id: str) -> CommandResult: previous = prior.get("result") or {} return CommandResult( command=previous.get("command", command), status=previous.get("status", "applied"), correlation_id=previous.get("correlation_id", correlation_id), evidence=previous.get("evidence") or {"status": "applied", "replay": True}, error=previous.get("error"), ) def _rejected(command: str, correlation_id: str, code: str, message: str, **evidence: Any) -> CommandResult: return CommandResult( command=command, status="rejected", correlation_id=correlation_id, evidence={"status": "rejected", **evidence}, error={"code": code, "message": message, "correlation_id": correlation_id}, ) def _is_uuid(value: str) -> bool: try: uuid.UUID(value) return True except ValueError: return False def _slugify(value: str) -> str: return re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") or "workplan" def _quoted(value: str) -> str: return json.dumps(value, ensure_ascii=False) def _today() -> date: return utc_today() def _patch_frontmatter(text: str, updates: dict[str, str]) -> str | None: match = _FRONTMATTER_RE.match(text) if not match: return None meta = match.group("meta") for key, value in updates.items(): rendered = value if key == "status" else _quoted(value) pattern = re.compile(rf"^({re.escape(key)}:\s*).*$", re.MULTILINE) if pattern.search(meta): meta = pattern.sub(rf"\g<1>{rendered}", meta, count=1) else: meta = f"{meta.rstrip()}\n{key}: {rendered}" return f"---\n{meta}\n---{match.group('body') or ''}" def _find_workplan(index: Any, workplan_id: str) -> Any | None: by_uuid = _is_uuid(workplan_id) for record in index.work_records: if record.kind != "workplan": continue if not by_uuid and record.id == workplan_id: return record if by_uuid and record.uuid and record.uuid.lower() == workplan_id.lower(): return record return None def _create_text( *, workplan_id: str, title: str, goal: str, status: str, domain: str, repo_slug: str, owner: str, topic_slug: str, ) -> str: today = _today().isoformat() workplan_uuid = derive_work_record_uuid(load_fleet_namespace(), workplan_id) return ( "---\n" f"id: {workplan_id}\n" "type: workplan\n" f"title: {_quoted(title)}\n" f"domain: {domain}\n" f"repo: {repo_slug}\n" f"status: {status}\n" f"owner: {owner}\n" f"topic_slug: {topic_slug}\n" f"created: {_quoted(today)}\n" f"updated: {_quoted(today)}\n" f"state_hub_workstream_id: {_quoted(str(workplan_uuid))}\n" "---\n\n" f"# {title}\n\n" "## Goal\n\n" f"{goal.strip()}\n" ) def mutate_workplan( repo_root: Path, workplan_id: str, *, operation: str, title: str | None = None, goal: str | None = None, status: str | None = None, owner: str | None = None, topic_slug: str | None = None, domain: str | None = None, filename: str | None = None, confirm_archive: bool = False, correlation_id: str | None = None, reason: str = "rmgr command", commit: bool = True, push: bool = False, expected_head_sha: str | None = None, idempotency_key: str | None = None, repo_slug: str | None = None, ) -> CommandResult: """Create, update, or archive a file-backed workplan. Update/archive accepts a canonical workplan id or State Hub UUID. ``archive`` is the recoverable implementation of the legacy delete operation. """ commands = { "create": "repo.work.create_workplan", "update": "repo.work.update_workplan", "archive": "repo.work.archive_workplan", } correlation_id = correlation_id or str(uuid.uuid4()) command = commands.get(operation, "repo.work.update_workplan") repo_root = repo_root.resolve() payload = { "repo_root": str(repo_root), "workplan_id": workplan_id, "operation": operation, "title": title, "goal": goal, "status": status, "owner": owner, "topic_slug": topic_slug, "domain": domain, "filename": filename, "confirm_archive": confirm_archive, "expected_head_sha": expected_head_sha, } payload_digest = idempotency.payload_hash(payload) if idempotency_key: prior = idempotency.get(idempotency_key) if prior: if prior.get("payload_hash") != payload_digest: return _rejected( command, correlation_id, "conflict", "idempotency key reused with different payload", ) return _result_from_prior(command, prior, correlation_id) if operation not in commands: return _rejected(command, correlation_id, "validation_error", f"invalid operation {operation!r}") if status is not None and status not in VALID_WORKPLAN_STATUSES: return _rejected( command, correlation_id, "validation_error", f"invalid workplan status {status!r}", ) if operation == "archive" and not confirm_archive: return _rejected( command, correlation_id, "confirmation_required", "archive requires confirm_archive=True", ) current_head = head_sha(repo_root) if expected_head_sha and current_head and expected_head_sha != current_head: result = _rejected( command, correlation_id, "precondition_failed", f"expected_head_sha {expected_head_sha} != {current_head}", head_sha=current_head, ) if result.error: result.error["retryable"] = True return result snapshot, index = observe_repository(repo_root, slug=repo_slug) slug = repo_slug or snapshot.get("slug") record = _find_workplan(index, workplan_id) touched: list[str] changes: dict[str, Any] if operation == "create": if classify_record_id("workplan", workplan_id) != "canonical": return _rejected( command, correlation_id, "validation_error", f"invalid canonical workplan id {workplan_id!r}", ) if record is not None: return _rejected(command, correlation_id, "conflict", f"workplan {workplan_id!r} exists") if not title or not title.strip() or not goal or not goal.strip(): return _rejected( command, correlation_id, "validation_error", "create requires non-empty title and goal", ) create_status = status or "proposed" if create_status not in VALID_WORKPLAN_STATUSES: return _rejected( command, correlation_id, "validation_error", f"invalid workplan status {create_status!r}", ) is_adhoc = "-WP-ADHOC-" in workplan_id adhoc_date = workplan_id.rsplit("-WP-ADHOC-", 1)[-1] name = filename or ( f"ADHOC-{adhoc_date}.md" if is_adhoc else f"{workplan_id}-{_slugify(title)}.md" ) expected_adhoc_name = f"ADHOC-{adhoc_date}.md" safe_name = ( Path(name).name == name and name.endswith(".md") and ( name.startswith(f"{workplan_id}-") or (is_adhoc and name == expected_adhoc_name) ) ) if not safe_name: return _rejected(command, correlation_id, "validation_error", "unsafe workplan filename") path = repo_root / "workplans" / name if path.exists(): return _rejected(command, correlation_id, "conflict", f"source file exists: workplans/{name}") classification = snapshot.get("classification") or {} create_domain = domain or classification.get("domain") or "infotech" path.parent.mkdir(parents=True, exist_ok=True) path.write_text( _create_text( workplan_id=workplan_id, title=title.strip(), goal=goal, status=create_status, domain=create_domain, repo_slug=slug, owner=owner or "codex", topic_slug=topic_slug or create_domain, ), encoding="utf-8", ) touched = [str(path.relative_to(repo_root))] changes = {"title": title.strip(), "status": create_status} else: if record is None: return _rejected( command, correlation_id, "not_found", f"workplan {workplan_id!r} not found in workplans", ) path = repo_root / record.source_path if not path.is_file(): return CommandResult( command=command, status="failed", correlation_id=correlation_id, evidence={"status": "failed"}, error={ "code": "not_found", "message": f"source file missing: {record.source_path}", "correlation_id": correlation_id, }, ) if operation == "update": updates = { key: value.strip() for key, value in { "title": title, "status": status, "owner": owner, "topic_slug": topic_slug, "domain": domain, }.items() if value is not None and value.strip() } if not updates: return _rejected( command, correlation_id, "validation_error", "update requires at least one field", ) changes = { key: value for key, value in updates.items() if key not in {"status", "title"} or (key == "status" and value != record.status) or (key == "title" and value != record.title) } if not changes: result = CommandResult( command=command, status="applied", correlation_id=correlation_id, evidence={ "status": "applied", "git_sha": current_head, "files_touched": [], "noop": True, "observed_at": index.observed_at, }, ) if idempotency_key: idempotency.put(idempotency_key, payload_digest, result.to_dict()) return result changes["updated"] = _today().isoformat() new_text = _patch_frontmatter(path.read_text(encoding="utf-8"), changes) if new_text is None: return _rejected(command, correlation_id, "conflict", "workplan frontmatter is invalid") path.write_text(new_text, encoding="utf-8") touched = [record.source_path] else: if record.status == "archived" and record.source_path.startswith("workplans/archived/"): result = CommandResult( command=command, status="applied", correlation_id=correlation_id, evidence={ "status": "applied", "git_sha": current_head, "files_touched": [], "noop": True, "observed_at": index.observed_at, }, ) if idempotency_key: idempotency.put(idempotency_key, payload_digest, result.to_dict()) return result archived_dir = repo_root / "workplans" / "archived" archived_dir.mkdir(parents=True, exist_ok=True) archived_path = archived_dir / f"{_today():%y%m%d}-{path.name}" if archived_path.exists(): return _rejected( command, correlation_id, "conflict", f"archive target exists: {archived_path.relative_to(repo_root)}", ) new_text = _patch_frontmatter( path.read_text(encoding="utf-8"), {"status": "archived", "updated": _today().isoformat()}, ) if new_text is None: return _rejected(command, correlation_id, "conflict", "workplan frontmatter is invalid") archived_path.write_text(new_text, encoding="utf-8") path.unlink() touched = [record.source_path, str(archived_path.relative_to(repo_root))] changes = {"status": "archived", "source_path": touched[-1]} git_sha: str | None = None push_ok: bool | None = None push_message: str | None = None if commit: try: git_sha = commit_paths( repo_root, touched, message=( f"{command} {workplan_id} ({operation})\n\n" f"correlation_id: {correlation_id}\nreason: {reason}\nsource: repo-manager\n" ), ) except GitError as exc: return CommandResult( command=command, status="failed", correlation_id=correlation_id, evidence={"status": "failed", "files_touched": touched, "git_sha": None}, error={"code": "internal", "message": str(exc), "correlation_id": correlation_id}, ) if push: push_ok, push_message = push_ff(repo_root) _snapshot_after, index_after = observe_repository(repo_root, slug=slug) event = { "type": "repo.command.applied", "command": command, "operation": operation, "correlation_id": correlation_id, "workplan_id": record.id if record else workplan_id, "workplan_uuid": record.uuid if record else None, "changes": changes, "git_sha": git_sha, "files_touched": touched, "expected_head_sha_before": current_head, "source": "repo-manager", } append_event(index_after, event) append_event( index_after, { "type": "repo.work.indexed", "correlation_id": correlation_id, "kind": "workplan", "id": record.id if record else workplan_id, "source_path": touched[-1], }, ) save_index(index_after) dual_run.record_mutation( source="repo-manager", kind=f"workplan_{operation}", repo_slug=slug, detail={ "workplan_id": record.id if record else workplan_id, "changes": changes, "git_sha": git_sha, "correlation_id": correlation_id, }, ) evidence: dict[str, Any] = { "status": "applied", "git_sha": git_sha, "files_touched": touched, "observed_at": index_after.observed_at, "index_path": str(default_index_path(repo_root)), "source": "repo-manager", } if push: evidence["push_ok"] = push_ok evidence["push_message"] = push_message result = CommandResult( command=command, status="applied", correlation_id=correlation_id, evidence=evidence, ) if idempotency_key: idempotency.put(idempotency_key, payload_digest, result.to_dict()) return result def create_workplan(repo_root: Path, workplan_id: str, title: str, goal: str, **kwargs: Any) -> CommandResult: return mutate_workplan( repo_root, workplan_id, operation="create", title=title, goal=goal, **kwargs, ) def update_workplan(repo_root: Path, workplan_id: str, **kwargs: Any) -> CommandResult: return mutate_workplan(repo_root, workplan_id, operation="update", **kwargs) def archive_workplan(repo_root: Path, workplan_id: str, **kwargs: Any) -> CommandResult: return mutate_workplan(repo_root, workplan_id, operation="archive", **kwargs)