diff --git a/src/cli.py b/src/cli.py index 2af0871c..338fa163 100644 --- a/src/cli.py +++ b/src/cli.py @@ -5284,6 +5284,85 @@ def _load_release_count_by_name(*, output_dir: Path, username: str) -> dict[str, return result +def _latest_audit_report_path(*, output_dir: Path, username: str) -> Path | None: + audit_files = sorted( + output_dir.glob(f"audit-report-{username}-*.json"), + key=lambda p: p.stat().st_mtime, + ) + return audit_files[-1] if audit_files else None + + +def _repo_status_entries_from_metadata( + repo_metadata: list[dict], *, source: str +) -> dict[str, dict]: + result: dict[str, dict] = {} + for metadata in repo_metadata: + name = str(metadata.get("name") or "").strip() + full_name = str(metadata.get("full_name") or "").strip() + archived = metadata.get("archived") + if not name or not isinstance(archived, bool): + continue + entry = {"archived": archived, "full_name": full_name, "source": source} + result[name] = entry + repo_name = full_name.rsplit("/", 1)[-1] if full_name else "" + if repo_name: + result.setdefault(repo_name, entry) + return result + + +def _load_live_repo_status_by_name( + *, + username: str, + token: str | None, + cache: ResponseCache | None, +) -> dict[str, dict] | None: + """Fetch current GitHub repo archived flags using the existing REST client.""" + import logging + + _log = logging.getLogger(__name__) + try: + repos = GitHubClient(token=token, cache=cache).list_repos(username) + except Exception as exc: # noqa: BLE001 + _log.warning( + "--portfolio-truth: could not fetch live GitHub repo status for %s: %s — " + "falling back to latest audit report metadata", + username, + exc, + ) + return None + return _repo_status_entries_from_metadata(repos, source="github_api") + + +def _load_repo_status_from_audit_by_name( + *, output_dir: Path, username: str +) -> dict[str, dict] | None: + """Load GitHub repo status metadata from the latest audit report JSON.""" + import logging + + _log = logging.getLogger(__name__) + audit_path = _latest_audit_report_path(output_dir=output_dir, username=username) + if audit_path is None: + return None + + try: + with audit_path.open() as fh: + data = json.load(fh) + except Exception as exc: # noqa: BLE001 + _log.warning( + "--portfolio-truth: could not read repo status overlay from %s: %s — skipping", + audit_path, + exc, + ) + return None + + repo_metadata: list[dict] = [] + for audit in data.get("audits") or []: + metadata = audit.get("metadata") or {} + if isinstance(metadata, dict): + repo_metadata.append(metadata) + return _repo_status_entries_from_metadata(repo_metadata, source="audit_report") + + def _load_security_alerts_by_name(*, output_dir: Path, username: str) -> dict[str, dict] | None: """Load per-repo GHAS alert counts from the latest output/ghas-alerts--*.json. @@ -5346,15 +5425,15 @@ def _warn_if_warehouse_report_stale(output_dir: Path, username: str) -> None: """ from datetime import date - reports = sorted(output_dir.glob(f"audit-report-{username}-*.json")) - if not reports: + report_path = _latest_audit_report_path(output_dir=output_dir, username=username) + if report_path is None: print_warning( f"No audit-report-{username}-*.json in {output_dir}: Notion's Repo Auditor " f"signal reads that warehouse report and this --portfolio-truth run did not " f"create one. Run `audit report {username}` to generate it (F2)." ) return - match = re.search(r"(\d{4}-\d{2}-\d{2})", reports[-1].name) + match = re.search(r"(\d{4}-\d{2}-\d{2})", report_path.name) if not match: return try: @@ -5364,7 +5443,7 @@ def _warn_if_warehouse_report_stale(output_dir: Path, username: str) -> None: age = (date.today() - report_date).days if age > WAREHOUSE_REPORT_STALE_DAYS: print_warning( - f"Newest warehouse report {reports[-1].name} is {age}d old: Notion's Repo " + f"Newest warehouse report {report_path.name} is {age}d old: Notion's Repo " f"Auditor signal reads it and is now stale. Run `audit report {username}` to " f"refresh the warehouse report (F2 — both artifacts kept live by decision)." ) @@ -5400,6 +5479,16 @@ def _run_portfolio_truth_mode(args) -> None: output_dir=output_dir, username=args.username, ) + repo_status_by_name = _load_live_repo_status_by_name( + username=args.username, + token=getattr(args, "token", None), + cache=None if getattr(args, "no_cache", False) else ResponseCache(), + ) + if repo_status_by_name is None: + repo_status_by_name = _load_repo_status_from_audit_by_name( + output_dir=output_dir, + username=args.username, + ) try: result = publish_portfolio_truth( @@ -5413,6 +5502,7 @@ def _run_portfolio_truth_mode(args) -> None: allow_empty_notion=getattr(args, "portfolio_truth_allow_empty_notion", False), release_count_by_name=release_count_by_name, security_alerts_by_name=security_alerts_by_name, + repo_status_by_name=repo_status_by_name, ) except PortfolioTruthPublishError as exc: raise SystemExit(str(exc)) from exc diff --git a/src/portfolio_truth_publish.py b/src/portfolio_truth_publish.py index 3be17cb7..07e915d4 100644 --- a/src/portfolio_truth_publish.py +++ b/src/portfolio_truth_publish.py @@ -88,6 +88,7 @@ def publish_portfolio_truth( allow_empty_notion: bool = False, release_count_by_name: dict[str, int] | None = None, security_alerts_by_name: dict[str, dict] | None = None, + repo_status_by_name: dict[str, dict] | None = None, ) -> PortfolioTruthPublishResult: validate_publish_targets( workspace_root=workspace_root, @@ -107,6 +108,7 @@ def publish_portfolio_truth( notion_context_fallback=notion_context_fallback, release_count_by_name=release_count_by_name, security_alerts_by_name=security_alerts_by_name, + repo_status_by_name=repo_status_by_name, ) validate_truth_snapshot(build_result.snapshot) diff --git a/src/portfolio_truth_reconcile.py b/src/portfolio_truth_reconcile.py index ef157113..9c014e8c 100644 --- a/src/portfolio_truth_reconcile.py +++ b/src/portfolio_truth_reconcile.py @@ -185,6 +185,7 @@ def build_portfolio_truth_snapshot( now: datetime | None = None, release_count_by_name: dict[str, int] | None = None, security_alerts_by_name: dict[str, dict] | None = None, + repo_status_by_name: dict[str, dict] | None = None, ) -> PortfolioTruthBuildResult: now = now or datetime.now(timezone.utc) catalog_data = load_portfolio_catalog(catalog_path) @@ -217,6 +218,7 @@ def build_portfolio_truth_snapshot( now=now, release_count_by_name=release_count_by_name, security_alerts_by_name=security_alerts_by_name, + repo_status_by_name=repo_status_by_name, ) for raw_project in workspace_projects ] @@ -241,6 +243,11 @@ def build_portfolio_truth_snapshot( "attention_state_counts": dict( Counter(project.derived.attention_state for project in projects) ), + "github_archived_count": sum( + 1 + for project in projects + if project.provenance.get("github.archived", {}).get("detail") == "true" + ), "duplicate_display_names": _duplicate_display_names(projects), "unresolved_duplicate_display_names": _unresolved_duplicate_display_names(projects), } @@ -370,6 +377,14 @@ def _select_security_entry( return lookup.get(repo_name) or lookup.get(display_name) +def _select_repo_status_entry( + lookup: dict[str, dict], repo_full_name: str | None, display_name: str +) -> dict | None: + """Join GitHub repo metadata by remote repo name, then local display name.""" + repo_name = (repo_full_name or "").rsplit("/", 1)[-1] + return lookup.get(repo_name) or lookup.get(display_name) + + def _build_truth_project( raw_project: dict[str, Any], *, @@ -379,6 +394,7 @@ def _build_truth_project( now: datetime, release_count_by_name: dict[str, int] | None = None, security_alerts_by_name: dict[str, dict] | None = None, + repo_status_by_name: dict[str, dict] | None = None, ) -> PortfolioTruthProject: relative_path = raw_project["path"] group_entry = group_entry_for_path(relative_path, catalog_data) @@ -453,9 +469,24 @@ def _build_truth_project( "detail": raw_project["context_quality"], } + status_entry = _select_repo_status_entry( + repo_status_by_name or {}, + raw_project.get("repo_full_name"), + raw_project["name"], + ) + github_archived = bool(status_entry and status_entry.get("archived") is True) + if status_entry is not None: + provenance["github.archived"] = { + "source": str(status_entry.get("source") or "audit_report"), + "detail": str(github_archived).lower(), + } + last_activity = raw_project["last_meaningful_activity_at"] activity_status = _activity_status_for( - last_activity, declared_values["lifecycle_state"], now=now + last_activity, + declared_values["lifecycle_state"], + now=now, + github_archived=github_archived, ) registry_status = _registry_status_for(activity_status) @@ -473,6 +504,7 @@ def _build_truth_project( ), }, context_quality=context_quality, + archived=github_archived, registry_status=registry_status, ) provenance["declared.operating_path"] = { @@ -525,6 +557,7 @@ def _build_truth_project( category=declared_values["category"], path_override=path_entry.get("path_override", ""), risk_entry=risk_entry, + github_archived=github_archived, ) declared = DeclaredFields( @@ -585,6 +618,10 @@ def _build_truth_project( "path_rationale", "Operating path currently requires investigate override." ) ) + if github_archived and declared_values["lifecycle_state"] != "archived": + warnings.append( + "GitHub metadata marks this repo archived/read-only; portfolio truth reconciled it as archived attention." + ) # ── Strict local-filesystem signals (Sprint 8.2) ───────────────────────── project_path: Path | None = raw_project.get("project_path") @@ -776,8 +813,9 @@ def _activity_status_for( lifecycle_state: str, *, now: datetime, + github_archived: bool = False, ) -> str: - if lifecycle_state == "archived": + if github_archived or lifecycle_state == "archived": return "archived" if last_activity is None: return "stale" @@ -804,8 +842,14 @@ def _attention_state_for( category: str, path_override: str, risk_entry: dict[str, Any], + github_archived: bool = False, ) -> str: - if registry_status == "archived" or lifecycle_state == "archived" or operating_path == "archive": + if ( + github_archived + or registry_status == "archived" + or lifecycle_state == "archived" + or operating_path == "archive" + ): return "archived" if ( operating_path == "experiment" diff --git a/tests/test_portfolio_truth.py b/tests/test_portfolio_truth.py index e487d764..c7d9e64f 100644 --- a/tests/test_portfolio_truth.py +++ b/tests/test_portfolio_truth.py @@ -472,6 +472,54 @@ def test_attention_state_classifier_separates_activity_from_operator_attention() ) +def test_github_archived_status_reconciles_to_archived_attention( + portfolio_workspace: Path, + portfolio_catalog: Path, + legacy_registry: Path, +) -> None: + now = datetime.fromtimestamp(1_700_200_000, tz=timezone.utc) + baseline = build_portfolio_truth_snapshot( + workspace_root=portfolio_workspace, + catalog_path=portfolio_catalog, + legacy_registry_path=legacy_registry, + include_notion=False, + now=now, + ) + result = build_portfolio_truth_snapshot( + workspace_root=portfolio_workspace, + catalog_path=portfolio_catalog, + legacy_registry_path=legacy_registry, + include_notion=False, + now=now, + repo_status_by_name={ + "Alpha": { + "full_name": "d/Alpha", + "archived": True, + } + }, + ) + + projects = {project.identity.display_name: project for project in result.snapshot.projects} + alpha = projects["Alpha"] + + assert alpha.declared.lifecycle_state == "active" + assert alpha.declared.operating_path == "maintain" + assert alpha.derived.path_confidence == "low" + assert alpha.derived.activity_status == "archived" + assert alpha.derived.registry_status == "archived" + assert alpha.derived.attention_state == "archived" + assert alpha.provenance["github.archived"] == { + "source": "audit_report", + "detail": "true", + } + assert result.snapshot.source_summary["github_archived_count"] == 1 + assert baseline.snapshot.source_summary["attention_state_counts"]["active-product"] == 1 + assert result.snapshot.source_summary["attention_state_counts"].get("active-product", 0) == 0 + assert result.snapshot.source_summary["attention_state_counts"].get( + "decision-needed", 0 + ) == baseline.snapshot.source_summary["attention_state_counts"].get("decision-needed", 0) + + def test_build_security_fields_maps_ghas_entry() -> None: from src.portfolio_truth_reconcile import _build_security_fields