GovOPlaN/govoplan
1
0
Fork 0
Code Issues 10 Pull Requests Actions 3 Packages Projects Releases Wiki Activity

fix(release): fail closed on dashboard parse errors

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-22 17:07:30 +02:00
parent b6bef410d6
commit 753dd2a3ee
9 changed files with 306 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
tools/release/govoplan_release/contracts.py
View File

@@ -13,6 +13,7 @@ from pathlib import Path
from .model import (
CompatibilityIssue,
DashboardCollectionError,
InterfaceProviderSnapshot,
InterfaceRequirementSnapshot,
ModuleContractSnapshot,
@@ -33,6 +34,43 @@ def collect_contracts(repositories: tuple[RepositorySnapshot, ...]) -> tuple[Mod
return tuple(contracts)
def collect_contracts_with_diagnostics(
repositories: tuple[RepositorySnapshot, ...],
) -> tuple[tuple[ModuleContractSnapshot, ...], tuple[DashboardCollectionError, ...]]:
"""Collect readable contracts and retain bounded parse diagnostics for the UI."""
contracts: list[ModuleContractSnapshot] = []
diagnostics: list[DashboardCollectionError] = []
for repo in repositories:
if not repo.exists or not repo.is_git:
continue
root = Path(repo.absolute_path)
for manifest in sorted(root.glob("src/**/backend/manifest.py")):
try:
contract = parse_manifest_contract(manifest, repo_name=repo.spec.name)
except Exception as exc: # noqa: BLE001 - fail-closed UI diagnostic.
relative_source = manifest.relative_to(root).as_posix()
diagnostics.append(
DashboardCollectionError(
code="module_contract_unreadable",
message=(
"Module contract metadata could not be read or parsed "
f"({type(exc).__name__})."
),
remediation=(
"Repair the manifest's Python syntax or read permissions, "
"then refresh the dashboard before planning a release."
),
repo=repo.spec.name,
source=f"{repo.spec.name}/{relative_source}",
)
)
continue
if contract is not None:
contracts.append(contract)
return tuple(contracts), tuple(diagnostics)
def validate_contracts(contracts: tuple[ModuleContractSnapshot, ...]) -> tuple[CompatibilityIssue, ...]:
"""Validate the statically collected module interface graph.

41
tools/release/govoplan_release/dashboard.py
View File

@@ -6,10 +6,10 @@ from datetime import UTC, datetime
from pathlib import Path
from .catalog import collect_catalog_snapshot
from .contracts import collect_contracts
from .contracts import collect_contracts_with_diagnostics
from .git_state import collect_repository_snapshot, read_pyproject_version
from .migrations import collect_migration_snapshots
from .model import DashboardSummary, ReleaseDashboard, RepositorySnapshot
from .model import DashboardCollectionError, DashboardSummary, ReleaseDashboard, RepositorySnapshot
from .workspace import META_ROOT, core_root, load_repository_specs, resolve_workspace_root
@@ -26,10 +26,25 @@ def build_dashboard(
channel: str = "stable",
) -> ReleaseDashboard:
resolved_workspace = resolve_workspace_root(workspace_root)
collection_errors: list[DashboardCollectionError] = []
if check_public_catalog is None:
check_public_catalog = online
if target_version is None:
target_version = read_pyproject_version(core_root(resolved_workspace))
try:
target_version = read_pyproject_version(core_root(resolved_workspace))
except Exception as exc: # noqa: BLE001 - bounded fail-closed diagnostic.
collection_errors.append(
DashboardCollectionError(
code="core_version_metadata_unreadable",
message=f"Core version metadata could not be read or parsed ({type(exc).__name__}).",
remediation=(
"Repair govoplan-core/pyproject.toml, refresh the dashboard, "
"then rebuild the release plan."
),
repo="govoplan-core",
source="govoplan-core/pyproject.toml",
)
)
target_tag = f"v{target_version}" if target_version else None
repositories = tuple(
@@ -38,7 +53,11 @@ def build_dashboard(
)
catalog = collect_catalog_snapshot(workspace_root=resolved_workspace, channel=channel, check_public=check_public_catalog)
migrations = collect_migration_snapshots() if include_migrations else ()
contracts = collect_contracts(repositories) if include_contracts else ()
if include_contracts:
contracts, contract_errors = collect_contracts_with_diagnostics(repositories)
collection_errors.extend(contract_errors)
else:
contracts = ()
return ReleaseDashboard(
generated_at=datetime.now(tz=UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z"),
meta_root=str(META_ROOT),
@@ -47,28 +66,34 @@ def build_dashboard(
target_tag=target_tag,
online=online or check_remote_tags or check_public_catalog,
include_migrations=include_migrations,
summary=summarize(repositories, target_tag=target_tag),
summary=summarize(repositories, target_tag=target_tag, collection_errors=tuple(collection_errors)),
repositories=repositories,
catalog=catalog,
migrations=migrations,
contracts=contracts,
collection_errors=tuple(collection_errors),
)
def summarize(repositories: tuple[RepositorySnapshot, ...], *, target_tag: str | None) -> DashboardSummary:
def summarize(
repositories: tuple[RepositorySnapshot, ...],
*,
target_tag: str | None,
collection_errors: tuple[DashboardCollectionError, ...] = (),
) -> DashboardSummary:
missing_count = sum(1 for repo in repositories if not repo.exists or not repo.is_git)
dirty_count = sum(1 for repo in repositories if repo.dirty)
ahead_count = sum(1 for repo in repositories if repo.ahead)
behind_count = sum(1 for repo in repositories if repo.behind)
no_head_count = sum(1 for repo in repositories if repo.exists and repo.is_git and not repo.has_head)
error_count = sum(1 for repo in repositories if repo.errors)
error_count = sum(1 for repo in repositories if repo.errors) + len(collection_errors)
safe_directory_count = sum(1 for repo in repositories if repo.safe_directory_required)
local_target_tag_missing_count = (
sum(1 for repo in repositories if repo.local_target_tag_exists is False and repo.versions.primary == target_tag.removeprefix("v"))
if target_tag
else 0
)
if missing_count or behind_count or safe_directory_count:
if missing_count or behind_count or safe_directory_count or collection_errors:
status = "blocked"
elif dirty_count or ahead_count or no_head_count or error_count or local_target_tag_missing_count:
status = "attention"

2
tools/release/govoplan_release/git_state.py
View File

@@ -70,7 +70,7 @@ def collect_repository_snapshot(
try:
versions = collect_versions(path)
except Exception as exc: # noqa: BLE001 - surfaced in the dashboard.
errors.append(str(exc))
errors.append(f"version metadata could not be read or parsed ({type(exc).__name__})")
versions = VersionSnapshot()
return RepositorySnapshot(

12
tools/release/govoplan_release/model.py
View File

@@ -111,6 +111,17 @@ class MigrationTrackSnapshot:
error: str | None = None
@dataclass(frozen=True, slots=True)
class DashboardCollectionError:
"""Bounded, actionable failure from read-only dashboard collection."""
code: str
message: str
remediation: str
repo: str | None = None
source: str | None = None
@dataclass(frozen=True, slots=True)
class DashboardSummary:
repository_count: int
@@ -139,6 +150,7 @@ class ReleaseDashboard:
catalog: CatalogSnapshot
migrations: tuple[MigrationTrackSnapshot, ...] = ()
contracts: tuple[ModuleContractSnapshot, ...] = ()
collection_errors: tuple[DashboardCollectionError, ...] = ()
@dataclass(frozen=True, slots=True)

11
tools/release/govoplan_release/planner.py
View File

@@ -18,6 +18,17 @@ def build_release_plan(dashboard: ReleaseDashboard) -> ReleasePlan:
target_version = dashboard.target_version
target_tag = dashboard.target_tag
for index, error in enumerate(dashboard.collection_errors, start=1):
actions.append(
PlanAction(
id=f"dashboard:{error.code}:{index}",
title=f"Release metadata is unreadable in {error.repo or 'the workspace'}",
detail=f"{error.message} {error.remediation}",
severity="blocker",
repo=error.repo,
)
)
for repo in dashboard.repositories:
if not repo.exists:
actions.append(

31
tools/release/govoplan_release/selective_planner.py
View File

@@ -55,8 +55,9 @@ def build_selective_release_plan(
compatibility = compatibility_issues(dashboard)
steps = dry_run_steps(units=units, dashboard=dashboard, channel=channel)
notes = release_notes(dashboard)
status = plan_status(units=units, compatibility=compatibility)
findings = tuple(finding for unit in units for finding in unit.gate_findings)
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)
@@ -74,11 +75,28 @@ def build_selective_release_plan(
findings=findings,
compatibility=compatibility,
),
source_preflight_ready=bool(units)
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, ...],
*,
@@ -620,9 +638,12 @@ def release_notes(dashboard: ReleaseDashboard) -> tuple[str, ...]:
def plan_status(
*, units: tuple[ReleasePlanUnit, ...], compatibility: tuple[CompatibilityIssue, ...]
*,
units: tuple[ReleasePlanUnit, ...],
compatibility: tuple[CompatibilityIssue, ...],
findings: tuple[ReleaseGateFinding, ...] = (),
) -> str:
if any(unit.blockers for unit in units) or any(
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"