"""Selective release planning for independently versioned packages.""" from __future__ import annotations from dataclasses import replace from datetime import UTC, datetime from pathlib import Path import shlex from .contracts import validate_contracts from .model import ( CompatibilityIssue, InterfaceProviderSnapshot, InterfaceRequirementSnapshot, ReleaseGateFinding, ReleaseDashboard, ReleasePlanStep, ReleasePlanUnit, ReleaseRecommendedAction, RepositorySnapshot, SelectiveReleasePlan, ModuleContractSnapshot, ) from .version_alignment import ( selected_release_webui_bundle_issues, selected_repository_version_issues, ) def build_selective_release_plan( dashboard: ReleaseDashboard, *, selected_repos: tuple[str, ...] = (), target_version: str | None = None, repo_versions: dict[str, str] | None = None, channel: str = "stable", ) -> SelectiveReleasePlan: repo_versions = repo_versions or {} repositories = selected_repositories(dashboard, selected_repos=selected_repos) contracts_by_repo = {contract.repo: contract for contract in dashboard.contracts} units = tuple( build_unit( repo, target_version=repo_versions.get(repo.spec.name) or target_version, contracts=contracts_by_repo.get(repo.spec.name), ) for repo in repositories ) units = apply_repository_version_gate( units, workspace=Path(dashboard.workspace_root) ) units = apply_release_webui_bundle_gate( units, workspace=Path(dashboard.workspace_root) ) compatibility = compatibility_issues(dashboard) steps = dry_run_steps(units=units, dashboard=dashboard, channel=channel) notes = release_notes(dashboard) collection_findings = dashboard_collection_findings(dashboard) findings = (*collection_findings, *(finding for unit in units for finding in unit.gate_findings)) status = plan_status(units=units, compatibility=compatibility, findings=findings) return SelectiveReleasePlan( generated_at=datetime.now(tz=UTC) .replace(microsecond=0) .isoformat() .replace("+00:00", "Z"), target_channel=channel, status=status, units=units, compatibility=compatibility, dry_run_steps=steps, notes=notes, gate_findings=findings, recommended_action=recommended_action( units=units, findings=findings, compatibility=compatibility, ), source_preflight_ready=not collection_findings and bool(units) and all(unit.source_preflight_ready for unit in units), ) def dashboard_collection_findings(dashboard: ReleaseDashboard) -> tuple[ReleaseGateFinding, ...]: return tuple( ReleaseGateFinding( code=error.code, severity="blocker", message=error.message, remediation=error.remediation, repo=error.repo, source=error.source, expected="readable release metadata", actual="unreadable", ) for error in dashboard.collection_errors ) def apply_repository_version_gate( units: tuple[ReleasePlanUnit, ...], *, workspace: Path, ) -> tuple[ReleasePlanUnit, ...]: issues_by_repo: dict[str, list[ReleaseGateFinding]] = {} for unit in units: try: issues = selected_repository_version_issues( repo_versions={unit.repo: unit.target_version}, workspace=workspace, ) except Exception as exc: # noqa: BLE001 - source errors become release findings. issues_by_repo.setdefault(unit.repo, []).append( ReleaseGateFinding( code="repository_version_metadata_unreadable", severity="blocker", message=( "Version-bearing source metadata could not be read or parsed " f"({type(exc).__name__})." ), remediation=( "Repair the malformed TOML or JSON version metadata, refresh the " "repository dashboard, then rebuild this plan." ), repo=unit.repo, source="repository version metadata", expected=unit.target_version, actual="unreadable", ) ) continue for issue in issues: issues_by_repo.setdefault(issue.repo, []).append( ReleaseGateFinding( code="repository_version_alignment", severity="blocker", message=issue.message, remediation=( f"Align {issue.source} and every other version-bearing source declaration to " f"{issue.expected!r}. Regenerate lockfiles instead of editing resolved entries by hand, " "commit the reviewed changes, then rebuild this plan." ), repo=issue.repo, source=issue.source, expected=issue.expected, actual=issue.actual, ) ) return tuple( replace( unit, status="blocked", blockers=( *unit.blockers, *( f"{finding.source}={finding.actual!r}, expected {finding.expected!r} ({finding.message})" for finding in issues_by_repo[unit.repo] ), ), gate_findings=(*unit.gate_findings, *issues_by_repo[unit.repo]), source_preflight_ready=False, ) if unit.repo in issues_by_repo else unit for unit in units ) def apply_release_webui_bundle_gate( units: tuple[ReleasePlanUnit, ...], *, workspace: Path, ) -> tuple[ReleasePlanUnit, ...]: issues_by_repo: dict[str, list[ReleaseGateFinding]] = {} try: issues = selected_release_webui_bundle_issues( repo_versions={unit.repo: unit.target_version for unit in units}, workspace=workspace, ) except Exception as exc: # noqa: BLE001 - source errors become release findings. for unit in units: issues_by_repo.setdefault(unit.repo, []).append( ReleaseGateFinding( code="release_webui_metadata_unreadable", severity="blocker", message=( "Release WebUI composition metadata could not be read or parsed " f"({type(exc).__name__})." ), remediation=( "Repair malformed JSON in the selected module WebUI package or " "Core's release package/lockfile, refresh the repository " "dashboard, then rebuild this plan." ), repo=unit.repo, source="release WebUI composition metadata", expected=unit.target_version, actual="unreadable", ) ) issues = () for issue in issues: issues_by_repo.setdefault(issue.repo, []).append( ReleaseGateFinding( code="release_webui_composition_alignment", severity="blocker", message=issue.message, remediation=( "Update Core's webui/package.release.json dependency to the selected module tag, " "regenerate webui/package-lock.release.json, commit both files, then rebuild this plan." ), repo=issue.repo, source=issue.source, expected=issue.expected, actual=issue.actual, ) ) return tuple( replace( unit, status="blocked", blockers=( *unit.blockers, *( f"{finding.source}={finding.actual!r}, expected {finding.expected!r} ({finding.message})" for finding in issues_by_repo[unit.repo] ), ), gate_findings=(*unit.gate_findings, *issues_by_repo[unit.repo]), source_preflight_ready=False, ) if unit.repo in issues_by_repo else unit for unit in units ) def selected_repositories( dashboard: ReleaseDashboard, *, selected_repos: tuple[str, ...] ) -> tuple[RepositorySnapshot, ...]: if selected_repos: wanted = set(selected_repos) return tuple( repo for repo in dashboard.repositories if repo.spec.name in wanted ) return tuple(repo for repo in dashboard.repositories if release_candidate(repo)) def release_candidate(repo: RepositorySnapshot) -> bool: return ( repo.dirty or bool(repo.ahead) or (repo.exists and repo.is_git and not repo.has_head) ) def build_unit( repo: RepositorySnapshot, *, target_version: str | None, contracts: ModuleContractSnapshot | None, ) -> ReleasePlanUnit: current_version = repo.versions.primary resolved_target = (target_version or current_version or "0.1.0").removeprefix("v") target_tag = f"v{resolved_target}" blockers: list[str] = [] warnings: list[str] = [] findings: list[ReleaseGateFinding] = [] if not repo.exists: blockers.append("repository is missing locally") findings.append( gate_finding( repo=repo.spec.name, code="repository_missing", message="Repository is missing locally.", remediation="Clone or bootstrap the registered repository, then refresh the dashboard.", ) ) elif not repo.is_git: blockers.append("path is not a Git repository") findings.append( gate_finding( repo=repo.spec.name, code="repository_not_git", message="Registered path is not a Git repository.", remediation="Restore the registered checkout as a Git repository, then refresh the dashboard.", ) ) elif repo.safe_directory_required: blockers.append("Git safe.directory approval is required") findings.append( gate_finding( repo=repo.spec.name, code="git_safe_directory", message="Git safe.directory approval is required.", remediation="Review checkout ownership and apply Git safe.directory approval only if this path is trusted.", ) ) elif repo.behind: blockers.append(f"repository is behind {repo.upstream}") findings.append( gate_finding( repo=repo.spec.name, code="repository_behind", message=f"Repository is behind {repo.upstream}.", remediation="Fetch and integrate the upstream changes, then rebuild the plan from the reviewed HEAD.", ) ) if repo.exists and repo.is_git and not repo.has_head: blockers.append("repository has no initial commit") findings.append( gate_finding( repo=repo.spec.name, code="repository_no_head", message="Repository has no initial commit.", remediation="Create and publish a reviewed initial commit before planning its first release.", ) ) local_versions = tuple( value for value in ( repo.versions.pyproject, repo.versions.package, repo.versions.webui_package, *repo.versions.manifests, *repo.versions.package_inits, ) if value ) if len(set(local_versions)) > 1: blockers.append( "backend, manifest, and frontend version metadata is not aligned" ) findings.append( gate_finding( repo=repo.spec.name, code="repository_version_alignment", message="Backend, manifest, and frontend version metadata is not aligned.", remediation=f"Align every version-bearing source declaration to {resolved_target!r}, regenerate lockfiles, and rebuild the plan.", ) ) if repo.exists and repo.is_git and repo.has_head and not repo.branch: blockers.append("repository is not on a named branch") findings.append( gate_finding( repo=repo.spec.name, code="repository_detached_head", message="Repository is not on a named branch.", remediation="Check out the intended release branch, then rebuild the plan.", ) ) if repo.dirty: warnings.append("uncommitted changes will need review before release") findings.append( gate_finding( repo=repo.spec.name, code="worktree_dirty", severity="warning", message="Uncommitted changes prevent source tagging.", remediation="Preview Prepare Changes, review the exact paths, then commit or discard them before rebuilding the plan.", ) ) if repo.ahead: warnings.append("unpushed commits should be pushed before catalog publication") findings.append( gate_finding( repo=repo.spec.name, code="repository_ahead", severity="warning", message="The branch contains unpushed commits.", remediation="Preview the source release; guarded publication can push the branch and annotated tag atomically.", ) ) if current_version is None: warnings.append( "no local version metadata found; initial release needs version metadata and catalog entry synthesis" ) findings.append( gate_finding( repo=repo.spec.name, code="version_metadata_missing", severity="warning", message="No local version metadata was discovered.", remediation=f"Add the required package and manifest version declarations for {resolved_target!r}, then rebuild the plan.", ) ) if ( current_version and current_version == resolved_target and repo.local_target_tag_exists ): warnings.append(f"local tag {target_tag} already exists") provides: tuple[InterfaceProviderSnapshot, ...] = () requires: tuple[InterfaceRequirementSnapshot, ...] = () if contracts is not None: provides = contracts.provides_interfaces requires = contracts.requires_interfaces return ReleasePlanUnit( repo=repo.spec.name, category=repo.spec.category, subtype=repo.spec.subtype, branch=repo.branch, current_version=current_version, target_version=resolved_target, target_tag=target_tag, status="blocked" if blockers else "attention" if warnings else "ready", blockers=tuple(blockers), warnings=tuple(warnings), provides_interfaces=provides, requires_interfaces=requires, gate_findings=tuple(findings), source_preflight_ready=( not blockers and not repo.dirty and repo.has_head and bool(repo.branch) and current_version == resolved_target ), ) def gate_finding( *, repo: str, code: str, message: str, remediation: str, severity: str = "blocker", ) -> ReleaseGateFinding: return ReleaseGateFinding( code=code, severity=severity, message=message, remediation=remediation, repo=repo, ) def recommended_action( *, units: tuple[ReleasePlanUnit, ...], findings: tuple[ReleaseGateFinding, ...], compatibility: tuple[CompatibilityIssue, ...], ) -> ReleaseRecommendedAction: blocker = next( (finding for finding in findings if finding.severity == "blocker"), None ) if blocker is not None: return ReleaseRecommendedAction( id="resolve_release_gate", title=f"Resolve {blocker.repo or 'release'} gate", detail=blocker.message, remediation=blocker.remediation, repo=blocker.repo, ) contract_blocker = next( (issue for issue in compatibility if issue.severity == "blocker"), None ) if contract_blocker is not None: return ReleaseRecommendedAction( id="resolve_compatibility_gate", title="Resolve interface compatibility gate", detail=contract_blocker.message, remediation="Align the selected provider and consumer interface contracts, then rebuild the plan before publication.", repo=contract_blocker.repo, ) worktree = next( (finding for finding in findings if finding.code == "worktree_dirty"), None ) if worktree is not None: return ReleaseRecommendedAction( id="prepare_changes", title=f"Prepare {worktree.repo} changes", detail=worktree.message, remediation=worktree.remediation, repo=worktree.repo, ) if not units: return ReleaseRecommendedAction( id="select_release_targets", title="Select release targets", detail="No repositories are selected for this release plan.", remediation="Select one or more repositories, confirm their target versions, then build the plan again.", ) return ReleaseRecommendedAction( id="preview_source_release", title="Preview source release tags", detail="All plan-visible source preflight gates pass for the selected repositories.", remediation="Run Preview Tag + Publish to verify manifest scope, configured remotes, and immutable local/remote tag state before mutation.", ) def compatibility_issues(dashboard: ReleaseDashboard) -> tuple[CompatibilityIssue, ...]: return validate_contracts(dashboard.contracts) def dry_run_steps( *, units: tuple[ReleasePlanUnit, ...], dashboard: ReleaseDashboard, channel: str ) -> tuple[ReleasePlanStep, ...]: steps: list[ReleasePlanStep] = [] snapshots = {repo.spec.name: repo for repo in dashboard.repositories} selected_names = {unit.repo for unit in units} if "govoplan-core" in selected_names and len(selected_names) > 1: steps.append( ReleasePlanStep( id="release:cross-repository-ordering", title="Resolve module/Core release ordering", detail=( "A combined Core/module release needs a dependency-aware DAG: " "local module tags, Core release-lock regeneration and commit, " "Core tag, full alignment, then atomic pushes. The current linear " "executor must not publish a module before that lock is committed." ), command=None, cwd=dashboard.meta_root, mutating=True, status="needs-executor", ) ) for unit in units: steps.append( ReleasePlanStep( id=f"{unit.repo}:preflight", title=f"Inspect {unit.repo}", detail="Verify local branch, dirty state, current version, and target tag before release.", command="git status --short --branch && git tag --list " + shlex.quote(unit.target_tag), cwd=f"{dashboard.workspace_root}/{unit.repo}", repo=unit.repo, ) ) if unit.current_version != unit.target_version: steps.append( ReleasePlanStep( id=f"{unit.repo}:version", title=f"Update {unit.repo} version metadata", detail=f"Prepare version files and manifest metadata for {unit.target_version}.", command=None, cwd=f"{dashboard.workspace_root}/{unit.repo}", mutating=True, repo=unit.repo, status="needs-executor", ) ) snapshot = snapshots.get(unit.repo) if unit.current_version != unit.target_version or ( snapshot is not None and snapshot.dirty ): steps.append( ReleasePlanStep( id=f"{unit.repo}:commit", title=f"Commit {unit.repo} release changes", detail="Commit reviewed changes for this release unit only.", command=f"git add -A && git commit -m {shlex.quote(f'Release {unit.repo} {unit.target_tag}')}", cwd=f"{dashboard.workspace_root}/{unit.repo}", mutating=True, repo=unit.repo, ) ) steps.append( ReleasePlanStep( id=f"{unit.repo}:tag", title=f"Create {unit.repo} tag {unit.target_tag}", detail="Create an annotated tag for this package release.", command=f"git tag -a {shlex.quote(unit.target_tag)} -m {shlex.quote(f'Release {unit.repo} {unit.target_tag}')}", cwd=f"{dashboard.workspace_root}/{unit.repo}", mutating=True, repo=unit.repo, ) ) steps.append( ReleasePlanStep( id=f"{unit.repo}:push", title=f"Push {unit.repo} release", detail="Push the branch and release tag for this package only.", command=f"git push --atomic origin HEAD:refs/heads/{shlex.quote(unit.branch or 'main')} refs/tags/{shlex.quote(unit.target_tag)}", cwd=f"{dashboard.workspace_root}/{unit.repo}", mutating=True, repo=unit.repo, ) ) if units: initial_units = tuple(unit for unit in units if unit.current_version is None) repo_version_args = " ".join( f"--repo-version {shlex.quote(unit.repo + '=' + unit.target_version)}" for unit in units ) if initial_units: steps.append( ReleasePlanStep( id="catalog:initial-version-metadata", title=f"Prepare version metadata before adding to {channel}", detail=( "Selected repositories have no source version metadata: " + ", ".join(unit.repo for unit in initial_units) + ". Initial catalog synthesis requires aligned pyproject, module manifest, and optional frontend versions." ), cwd=dashboard.meta_root, mutating=True, status="needs-version-metadata", ) ) else: steps.append( ReleasePlanStep( id="catalog:selective-generator", title=f"Update {channel} release channel catalog", detail=( "Generate a signed candidate catalog that preserves unchanged package versions " "and updates only selected release units." ), command=( 'KEY_DIR="$HOME/.config/govoplan/release-keys" ' f"tools/release/release-catalog.py selective --channel {shlex.quote(channel)} " f"{repo_version_args} " '--catalog-signing-key "release-key-1=$KEY_DIR/release-key-1.pem"' ), cwd=dashboard.meta_root, mutating=True, status="planned", ) ) steps.append( ReleasePlanStep( id="catalog:validate-sign-publish", title=f"Validate, sign, and publish {channel}", detail=( "The candidate writer validates and signs locally. Preview and guarded apply/tag/push executors " "are available in the Signed Website Catalog panel and the release-catalog CLI." ), command=( "tools/release/release-catalog.py publish-candidate " f'--candidate-dir "$CANDIDATE_DIR" --channel {shlex.quote(channel)}' ), cwd=dashboard.meta_root, mutating=True, status="planned", ) ) return tuple(steps) def release_notes(dashboard: ReleaseDashboard) -> tuple[str, ...]: notes = [ "Release units are independent; selecting one package should not force every repository to a shared tag.", "Channel publication must preserve unchanged package versions and only advance selected release units.", "Compatibility is checked from currently discovered manifest interface providers and requirements.", ] if dashboard.catalog.public_checked: notes.append( "Local catalog/keyring hashes are compared with the published channel and keyring." ) return tuple(notes) def plan_status( *, units: tuple[ReleasePlanUnit, ...], compatibility: tuple[CompatibilityIssue, ...], findings: tuple[ReleaseGateFinding, ...] = (), ) -> str: if any(finding.severity == "blocker" for finding in findings) or any(unit.blockers for unit in units) or any( issue.severity == "blocker" for issue in compatibility ): return "blocked" if units or compatibility: return "attention" return "ready" def version_satisfies(version: str, requirement: InterfaceRequirementSnapshot) -> bool: parsed = parse_version(version) if parsed is None: return False if ( requirement.version_min and (minimum := parse_version(requirement.version_min)) is not None and parsed < minimum ): return False if ( requirement.version_max_exclusive and (maximum := parse_version(requirement.version_max_exclusive)) is not None and parsed >= maximum ): return False return True def parse_version(value: str) -> tuple[int, ...] | None: parts = value.split(".") if not parts or any(not part.isdigit() for part in parts): return None return tuple(int(part) for part in parts) def format_requirement(requirement: InterfaceRequirementSnapshot) -> str: parts: list[str] = [] if requirement.version_min: parts.append(f">={requirement.version_min}") if requirement.version_max_exclusive: parts.append(f"<{requirement.version_max_exclusive}") return ", ".join(parts) if parts else "any version"