diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 58cb70a..a1020c8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -24,7 +24,7 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip - pip install -e ".[dev,serve,semantic]" + pip install -e ".[dev,serve,semantic,build]" - name: Verify portable PortfolioTruth consumer contract run: python scripts/generate_portfolio_truth_contract_fixture.py --check - name: Run tests diff --git a/CHANGELOG.md b/CHANGELOG.md index e1b64cd..0a0c0a4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,19 @@ Format: [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) ## [Unreleased] +- Reject reckoning path collisions under the saved truth workspace root and + validate canonical truth before exporting private overlay workbooks. + +- Prefer live checkouts over archive copies, validate private overlay inputs in + CLI preflight, preserve overlays on shared workbook refresh, and isolate + synthetic gates and exports from private runtime settings. + +- Show every reckoning family once, aggregate matched truth activity and attention, + and expose unmatched members and unreckoned truth projects in the workbook. + +- Read private runtime reckoning verdicts into an optional visible workbook sheet + without changing canonical truth; discover `_archive` children with archived defaults. + - Refuse new portfolio generation publication when staged truth lacks valid GitHub security receipt and content identities or disagrees with the admitted manifest; preserve historical generation readback compatibility. diff --git a/README.md b/README.md index e441f7b..c159591 100644 --- a/README.md +++ b/README.md @@ -105,6 +105,34 @@ audit triage --approval-center Treat campaign/writeback, GitHub Projects, Notion sync, catalog overrides, scorecards overrides, and `--excel-mode template` as advanced paths. +Private family verdicts can be displayed in a visible `Reckoning` workbook sheet +with `--reckoning-overlay PATH` or `GHRA_RECKONING_OVERLAY` (the flag takes +precedence). The YAML's `schema` value is `reckoning-verdicts`, then a slash, then +`1`; invalid verdicts stop CLI preflight before the GitHub audit. The overlay is +read once at startup, so a long `--watch` run picks up edits after a restart. Export also requires the saved +`portfolio-truth-latest.json` beside the audit report, or in the exporter API's +`truth_dir`. Each overlay family gets one row in overlay order, including +families with no saved truth match. `Members` lists the family's member keys; +`Matched projects` counts matched truth projects and `Unmatched members` lists +member keys with no match. Matching retains a GitHub repository ID tier, but +canonical truth carries no repository ID today, so current truth matches by +`owner/repo`, then workspace-relative path. `Truth attention state` is a single +value using the existing urgency order (`decision-needed`, `active-product`, +`active-infra`), with unranked ties resolved in saved-truth order. `Attention states` +lists all distinct matched states. Last activity is the +latest matched timestamp. An `Unreckoned projects` block below the family table +reports the count and first 50 keys of truth projects with no family. + +Keep the overlay outside this repository and published artifacts. It is read +only, is never copied into output, and adds no fields to canonical truth. The +resulting workbook contains private verdicts and should be treated as private. +Without an overlay, no `Reckoning` sheet is added. Both workbook modes can be +checked without private inputs using +`make workbook-gate ARGS="--synthetic-reckoning-overlay"`. This gate writes to +`output/workbook-gate/synthetic-reckoning/`, keeping the normal desktop signoff +artifacts separate. The environment variable is read only by the audit CLI; +demo, benchmark, and direct exporter calls require an explicit overlay argument. + ## Demo and Guides - Safe demo path: run `make demo` after a local clone to generate sample artifacts from the committed fixture without a GitHub token. diff --git a/config/portfolio-catalog.yaml b/config/portfolio-catalog.yaml index d3a4581..3f741da 100644 --- a/config/portfolio-catalog.yaml +++ b/config/portfolio-catalog.yaml @@ -7,6 +7,12 @@ defaults: maturity_program: maintain target_maturity: operating groups: + archive: + label: Archive + path_prefixes: + - _archive + lifecycle_state: archived + operating_path: archive itprjsviaclaude: label: IT Tools section_marker: ITPRJsViaClaude/ diff --git a/src/github_repo_auditor/app/run_audit.py b/src/github_repo_auditor/app/run_audit.py index 031cdc6..9e922d0 100644 --- a/src/github_repo_auditor/app/run_audit.py +++ b/src/github_repo_auditor/app/run_audit.py @@ -1043,6 +1043,7 @@ def _write_report_outputs( collection=args.collection, excel_mode=args.excel_mode, truth_dir=output_dir, + reckoning_overlay=getattr(args, "reckoning_overlay", None), ) report.runtime_breakdown["workbook_build_seconds"] = round(perf_counter() - workbook_start, 3) md_path = write_markdown_report(report, output_dir, diff_data=diff_dict) diff --git a/src/github_repo_auditor/cli.py b/src/github_repo_auditor/cli.py index 12dbe10..90abe97 100644 --- a/src/github_repo_auditor/cli.py +++ b/src/github_repo_auditor/cli.py @@ -60,11 +60,42 @@ DEFAULT_EXPECTED_GITHUB_COHORT_COUNT, DEFAULT_MAX_COHORT_SIZE, ) +from github_repo_auditor.reckoning_overlay import ( + ReckoningOverlayError, + load_reckoning_overlay, +) # Emitted at most once per process when legacy flat invocation is used. _LEGACY_WARNING_EVENTS: set[str] = set() + +def _prepare_reckoning_export(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None: + """Load requested private workbook inputs before any audit or shared refresh.""" + overlay_path = getattr(args, "reckoning_overlay", None) + if overlay_path is None: + configured_path = os.environ.get("GHRA_RECKONING_OVERLAY") + overlay_path = Path(configured_path).expanduser() if configured_path else None + + try: + overlay = load_reckoning_overlay(overlay_path) + if overlay is None: + args.reckoning_overlay = None + return + + # A dry run never exports a workbook, so only the overlay itself is checked. + if not getattr(args, "dry_run", False): + from github_repo_auditor.excel_export_truth_helpers import load_reckoning_truth + + load_reckoning_truth(Path(args.output_dir)) + except (ReckoningOverlayError, OSError, ValueError) as exc: + parser.error(str(exc)) + + # export_excel accepts an already validated object, avoiding environment + # lookup or overlay parsing anywhere below the CLI boundary. + args.reckoning_overlay = overlay + + DEFAULT_ANALYSIS_WORKERS = 1 MAX_ANALYSIS_WORKERS = 8 DEFAULT_PORTFOLIO_WORKSPACE = Path.home() / "Projects" @@ -170,6 +201,10 @@ def _build_run_subparser(subparsers: argparse._SubParsersAction) -> None: # typ formatter_class=argparse.RawDescriptionHelpFormatter, ) _add_global_flags(p) + p.add_argument( + "--reckoning-overlay", type=Path, metavar="PATH", + help=argparse.SUPPRESS, # Advanced option; keep primary subcommand help bounded. + ) p.add_argument( "--repos", nargs="+", @@ -574,6 +609,12 @@ def _build_report_subparser(subparsers: argparse._SubParsersAction) -> None: # help="Run context quality triage across the portfolio", ) _add_automation_proposal_flags(p) + p.add_argument( + "--reckoning-overlay", + type=Path, + metavar="PATH", + help=argparse.SUPPRESS, # Documented in flat help and the workbook guide. + ) p.add_argument( "--excel-mode", choices=["template", "standard"], @@ -1053,6 +1094,12 @@ def build_parser() -> argparse.ArgumentParser: action="store_true", help="Generate PDF audit report", ) + parser.add_argument( + "--reckoning-overlay", + type=Path, + metavar="PATH", + help="Private verdict overlay for the workbook (or GHRA_RECKONING_OVERLAY)", + ) parser.add_argument( "--excel-mode", choices=["template", "standard"], @@ -2188,6 +2235,7 @@ def main() -> None: or args.review_governance or args.review_packet ): + _prepare_reckoning_export(args, parser) _run_approval_capture_mode(args, parser) return @@ -2305,6 +2353,7 @@ def main() -> None: return if args.watch: + _prepare_reckoning_export(args, parser) _run_watch_mode(args, config_inspection) return @@ -2314,6 +2363,7 @@ def main() -> None: run_semantic_search_mode(args, query) return + _prepare_reckoning_export(args, parser) _run_main_audit_cycle(args, config_inspection) diff --git a/src/github_repo_auditor/excel_export.py b/src/github_repo_auditor/excel_export.py index b239b9e..23bbd1a 100644 --- a/src/github_repo_auditor/excel_export.py +++ b/src/github_repo_auditor/excel_export.py @@ -10,6 +10,12 @@ import json from pathlib import Path +from github_repo_auditor.reckoning_overlay import ( + ReckoningOverlay, + load_reckoning_overlay, +) +from github_repo_auditor.excel_portfolio_sheet_helpers import build_reckoning_sheet + from openpyxl import Workbook from openpyxl.chart import BubbleChart, Reference from openpyxl.formatting.rule import ColorScaleRule, DataBarRule, IconSetRule @@ -1760,6 +1766,8 @@ def _build_excel_workbook( collection: str | None = None, excel_mode: str = "standard", risk_lookup: dict[str, str] | None = None, + reckoning_overlay: ReckoningOverlay | None = None, + portfolio_truth: dict | None = None, ) -> None: _build_excel_workbook_runner( wb, @@ -1771,7 +1779,11 @@ def _build_excel_workbook( collection=collection, excel_mode=excel_mode, risk_lookup=risk_lookup, + reckoning_overlay=reckoning_overlay, + portfolio_truth=portfolio_truth, **_build_excel_workbook_runtime( + build_reckoning_sheet=_build_reckoning_sheet, + reckoning_present=reckoning_overlay is not None, build_dashboard=_build_dashboard, build_all_repos=_build_all_repos, build_portfolio_explorer=_build_portfolio_explorer, @@ -1824,6 +1836,25 @@ def _build_excel_workbook( ) +def _build_reckoning_sheet( + wb: Workbook, + truth: dict | None, + overlay: ReckoningOverlay | None, +) -> None: + build_reckoning_sheet( + wb, + truth, + overlay, + get_or_create_sheet=_get_or_create_sheet, + set_sheet_header=_set_sheet_header, + style_header_row=style_header_row, + style_data_cell=style_data_cell, + apply_zebra_stripes=apply_zebra_stripes, + add_table=_add_table, + auto_width=auto_width, + ) + + def _build_risk_summary_sheet(wb: Workbook, risk_posture: dict) -> None: ws = _get_or_create_sheet(wb, "Risk Summary") ws.sheet_properties.tabColor = "DC2626" @@ -1856,8 +1887,14 @@ def export_excel( excel_mode: str = "standard", template_path: Path | None = None, truth_dir: Path | None = None, + reckoning_overlay: Path | ReckoningOverlay | None = None, ) -> Path: """Generate the flagship Excel dashboard.""" + overlay = ( + reckoning_overlay + if isinstance(reckoning_overlay, ReckoningOverlay) + else load_reckoning_overlay(reckoning_overlay) + ) return _export_excel_workbook( report_path, output_path, @@ -1869,6 +1906,7 @@ def export_excel( excel_mode=excel_mode, template_path=template_path, truth_dir=truth_dir, + reckoning_overlay=overlay, load_risk_truth=_load_risk_truth, default_template_path=DEFAULT_TEMPLATE_PATH, resolve_template_path=resolve_template_path, diff --git a/src/github_repo_auditor/excel_export_registry_helpers.py b/src/github_repo_auditor/excel_export_registry_helpers.py index e9ba198..20753ba 100644 --- a/src/github_repo_auditor/excel_export_registry_helpers.py +++ b/src/github_repo_auditor/excel_export_registry_helpers.py @@ -5,7 +5,7 @@ from typing import Any, Callable from github_repo_auditor.excel_workbook_helpers import ( - CORE_VISIBLE_SHEETS, + core_visible_sheets_for_overlay, DEFAULT_PREFERRED_SHEET_ORDER, ) from github_repo_auditor.excel_workbook_helpers import ( @@ -70,11 +70,17 @@ def build_excel_workbook_runtime( finalize_workbook_structure: Callable[..., Any] | None = None, core_visible_sheets: set[str] | None = None, preferred_order: list[str] | None = None, + build_reckoning_sheet: Callable[..., Any] | None = None, + reckoning_present: bool = False, ) -> dict[str, Any]: build_steps_runner = run_workbook_build_steps or default_run_workbook_build_steps - workbook_finalizer = finalize_workbook_structure or default_finalize_workbook_structure + workbook_finalizer = ( + finalize_workbook_structure or default_finalize_workbook_structure + ) visible_sheets = set( - CORE_VISIBLE_SHEETS if core_visible_sheets is None else core_visible_sheets + core_visible_sheets_for_overlay(reckoning_present) + if core_visible_sheets is None + else core_visible_sheets ) sheet_order = list( DEFAULT_PREFERRED_SHEET_ORDER if preferred_order is None else preferred_order @@ -82,6 +88,7 @@ def build_excel_workbook_runtime( return { "run_workbook_build_steps": build_steps_runner, + "build_reckoning_sheet": build_reckoning_sheet, "build_dashboard": build_dashboard, "build_all_repos": build_all_repos, "build_portfolio_explorer": build_portfolio_explorer, diff --git a/src/github_repo_auditor/excel_export_runner_helpers.py b/src/github_repo_auditor/excel_export_runner_helpers.py index 6ebf10d..8ff2ee1 100644 --- a/src/github_repo_auditor/excel_export_runner_helpers.py +++ b/src/github_repo_auditor/excel_export_runner_helpers.py @@ -2,11 +2,14 @@ import json from pathlib import Path -from typing import Any +from typing import Any, Callable from openpyxl import Workbook from openpyxl.utils import get_column_letter +from github_repo_auditor.excel_export_truth_helpers import load_reckoning_truth +from github_repo_auditor.reckoning_overlay import ReckoningOverlay + def build_excel_workbook( wb: Workbook, @@ -71,6 +74,9 @@ def build_excel_workbook( core_visible_sheets: set[str], template_info_sheet: str, preferred_order: list[str], + build_reckoning_sheet: Callable[..., Any] | None = None, + reckoning_overlay: ReckoningOverlay | None = None, + portfolio_truth: dict[str, Any] | None = None, ) -> None: run_workbook_build_steps( [ @@ -165,6 +171,10 @@ def build_excel_workbook( ), ] ) + if reckoning_overlay is not None: + if build_reckoning_sheet is None: + raise ValueError("Reckoning requires a sheet builder") + build_reckoning_sheet(wb, portfolio_truth, reckoning_overlay) inject_sheet_navigation(wb) apply_workbook_named_ranges( wb, @@ -202,9 +212,14 @@ def export_excel_workbook( build_risk_summary_sheet, inject_native_sparklines, build_template_sparkline_specs, + reckoning_overlay: ReckoningOverlay | None = None, ) -> Path: data = json.loads(report_path.read_text()) + portfolio_truth = None + if reckoning_overlay is not None: + portfolio_truth = load_reckoning_truth(truth_dir or report_path.parent) + risk_lookup, risk_posture = load_risk_truth(truth_dir) if excel_mode not in {"template", "standard"}: @@ -217,6 +232,12 @@ def export_excel_workbook( else: wb = Workbook() + reckoning_kwargs = {} + if reckoning_overlay is not None: + reckoning_kwargs = { + "reckoning_overlay": reckoning_overlay, + "portfolio_truth": portfolio_truth, + } build_excel_workbook( wb, data, @@ -227,6 +248,7 @@ def export_excel_workbook( collection=collection, excel_mode=excel_mode, risk_lookup=risk_lookup, + **reckoning_kwargs, ) if risk_posture: diff --git a/src/github_repo_auditor/excel_export_truth_helpers.py b/src/github_repo_auditor/excel_export_truth_helpers.py index 35d35fd..3e1c310 100644 --- a/src/github_repo_auditor/excel_export_truth_helpers.py +++ b/src/github_repo_auditor/excel_export_truth_helpers.py @@ -4,8 +4,38 @@ import json from pathlib import Path +from typing import Any -from github_repo_auditor.portfolio_truth_types import truth_latest_path +from github_repo_auditor.portfolio_truth_types import SCHEMA_VERSION, truth_latest_path +from github_repo_auditor.portfolio_truth_validate import ( + canonicalize_truth_snapshot_payload, +) + + +def load_reckoning_truth(truth_dir: Path) -> dict[str, Any]: + """Require a saved truth snapshot before a requested Reckoning export.""" + truth_path = truth_latest_path(truth_dir) + try: + truth = json.loads(truth_path.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise ValueError( + f"Reckoning requires a readable {truth_path.name} in truth_dir (or beside the report)" + ) from exc + if not isinstance(truth, dict): + raise ValueError( + f"Reckoning requires valid {truth_path.name}: root must be an object" + ) + if truth.get("schema_version") != SCHEMA_VERSION: + raise ValueError( + f"Reckoning requires valid {truth_path.name}: unsupported schema_version " + f"{truth.get('schema_version')!r}; expected {SCHEMA_VERSION}" + ) + try: + # The portable validator pins /demo-workspace; saved operator truth may + # use any workspace root, so use its shared model-validation entrypoint. + return canonicalize_truth_snapshot_payload(truth) + except ValueError as exc: + raise ValueError(f"Reckoning requires valid {truth_path.name}: {exc}") from exc def load_risk_truth(truth_dir: Path | None) -> tuple[dict[str, str], dict[str, int]]: diff --git a/src/github_repo_auditor/excel_portfolio_sheet_helpers.py b/src/github_repo_auditor/excel_portfolio_sheet_helpers.py index 0ff606f..1a4bb2f 100644 --- a/src/github_repo_auditor/excel_portfolio_sheet_helpers.py +++ b/src/github_repo_auditor/excel_portfolio_sheet_helpers.py @@ -2,7 +2,209 @@ from __future__ import annotations -from typing import Any +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable + +from openpyxl import Workbook + +from github_repo_auditor.reckoning_overlay import ReckoningOverlay +from github_repo_auditor.weekly_command_center import ATTENTION_STATE_RANK + +RECKONING_HEADERS = [ + "Family", + "Title", + "Cluster", + "Members", + "Fate", + "Assimilate into", + "Tags", + "Gut", + "Status", + "Rationale", + "Dossier", + "Seed brief", + "Visibility after", + "Wave", + "Decided at", + "Truth attention state", + "Attention states", + "Last activity", + "Matched projects", + "Unmatched members", +] + + +def build_reckoning_rows( + truth: dict[str, Any], + overlay: ReckoningOverlay, +) -> list[list[Any]]: + """One row per overlay family, including families absent from saved truth.""" + return _build_reckoning_content(truth, overlay)[0] + + +def _latest_activity(values: list[str]) -> str: + def timestamp(value: str) -> datetime: + parsed = datetime.fromisoformat(value) + return parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed + + return max(values, key=timestamp) if values else "" + + +def _family_attention(states: list[str]) -> str: + # Unranked states tie and retain saved-truth order; expose all separately. + return ( + min(states, key=lambda state: ATTENTION_STATE_RANK.get(state, 9)) + if states else "" + ) + + +def _build_reckoning_content( + truth: dict[str, Any], overlay: ReckoningOverlay +) -> tuple[list[list[Any]], list[str]]: + projects_by_family: dict[str, list[dict[str, Any]]] = { + family_id: [] for family_id in overlay.families + } + matched_members: dict[str, set[str]] = { + family_id: set() for family_id in overlay.families + } + unreckoned: list[str] = [] + workspace_root = ( + Path(truth["workspace_root"]) if truth.get("workspace_root") else None + ) + for project in truth["projects"]: + identity = project["identity"] + matches = overlay.lookup_members( + github_id=identity.get("github_id") + or project.get("repository_state", {}).get("github_id"), + repo_full_name=identity.get("repo_full_name", ""), + local_path=identity.get("path", ""), + workspace_root=workspace_root, + ) + if not matches: + unreckoned.append( + identity.get("project_key") + or identity.get("repo_full_name") + or identity.get("path", "") + ) + continue + # Preserve lookup's first-family choice for ambiguous normalized paths. + family_id = matches[0][0] + projects_by_family[family_id].append(project) + matched_members[family_id].update( + member["key"] for matched_id, member in matches if matched_id == family_id + ) + + rows: list[list[Any]] = [] + for family_id, family in overlay.families.items(): + projects = projects_by_family[family_id] + states = [ + project.get("derived", {}).get("attention_state") for project in projects + ] + activities = [ + project.get("derived", {}).get("last_meaningful_activity_at") + for project in projects + ] + member_keys = [member["key"] for member in family["members"]] + evidence = family.get("evidence") or {} + github = family.get("github") or {} + rows.append( + [ + family_id, + family.get("title") or family_id, + family.get("cluster", ""), + ", ".join(member_keys), + family.get("fate", ""), + family.get("assimilate_into", ""), + ", ".join(family.get("tags", [])), + family.get("gut", ""), + family.get("status", ""), + family.get("rationale", ""), + evidence.get("dossier", ""), + evidence.get("seed_brief", ""), + github.get("visibility_after", ""), + family.get("wave", ""), + str(family.get("decided_at") or ""), + _family_attention([state for state in states if state]), + ", ".join(dict.fromkeys(state for state in states if state)), + _latest_activity([activity for activity in activities if activity]), + len(projects), + ", ".join( + key for key in member_keys if key not in matched_members[family_id] + ), + ] + ) + return rows, unreckoned + + +def build_reckoning_sheet( + wb: Workbook, + truth: dict[str, Any] | None, + overlay: ReckoningOverlay | None, + *, + get_or_create_sheet: Callable[..., Any], + set_sheet_header: Callable[..., Any], + style_header_row: Callable[..., Any], + style_data_cell: Callable[..., Any], + apply_zebra_stripes: Callable[..., Any], + add_table: Callable[..., Any], + auto_width: Callable[..., Any], +) -> None: + if overlay is None: + return + if truth is None: + raise ValueError("Reckoning requires the full portfolio truth snapshot") + ws = get_or_create_sheet(wb, "Reckoning") + ws.sheet_properties.tabColor = "7C3AED" + set_sheet_header( + ws, + "Reckoning", + "Private family verdicts alongside saved portfolio truth; one row per family. Coverage gaps appear below and in Unmatched members.", + width=len(RECKONING_HEADERS), + ) + rows, unreckoned = _build_reckoning_content(truth, overlay) + end_row = write_portfolio_table( + ws, + rows, + start_row=5, + headers=RECKONING_HEADERS, + freeze_panes="E6", + table_name="tblReckoning", + centered_columns=set(), + style_header_row=style_header_row, + style_data_cell=style_data_cell, + apply_zebra_stripes=apply_zebra_stripes, + add_table=add_table, + auto_width=auto_width, + ) + block_row = end_row + 3 + family_filter = ws.auto_filter.ref + ws.cell(row=block_row, column=1, value="Unreckoned projects") + style_header_row(ws, block_row, 2) + style_data_cell(ws.cell(row=block_row + 1, column=1, value="Count"), "left") + style_data_cell(ws.cell(row=block_row + 1, column=2, value=len(unreckoned)), "left") + write_portfolio_table( + ws, + [[key] for key in unreckoned[:50]], + start_row=block_row + 2, + headers=["Project key (first 50)"], + freeze_panes="E6", + table_name="tblUnreckonedProjects", + centered_columns=set(), + style_header_row=style_header_row, + style_data_cell=style_data_cell, + apply_zebra_stripes=apply_zebra_stripes, + add_table=add_table, + auto_width=auto_width, + ) + # Visible sheets use a single worksheet filter; keep it on family rows. + ws.auto_filter.ref = family_filter + # Private prose is data, even when it starts with Excel's formula prefix. + for row in ws.iter_rows(min_row=6): + for cell in row: + if isinstance(cell.value, str): + cell.data_type = "s" + PORTFOLIO_EXPLORER_HEADERS = [ "Repo", @@ -76,7 +278,8 @@ def build_portfolio_explorer_rows( ", ".join(entry["collections"]), audit.get("portfolio_catalog", {}).get("lifecycle_state", "") or "—", audit.get("portfolio_catalog", {}).get("criticality", "") or "—", - audit.get("portfolio_catalog", {}).get("intended_disposition", "") or "—", + audit.get("portfolio_catalog", {}).get("intended_disposition", "") + or "—", audit.get("portfolio_catalog", {}).get("operating_path", "") or "—", audit.get("scorecard", {}).get("maturity_level", "") or "—", build_maturity_gap_summary(audit), @@ -91,7 +294,9 @@ def build_portfolio_explorer_rows( def build_portfolio_catalog_table_rows(audits: list[dict[str, Any]]) -> list[list[Any]]: rows: list[list[Any]] = [] - for audit in sorted(audits, key=lambda item: item.get("metadata", {}).get("name", "")): + for audit in sorted( + audits, key=lambda item: item.get("metadata", {}).get("name", "") + ): catalog = audit.get("portfolio_catalog", {}) rows.append( [ @@ -117,7 +322,9 @@ def build_portfolio_catalog_table_rows(audits: list[dict[str, Any]]) -> list[lis def build_scorecard_table_rows(audits: list[dict[str, Any]]) -> list[list[Any]]: rows: list[list[Any]] = [] - for audit in sorted(audits, key=lambda item: item.get("metadata", {}).get("name", "")): + for audit in sorted( + audits, key=lambda item: item.get("metadata", {}).get("name", "") + ): scorecard = audit.get("scorecard", {}) rows.append( [ diff --git a/src/github_repo_auditor/excel_workbook_helpers.py b/src/github_repo_auditor/excel_workbook_helpers.py index 7849e0d..b53968a 100644 --- a/src/github_repo_auditor/excel_workbook_helpers.py +++ b/src/github_repo_auditor/excel_workbook_helpers.py @@ -10,6 +10,7 @@ "Dashboard", "Review Queue", "Portfolio Explorer", + "Reckoning", "Implementation Hotspots", "Operator Outcomes", "Approval Ledger", @@ -49,6 +50,10 @@ } +def core_visible_sheets_for_overlay(present: bool) -> set[str]: + return CORE_VISIBLE_SHEETS | ({"Reckoning"} if present else set()) + + def set_autofilter(ws, max_col: int, max_row: int, start_row: int = 1) -> None: """Attach a plain AutoFilter to an already-populated range.""" if max_row <= start_row: diff --git a/src/github_repo_auditor/portfolio_truth_sources.py b/src/github_repo_auditor/portfolio_truth_sources.py index 2c32d7e..c36c865 100644 --- a/src/github_repo_auditor/portfolio_truth_sources.py +++ b/src/github_repo_auditor/portfolio_truth_sources.py @@ -195,7 +195,7 @@ def discover_workspace_projects( if exclusion_reason is not None: _record_exclusion(exclusion_counts, exclusion_reason) continue - if _is_project_dir(child): + if child.name != "_archive" and _is_project_dir(child): discovered.append( _inspect_project_dir(child, workspace_root, catalog_data=catalog_data, now=now) ) @@ -421,8 +421,10 @@ def _checkout_representative( return min( group, key=lambda project: ( + project.get("_external_worktree") is True, _checkout_observation(project).get("state") != "observed", _checkout_observation(project).get("bare") is True, + "_archive" in Path(str(project.get("path", ""))).parts, str(project.get("name", "")).lower() != repo_base, len(Path(str(project.get("path", ""))).parts), len(str(project.get("path", ""))), @@ -439,6 +441,7 @@ def _checkout_identity_project( return min( group, key=lambda project: ( + "_archive" in Path(str(project.get("path", ""))).parts, str(project.get("name", "")).lower() != repo_base, len(Path(str(project.get("path", ""))).parts), len(str(project.get("path", ""))), diff --git a/src/github_repo_auditor/reckoning_overlay.py b/src/github_repo_auditor/reckoning_overlay.py new file mode 100644 index 0000000..1f0a4dd --- /dev/null +++ b/src/github_repo_auditor/reckoning_overlay.py @@ -0,0 +1,311 @@ +"""Read-only private family verdicts; never part of canonical portfolio truth.""" + +from __future__ import annotations + +import posixpath +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Any + +import yaml + + +class ReckoningOverlayError(ValueError): + """An explicitly requested overlay could not be read or validated.""" + + +class _UniqueKeyLoader(yaml.SafeLoader): + pass + + +def _unique_mapping(loader: Any, node: Any, deep: bool = False) -> dict[Any, Any]: + mapping: dict[Any, Any] = {} + for key_node, value_node in node.value: + key = loader.construct_object(key_node, deep=deep) + try: + duplicate = key in mapping + except TypeError as exc: + raise ReckoningOverlayError( + f"Invalid YAML mapping key at line {key_node.start_mark.line + 1}" + ) from exc + if duplicate: + raise ReckoningOverlayError( + f"duplicate YAML mapping key at line {key_node.start_mark.line + 1}" + ) + mapping[key] = loader.construct_object(value_node, deep=deep) + return mapping + + +_UniqueKeyLoader.add_constructor( + yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, _unique_mapping +) + + +def _path_key(value: str, workspace_root: Path | None) -> str: + # Normalize lexically: overlay validation must not read or resolve checkouts. + path = posixpath.normpath(Path(value).expanduser().as_posix()).casefold() + root = posixpath.normpath( + (workspace_root or Path.home() / "Projects").expanduser().as_posix() + ).casefold() + if path.startswith(root + "/"): + path = path[len(root) + 1 :] + return path + + +@dataclass(frozen=True) +class ReckoningOverlay: + # Retain the original mappings, including forward-compatible unknown fields. + data: dict[str, Any] + + @property + def families(self) -> dict[str, dict[str, Any]]: + return self.data["families"] + + def lookup( + self, + *, + github_id: int | None = None, + repo_full_name: str = "", + local_path: str = "", + workspace_root: Path | None = None, + ) -> tuple[str, dict[str, Any]] | None: + """Match globally by ID, then owner/repo, then workspace-relative path.""" + matches = self.lookup_members( + github_id=github_id, + repo_full_name=repo_full_name, + local_path=local_path, + workspace_root=workspace_root, + ) + if matches: + family_id = matches[0][0] + return family_id, self.families[family_id] + return None + + def lookup_members( + self, + *, + github_id: int | None = None, + repo_full_name: str = "", + local_path: str = "", + workspace_root: Path | None = None, + ) -> list[tuple[str, dict[str, Any]]]: + """Resolve the family by ID > name > path, then match its member aliases.""" + members = [ + (family_id, member) + for family_id, family in self.families.items() + for member in family["members"] + ] + if workspace_root is not None: + paths: dict[str, str] = {} + for family_id, member in members: + for path in [member["key"], *member.get("local_paths", [])]: + normalized = _path_key(path, workspace_root) + previous = paths.get(normalized) + if previous is not None and previous != family_id: + raise ReckoningOverlayError( + f"Invalid reckoning overlay: normalized path {normalized!r} " + f"belongs to families {previous!r} and {family_id!r} " + "under the truth workspace root" + ) + paths[normalized] = family_id + matches: list[tuple[str, dict[str, Any]]] = [] + if github_id is not None: + matches.extend( + [ + (family_id, member) + for family_id, member in members + if member.get("github_id") == github_id + ] + ) + if repo_full_name: + matches.extend( + [ + (family_id, member) + for family_id, member in members + if member["key"].casefold() == repo_full_name.casefold() + ] + ) + if local_path: + target = _path_key(local_path, workspace_root) + matches.extend( + [ + (family_id, member) + for family_id, member in members + if any( + _path_key(path, workspace_root) == target + for path in [member["key"], *member.get("local_paths", [])] + ) + ] + ) + if not matches: + return [] + family_id = matches[0][0] + matched_keys = { + member["key"] for matched_id, member in matches if matched_id == family_id + } + return [ + (family_id, member) + for matched_id, member in members + if matched_id == family_id and member["key"] in matched_keys + ] + + +def _validate(data: Any) -> dict[str, Any]: + errors: list[str] = [] + if not isinstance(data, dict): + raise ReckoningOverlayError("Invalid reckoning overlay: root must be a mapping") + if data.get("schema") != "reckoning-verdicts/1": + errors.append("schema must be reckoning-verdicts/1") + _validate_timestamp(data.get("updated_at"), "updated_at", errors) + families = data.get("families") + if not isinstance(families, dict): + errors.append("families must be a mapping") + families = {} + identities: dict[tuple[str, Any], str] = {} + for family_id, family in families.items(): + prefix = f"families.{family_id}" + if not isinstance(family_id, str) or not family_id.strip(): + errors.append("family IDs must be non-empty strings") + if not isinstance(family, dict): + errors.append(f"{prefix} must be a mapping") + continue + enums: dict[str, set[str | None]] = { + "fate": {"keep", "archive", "assimilate", "pending"}, + "status": {"proposed", "approved", "executed"}, + "gut": {"love", "meh", "done", None}, + } + for field, allowed in enums.items(): + value = family.get(field) + if not isinstance(value, (str, type(None))) or value not in allowed: + errors.append(f"{prefix}.{field} has an invalid value") + for field in ("title", "cluster", "rationale", "assimilate_into"): + if family.get(field) is not None and not isinstance(family[field], str): + errors.append(f"{prefix}.{field} must be text or null") + _validate_timestamp(family.get("decided_at"), f"{prefix}.decided_at", errors) + if family.get("fate") == "assimilate": + target = family.get("assimilate_into") + if not isinstance(target, str) or target not in families: + errors.append( + f"{prefix}.assimilate_into must reference an existing family" + ) + tags = family.get("tags", []) + if not isinstance(tags, list) or any(not isinstance(tag, str) for tag in tags): + errors.append(f"{prefix}.tags must be a list of strings") + tags = [] + evidence = family.get("evidence", {}) + if evidence is None: + evidence = {} + if not isinstance(evidence, dict): + errors.append(f"{prefix}.evidence must be a mapping") + evidence = {} + for field in ("dossier", "biography", "seed_brief"): + if evidence.get(field) is not None and not isinstance(evidence[field], str): + errors.append(f"{prefix}.evidence.{field} must be text or null") + if ( + "reborn-seed" in tags + and family.get("status") in ("approved", "executed") + and not evidence.get("seed_brief") + ): + errors.append( + f"{prefix}.evidence.seed_brief is required for an approved reborn-seed" + ) + github = family.get("github", {}) + if github is None: + github = {} + if not isinstance(github, dict): + errors.append(f"{prefix}.github must be a mapping") + elif github.get("visibility_after", "unchanged") not in ( + "unchanged", + "public", + "private", + ): + errors.append(f"{prefix}.github.visibility_after has an invalid value") + if "wave" in family and (type(family["wave"]) is not int or family["wave"] < 0): + errors.append(f"{prefix}.wave must be a non-negative integer") + members = family.get("members") + if not isinstance(members, list) or not members: + errors.append(f"{prefix}.members must be a non-empty list") + continue + for index, member in enumerate(members): + member_prefix = f"{prefix}.members[{index}]" + if not isinstance(member, dict): + errors.append(f"{member_prefix} must be a mapping") + continue + key = member.get("key") + if not isinstance(key, str) or not key.strip(): + errors.append(f"{member_prefix}.key must be a non-empty string") + continue + member_ids: list[tuple[str, Any]] = [ + ("key", key.casefold()), + ("path", _path_key(key, None)), + ] + github_id = member.get("github_id") + if github_id is not None: + if type(github_id) is not int or github_id <= 0: + errors.append( + f"{member_prefix}.github_id must be a positive integer" + ) + else: + member_ids.append(("github_id", github_id)) + paths = member.get("local_paths", []) + if not isinstance(paths, list) or any( + not isinstance(path, str) or not path for path in paths + ): + errors.append( + f"{member_prefix}.local_paths must be a list of non-empty strings" + ) + else: + member_ids.extend(("path", _path_key(path, None)) for path in paths) + if member.get("role", "primary") not in ( + "primary", + "mirror", + "variant", + "worktree", + "leftover", + ): + errors.append(f"{member_prefix}.role has an invalid value") + for identity in member_ids: + previous = identities.get(identity) + if previous is not None and previous != family_id: + errors.append( + f"{member_prefix}.{identity[0]} also appears in family {previous}" + ) + identities[identity] = family_id + if errors: + raise ReckoningOverlayError( + "Invalid reckoning overlay:\n- " + "\n- ".join(errors) + ) + return data + + +def _validate_timestamp(value: Any, field: str, errors: list[str]) -> None: + if value is None or isinstance(value, datetime): + return + if isinstance(value, str): + try: + datetime.fromisoformat(value.replace("Z", "+00:00")) + return + except ValueError: + pass + errors.append(f"{field} must be an ISO timestamp or null") + + +def load_reckoning_overlay(path: Path | None = None) -> ReckoningOverlay | None: + """Load an explicit overlay path; environment selection belongs to the CLI.""" + if path is None: + return None + overlay_path = path.expanduser() + try: + data = yaml.load( + overlay_path.read_text(encoding="utf-8"), Loader=_UniqueKeyLoader + ) + except (OSError, UnicodeError) as exc: + raise ReckoningOverlayError(f"Cannot read reckoning overlay: {exc}") from exc + except yaml.YAMLError as exc: + mark = getattr(exc, "problem_mark", None) + location = f" at line {mark.line + 1}" if mark else "" + raise ReckoningOverlayError( + f"Invalid reckoning overlay YAML{location}" + ) from exc + return ReckoningOverlay(_validate(data)) diff --git a/src/github_repo_auditor/report_shared_artifacts.py b/src/github_repo_auditor/report_shared_artifacts.py index c953af4..41de5c7 100644 --- a/src/github_repo_auditor/report_shared_artifacts.py +++ b/src/github_repo_auditor/report_shared_artifacts.py @@ -44,6 +44,7 @@ def refresh_shared_artifacts_from_report( trend_data=trend_data, diff_data=diff_dict, score_history=score_history, portfolio_profile=args.portfolio_profile, collection=args.collection, excel_mode=args.excel_mode, truth_dir=output_dir, + reckoning_overlay=getattr(args, "reckoning_overlay", None), ) export_review_pack(report.to_dict(), output_dir, diff_data=diff_dict, portfolio_profile=args.portfolio_profile, collection=args.collection) diff --git a/src/github_repo_auditor/weekly_command_center.py b/src/github_repo_auditor/weekly_command_center.py index 9b6c3bf..09a29a2 100644 --- a/src/github_repo_auditor/weekly_command_center.py +++ b/src/github_repo_auditor/weekly_command_center.py @@ -23,6 +23,7 @@ MAX_REPO_BRIEFINGS = 3 MAX_RISK_ATTENTION_ITEMS = 5 MAX_SECURITY_ATTENTION_ITEMS = 5 +ATTENTION_STATE_RANK = {"decision-needed": 0, "active-product": 1, "active-infra": 2} _PERSISTED_DIGEST_SCHEMA = "weekly_command_center_digest_v2" _PERSISTED_REDACTED_TEXT = "" _PERSISTED_WEEKLY_DIGEST = { @@ -614,7 +615,6 @@ def _build_path_attention_items( ) -> list[dict[str, Any]]: projects = list(portfolio_truth.get("projects") or []) candidates: list[dict[str, Any]] = [] - status_rank = {"decision-needed": 0, "active-product": 1, "active-infra": 2} context_rank = { "none": 0, "boilerplate": 1, @@ -665,7 +665,7 @@ def _build_path_attention_items( candidates.sort( key=lambda item: ( - status_rank.get(item["attention_state"], 9), + ATTENTION_STATE_RANK.get(item["attention_state"], 9), 0 if item["operating_path"] == "unspecified" else 1, 0 if item["path_confidence"] == "low" else 1, context_rank.get(item["context_quality"], 9), diff --git a/src/github_repo_auditor/workbook_gate.py b/src/github_repo_auditor/workbook_gate.py index 3238ae5..134562c 100644 --- a/src/github_repo_auditor/workbook_gate.py +++ b/src/github_repo_auditor/workbook_gate.py @@ -9,8 +9,10 @@ from xml.etree import ElementTree as ET from github_repo_auditor.excel_export import export_excel -from github_repo_auditor.excel_template import DEFAULT_TEMPLATE_PATH, load_workbook_allowing_native_sparklines -from github_repo_auditor.excel_workbook_helpers import CORE_VISIBLE_SHEETS +from github_repo_auditor.excel_template import ( + DEFAULT_TEMPLATE_PATH, + load_workbook_allowing_native_sparklines, +) DEFAULT_GATE_DIR = Path("output") / "workbook-gate" RESULT_FILENAME = "workbook-gate-result.json" @@ -43,7 +45,13 @@ def _sample_report_data() -> dict: "security_posture": {"score": 0.82, "summary": "Healthy"}, }, "security_posture": {"label": "healthy", "score": 0.82}, - "hotspots": [{"title": "Keep momentum", "severity": 0.2, "category": "finish-line"}], + "hotspots": [ + { + "title": "Keep momentum", + "severity": 0.2, + "category": "finish-line", + } + ], "action_candidates": [{"title": "Protect current momentum"}], "analyzer_results": [], }, @@ -66,7 +74,9 @@ def _sample_report_data() -> dict: "security_posture": {"score": 0.66, "summary": "Watch"}, }, "security_posture": {"label": "watch", "score": 0.66}, - "hotspots": [{"title": "Finish testing", "severity": 0.54, "category": "quality"}], + "hotspots": [ + {"title": "Finish testing", "severity": 0.54, "category": "quality"} + ], "action_candidates": [{"title": "Finish the remaining delivery work"}], "analyzer_results": [], }, @@ -89,7 +99,13 @@ def _sample_report_data() -> dict: "security_posture": {"score": 0.29, "summary": "Risky"}, }, "security_posture": {"label": "critical", "score": 0.29}, - "hotspots": [{"title": "Security posture needs attention", "severity": 0.83, "category": "security-debt"}], + "hotspots": [ + { + "title": "Security posture needs attention", + "severity": 0.83, + "category": "security-debt", + } + ], "action_candidates": [{"title": "Preview governance controls"}], "analyzer_results": [], }, @@ -99,13 +115,27 @@ def _sample_report_data() -> dict: "average_score": 0.63, "portfolio_grade": "B", "portfolio_health_score": 0.67, - "tier_distribution": {"shipped": 1, "functional": 1, "wip": 1, "skeleton": 0, "abandoned": 0}, + "tier_distribution": { + "shipped": 1, + "functional": 1, + "wip": 1, + "skeleton": 0, + "abandoned": 0, + }, "language_distribution": {"Python": 2, "TypeScript": 1}, "collections": { - "showcase": {"description": "Best examples", "repos": [{"name": "RepoA", "reason": "Strong showcase"}]}, + "showcase": { + "description": "Best examples", + "repos": [{"name": "RepoA", "reason": "Strong showcase"}], + }, }, "profiles": {"default": {"description": "Balanced"}}, - "lenses": {"ship_readiness": {"description": "Delivery readiness", "average_score": 0.71}}, + "lenses": { + "ship_readiness": { + "description": "Delivery readiness", + "average_score": 0.71, + } + }, "scenario_summary": { "top_levers": [ { @@ -180,7 +210,11 @@ def _sample_report_data() -> dict: "drift_state": "managed-issue-edited", } ], - "rollback_preview": {"available": True, "item_count": 1, "fully_reversible_count": 1}, + "rollback_preview": { + "available": True, + "item_count": 1, + "fully_reversible_count": 1, + }, "review_summary": { "review_id": "sample-review-1", "status": "open", @@ -270,7 +304,9 @@ def _sample_report_data() -> dict: "resolution_evidence_summary": "The last intervention was drifted for RepoC: Security posture needs attention, but the item is still open.", "next_action_confidence_score": 0.75, "next_action_confidence_label": "high", - "next_action_confidence_reasons": ["The next step is tied directly to the current top target."], + "next_action_confidence_reasons": [ + "The next step is tied directly to the current top target." + ], "primary_target_trust_policy": "act-with-review", "primary_target_trust_policy_reason": "Urgent work has enough tuned confidence to act, with a quick operator review.", "next_action_trust_policy": "act-with-review", @@ -678,7 +714,12 @@ def _sample_report_data() -> dict: "primary_target_reason": "This urgent item is already being watched across recent runs, so it stays ahead of ready work until it clears.", "primary_target_done_criteria": "Complete the recommended action and confirm the item exits the blocked or urgent queue on the next run.", "closure_guidance": "Preview governance controls. Treat this as done only when complete the recommended action and confirm the item exits the blocked or urgent queue on the next run.", - "attention_age_bands": {"0-1 days": 0, "2-7 days": 1, "8-21 days": 0, "22+ days": 0}, + "attention_age_bands": { + "0-1 days": 0, + "2-7 days": 1, + "8-21 days": 0, + "22+ days": 0, + }, "chronic_item_count": 0, "newly_stale_count": 0, "longest_persisting_item": { @@ -844,7 +885,9 @@ def _sample_report_data() -> dict: } -def _sheet_targets(workbook_path: Path) -> tuple[list[tuple[str, str]], list[tuple[str, str]]]: +def _sheet_targets( + workbook_path: Path, +) -> tuple[list[tuple[str, str]], list[tuple[str, str]]]: with zipfile.ZipFile(workbook_path) as archive: workbook_root = ET.fromstring(archive.read("xl/workbook.xml")) workbook_rels = ET.fromstring(archive.read("xl/_rels/workbook.xml.rels")) @@ -855,9 +898,15 @@ def _sheet_targets(workbook_path: Path) -> tuple[list[tuple[str, str]], list[tup } visible: list[tuple[str, str]] = [] hidden: list[tuple[str, str]] = [] - for sheet in workbook_root.findall("{http://schemas.openxmlformats.org/spreadsheetml/2006/main}sheets/{http://schemas.openxmlformats.org/spreadsheetml/2006/main}sheet"): + for sheet in workbook_root.findall( + "{http://schemas.openxmlformats.org/spreadsheetml/2006/main}sheets/{http://schemas.openxmlformats.org/spreadsheetml/2006/main}sheet" + ): title = sheet.attrib["name"] - target = rel_targets[sheet.attrib["{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id"]] + target = rel_targets[ + sheet.attrib[ + "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id" + ] + ] if sheet.attrib.get("state") == "hidden": hidden.append((title, target)) else: @@ -876,7 +925,9 @@ def _check_result(name: str, status: str, details: str) -> dict: def _section_result(name: str, checks: list[dict]) -> dict: return { "name": name, - "status": "passed" if all(check["status"] == "passed" for check in checks) else "failed", + "status": "passed" + if all(check["status"] == "passed" for check in checks) + else "failed", "checks": checks, } @@ -907,11 +958,21 @@ def _validate_workbook_artifact( ) ) - hidden_data_sheets = [ws.title for ws in wb.worksheets if ws.title.startswith("Data_")] + hidden_data_sheets = [ + ws.title for ws in wb.worksheets if ws.title.startswith("Data_") + ] if not hidden_data_sheets: - checks.append(_check_result("hidden-data-sheets", "failed", "No hidden Data_* sheets were found.")) + checks.append( + _check_result( + "hidden-data-sheets", "failed", "No hidden Data_* sheets were found." + ) + ) elif any(wb[name].sheet_state != "hidden" for name in hidden_data_sheets): - checks.append(_check_result("hidden-data-sheets", "failed", "One or more Data_* sheets are visible.")) + checks.append( + _check_result( + "hidden-data-sheets", "failed", "One or more Data_* sheets are visible." + ) + ) else: checks.append( _check_result( @@ -922,7 +983,9 @@ def _validate_workbook_artifact( ) visible_targets, hidden_targets = _sheet_targets(workbook_path) - hidden_data_targets = [target for title, target in hidden_targets if title.startswith("Data_")] + hidden_data_targets = [ + target for title, target in hidden_targets if title.startswith("Data_") + ] visible_table_parts: list[str] = [] hidden_table_parts_ok = False with zipfile.ZipFile(workbook_path) as archive: @@ -1075,7 +1138,11 @@ def _manual_signoff_template(standard_path: Path) -> dict: def _manual_check_lookup(manual_signoff: dict) -> dict[str, dict]: - return {item.get("id", ""): item for item in manual_signoff.get("checks", []) if item.get("id")} + return { + item.get("id", ""): item + for item in manual_signoff.get("checks", []) + if item.get("id") + } def _release_status(result: dict) -> str: @@ -1113,9 +1180,9 @@ def _write_manual_checklist(output_dir: Path, manual_signoff: dict) -> Path: "", f"- Workbook: `{manual_signoff['workbook']}`", f"- Status: `{manual_signoff['status']}`", - *( [f"- Reviewer: `{reviewer}`"] if reviewer else [] ), - *( [f"- Reviewed At: `{reviewed_at}`"] if reviewed_at else [] ), - *( [f"- Notes: {notes}"] if notes else [] ), + *([f"- Reviewer: `{reviewer}`"] if reviewer else []), + *([f"- Reviewed At: `{reviewed_at}`"] if reviewed_at else []), + *([f"- Notes: {notes}"] if notes else []), "", "## Required Checks", "", @@ -1163,14 +1230,24 @@ def _write_gate_summary(output_dir: Path, result: dict) -> Path: "", f"- Status: `{manual.get('status', 'pending')}`", f"- Authority: `{manual.get('authority', 'desktop-excel')}`", - *( [f"- Reviewer: `{manual.get('reviewer', '')}`"] if manual.get("reviewer") else [] ), - *( [f"- Reviewed At: `{manual.get('reviewed_at', '')}`"] if manual.get("reviewed_at") else [] ), - *( [f"- Notes: {manual.get('notes', '')}"] if manual.get("notes") else [] ), + *( + [f"- Reviewer: `{manual.get('reviewer', '')}`"] + if manual.get("reviewer") + else [] + ), + *( + [f"- Reviewed At: `{manual.get('reviewed_at', '')}`"] + if manual.get("reviewed_at") + else [] + ), + *([f"- Notes: {manual.get('notes', '')}"] if manual.get("notes") else []), "", ] ) for check in manual.get("checks", []): - lines.append(f"- {_format_manual_check_marker(check.get('status', 'pending'))} {check.get('label', '')}") + lines.append( + f"- {_format_manual_check_marker(check.get('status', 'pending'))} {check.get('label', '')}" + ) history = result.get("manual_signoff_history", []) or [] if history: lines.extend( @@ -1195,28 +1272,186 @@ def _flatten_failed_checks(sections: list[dict]) -> list[str]: for section in sections: for check in section.get("checks", []): if check.get("status") == "failed": - errors.append(f"{section.get('name', 'section')}::{check.get('name', 'check')} - {check.get('details', '')}") + errors.append( + f"{section.get('name', 'section')}::{check.get('name', 'check')} - {check.get('details', '')}" + ) return errors -def run_workbook_gate(output_dir: Path = DEFAULT_GATE_DIR) -> dict: +def _sample_reckoning_truth() -> dict[str, Any]: + """Use the production truth contract for the synthetic overlay gate.""" + from github_repo_auditor.demo_portfolio import DemoProject, build_snapshot + + return build_snapshot( + datetime(2026, 2, 3, tzinfo=timezone.utc), + project_specs=( + DemoProject( + "LocalOnly", "archive", ("Python",), "archived", "stale", "unknown" + ), + DemoProject( + "LocalMirror", + "lab", + ("Python",), + "decision-needed", + "active", + "complete", + (0, 1, 0, 0, 0, 0, 0), + ), + DemoProject( + "Unassigned", "studio", ("Python",), "manual-only", "recent", "unknown" + ), + ), + ) + + +def run_workbook_gate( + output_dir: Path = DEFAULT_GATE_DIR, + *, + synthetic_reckoning_overlay: bool = False, +) -> dict: + from tempfile import TemporaryDirectory + + import yaml + + from github_repo_auditor.excel_portfolio_sheet_helpers import RECKONING_HEADERS + from github_repo_auditor.excel_workbook_helpers import ( + core_visible_sheets_for_overlay, + ) + from github_repo_auditor.reckoning_overlay import load_reckoning_overlay + from github_repo_auditor.portfolio_truth_types import truth_latest_path + + if synthetic_reckoning_overlay: + output_dir = output_dir / "synthetic-reckoning" + output_dir.mkdir(parents=True, exist_ok=True) report_path = output_dir / "workbook-gate-report.json" report_path.write_text(json.dumps(_sample_report_data(), indent=2)) - standard_path = export_excel(report_path, output_dir / "workbook-gate-standard.xlsx", excel_mode="standard") - template_path = export_excel( - report_path, - output_dir / "workbook-gate-template.xlsx", - excel_mode="template", - template_path=DEFAULT_TEMPLATE_PATH, - ) + # Synthetic private inputs live only in a temporary directory, never output/. + with TemporaryDirectory(prefix="ghra-workbook-inputs-") as temporary: + input_dir = Path(temporary) + overlay_path = None + if synthetic_reckoning_overlay: + overlay_path = input_dir / "verdicts.yaml" + overlay_path.write_text( + yaml.safe_dump( + { + "schema": "reckoning-verdicts/1", + "families": { + "local-family": { + "title": "Local project", + "members": [ + {"key": "archive/localonly"}, + {"key": "lab/localmirror"}, + ], + "fate": "archive", + "status": "approved", + "tags": ["strip-for-parts"], + }, + "remote-family": { + "title": "Remote-only project", + "members": [{"key": "fixture-owner/remote-only"}], + "fate": "pending", + "status": "proposed", + }, + }, + } + ) + ) + truth_latest_path(input_dir).write_text( + json.dumps(_sample_reckoning_truth()) + ) + overlay = load_reckoning_overlay(overlay_path) + present = overlay is not None + standard_path = export_excel( + report_path, + output_dir / "workbook-gate-standard.xlsx", + excel_mode="standard", + reckoning_overlay=overlay, + truth_dir=input_dir, + ) + template_path = export_excel( + report_path, + output_dir / "workbook-gate-template.xlsx", + excel_mode="template", + template_path=DEFAULT_TEMPLATE_PATH, + reckoning_overlay=overlay, + truth_dir=input_dir, + ) + + expected_visible = core_visible_sheets_for_overlay(present) + if present: + # Providing saved truth also enables the exporter's existing Risk Summary. + expected_visible.add("Risk Summary") automated_sections = [ - _validate_workbook_artifact(standard_path, workbook_mode="standard", expected_visible=CORE_VISIBLE_SHEETS), - _validate_workbook_artifact(template_path, workbook_mode="template", expected_visible=CORE_VISIBLE_SHEETS), + _validate_workbook_artifact( + standard_path, workbook_mode="standard", expected_visible=expected_visible + ), + _validate_workbook_artifact( + template_path, workbook_mode="template", expected_visible=expected_visible + ), _validate_parity(standard_path, template_path), ] + checks = [] + workbooks = [ + load_workbook_allowing_native_sparklines(path) + for path in (standard_path, template_path) + ] + if present: + values = [ + list(wb["Reckoning"].values) if "Reckoning" in wb.sheetnames else [] + for wb in workbooks + ] + valid = bool(values[0]) and values[0] == values[1] + checks.append( + _check_result( + "reckoning-parity", + "passed" if valid else "failed", + "Reckoning cell values must match across both modes.", + ) + ) + valid = all( + "Reckoning" in wb.sheetnames + and wb["Reckoning"].sheet_state == "visible" + and [cell.value for cell in wb["Reckoning"][5]] == RECKONING_HEADERS + and wb["Reckoning"]["A6"].value == "local-family" + and wb["Reckoning"]["D6"].value == "archive/localonly, lab/localmirror" + and wb["Reckoning"]["E6"].value == "archive" + and wb["Reckoning"]["P6"].value == "decision-needed" + and wb["Reckoning"]["Q6"].value == "archived, decision-needed" + and wb["Reckoning"]["R6"].value == "2026-02-01T00:00:00+00:00" + and wb["Reckoning"]["S6"].value == 2 + and wb["Reckoning"]["T6"].value is None + and wb["Reckoning"]["A7"].value == "remote-family" + and wb["Reckoning"]["S7"].value == 0 + and wb["Reckoning"]["T7"].value == "fixture-owner/remote-only" + and wb["Reckoning"].auto_filter.ref == "A5:T7" + and wb["Reckoning"]["A10"].value == "Unreckoned projects" + and wb["Reckoning"]["B11"].value == 1 + and wb["Reckoning"]["A13"].value == "studio/unassigned" + and wb["Reckoning"].max_row == 13 + for wb in workbooks + ) + checks.append( + _check_result( + "reckoning-population", + "passed" if valid else "failed", + "Every synthetic family must appear once with aggregated truth and unmatched members; unreckoned truth projects must appear only below the family table.", + ) + ) + else: + valid = all("Reckoning" not in wb.sheetnames for wb in workbooks) + checks.append( + _check_result( + "reckoning-absent", + "passed" if valid else "failed", + "No overlay means no Reckoning sheet.", + ) + ) + for wb in workbooks: + wb.close() + automated_sections.append(_section_result("reckoning", checks)) validation_errors = _flatten_failed_checks(automated_sections) manual_signoff = _manual_signoff_template(standard_path) checklist_path = _write_manual_checklist(output_dir, manual_signoff) @@ -1236,7 +1471,9 @@ def run_workbook_gate(output_dir: Path = DEFAULT_GATE_DIR) -> dict: "generated_at": datetime.now(timezone.utc).isoformat(), "workbook_modes": ["standard", "template"], "parity_status": automated_sections[2]["status"], - "invariant_status": "passed" if all(section["status"] == "passed" for section in automated_sections[:2]) else "failed", + "invariant_status": "passed" + if all(section["status"] == "passed" for section in automated_sections[:2]) + else "failed", }, "automated_checks": { "status": "passed" if not validation_errors else "failed", @@ -1257,29 +1494,41 @@ def run_workbook_gate(output_dir: Path = DEFAULT_GATE_DIR) -> dict: def _load_gate_result(output_dir: Path) -> dict: result_path = output_dir / RESULT_FILENAME if not result_path.exists(): - raise FileNotFoundError("No workbook gate result exists yet. Run workbook-gate first.") + raise FileNotFoundError( + "No workbook gate result exists yet. Run workbook-gate first." + ) return json.loads(result_path.read_text()) -def _validate_signoff_args(result: dict, *, reviewer: str, outcome: str, checks: list[str]) -> dict[str, str]: +def _validate_signoff_args( + result: dict, *, reviewer: str, outcome: str, checks: list[str] +) -> dict[str, str]: if result.get("automated_checks", {}).get("status") != "passed": - raise ValueError("Workbook signoff cannot be recorded while automated checks are failing.") + raise ValueError( + "Workbook signoff cannot be recorded while automated checks are failing." + ) if not reviewer.strip(): raise ValueError("--reviewer is required when recording signoff.") if outcome not in MANUAL_SIGNOFF_OUTCOMES: raise ValueError(f"--outcome must be one of {sorted(MANUAL_SIGNOFF_OUTCOMES)}.") if not checks: - raise ValueError("At least one --check entry is required when recording signoff.") + raise ValueError( + "At least one --check entry is required when recording signoff." + ) parsed: dict[str, str] = {} for raw in checks: if "=" not in raw: - raise ValueError(f"Invalid --check value `{raw}`. Use =passed|failed|n/a.") + raise ValueError( + f"Invalid --check value `{raw}`. Use =passed|failed|n/a." + ) check_id, status = raw.split("=", 1) check_id = check_id.strip() status = status.strip() if not check_id: - raise ValueError(f"Invalid --check value `{raw}`. Check id cannot be empty.") + raise ValueError( + f"Invalid --check value `{raw}`. Check id cannot be empty." + ) if status not in MANUAL_CHECK_STATES: raise ValueError(f"Invalid check status `{status}` for `{check_id}`.") if check_id in parsed: @@ -1298,9 +1547,13 @@ def _validate_signoff_args(result: dict, *, reviewer: str, outcome: str, checks: statuses = list(parsed.values()) if outcome == "passed" and any(status != "passed" for status in statuses): - raise ValueError("A passing signoff requires every manual check to be marked passed.") + raise ValueError( + "A passing signoff requires every manual check to be marked passed." + ) if outcome == "failed" and "failed" not in statuses: - raise ValueError("A failing signoff requires at least one manual check to be marked failed.") + raise ValueError( + "A failing signoff requires at least one manual check to be marked failed." + ) return parsed @@ -1313,7 +1566,9 @@ def record_manual_signoff( notes: str = "", ) -> dict: result = _load_gate_result(output_dir) - parsed_checks = _validate_signoff_args(result, reviewer=reviewer, outcome=outcome, checks=checks) + parsed_checks = _validate_signoff_args( + result, reviewer=reviewer, outcome=outcome, checks=checks + ) reviewed_at = datetime.now(timezone.utc).isoformat() manual_signoff = dict(result.get("manual_signoff") or {}) @@ -1340,7 +1595,10 @@ def record_manual_signoff( "checks": {item["id"]: item["status"] for item in updated_checks}, } result["manual_signoff"] = manual_signoff - result["manual_signoff_history"] = [*(result.get("manual_signoff_history") or []), history_entry] + result["manual_signoff_history"] = [ + *(result.get("manual_signoff_history") or []), + history_entry, + ] result["release_status"] = _release_status(result) checklist_path = _write_manual_checklist(output_dir, manual_signoff) @@ -1373,7 +1631,9 @@ def format_gate_result(result: dict) -> str: elif result.get("release_status") == "ready": lines.append("Validation checks passed. Workbook release gate is ready.") else: - lines.append("Validation checks passed. Final release step: complete the manual desktop Excel checklist.") + lines.append( + "Validation checks passed. Final release step: complete the manual desktop Excel checklist." + ) return "\n".join(lines) @@ -1392,6 +1652,11 @@ def main() -> None: action="store_true", help="Record the outcome of the manual desktop Excel signoff against the existing workbook gate artifact.", ) + parser.add_argument( + "--synthetic-reckoning-overlay", + action="store_true", + help="Check the private Reckoning sheet using temporary synthetic verdicts and truth.", + ) parser.add_argument( "--reviewer", default="", @@ -1424,7 +1689,10 @@ def main() -> None: notes=args.notes, ) else: - result = run_workbook_gate(Path(args.output_dir)) + result = run_workbook_gate( + Path(args.output_dir), + synthetic_reckoning_overlay=args.synthetic_reckoning_overlay, + ) print(format_gate_result(result)) if result["status"] != "ok": raise SystemExit(1) diff --git a/tests/conftest.py b/tests/conftest.py index 11d1d33..09c14f8 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -8,6 +8,12 @@ from github_repo_auditor.models import RepoMetadata +@pytest.fixture(autouse=True) +def clear_reckoning_overlay_env(monkeypatch: pytest.MonkeyPatch) -> None: + """Private runtime settings must not affect fixture exports.""" + monkeypatch.delenv("GHRA_RECKONING_OVERLAY", raising=False) + + @pytest.fixture def sample_metadata() -> RepoMetadata: """A realistic RepoMetadata for testing.""" diff --git a/tests/test_cli_reckoning.py b/tests/test_cli_reckoning.py new file mode 100644 index 0000000..92dd755 --- /dev/null +++ b/tests/test_cli_reckoning.py @@ -0,0 +1,179 @@ +from __future__ import annotations + +import sys +import json +from datetime import datetime, timezone +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from github_repo_auditor import cli + + +def _write_overlay(path: Path) -> Path: + path.write_text( + "schema: reckoning-verdicts/1\nupdated_at: null\nfamilies: {}\n", + encoding="utf-8", + ) + return path + + +def _main_args(monkeypatch, output_dir: Path, *extra: str) -> None: + monkeypatch.setattr( + sys, + "argv", + ["audit", "run", "fixture-user", "--output-dir", str(output_dir), *extra], + ) + monkeypatch.setattr( + "github_repo_auditor.config.inspect_config", + lambda _path: SimpleNamespace(data={}), + ) + + +@pytest.mark.parametrize("truth", [None, "not JSON", {}, {"projects": [{}]}]) +def test_reckoning_overlay_truth_failure_precedes_audit( + tmp_path: Path, monkeypatch, truth +) -> None: + overlay_path = _write_overlay(tmp_path / "overlay.yaml") + audit_called = False + + def run_audit(*_args): + nonlocal audit_called + audit_called = True + + monkeypatch.setattr(cli, "_run_main_audit_cycle", run_audit) + output_dir = tmp_path / "output" + if truth is not None: + output_dir.mkdir() + (output_dir / "portfolio-truth-latest.json").write_text( + truth if isinstance(truth, str) else json.dumps(truth) + ) + _main_args( + monkeypatch, + output_dir, + "--reckoning-overlay", + str(overlay_path), + "--preflight-mode", + "off", + ) + + with pytest.raises(SystemExit) as exc_info: + cli.main() + + assert exc_info.value.code == 2 + assert not audit_called + + +def test_invalid_environment_overlay_fails_before_audit(tmp_path, monkeypatch) -> None: + path = tmp_path / "invalid.yaml" + path.write_text("schema: wrong\nfamilies: {}\n") + monkeypatch.setenv("GHRA_RECKONING_OVERLAY", str(path)) + monkeypatch.setattr( + cli, "_run_main_audit_cycle", lambda *_args: pytest.fail("audit must not start") + ) + _main_args(monkeypatch, tmp_path, "--preflight-mode", "off") + with pytest.raises(SystemExit) as error: + cli.main() + assert error.value.code == 2 + + +def test_cli_explicit_overlay_wins_and_loaded_object_reaches_audit( + tmp_path: Path, monkeypatch +) -> None: + from github_repo_auditor import excel_export_truth_helpers + + explicit_path = tmp_path / "explicit.yaml" + env_path = tmp_path / "environment.yaml" + marker = object() + loaded_paths: list[Path | None] = [] + truth_dirs: list[Path] = [] + audit_args = [] + + def load_overlay(path): + loaded_paths.append(path) + return marker + + monkeypatch.setattr(cli, "load_reckoning_overlay", load_overlay) + monkeypatch.setattr( + excel_export_truth_helpers, + "load_reckoning_truth", + lambda truth_dir: truth_dirs.append(truth_dir), + ) + monkeypatch.setattr( + cli, "_run_main_audit_cycle", lambda args, _config: audit_args.append(args) + ) + monkeypatch.setenv("GHRA_RECKONING_OVERLAY", str(env_path)) + _main_args( + monkeypatch, + tmp_path / "output", + "--reckoning-overlay", + str(explicit_path), + "--preflight-mode", + "off", + ) + + cli.main() + + assert loaded_paths == [explicit_path] + assert truth_dirs == [tmp_path / "output"] + assert len(audit_args) == 1 + assert audit_args[0].reckoning_overlay is marker + + +def test_shared_artifact_refresh_passes_overlay_to_workbook( + monkeypatch, tmp_path: Path +) -> None: + from github_repo_auditor import report_shared_artifacts as shared + + captured = {} + report = SimpleNamespace( + username="fixture-user", + generated_at=datetime(2026, 10, 12, tzinfo=timezone.utc), + operator_summary={}, + operator_queue=[], + to_dict=lambda: {}, + ) + args = SimpleNamespace( + approval_view="all", + portfolio_profile="default", + collection=None, + excel_mode="standard", + reckoning_overlay=object(), + ) + json_path = tmp_path / "report.json" + monkeypatch.setattr( + shared, + "write_approval_center_artifacts", + lambda *_a, **_k: (Path("approval.json"), Path("approval.md"), {}), + ) + monkeypatch.setattr(shared, "write_json_report", lambda *_a, **_k: json_path) + monkeypatch.setattr(shared, "write_markdown_report", lambda *_a, **_k: None) + monkeypatch.setattr(shared, "write_pcc_export", lambda *_a, **_k: None) + monkeypatch.setattr(shared, "write_raw_metadata", lambda *_a, **_k: None) + monkeypatch.setattr(shared, "load_trend_data", lambda: []) + monkeypatch.setattr(shared, "load_repo_score_history", lambda: {}) + monkeypatch.setattr( + shared, "export_excel", lambda *_a, **kwargs: captured.update(kwargs) + ) + monkeypatch.setattr(shared, "export_review_pack", lambda *_a, **_k: None) + monkeypatch.setattr(shared, "export_html_dashboard", lambda *_a, **_k: None) + monkeypatch.setattr( + shared, "report_artifact_datetime", lambda *_a: report.generated_at + ) + monkeypatch.setattr( + shared, + "write_control_center_artifacts", + lambda *_a, **_k: ( + Path("control.json"), + Path("control.md"), + Path("weekly.json"), + Path("weekly.md"), + {}, + ), + ) + monkeypatch.setattr(shared, "write_warehouse_snapshot", lambda *_a, **_k: None) + + shared.refresh_shared_artifacts_from_report(report, tmp_path, args) + + assert captured["reckoning_overlay"] is args.reckoning_overlay diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 1c6874e..0909c1b 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -1,9 +1,13 @@ from __future__ import annotations import json +import os +import re import subprocess import sys +import tarfile import tomllib +import zipfile from pathlib import Path import pytest @@ -12,6 +16,94 @@ ROOT = Path(__file__).resolve().parents[1] +_FAMILY_VERDICT_BLOCK = re.compile(rb"(?ms)^families:[ \t]*\r?\n.*?(?=^\S|\Z)") +_FAMILY_VERDICT_KEY = re.compile(rb"(?m)^[ \t]+(?:fate|rationale):") + + +def _private_reckoning_signatures(contents: bytes) -> list[str]: + signatures: list[str] = [] + if b"reckoning-verdicts/1" in contents: + signatures.append("reckoning-verdicts/1") + if any( + _FAMILY_VERDICT_KEY.search(block.group()) + for block in _FAMILY_VERDICT_BLOCK.finditer(contents) + ): + signatures.append("families verdict fields") + return signatures + + +def test_built_distributions_exclude_private_reckoning_data(tmp_path): + private = tmp_path / "reckoning-verdicts.yaml" + private.write_text("schema: reckoning-verdicts/1\nfamilies: {}\n") + result = subprocess.run( + [sys.executable, "-m", "build", "--outdir", str(tmp_path / "dist")], + cwd=ROOT, + capture_output=True, + text=True, + env={ + **os.environ, + "GHRA_RECKONING_OVERLAY": str(private), + "TMPDIR": str(tmp_path), + "PIP_CACHE_DIR": str(tmp_path / "pip-cache"), + }, + ) + assert result.returncode == 0, result.stdout + result.stderr + (wheel,) = (tmp_path / "dist").glob("*.whl") + (sdist,) = (tmp_path / "dist").glob("*.tar.gz") + with zipfile.ZipFile(wheel) as archive: + wheel_files = [ + (item.filename, archive.read(item)) + for item in archive.infolist() + if not item.is_dir() + ] + with tarfile.open(sdist) as archive: + sdist_files = [ + (member.name, extracted.read()) + for member in archive.getmembers() + if member.isfile() + and (extracted := archive.extractfile(member)) is not None + ] + + for files in (wheel_files, sdist_files): + names = [name for name, _ in files] + assert any( + name.endswith("github_repo_auditor/reckoning_overlay.py") for name in names + ) + assert not any( + ("reckoning" in name.lower() or "verdicts" in name.lower()) + and Path(name).name not in { + "reckoning_overlay.py", "test_reckoning_overlay.py", "test_cli_reckoning.py", + } + for name in names + ) + for name, contents in files: + if not name.endswith(".py"): + assert _private_reckoning_signatures(contents) == [], name + + +@pytest.mark.parametrize( + "contents, expected", + [ + (b"schema: reckoning-verdicts/1\n", ["reckoning-verdicts/1"]), + ( + b"Documentation mentions reckoning-verdicts/1 inline.\n", + ["reckoning-verdicts/1"], + ), + ( + b"families:\n Example:\n fate: archive\n", + ["families verdict fields"], + ), + ( + b"families:\n Example:\n rationale: unsupported\n", + ["families verdict fields"], + ), + (b"families:\n Example:\n fate: keep", ["families verdict fields"]), + (b"families:\n Example:\n owner: team\nother: value\n", []), + ], +) +def test_private_reckoning_signature_scanner(contents: bytes, expected: list[str]): + assert _private_reckoning_signatures(contents) == expected + def _project_dependency_names() -> set[str]: data = tomllib.loads((ROOT / "pyproject.toml").read_text()) diff --git a/tests/test_portfolio_truth_sources.py b/tests/test_portfolio_truth_sources.py index a78f54f..329b583 100644 --- a/tests/test_portfolio_truth_sources.py +++ b/tests/test_portfolio_truth_sources.py @@ -12,6 +12,9 @@ from datetime import datetime, timezone from pathlib import Path +from github_repo_auditor.portfolio_catalog import load_portfolio_catalog +from github_repo_auditor.portfolio_truth_reconcile import build_portfolio_truth_snapshot + from github_repo_auditor.portfolio_truth_sources import ( WORKSPACE_EXCLUSION_REASONS, _dedupe_checkouts_by_origin, @@ -22,6 +25,66 @@ ) +def test_archive_is_a_container_with_index_and_catalog_defaults(tmp_path) -> None: + archive = tmp_path / "_archive" + archive.mkdir() + (archive / "README.md").write_text("Archive index") + for name in ("Local", "Override"): + project = archive / name + project.mkdir() + (project / "README.md").write_text("Local project") + catalog_path = Path(__file__).resolve().parents[1] / "config/portfolio-catalog.yaml" + catalog = load_portfolio_catalog(catalog_path) + assert catalog["errors"] == [] + discovered = discover_workspace_projects(tmp_path, catalog_data=catalog) + assert {project["path"] for project in discovered} == { + "_archive/Local", + "_archive/Override", + } + assert all( + project["group_entry"]["lifecycle_state"] == "archived" + for project in discovered + ) + assert all( + project["group_entry"]["operating_path"] == "archive" for project in discovered + ) + # Use only the public group's data with synthetic repo declarations. + import yaml + + synthetic_catalog = tmp_path / "catalog.yaml" + synthetic_catalog.write_text( + yaml.safe_dump( + { + "groups": { + "archive": { + "path_prefixes": ["_archive"], + "lifecycle_state": "archived", + "operating_path": "archive", + } + }, + "repos": { + "_archive/Override": { + "lifecycle_state": "maintenance", + "operating_path": "maintain", + } + }, + } + ) + ) + result = build_portfolio_truth_snapshot( + workspace_root=tmp_path, + catalog_path=synthetic_catalog, + include_notion=False, + ) + by_name = { + project.identity.display_name: project for project in result.snapshot.projects + } + assert by_name["Local"].declared.lifecycle_state == "archived" + assert by_name["Local"].declared.operating_path == "archive" + assert by_name["Override"].declared.lifecycle_state == "maintenance" + assert by_name["Override"].declared.operating_path == "maintain" + + def _p( name: str, repo_full_name: str = "", @@ -300,6 +363,47 @@ def test_working_checkout_is_preferred_over_bare_same_origin_repo() -> None: ] +def test_live_nested_checkout_wins_identity_and_representation_over_archive() -> None: + collisions: list[dict] = [] + head = "1" * 40 + archive = _p( + "Repo", "owner/Repo", path="_archive/Repo", + head=head, common_dir="/git/archive/.git", + ) + archive["group_entry"] = {"lifecycle_state": "archived", "operating_path": "archive"} + live = _p( + "Renamed", "owner/Repo", path="NestedProjects/Longer/Renamed", + head=head, common_dir="/git/live/.git", + ) + live["group_entry"] = {"lifecycle_state": "maintenance", "operating_path": "maintain"} + + result = _dedupe_checkouts_by_origin([archive, live], checkout_collisions=collisions) + + assert result == [live] + assert result[0]["name"] == "Renamed" + assert result[0]["group_entry"]["lifecycle_state"] != "archived" + assert collisions[0]["selection"]["representative_path"] == live["path"] + + +def test_archive_clone_beats_external_worktree_placeholder_as_representative() -> None: + from github_repo_auditor.portfolio_truth_sources import _checkout_representative + + archive = _p( + "Repo", "owner/Repo", path="_archive/Repo", + head="1" * 40, common_dir="/git/archive/.git", + ) + placeholder = { + "name": "external-worktree", + "path": "external-worktree", + "repo_full_name": "owner/Repo", + "_external_worktree": True, + "_checkout_observation": {"state": "unknown", "bare": None}, + } + + assert _checkout_representative([placeholder, archive], "owner/Repo") is archive + assert _checkout_representative([archive, placeholder], "owner/Repo") is archive + + def test_observed_checkout_supplies_evidence_without_replacing_basename_identity() -> None: collisions: list[dict] = [] head = "1" * 40 diff --git a/tests/test_reckoning_overlay.py b/tests/test_reckoning_overlay.py new file mode 100644 index 0000000..20a7ff2 --- /dev/null +++ b/tests/test_reckoning_overlay.py @@ -0,0 +1,522 @@ +from __future__ import annotations + +import copy +import json +from pathlib import Path + +import pytest +import yaml +from openpyxl import Workbook + +from github_repo_auditor.cli import build_parser, build_subcommand_parser +from github_repo_auditor.excel_export import _build_reckoning_sheet, export_excel +from github_repo_auditor.excel_export_truth_helpers import load_reckoning_truth +from github_repo_auditor.excel_portfolio_sheet_helpers import ( + RECKONING_HEADERS, + build_reckoning_rows, +) +from github_repo_auditor.excel_template import load_workbook_allowing_native_sparklines +from github_repo_auditor.excel_workbook_helpers import CORE_VISIBLE_SHEETS +from github_repo_auditor.portfolio_truth_types import SCHEMA_VERSION +from github_repo_auditor.reckoning_overlay import ( + ReckoningOverlayError, + ReckoningOverlay, + load_reckoning_overlay, +) +from github_repo_auditor.workbook_gate import ( + _sample_reckoning_truth, + _sample_report_data, + record_manual_signoff, + run_workbook_gate, +) + + +def _family(key: str, **changes: object) -> dict: + return { + "title": key, + "members": [{"key": key}], + "fate": "pending", + "status": "proposed", + **changes, + } + + +def _write_overlay(path: Path, families: dict) -> Path: + path.write_text( + yaml.safe_dump( + {"schema": "reckoning-verdicts/1", "families": families}, sort_keys=False + ) + ) + return path + + +def test_absent_env_and_cli_path_selection(tmp_path, monkeypatch) -> None: + monkeypatch.setattr("github_repo_auditor.cli._gh_auth_token", lambda: None) + monkeypatch.delenv("GHRA_RECKONING_OVERLAY", raising=False) + assert load_reckoning_overlay() is None + env_path = _write_overlay(tmp_path / "env.yaml", {"env": _family("Env")}) + explicit = _write_overlay( + tmp_path / "explicit.yaml", + {"explicit": _family("Explicit", future={"preserved": True})}, + ) + monkeypatch.setenv("GHRA_RECKONING_OVERLAY", str(env_path)) + assert load_reckoning_overlay() is None + assert "env" in load_reckoning_overlay(env_path).families + overlay = load_reckoning_overlay(explicit) + assert overlay.families["explicit"]["future"] == {"preserved": True} + assert ( + build_parser() + .parse_args(["fixture-owner", "--reckoning-overlay", str(explicit)]) + .reckoning_overlay + == explicit + ) + assert ( + build_subcommand_parser() + .parse_args(["run", "fixture-owner", "--reckoning-overlay", str(explicit)]) + .reckoning_overlay + == explicit + ) + + +def test_matching_precedence_and_archive_path(tmp_path) -> None: + overlay = load_reckoning_overlay( + _write_overlay( + tmp_path / "private.yaml", + { + "by-path": _family( + "Local", + members=[{"key": "Local", "local_paths": ["_archive/Local"]}], + ), + "by-name": _family("fixture-owner/repo"), + "by-id": _family( + "fixture-owner/renamed", + members=[{"key": "fixture-owner/renamed", "github_id": 17}], + ), + }, + ) + ) + original = copy.deepcopy(overlay.data) + arguments = { + "github_id": 17, + "repo_full_name": "FIXTURE-OWNER/REPO", + "local_path": "_archive/Local", + } + assert overlay.lookup(**arguments)[0] == "by-id" + arguments["github_id"] = 18 + assert overlay.lookup(**arguments)[0] == "by-name" + arguments["repo_full_name"] = "" + assert overlay.lookup(**arguments)[0] == "by-path" + assert ( + overlay.lookup( + local_path=str(tmp_path / "_archive/Local"), workspace_root=tmp_path + )[0] + == "by-path" + ) + assert overlay.lookup(local_path="Other") is None + assert overlay.data == original + + +def test_family_counts_projects_once_and_preserves_member_matching_precedence() -> None: + overlay = ReckoningOverlay( + { + "families": { + "by-id": _family( + "fixture-owner/original", + members=[ + {"key": "fixture-owner/original", "github_id": 17}, + {"key": "Mirror", "github_id": 17}, + {"key": "LocalAlias"}, + {"key": "MissingFolder"}, + ], + ), + "by-name": _family("fixture-owner/other"), + "by-path": _family("Local"), + } + } + ) + truth = { + "projects": [ + { + "identity": { + "github_id": 17, + "repo_full_name": "fixture-owner/other", + "path": "Local", + }, + "derived": {"attention_state": "active-infra"}, + }, + { + "identity": { + "repo_full_name": "fixture-owner/original", + "path": "LocalAlias", + }, + "derived": {"attention_state": "active-product"}, + }, + ] + } + before = copy.deepcopy((truth, overlay.data)) + rows = [ + dict(zip(RECKONING_HEADERS, row)) + for row in build_reckoning_rows(truth, overlay) + ] + assert [row["Family"] for row in rows] == ["by-id", "by-name", "by-path"] + assert rows[0]["Matched projects"] == 2 + assert rows[0]["Unmatched members"] == "MissingFolder" + assert rows[0]["Truth attention state"] == "active-product" + assert rows[1]["Matched projects"] == rows[2]["Matched projects"] == 0 + assert rows[1]["Unmatched members"] == "fixture-owner/other" + assert rows[2]["Unmatched members"] == "Local" + assert (truth, overlay.data) == before + + +def test_unranked_attention_has_single_value_and_lists_distinct_states() -> None: + overlay = ReckoningOverlay( + { + "families": { + "local": _family("Local", members=[{"key": "Local"}, {"key": "Mirror"}]) + } + } + ) + truth = { + "projects": [ + {"identity": {"path": path}, "derived": {"attention_state": state}} + for path, state in [ + ("Local", "archived"), + ("Mirror", "parked"), + ("Local", "archived"), + ] + ] + } + row = dict(zip(RECKONING_HEADERS, build_reckoning_rows(truth, overlay)[0])) + assert row["Truth attention state"] == "archived" + assert row["Attention states"] == "archived, parked" + assert row["Matched projects"] == 3 + assert row["Last activity"] == "" + + +@pytest.mark.parametrize("count", [0, 55]) +def test_unreckoned_block_counts_all_projects_and_lists_first_50(count) -> None: + truth = { + "projects": [ + {"identity": {"project_key": f"=project-{index}", "path": f"Local{index}"}} + for index in range(count) + ] + } + wb = Workbook() + _build_reckoning_sheet(wb, truth, ReckoningOverlay({"families": {}})) + sheet = wb["Reckoning"] + assert sheet["A8"].value == "Unreckoned projects" + assert sheet["B9"].value == count + assert sheet.max_row == 10 + min(count, 50) + assert sheet.auto_filter.ref is None + assert [ + sheet.cell(row=11 + index, column=1).value for index in range(min(count, 50)) + ] == [f"=project-{index}" for index in range(min(count, 50))] + if count: + assert sheet["A11"].data_type == "s" + wb.close() + + +def test_validation_lists_all_schema_rule_problems(tmp_path) -> None: + path = _write_overlay( + tmp_path / "bad.yaml", + { + "first": _family( + "Duplicate", + fate="assimilate", + assimilate_into="missing", + tags=["reborn-seed"], + status="approved", + ), + "second": _family("Duplicate"), + }, + ) + with pytest.raises(ReckoningOverlayError) as error: + load_reckoning_overlay(path) + message = str(error.value) + assert "assimilate_into" in message + assert "seed_brief" in message + assert "also appears" in message + + +@pytest.mark.parametrize( + "first,second", + [ + ({"key": "Foo"}, {"key": "~/Projects/Foo"}), + ({"key": "Foo"}, {"key": "foo"}), + ({"key": "~/Projects/Foo"}, {"key": "foo"}), + ({"key": "owner/first", "local_paths": ["Foo"]}, {"key": "foo"}), + ({"key": "Foo"}, {"key": "owner/second", "local_paths": ["~/Projects/foo"]}), + ( + {"key": "owner/first", "local_paths": ["./Foo/"]}, + {"key": "owner/second", "local_paths": ["~/Projects/foo"]}, + ), + ], +) +def test_folder_aliases_cannot_belong_to_two_families(tmp_path, first, second) -> None: + # Exercise an absolute home path as well as a relative, case-folded alias. + first = copy.deepcopy(first) + if first["key"].startswith("~/"): + first["key"] = str(Path(first["key"]).expanduser()) + path = _write_overlay( + tmp_path / "duplicate.yaml", + { + "first": _family(first["key"], members=[first]), + "second": _family(second["key"], members=[second]), + }, + ) + with pytest.raises(ReckoningOverlayError, match="also appears"): + load_reckoning_overlay(path) + + +def test_path_matching_uses_casefolded_workspace_relative_aliases(tmp_path) -> None: + overlay = load_reckoning_overlay( + _write_overlay(tmp_path / "paths.yaml", {"local": _family("Foo")}) + ) + assert ( + overlay.lookup(local_path=str(Path.home() / "Projects" / "foo"))[0] == "local" + ) + + +@pytest.mark.parametrize( + "first,second", + [ + ({"key": "/mnt/work/Foo"}, {"key": "foo"}), + ({"key": "owner/first", "local_paths": ["/mnt/work/Foo"]}, {"key": "foo"}), + ( + {"key": "/mnt/work/Foo"}, + {"key": "owner/second", "local_paths": ["./foo/"]}, + ), + ], +) +def test_truth_workspace_path_collisions_fail_matching(tmp_path, first, second) -> None: + overlay = load_reckoning_overlay( + _write_overlay( + tmp_path / "paths.yaml", + { + "first": _family(first["key"], members=[first]), + "second": _family(second["key"], members=[second]), + }, + ) + ) + # Early validation retains the default root; the saved truth supplies this one. + with pytest.raises(ReckoningOverlayError, match="families 'first' and 'second'"): + build_reckoning_rows( + { + "workspace_root": "/mnt/work", + "projects": [{"identity": {"path": "Foo"}}], + }, + overlay, + ) + + +def test_truth_workspace_path_aliases_within_one_family_are_allowed(tmp_path) -> None: + overlay = load_reckoning_overlay( + _write_overlay( + tmp_path / "paths.yaml", + { + "local": _family( + "/mnt/work/Foo", + members=[{"key": "/mnt/work/Foo", "local_paths": ["./foo/"]}], + ) + }, + ) + ) + assert ( + overlay.lookup(local_path="FOO", workspace_root=Path("/mnt/work"))[0] == "local" + ) + + +@pytest.mark.parametrize( + "contents", + [ + "schema: wrong\nfamilies: {}", + "[]", + "families: [", + "schema: reckoning-verdicts/1\nfamilies: {}\nfamilies: {}", + "schema: reckoning-verdicts/1\nfamilies: {bad: {fate: [], status: [], members: [3]}}", + ], +) +def test_invalid_overlay_is_never_silently_skipped(tmp_path, contents) -> None: + path = tmp_path / "invalid.yaml" + path.write_text(contents) + with pytest.raises(ReckoningOverlayError): + load_reckoning_overlay(path) + with pytest.raises(ReckoningOverlayError, match="Cannot read"): + load_reckoning_overlay(tmp_path / "missing.yaml") + + +@pytest.mark.parametrize( + "field,value", + [ + ("evidence", []), + ("evidence", False), + ("github", []), + ("github", False), + ("decided_at", {}), + ("decided_at", "not-a-date"), + ], +) +def test_invalid_optional_fields_are_not_silently_defaulted( + tmp_path, field, value +) -> None: + path = _write_overlay( + tmp_path / "invalid.yaml", {"local": _family("Local", **{field: value})} + ) + with pytest.raises(ReckoningOverlayError, match=field): + load_reckoning_overlay(path) + + +@pytest.mark.parametrize("mode", ["standard", "template"]) +def test_workbook_uses_full_truth_without_mutating_inputs( + tmp_path, monkeypatch, mode +) -> None: + monkeypatch.delenv("GHRA_RECKONING_OVERLAY", raising=False) + report = tmp_path / "report.json" + report.write_text(json.dumps(_sample_report_data())) + overlay_path = _write_overlay( + tmp_path / "private.yaml", + { + "local": _family( + "archive/localonly", + members=[{"key": "archive/localonly"}, {"key": "lab/localmirror"}], + fate="archive", + title="=Private text", + tags=["custom-tag"], + evidence={ + "dossier": "dossiers/local.md", + "seed_brief": "seeds/local.md", + }, + github={"visibility_after": "private"}, + wave=2, + decided_at="2026-10-11T00:00:00Z", + ), + "remote": _family("fixture-owner/remote-only"), + }, + ) + truth_path = tmp_path / "portfolio-truth-latest.json" + truth_path.write_text(json.dumps(_sample_reckoning_truth())) + before = {path: path.read_bytes() for path in (report, overlay_path, truth_path)} + monkeypatch.setenv("GHRA_RECKONING_OVERLAY", str(overlay_path)) + absent_path = export_excel(report, tmp_path / "absent.xlsx", excel_mode=mode) + absent = load_workbook_allowing_native_sparklines(absent_path) + assert "Reckoning" not in absent.sheetnames + assert { + ws.title for ws in absent.worksheets if ws.sheet_state == "visible" + } == CORE_VISIBLE_SHEETS + present_path = export_excel( + report, + tmp_path / "present.xlsx", + excel_mode=mode, + reckoning_overlay=load_reckoning_overlay(overlay_path), + ) + present = load_workbook_allowing_native_sparklines(present_path) + assert set(present.sheetnames) == set(absent.sheetnames) | {"Reckoning"} + sheet = present["Reckoning"] + assert sheet.sheet_state == "visible" + assert [cell.value for cell in sheet[5]] == RECKONING_HEADERS + assert [cell.value for cell in sheet[6]] == [ + "local", + "=Private text", + None, + "archive/localonly, lab/localmirror", + "archive", + None, + "custom-tag", + None, + "proposed", + None, + "dossiers/local.md", + "seeds/local.md", + "private", + 2, + "2026-10-11T00:00:00Z", + "decision-needed", + "archived, decision-needed", + "2026-02-01T00:00:00+00:00", + 2, + None, + ] + assert sheet["B6"].data_type == "s" + assert sheet["A7"].value == "remote" + assert sheet["D7"].value == "fixture-owner/remote-only" + assert all(sheet[f"{column}7"].value is None for column in ("P", "Q", "R")) + assert sheet["S7"].value == 0 + assert sheet["T7"].value == "fixture-owner/remote-only" + assert sheet.auto_filter.ref == "A5:T7" + assert sheet["A10"].value == "Unreckoned projects" + assert sheet["A11"].value == "Count" and sheet["B11"].value == 1 + assert sheet["A13"].value == "studio/unassigned" + assert sheet.max_row == 13 + for path, content in before.items(): + assert path.read_bytes() == content + assert list(tmp_path.glob("*verdict*")) == [] + absent.close() + present.close() + + +@pytest.mark.parametrize("mode", ["standard", "template"]) +@pytest.mark.parametrize( + "contents", + [ + None, + "not JSON", + json.dumps({"projects": [{"identity": {}}]}), + json.dumps({"schema_version": SCHEMA_VERSION, "projects": [{"identity": {}}]}), + json.dumps({"schema_version": "999.0.0", "projects": []}), + ], +) +def test_requested_overlay_requires_saved_truth_before_writing_workbook( + tmp_path, monkeypatch, mode, contents +) -> None: + monkeypatch.delenv("GHRA_RECKONING_OVERLAY", raising=False) + report = tmp_path / "report.json" + report.write_text(json.dumps(_sample_report_data())) + overlay_path = _write_overlay( + tmp_path / "private.yaml", {"local": _family("Local")} + ) + output = tmp_path / "workbook.xlsx" + if contents is not None: + (tmp_path / "portfolio-truth-latest.json").write_text(contents) + with pytest.raises(ValueError, match="portfolio-truth-latest.json"): + export_excel(report, output, excel_mode=mode, reckoning_overlay=overlay_path) + assert not output.exists() + + +def test_reckoning_truth_rejects_unsupported_schema(tmp_path) -> None: + truth = _sample_reckoning_truth() + truth["schema_version"] = "999.0.0" + (tmp_path / "portfolio-truth-latest.json").write_text(json.dumps(truth)) + with pytest.raises(ValueError, match="unsupported schema_version '999.0.0'"): + load_reckoning_truth(tmp_path) + + +def test_reckoning_truth_validates_required_fields(tmp_path) -> None: + (tmp_path / "portfolio-truth-latest.json").write_text( + json.dumps({"schema_version": SCHEMA_VERSION, "projects": [{"identity": {}}]}) + ) + with pytest.raises(ValueError, match="missing field: derived"): + load_reckoning_truth(tmp_path) + + +def test_reckoning_truth_accepts_valid_operator_workspace(tmp_path) -> None: + truth = _sample_reckoning_truth() + truth["workspace_root"] = "/mnt/work" + truth["source_summary"]["workspace_root"] = "/mnt/work" + path = tmp_path / "portfolio-truth-latest.json" + path.write_text(json.dumps(truth)) + before = path.read_bytes() + assert load_reckoning_truth(tmp_path) == truth + assert path.read_bytes() == before + + +def test_workbook_gate_with_synthetic_overlay(tmp_path, monkeypatch) -> None: + monkeypatch.delenv("GHRA_RECKONING_OVERLAY", raising=False) + result = run_workbook_gate(tmp_path, synthetic_reckoning_overlay=True) + assert result["status"] == "ok" + section = result["automated_checks"]["sections"][-1] + assert section["name"] == "reckoning" + assert section["status"] == "passed" + assert (tmp_path / "synthetic-reckoning" / "workbook-gate-standard.xlsx").is_file() + assert not (tmp_path / "workbook-gate-result.json").exists() + with pytest.raises(FileNotFoundError, match="No workbook gate result"): + record_manual_signoff(tmp_path, reviewer="Fixture", outcome="passed", checks=[])