Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
98 changes: 94 additions & 4 deletions src/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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-<username>-*.json.

Expand Down Expand Up @@ -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:
Expand All @@ -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)."
)
Expand Down Expand Up @@ -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:

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Fall back when live repo data is partial

When --portfolio-truth runs without an owner-scoped token, or with a token for a different user, GitHubClient.list_repos() still succeeds via the public /users/{username}/repos path and returns only public repos rather than None. Because the audit-report fallback only runs on None, archived private repos present in the local workspace and in the last authenticated audit report get no status overlay and can remain active/decision-needed. Consider merging the audit-report overlay for repos missing from the live result, or only treating live data as authoritative when it is owner-private.

Useful? React with 👍 / 👎.

repo_status_by_name = _load_repo_status_from_audit_by_name(
output_dir=output_dir,
username=args.username,
)

try:
result = publish_portfolio_truth(
Expand All @@ -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
Expand Down
2 changes: 2 additions & 0 deletions src/portfolio_truth_publish.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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)

Expand Down
50 changes: 47 additions & 3 deletions src/portfolio_truth_reconcile.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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
]
Expand All @@ -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),
}
Expand Down Expand Up @@ -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)
Comment on lines +384 to +385

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Match archived status by full name first

When the workspace contains a checkout from another owner whose repo basename matches one of the audited user's repos, this lookup still uses only the basename/display name, so other/Alpha can inherit saagpatel/Alpha's archived flag and be forced into archived attention incorrectly. The status entries already carry full_name; key and select by the full owner/repo before falling back to basename-only matching.

Useful? React with 👍 / 👎.



def _build_truth_project(
raw_project: dict[str, Any],
*,
Expand All @@ -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)
Expand Down Expand Up @@ -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)

Expand All @@ -473,6 +504,7 @@ def _build_truth_project(
),
},
context_quality=context_quality,
archived=github_archived,
registry_status=registry_status,
)
provenance["declared.operating_path"] = {
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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"
Expand All @@ -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"
Expand Down
48 changes: 48 additions & 0 deletions tests/test_portfolio_truth.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down