feat(RMGR-WP-0008): add workplan and register receiving surfaces

src/repo_manager/commands/register.py

@@ -0,0 +1,311 @@
+"""Governed mutation spine for repository-owned registers."""
+
+from __future__ import annotations
+
+import re
+import uuid
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+from repo_manager import dual_run, idempotency
+from repo_manager.commands.workplan import CommandResult
+from repo_manager.gitops import GitError, commit_paths, head_sha, push_ff
+from repo_manager.index_store import append_event, default_index_path, save_index
+from repo_manager.observe import observe_repository
+from repo_manager.parse.register import REGISTER_SCHEMA, SUPPORTED_REGISTER_KINDS, register_path
+
+_ENTRY_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{1,127}$")
+_PROTECTED_FIELDS = frozenset({"id", "title", "status", "notes", "created", "updated"})
+_REQUIRED_DATA = {
+    "sbom-inventory": frozenset({"package_name", "ecosystem"}),
+    "repo-goals": frozenset({"description"}),
+    "upstream-contributions": frozenset({"type"}),
+    "technical-debt": frozenset(),
+    "extension-points": frozenset(),
+    "register-entries": frozenset({"register_kind"}),
+}
+
+
+def _now() -> str:
+    return datetime.now(UTC).isoformat()
+
+
+def _reject(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 _load(path: Path, kind: str) -> dict[str, Any]:
+    if not path.is_file():
+        return {"schema": REGISTER_SCHEMA, "kind": kind, "entries": []}
+    try:
+        data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
+    except yaml.YAMLError as exc:
+        raise ValueError(f"invalid register YAML: {exc}") from exc
+    if not isinstance(data, dict) or not isinstance(data.get("entries", []), list):
+        raise TypeError("register must be a mapping with an entries list")
+    if data.get("kind", kind) != kind:
+        raise ValueError(f"register kind does not match filename {kind!r}")
+    data.setdefault("schema", REGISTER_SCHEMA)
+    data.setdefault("kind", kind)
+    data.setdefault("entries", [])
+    return data
+
+
+def _save(path: Path, document: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(
+        yaml.safe_dump(document, sort_keys=False, allow_unicode=True),
+        encoding="utf-8",
+    )
+
+
+def mutate_register_entry(
+    repo_root: Path,
+    kind: str,
+    entry_id: str,
+    *,
+    operation: str,
+    title: str | None = None,
+    status: str | None = None,
+    data: dict[str, Any] | None = None,
+    note: str | None = None,
+    note_author: str | None = None,
+    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:
+    """Upsert, defer, or annotate an entry in one of the shared register kinds."""
+    command_map = {
+        "upsert": "repo.register.upsert_entry",
+        "defer": "repo.register.defer_entry",
+        "note": "repo.register.add_note",
+    }
+    correlation_id = correlation_id or str(uuid.uuid4())
+    command = command_map.get(operation, "repo.register.upsert_entry")
+    repo_root = repo_root.resolve()
+    payload = {
+        "repo_root": str(repo_root),
+        "kind": kind,
+        "entry_id": entry_id,
+        "operation": operation,
+        "title": title,
+        "status": status,
+        "data": data,
+        "note": note,
+        "note_author": note_author,
+        "expected_head_sha": expected_head_sha,
+    }
+    digest = idempotency.payload_hash(payload)
+    if idempotency_key:
+        prior = idempotency.get(idempotency_key)
+        if prior:
+            if prior.get("payload_hash") != digest:
+                return _reject(command, correlation_id, "conflict", "idempotency key reused")
+            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"),
+            )
+
+    if operation not in command_map:
+        return _reject(command, correlation_id, "validation_error", f"invalid operation {operation!r}")
+    if kind not in SUPPORTED_REGISTER_KINDS:
+        return _reject(command, correlation_id, "validation_error", f"unsupported register {kind!r}")
+    if not _ENTRY_ID_RE.fullmatch(entry_id):
+        return _reject(command, correlation_id, "validation_error", f"invalid entry id {entry_id!r}")
+    if status is not None and not status.strip():
+        return _reject(command, correlation_id, "validation_error", "status cannot be empty")
+    if data is not None and not isinstance(data, dict):
+        return _reject(command, correlation_id, "validation_error", "data must be an object")
+    protected = _PROTECTED_FIELDS.intersection(data or {})
+    if protected:
+        return _reject(
+            command,
+            correlation_id,
+            "validation_error",
+            f"data contains protected fields: {', '.join(sorted(protected))}",
+        )
+
+    current_head = head_sha(repo_root)
+    if expected_head_sha and current_head and expected_head_sha != current_head:
+        result = _reject(
+            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")
+    path = register_path(repo_root, kind)
+    try:
+        document = _load(path, kind)
+    except (TypeError, ValueError) as exc:
+        return _reject(command, correlation_id, "conflict", str(exc))
+    entries = document["entries"]
+    matches = [item for item in entries if isinstance(item, dict) and str(item.get("id")) == entry_id]
+    if len(matches) > 1:
+        return _reject(command, correlation_id, "conflict", f"duplicate entry id {entry_id!r}")
+    entry = matches[0] if matches else None
+    before = yaml.safe_dump(entry, sort_keys=True) if entry is not None else None
+    now = _now()
+
+    if operation == "upsert":
+        if entry is None:
+            if not title or not title.strip():
+                return _reject(command, correlation_id, "validation_error", "new entry requires title")
+            missing = sorted(key for key in _REQUIRED_DATA[kind] if key not in (data or {}))
+            if missing:
+                return _reject(
+                    command,
+                    correlation_id,
+                    "validation_error",
+                    f"{kind} entry missing data fields: {', '.join(missing)}",
+                )
+            entry = {
+                "id": entry_id,
+                "title": title.strip(),
+                "status": (status or "open").strip(),
+                **(data or {}),
+                "created": now,
+                "updated": now,
+            }
+            entries.append(entry)
+        else:
+            if title is not None:
+                if not title.strip():
+                    return _reject(command, correlation_id, "validation_error", "title cannot be empty")
+                entry["title"] = title.strip()
+            if status is not None:
+                entry["status"] = status.strip()
+            entry.update(data or {})
+            entry["updated"] = now
+    elif operation == "defer":
+        if entry is None:
+            return _reject(command, correlation_id, "not_found", f"entry {entry_id!r} not found")
+        entry["status"] = (status or "deferred").strip()
+        entry["updated"] = now
+    else:
+        if entry is None:
+            return _reject(command, correlation_id, "not_found", f"entry {entry_id!r} not found")
+        if not note or not note.strip():
+            return _reject(command, correlation_id, "validation_error", "note requires content")
+        entry.setdefault("notes", []).append(
+            {"content": note.strip(), "author": note_author or "codex", "created": now}
+        )
+        entry["updated"] = now
+
+    after = yaml.safe_dump(entry, sort_keys=True)
+    if before == after:
+        result = CommandResult(
+            command=command,
+            status="applied",
+            correlation_id=correlation_id,
+            evidence={"status": "applied", "git_sha": current_head, "files_touched": [], "noop": True},
+        )
+        if idempotency_key:
+            idempotency.put(idempotency_key, digest, result.to_dict())
+        return result
+
+    _save(path, document)
+    relative_path = str(path.relative_to(repo_root))
+    git_sha: str | None = None
+    push_ok: bool | None = None
+    push_message: str | None = None
+    if commit:
+        try:
+            git_sha = commit_paths(
+                repo_root,
+                [relative_path],
+                message=(
+                    f"{command} {kind}/{entry_id}\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": [relative_path], "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)
+    append_event(
+        index_after,
+        {
+            "type": "repo.command.applied",
+            "command": command,
+            "operation": operation,
+            "correlation_id": correlation_id,
+            "register_kind": kind,
+            "entry_id": entry_id,
+            "git_sha": git_sha,
+            "files_touched": [relative_path],
+            "source": "repo-manager",
+        },
+    )
+    append_event(
+        index_after,
+        {
+            "type": "repo.work.indexed",
+            "correlation_id": correlation_id,
+            "kind": f"register:{kind}",
+            "id": entry_id,
+            "source_path": relative_path,
+        },
+    )
+    save_index(index_after)
+    dual_run.record_mutation(
+        source="repo-manager",
+        kind=f"register_{operation}",
+        repo_slug=slug,
+        detail={
+            "register_kind": kind,
+            "entry_id": entry_id,
+            "git_sha": git_sha,
+            "correlation_id": correlation_id,
+        },
+    )
+    evidence: dict[str, Any] = {
+        "status": "applied",
+        "git_sha": git_sha,
+        "files_touched": [relative_path],
+        "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, digest, result.to_dict())
+    return result

src/repo_manager/commands/workplan.py

@@ -0,0 +1,498 @@
+"""Governed workplan create, update, and archive commands."""
+
+from __future__ import annotations
+
+import json
+import re
+import uuid
+from dataclasses import dataclass
+from datetime import UTC, date, datetime
+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.index_store import append_event, default_index_path, save_index
+from repo_manager.observe import observe_repository
+
+VALID_WORKPLAN_STATUSES = frozenset(
+    {"proposed", "ready", "active", "blocked", "backlog", "finished", "archived"}
+)
+_WORKPLAN_ID_RE = re.compile(r"^[A-Z][A-Z0-9]*-WP-[0-9]{4,}$")
+_FRONTMATTER_RE = re.compile(r"\A---\r?\n(?P<meta>.*?)\r?\n---(?P<body>\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 datetime.now(UTC).date()
+
+
+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()
+    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"
+        "---\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 not _WORKPLAN_ID_RE.fullmatch(workplan_id):
+            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}",
+            )
+        name = filename or f"{workplan_id}-{_slugify(title)}.md"
+        if Path(name).name != name or not name.endswith(".md") or not name.startswith(f"{workplan_id}-"):
+            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)