"""Governed command: create flavor-correct repository files (RMGR-WP-0004-T03).""" from __future__ import annotations import uuid from dataclasses import dataclass from pathlib import Path from typing import Any from repo_manager.gitops import GitError, commit_paths, is_git_repo from repo_manager.hub_access import apply_hub_access_block, render_hub_access_block from repo_manager.standards import FLAVOR_MARKER_PREFIX, expected_workplan_prefix from repo_manager.time import utc_today DURABLE_FLAVORS = ("experimental", "research", "tooling", "product", "business") FLAVORS = (*DURABLE_FLAVORS, "project") @dataclass class CommandResult: status: str evidence: dict[str, Any] error: dict[str, Any] | None = None correlation_id: str = "" def to_dict(self) -> dict[str, Any]: out = { "command": "repo.scaffold", "status": self.status, "correlation_id": self.correlation_id, "evidence": self.evidence, } if self.error: out["error"] = self.error return out def _write(path: Path, content: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) if not content.endswith("\n"): content += "\n" path.write_text(content) def scaffold_repository( path: Path, *, flavor: str, slug: str | None = None, domain: str = "infotech", wp_prefix: str | None = None, force: bool = False, commit: bool = True, ) -> CommandResult: cid = str(uuid.uuid4()) dest = path.expanduser().resolve() dest.mkdir(parents=True, exist_ok=True) slug = slug or dest.name flavor = flavor.strip().lower() if flavor not in FLAVORS: return CommandResult( "rejected", {"path": str(dest)}, {"message": f"unknown flavor {flavor!r}"}, cid, ) prj = flavor == "project" or slug.startswith("prj-") if prj: flavor = "project" prefix = (wp_prefix or "").strip().upper() if prefix and not prefix.endswith("-WP"): prefix = prefix.rstrip("-") + "-WP" if not prefix: derived = expected_workplan_prefix(dest, slug=slug) return CommandResult( "rejected", {"path": str(dest), "suggested_prefix": derived}, { "message": ( "project flavor requires --wp-prefix derived from the " "project identity (never PRJ-WP-)" ) }, cid, ) if prefix == FLAVOR_MARKER_PREFIX or prefix.startswith("PRJ-"): return CommandResult( "rejected", {"path": str(dest)}, {"message": "refusing flavor-derived prefix PRJ-WP-"}, cid, ) else: prefix = (wp_prefix or expected_workplan_prefix(dest, slug=slug) or "REPO-WP").upper() if not prefix.endswith("-WP"): prefix = prefix.rstrip("-") + "-WP" written: list[str] = [] def put(rel: str, content: str) -> None: target = dest / rel if target.exists() and not force: return _write(target, content) written.append(rel) put( ".repo-classification.yaml", "repo_classification:\n" f" category: {'project' if prj else flavor}\n" f" domain: {domain}\n" " secondary_domains: []\n" " capability_tags: []\n", ) put("README.md", f"# {slug}\n\nSee {'GOAL.md' if prj else 'INTENT.md'} and SCOPE.md.\n") put( "SCOPE.md", f"# Scope\n\n## In scope\n\n- {slug}\n\n## Out of scope\n\n- TBD\n", ) put( "AGENTS.md", f"# Agent instructions — {slug}\n\n" f"Orient: {'GOAL.md' if prj else 'INTENT.md'} → SCOPE.md → workplans/.\n" f"Workplan prefix: `{prefix}-`.\n\n" "## State Hub Integration\n\n" f"{render_hub_access_block()}\n\n" "## Cross-repo waits\n\n" "If blocked on another repo, look for a `prj-` driver workplan and " "follow `coordination-engine` `spec/cross-owner-wait-mode-v0.1.md`. " "Do not mint a stranded `wait`, implement `flavor: residual`, or " "message `ops-warden` for secrets.\n", ) if prj: today = utc_today().isoformat() put( "GOAL.md", "---\n" f"repo: {slug}\n" "repo_flavor: project\n" "project_status: draft\n" f'started: "{today}"\n' "---\n\n" f"# Goal — {slug}\n\n" "## Outcome\n\nWhat success looks like.\n\n" "## Invariants\n\nRules that must hold during the project.\n\n" "## Success gates\n\nCheckable conditions for goal achieved.\n\n" "## Project retirement\n\n" "Archive this repository when gates are met and residuals have live owners.\n", ) if (dest / "INTENT.md").exists() and not force: pass else: put( "INTENT.md", f"# INTENT\n\n> Why `{slug}` exists.\n\n## One-liner\n\nTBD.\n", ) (dest / "workplans" / "archived").mkdir(parents=True, exist_ok=True) if not prj: bootstrap = dest / "workplans" / f"{prefix}-0001-foundation.md" if force or not bootstrap.exists(): _write( bootstrap, "---\n" f"id: {prefix}-0001\n" "type: workplan\n" f'title: "Foundation"\n' f"domain: {domain}\n" f"repo: {slug}\n" "status: proposed\n" "flavor: planning\n" "depends_on: []\n" "---\n\n" "# Foundation\n\n" "```task\n" f"id: {prefix}-0001-T01\n" "status: todo\n" "priority: high\n" "```\n\n" "Establish the repository baseline.\n", ) written.append(str(bootstrap.relative_to(dest))) evidence: dict[str, Any] = { "path": str(dest), "flavor": flavor, "slug": slug, "wp_prefix": prefix, "written": written, "prj_layout": prj, } if commit and is_git_repo(dest) and written: try: sha = commit_paths(dest, written, f"chore: scaffold {flavor} repository baseline") evidence["git_sha"] = sha except GitError as exc: return CommandResult("failed", evidence, {"message": str(exc)}, cid) if not written: evidence["noop"] = True return CommandResult("applied", evidence, None, cid) def refresh_hub_access(path: Path, *, commit: bool = True) -> CommandResult: """Replace the State Hub access table in AGENTS.md from the canonical map.""" cid = str(uuid.uuid4()) dest = path.expanduser().resolve() agents = dest / "AGENTS.md" evidence: dict[str, Any] = {"path": str(dest), "file": "AGENTS.md"} if not agents.is_file(): return CommandResult( "rejected", evidence, {"message": "AGENTS.md is missing; scaffold the repository first"}, cid, ) original = agents.read_text(encoding="utf-8") updated, action = apply_hub_access_block(original) evidence["action"] = action if action == "noop": evidence["noop"] = True return CommandResult("applied", evidence, None, cid) agents.write_text(updated, encoding="utf-8") evidence["written"] = ["AGENTS.md"] if commit and is_git_repo(dest): try: evidence["git_sha"] = commit_paths( dest, ["AGENTS.md"], "docs(agents): refresh State Hub access map from repo-manager", ) except GitError as exc: return CommandResult("failed", evidence, {"message": str(exc)}, cid) return CommandResult("applied", evidence, None, cid)