From 71ead226c8849345c908b73821f999c91caf2abd Mon Sep 17 00:00:00 2001 From: saagpatel Date: Tue, 4 Aug 2026 08:59:36 -0700 Subject: [PATCH 1/4] fix(portfolio): bind security cohort to live identities --- config/portfolio-catalog.yaml | 8 +- src/github_security_coverage.py | 18 +- src/portfolio_truth_reconcile.py | 68 ++- src/portfolio_truth_validate.py | 5 + tests/test_github_security_coverage.py | 48 +- tests/test_portfolio_catalog.py | 27 + tests/test_portfolio_truth.py | 555 ++++++++++++++++++ .../test_portfolio_truth_contract_fixture.py | 15 + 8 files changed, 716 insertions(+), 28 deletions(-) diff --git a/config/portfolio-catalog.yaml b/config/portfolio-catalog.yaml index 0bbc7147..fe8b6dc5 100644 --- a/config/portfolio-catalog.yaml +++ b/config/portfolio-catalog.yaml @@ -971,6 +971,8 @@ repos: automation_eligible: false doctor_standard: basic notes: Lane is live end-to-end as of 2026-06-12; keep residual hardening, policy widening, and hook deployment under explicit operator review. + aliases: + - egress-guard-oss da-scaffold: owner: d lifecycle_state: archived @@ -1271,12 +1273,6 @@ repos: review_cadence: weekly operating_path: finish maturity_program: finish - egress-guard-oss: - owner: d - lifecycle_state: active - review_cadence: weekly - operating_path: finish - maturity_program: finish proof-pr: owner: d lifecycle_state: active diff --git a/src/github_security_coverage.py b/src/github_security_coverage.py index 6363051b..21456abc 100644 --- a/src/github_security_coverage.py +++ b/src/github_security_coverage.py @@ -31,7 +31,7 @@ DEFAULT_ATTENTION_STATES = frozenset( {"active-product", "active-infra", "decision-needed"} ) -DEFAULT_EXPECTED_GITHUB_COHORT_COUNT = 9 +DEFAULT_EXPECTED_GITHUB_COHORT_COUNT = 11 PROVIDER_NAMES = ("dependabot", "code_scanning", "secret_scanning") ELIGIBILITY_SOURCE = "github-account-repository-preflight-v1" ELIGIBILITY_REASON = "private_user_repo_plan_unavailable" @@ -431,19 +431,23 @@ def derive_default_attention_cohort( *, expected_count: int = DEFAULT_EXPECTED_GITHUB_COHORT_COUNT, ) -> tuple[str, ...]: - """Return the repo-backed default-attention cohort, failing on expansion.""" + """Return the repo-backed default-attention cohort, failing on size drift.""" repos: list[str] = [] for project in portfolio_truth.get("projects") or []: if not isinstance(project, dict): continue + identity = _mapping(project.get("identity")) + project_key = _text(identity.get("project_key")) + repo_full_name = _text(identity.get("repo_full_name")) + if project_key.startswith("supp:") and repo_full_name: + raise SecurityCoverageError( + "supplementary project identity cannot declare a repository: " + f"{project_key}" + ) derived = _mapping(project.get("derived")) if derived.get("attention_state") not in DEFAULT_ATTENTION_STATES: continue - identity = _mapping(project.get("identity")) - repo_full_name = identity.get("repo_full_name") - if not _text(repo_full_name) and _text(identity.get("project_key")).startswith( - "supp:" - ): + if project_key.startswith("supp:"): # Supplementary projects such as personal-ops are real portfolio # identities, but they do not have a GitHub repository to query. continue diff --git a/src/portfolio_truth_reconcile.py b/src/portfolio_truth_reconcile.py index 7f042c29..ca9abd92 100644 --- a/src/portfolio_truth_reconcile.py +++ b/src/portfolio_truth_reconcile.py @@ -8,6 +8,11 @@ from pathlib import Path from typing import Any +from src.github_security_coverage import ( + DEFAULT_ATTENTION_STATES, + SecurityCoverageError, + derive_default_attention_cohort, +) from src.portfolio_catalog import ( catalog_entry_for_repo, group_entry_for_path, @@ -188,6 +193,54 @@ class PortfolioTruthBuildResult: legacy_rows: dict[str, dict[str, str]] +def _validate_security_receipt_cohort_identity( + *, + projects: list[PortfolioTruthProject], + security_alerts_by_name: dict[str, dict], +) -> None: + """Require receipt membership to match freshly derived default attention.""" + receipt_repositories = tuple(sorted(security_alerts_by_name, key=str.lower)) + try: + derived_repositories = derive_default_attention_cohort( + { + "projects": [ + { + "identity": { + "project_key": project.identity.project_key, + "repo_full_name": project.identity.repo_full_name, + }, + "derived": { + "attention_state": project.derived.attention_state, + }, + } + for project in projects + ] + }, + expected_count=len(receipt_repositories), + ) + except SecurityCoverageError as exc: + raise ValueError( + "PortfolioTruth GitHub security receipt cohort cannot match freshly " + f"derived default attention: {exc}." + ) from exc + + if receipt_repositories == derived_repositories: + return + + receipt_only = sorted( + set(receipt_repositories) - set(derived_repositories), + key=str.lower, + ) + derived_only = sorted( + set(derived_repositories) - set(receipt_repositories), + key=str.lower, + ) + raise ValueError( + "PortfolioTruth GitHub security receipt cohort differs from freshly derived " + f"default attention: receipt_only={receipt_only}; derived_only={derived_only}." + ) + + def build_portfolio_truth_snapshot( *, workspace_root: Path, @@ -253,6 +306,11 @@ def build_portfolio_truth_snapshot( ) for raw_project in workspace_projects ] + if security_coverage_metadata is not None and security_alerts_by_name is not None: + _validate_security_receipt_cohort_identity( + projects=projects, + security_alerts_by_name=security_alerts_by_name, + ) projects.sort( key=lambda item: ( item.identity.section_marker.lower(), @@ -774,11 +832,11 @@ def _build_truth_project( security_high_alerts=security.dependabot_high or 0, security_critical_alerts=security.dependabot_critical or 0, ) - if not security.receipt_schema_version and attention_state in { - "active-product", - "active-infra", - "decision-needed", - } and not identity.project_key.startswith("supp:"): + if ( + not security.receipt_schema_version + and attention_state in DEFAULT_ATTENTION_STATES + and not identity.project_key.startswith("supp:") + ): security = replace( security, cohort_member=True, diff --git a/src/portfolio_truth_validate.py b/src/portfolio_truth_validate.py index e08b685f..779ab915 100644 --- a/src/portfolio_truth_validate.py +++ b/src/portfolio_truth_validate.py @@ -108,6 +108,11 @@ def validate_truth_snapshot( raise ValueError( f"Project key for {key} must exactly match its identity path." ) + if key.startswith("supp:") and project.identity.repo_full_name: + raise ValueError( + "PortfolioTruth supplementary project identity cannot declare " + f"a repository: {key}." + ) required_identity = ( project.identity.project_key, project.identity.display_name, diff --git a/tests/test_github_security_coverage.py b/tests/test_github_security_coverage.py index 2afb58de..61f40f8d 100644 --- a/tests/test_github_security_coverage.py +++ b/tests/test_github_security_coverage.py @@ -759,8 +759,8 @@ def test_default_attention_cohort_is_exact_and_fail_closed() -> None: assert len(cohort) == DEFAULT_EXPECTED_GITHUB_COHORT_COUNT assert "owner/parked" not in cohort - with pytest.raises(SecurityCoverageError, match="expected 9, observed 10"): - derive_default_attention_cohort(_truth(10)) + with pytest.raises(SecurityCoverageError, match="expected 11, observed 12"): + derive_default_attention_cohort(_truth(12)) def test_repo_less_non_supplementary_attention_identity_fails_closed() -> None: @@ -778,6 +778,28 @@ def test_repo_less_non_supplementary_attention_identity_fails_closed() -> None: derive_default_attention_cohort(truth) +@pytest.mark.parametrize("attention_state", ("active-infra", "parked")) +def test_repo_backed_supplementary_identity_fails_closed_regardless_attention( + attention_state: str, +) -> None: + truth = _truth(DEFAULT_EXPECTED_GITHUB_COHORT_COUNT) + truth["projects"].append( + { + "identity": { + "project_key": "supp:repo-backed", + "repo_full_name": "owner/repo-backed", + }, + "derived": {"attention_state": attention_state}, + } + ) + + with pytest.raises( + SecurityCoverageError, + match="supplementary project identity cannot declare a repository", + ): + derive_default_attention_cohort(truth) + + def test_no_token_writes_exact_fail_closed_outcomes_without_network() -> None: session = _Session() @@ -828,13 +850,17 @@ def test_valid_prior_for_old_cohort_is_ignored_during_contraction() -> None: cohort_count=DEFAULT_EXPECTED_GITHUB_COHORT_COUNT, ) - assert receipt["cohort"]["repository_count"] == 9 - assert len(session.calls) == 28 + provider_request_count = DEFAULT_EXPECTED_GITHUB_COHORT_COUNT * 3 + assert ( + receipt["cohort"]["repository_count"] + == DEFAULT_EXPECTED_GITHUB_COHORT_COUNT + ) + assert len(session.calls) == provider_request_count + 1 assert all( kwargs.get("headers") == {} - for _, kwargs in session.calls[:27] + for _, kwargs in session.calls[:provider_request_count] ) - assert "headers" not in session.calls[27][1] + assert "headers" not in session.calls[provider_request_count][1] def test_invalid_prior_for_old_cohort_still_fails_closed() -> None: @@ -902,10 +928,11 @@ def test_collector_is_serial_count_only_and_bounded_to_48_base_requests() -> Non ) -def test_current_nine_repository_cut_binds_remote_branch_and_head() -> None: +def test_current_eleven_repository_cut_binds_remote_branch_and_head() -> None: + provider_request_count = DEFAULT_EXPECTED_GITHUB_COHORT_COUNT * 3 session = _Session( [ - *[_Response() for _ in range(27)], + *[_Response() for _ in range(provider_request_count)], _remote_graphql_response(DEFAULT_EXPECTED_GITHUB_COHORT_COUNT), ] ) @@ -914,7 +941,7 @@ def test_current_nine_repository_cut_binds_remote_branch_and_head() -> None: cohort_count=DEFAULT_EXPECTED_GITHUB_COHORT_COUNT, ) - assert len(session.calls) == 28 + assert len(session.calls) == provider_request_count + 1 assert receipt["request_budget"]["stop_reason"] is None assert all( repository["repository"]["state"] == "observed" @@ -1094,9 +1121,10 @@ def test_shared_upstream_validator_matches_git_ref_format(upstream: str) -> None def test_graphql_rate_limit_marks_remote_cut_with_exact_reason_code() -> None: outcome = OUTCOME_FIXTURES["rate_limited"] + provider_request_count = DEFAULT_EXPECTED_GITHUB_COHORT_COUNT * 3 session = _Session( [ - *[_Response() for _ in range(27)], + *[_Response() for _ in range(provider_request_count)], _Response( 200, { diff --git a/tests/test_portfolio_catalog.py b/tests/test_portfolio_catalog.py index fda25502..b71a37ef 100644 --- a/tests/test_portfolio_catalog.py +++ b/tests/test_portfolio_catalog.py @@ -255,6 +255,33 @@ def test_live_catalog_matches_operator_attention_reconciliation() -> None: assert catalog["repos"]["gpt_rag"]["lifecycle_state"] == "dormant" +def test_live_catalog_resolves_egress_alias_to_canonical_manual_only_entry() -> None: + catalog_path = Path(__file__).parents[1] / "config" / "portfolio-catalog.yaml" + catalog = load_portfolio_catalog(catalog_path) + + assert catalog["errors"] == [] + assert catalog["warnings"] == [] + assert catalog["repos"]["egress-guard-oss"]["catalog_key"] == ( + "cross-provider-egress-guard" + ) + + entry = catalog_entry_for_repo( + { + "name": "egress-guard-oss", + "full_name": "saagpatel/cross-provider-egress-guard", + "path": "egress-guard-oss", + }, + catalog, + ) + + assert entry["catalog_key"] == "cross-provider-egress-guard" + assert entry["matched_by"] == "path" + assert entry["lifecycle_state"] == "manual-only" + assert entry["operating_path"] == "maintain" + assert entry["category"] == "infrastructure" + assert entry["maturity_program"] == "maintain" + + def test_catalog_entry_matches_full_name_then_bare_name(): catalog = { "repos": { diff --git a/tests/test_portfolio_truth.py b/tests/test_portfolio_truth.py index ad5a0465..6510b813 100644 --- a/tests/test_portfolio_truth.py +++ b/tests/test_portfolio_truth.py @@ -17,6 +17,7 @@ _provider_result, _remote_repository_result, collect_security_coverage, + derive_default_attention_cohort, load_security_coverage_receipt, write_security_coverage_receipt, ) @@ -704,6 +705,90 @@ def test_live_catalog_produces_exact_tier_zero_attention_semantics( ) +def test_live_catalog_resolves_current_eleven_repo_cohort_and_egress_alias( + tmp_path: Path, +) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + now = datetime(2026, 8, 4, 12, tzinfo=timezone.utc) + expected_repositories = { + "agent-permission-diff-bot": "saagpatel/agent-permission-diff-bot", + "AIGCCore": "saagpatel/AIGCCore", + "bridge-db": "saagpatel/bridge-db", + "GithubRepoAuditor": "saagpatel/GithubRepoAuditor", + "mcp-trust": "saagpatel/mcp-trust", + "MCPAudit": "saagpatel/MCPAudit", + "operant-public": "saagpatel/operant", + "operator-os-explainer": "saagpatel/operator-os-explainer", + "portfolio-index": "saagpatel/portfolio-index", + "PortfolioCommandCenter": "saagpatel/PortfolioCommandCenter", + "proof-pr": "saagpatel/proof-pr", + } + excluded_egress_repositories = { + "cross-provider-egress-guard": ( + "saagpatel/cross-provider-egress-guard-private" + ), + "egress-guard-oss": "saagpatel/cross-provider-egress-guard", + } + + for name, remote in { + **expected_repositories, + **excluded_egress_repositories, + }.items(): + project = workspace / name + project.mkdir() + readme = project / "README.md" + _write(readme, f"# {name}\n\nCurrent cohort fixture.\n") + observed_at = ( + now - timedelta(days=31) + if name in {"agent-permission-diff-bot", "proof-pr"} + or name in excluded_egress_repositories + else now + ) + _set_mtime(readme, observed_at.timestamp()) + subprocess.run( + ["git", "init"], + cwd=project, + capture_output=True, + check=True, + ) + subprocess.run( + [ + "git", + "remote", + "add", + "origin", + f"https://github.com/{remote}.git", + ], + cwd=project, + capture_output=True, + check=True, + ) + + result = build_portfolio_truth_snapshot( + workspace_root=workspace, + catalog_path=Path(__file__).parents[1] / "config" / "portfolio-catalog.yaml", + include_notion=False, + now=now, + ) + by_display_name = { + project.identity.display_name: project for project in result.snapshot.projects + } + + for name in excluded_egress_repositories: + assert by_display_name[name].derived.attention_state == "manual-only" + assert ( + by_display_name["egress-guard-oss"] + .provenance["declared.lifecycle_state"]["detail"] + == "cross-provider-egress-guard" + ) + for name in ("agent-permission-diff-bot", "proof-pr"): + assert by_display_name[name].derived.attention_state == "decision-needed" + assert derive_default_attention_cohort( + result.snapshot.to_dict(), expected_count=11 + ) == tuple(sorted(expected_repositories.values(), key=str.lower)) + + def test_discovered_personal_ops_replaces_supplementary_registry_identity( tmp_path: Path, ) -> None: @@ -827,6 +912,41 @@ def test_truth_snapshot_operating_path_catalog_field_matches_legacy_disposition( ) +def test_finish_attention_flips_exactly_at_31_day_activity_boundary() -> None: + from zoneinfo import ZoneInfo + + from src.portfolio_truth_decisions import derive_attention_state + from src.portfolio_truth_reconcile import _activity_status_for + + last_activity = datetime(2026, 7, 4, 7, 44, 49, tzinfo=timezone.utc) + before_boundary = last_activity + timedelta(days=31) - timedelta(microseconds=1) + at_boundary = last_activity + timedelta(days=31) + + assert _activity_status_for(last_activity, now=before_boundary) == "recent" + assert _activity_status_for(last_activity, now=at_boundary) == "stale" + assert ( + _activity_status_for( + last_activity.astimezone(ZoneInfo("America/Los_Angeles")), + now=at_boundary.astimezone(ZoneInfo("America/Los_Angeles")), + ) + == "stale" + ) + + def attention_at(now: datetime) -> str: + return derive_attention_state( + activity_status=_activity_status_for(last_activity, now=now), + archived=False, + lifecycle_state="active", + operating_path="finish", + category="vanity", + path_override="", + risk_entry={"security_risk": False}, + ) + + assert attention_at(before_boundary) == "manual-only" + assert attention_at(at_boundary) == "decision-needed" + + def test_attention_state_classifier_separates_activity_from_operator_attention() -> ( None ): @@ -1567,6 +1687,425 @@ def test_bound_security_identity_and_high_findings_reach_decision_queue( validate_truth_snapshot(result.snapshot) +def test_security_receipt_rejects_same_count_identity_rollover(tmp_path: Path) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + catalog_path = tmp_path / "portfolio-catalog.yaml" + catalog_path.write_text( + """ +repos: + Old: + owner: d + lifecycle_state: manual-only + review_cadence: weekly + operating_path: maintain + category: infrastructure + New: + owner: d + lifecycle_state: active + review_cadence: weekly + operating_path: finish + category: vanity +""" + ) + now = datetime(2026, 8, 4, 12, tzinfo=timezone.utc) + for name in ("Old", "New"): + project = workspace / name + project.mkdir() + readme = project / "README.md" + _write(readme, f"# {name}\n\nCohort rollover fixture.\n") + _set_mtime(readme, (now - timedelta(days=31)).timestamp()) + subprocess.run( + ["git", "init"], cwd=project, capture_output=True, check=True + ) + subprocess.run( + [ + "git", + "remote", + "add", + "origin", + f"https://github.com/d/{name}.git", + ], + cwd=project, + capture_output=True, + check=True, + ) + + observed_at = now.isoformat() + security = { + "d/Old": { + "repo_full_name": "d/Old", + "cohort_member": True, + "cohort_policy": "portfolio-default-attention-v1", + "receipt_schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "receipt_state": "fresh", + "source_produced_at": observed_at, + "providers": {}, + } + } + metadata = { + "source_id": "github-security-coverage-receipt", + "schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "produced_at": observed_at, + "state": "fresh", + "age_hours": 0.0, + "producer_commit": "a" * 40, + "cohort_policy": "portfolio-default-attention-v1", + "cohort_repository_count": 1, + "path": "/evidence/github-security-coverage-latest.json", + } + + with pytest.raises( + ValueError, + match=( + "receipt cohort differs from freshly derived default attention: " + "receipt_only=\\['d/Old'\\]; derived_only=\\['d/New'\\]" + ), + ): + build_portfolio_truth_snapshot( + workspace_root=workspace, + catalog_path=catalog_path, + include_notion=False, + now=now, + security_alerts_by_name=security, + security_coverage_metadata=metadata, + ) + + +def test_security_cohort_identity_skips_repo_less_supplementary() -> None: + from types import SimpleNamespace + + from src.portfolio_truth_reconcile import ( + _validate_security_receipt_cohort_identity, + ) + + projects = [ + SimpleNamespace( + identity=SimpleNamespace( + project_key="alpha", + repo_full_name="d/Alpha", + ), + derived=SimpleNamespace(attention_state="active-infra"), + ), + SimpleNamespace( + identity=SimpleNamespace( + project_key="supp:repo-less", + repo_full_name="", + ), + derived=SimpleNamespace(attention_state="active-infra"), + ), + ] + + _validate_security_receipt_cohort_identity( + projects=projects, + security_alerts_by_name={"d/Alpha": {}}, + ) + + +def test_security_cohort_identity_rejects_repo_backed_supplementary() -> None: + from types import SimpleNamespace + + from src.portfolio_truth_reconcile import ( + _validate_security_receipt_cohort_identity, + ) + + project = SimpleNamespace( + identity=SimpleNamespace( + project_key="supp:repo-backed", + repo_full_name="d/Supp", + ), + derived=SimpleNamespace(attention_state="active-infra"), + ) + + with pytest.raises( + ValueError, + match="supplementary project identity cannot declare a repository", + ): + _validate_security_receipt_cohort_identity( + projects=[project], + security_alerts_by_name={"d/Supp": {}}, + ) + + +def test_empty_security_metadata_cannot_bypass_cohort_identity( + tmp_path: Path, +) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + catalog_path = tmp_path / "portfolio-catalog.yaml" + catalog_path.write_text("repos: {}\n") + + with pytest.raises(ValueError, match="expected 1, observed 0"): + build_portfolio_truth_snapshot( + workspace_root=workspace, + catalog_path=catalog_path, + include_notion=False, + now=datetime(2026, 8, 4, 12, tzinfo=timezone.utc), + security_alerts_by_name={"d/Stale": {}}, + security_coverage_metadata={}, + ) + + +def test_receipt_backed_snapshot_excludes_repo_less_supplementary_from_cohort( + tmp_path: Path, +) -> None: + workspace = tmp_path / "workspace" + alpha = workspace / "Alpha" + alpha.mkdir(parents=True) + now = datetime(2026, 8, 4, 12, tzinfo=timezone.utc) + _write(alpha / "README.md", "# Alpha\n\nReceipt-backed cohort fixture.\n") + _set_mtime(alpha / "README.md", now.timestamp()) + subprocess.run(["git", "init"], cwd=alpha, capture_output=True, check=True) + subprocess.run( + [ + "git", + "remote", + "add", + "origin", + "https://github.com/d/Alpha.git", + ], + cwd=alpha, + capture_output=True, + check=True, + ) + catalog_path = tmp_path / "portfolio-catalog.yaml" + catalog_path.write_text( + """ +repos: + Alpha: + owner: d + lifecycle_state: active + review_cadence: weekly + operating_path: maintain + category: infrastructure + personal-ops: + owner: d + lifecycle_state: active + review_cadence: weekly + operating_path: maintain + category: infrastructure +""" + ) + observed_at = now.isoformat() + zero_counts = { + "dependabot": {"critical": 0, "high": 0, "medium": 0, "low": 0}, + "code_scanning": {"critical": 0, "high": 0, "warning": 0, "note": 0}, + "secret_scanning": {"open": 0}, + } + security = { + "d/Alpha": { + "repo_full_name": "d/Alpha", + "cohort_member": True, + "cohort_policy": "portfolio-default-attention-v1", + "receipt_schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "receipt_state": "fresh", + "source_produced_at": observed_at, + "repository": _remote_repository_result( + state="observed", + observed_at=observed_at, + default_branch="main", + head_sha="b" * 40, + archived=False, + ), + "providers": { + provider: _provider_result( + provider, + state="observed", + observed_at=observed_at, + http_status=200, + pagination_complete=True, + counts=counts, + ) + for provider, counts in zero_counts.items() + }, + } + } + metadata = { + "source_id": "github-security-coverage-receipt", + "schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "produced_at": observed_at, + "state": "fresh", + "age_hours": 0.0, + "producer_commit": "a" * 40, + "cohort_policy": "portfolio-default-attention-v1", + "cohort_repository_count": 1, + "path": "/evidence/github-security-coverage-latest.json", + } + + result = build_portfolio_truth_snapshot( + workspace_root=workspace, + catalog_path=catalog_path, + include_notion=False, + now=now, + security_alerts_by_name=security, + security_coverage_metadata=metadata, + ) + validate_truth_snapshot(result.snapshot) + projects = { + project.identity.display_name: project for project in result.snapshot.projects + } + + assert projects["Alpha"].security.cohort_member is True + assert projects["personal-ops"].identity.project_key == "supp:personal-ops" + assert projects["personal-ops"].identity.repo_full_name == "" + assert projects["personal-ops"].security.cohort_member is False + assert result.snapshot.rollups.security["cohort_repository_count"] == 1 + + +def test_security_cohort_identity_rejects_case_only_drift() -> None: + from types import SimpleNamespace + + from src.portfolio_truth_reconcile import ( + _validate_security_receipt_cohort_identity, + ) + + project = SimpleNamespace( + identity=SimpleNamespace(project_key="alpha", repo_full_name="d/Alpha"), + derived=SimpleNamespace(attention_state="active-product"), + ) + + with pytest.raises( + ValueError, + match=( + "receipt_only=\\['D/Alpha'\\]; derived_only=\\['d/Alpha'\\]" + ), + ): + _validate_security_receipt_cohort_identity( + projects=[project], + security_alerts_by_name={"D/Alpha": {}}, + ) + + +@pytest.mark.parametrize( + ("receipt_repositories", "derived_repositories", "expected_message"), + ( + (("d/Alpha",), ("d/Alpha", "d/Beta"), "expected 1, observed 2"), + (("d/Alpha", "d/Beta"), ("d/Alpha",), "expected 2, observed 1"), + ), +) +def test_security_cohort_identity_rejects_expansion_and_contraction( + receipt_repositories: tuple[str, ...], + derived_repositories: tuple[str, ...], + expected_message: str, +) -> None: + from types import SimpleNamespace + + from src.portfolio_truth_reconcile import ( + _validate_security_receipt_cohort_identity, + ) + + projects = [ + SimpleNamespace( + identity=SimpleNamespace( + project_key=repository.rsplit("/", 1)[-1], + repo_full_name=repository, + ), + derived=SimpleNamespace(attention_state="active-product"), + ) + for repository in derived_repositories + ] + + with pytest.raises(ValueError, match=expected_message): + _validate_security_receipt_cohort_identity( + projects=projects, + security_alerts_by_name={ + repository: {} for repository in receipt_repositories + }, + ) + + +def test_security_cohort_identity_rejects_missing_repository_name() -> None: + from types import SimpleNamespace + + from src.portfolio_truth_reconcile import ( + _validate_security_receipt_cohort_identity, + ) + + project = SimpleNamespace( + identity=SimpleNamespace(project_key="missing", repo_full_name=""), + derived=SimpleNamespace(attention_state="active-infra"), + ) + + with pytest.raises(ValueError, match="invalid canonical repository name"): + _validate_security_receipt_cohort_identity( + projects=[project], + security_alerts_by_name={}, + ) + + +def test_security_cohort_identity_accepts_receipt_risk_driven_attention( + tmp_path: Path, +) -> None: + workspace = tmp_path / "workspace" + project = workspace / "Manual" + project.mkdir(parents=True) + _write(project / "README.md", "# Manual\n\nRisk feedback fixture.\n") + subprocess.run(["git", "init"], cwd=project, capture_output=True, check=True) + subprocess.run( + ["git", "remote", "add", "origin", "https://github.com/d/Manual.git"], + cwd=project, + capture_output=True, + check=True, + ) + catalog_path = tmp_path / "portfolio-catalog.yaml" + catalog_path.write_text( + """ +repos: + Manual: + owner: d + lifecycle_state: manual-only + review_cadence: weekly + operating_path: maintain + category: infrastructure +""" + ) + now = datetime(2026, 8, 4, 12, tzinfo=timezone.utc) + observed_at = now.isoformat() + security = { + "d/Manual": { + "repo_full_name": "d/Manual", + "cohort_member": True, + "cohort_policy": "portfolio-default-attention-v1", + "receipt_schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "receipt_state": "fresh", + "source_produced_at": observed_at, + "providers": { + "dependabot": { + "state": "observed", + "observed_at": observed_at, + "pagination_complete": True, + "counts": {"critical": 0, "high": 1, "medium": 0, "low": 0}, + } + }, + } + } + metadata = { + "source_id": "github-security-coverage-receipt", + "schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "produced_at": observed_at, + "state": "fresh", + "age_hours": 0.0, + "producer_commit": "a" * 40, + "cohort_policy": "portfolio-default-attention-v1", + "cohort_repository_count": 1, + "path": "/evidence/github-security-coverage-latest.json", + } + + result = build_portfolio_truth_snapshot( + workspace_root=workspace, + catalog_path=catalog_path, + include_notion=False, + now=now, + security_alerts_by_name=security, + security_coverage_metadata=metadata, + ) + manual = result.snapshot.projects[0] + + assert manual.declared.lifecycle_state == "manual-only" + assert manual.security.dependabot_high == 1 + assert manual.derived.attention_state == "decision-needed" + + def test_security_overlay_absent_leaves_repos_unscanned( portfolio_workspace: Path, portfolio_catalog: Path, @@ -2733,6 +3272,14 @@ def test_publish_refuses_receipt_pointer_replacement_after_load( monkeypatch: pytest.MonkeyPatch, ) -> None: now = datetime.now(timezone.utc).replace(microsecond=0) + alpha_path = portfolio_workspace / "Alpha" + subprocess.run(["git", "init"], cwd=alpha_path, capture_output=True, check=True) + subprocess.run( + ["git", "remote", "add", "origin", "https://github.com/d/Alpha.git"], + cwd=alpha_path, + capture_output=True, + check=True, + ) output_dir = tmp_path / "output" output_dir.mkdir() receipt_path = output_dir / "github-security-coverage-latest.json" @@ -2842,6 +3389,14 @@ def test_publish_refuses_nested_evidence_that_expires_after_snapshot( from contextlib import contextmanager now = datetime.now(timezone.utc).replace(microsecond=0) + alpha_path = portfolio_workspace / "Alpha" + subprocess.run(["git", "init"], cwd=alpha_path, capture_output=True, check=True) + subprocess.run( + ["git", "remote", "add", "origin", "https://github.com/d/Alpha.git"], + cwd=alpha_path, + capture_output=True, + check=True, + ) nested_observed_at = now - timedelta(hours=24) + timedelta(milliseconds=500) output_dir = tmp_path / "output" output_dir.mkdir() diff --git a/tests/test_portfolio_truth_contract_fixture.py b/tests/test_portfolio_truth_contract_fixture.py index 811f6663..afae89a9 100644 --- a/tests/test_portfolio_truth_contract_fixture.py +++ b/tests/test_portfolio_truth_contract_fixture.py @@ -757,6 +757,20 @@ def test_contract_rejects_forged_supplementary_project_key() -> None: validate_truth_snapshot_payload(fixture) +def test_contract_rejects_repo_backed_supplementary_identity() -> None: + fixture = build_contract_fixture() + identity = fixture["projects"][0]["identity"] + identity["project_key"] = "supp:repo-backed" + identity["path"] = "supp:repo-backed" + identity["repo_full_name"] = "d/repo-backed" + + with pytest.raises( + ValueError, + match="supplementary project identity cannot declare a repository", + ): + validate_truth_snapshot_payload(fixture) + + @pytest.mark.parametrize( ("field", "value", "message"), ( @@ -850,6 +864,7 @@ def _append_hidden_duplicate(fixture: dict[str, object]) -> None: duplicate = deepcopy(projects[0]) duplicate["identity"]["project_key"] = "supp:duplicate-hidden" duplicate["identity"]["path"] = "supp:duplicate-hidden" + duplicate["identity"]["repo_full_name"] = "" projects.append(duplicate) From 39f1dd8728877170d414c0ceb0c00610d0cb0246 Mon Sep 17 00:00:00 2001 From: saagpatel Date: Tue, 4 Aug 2026 09:53:19 -0700 Subject: [PATCH 2/4] fix(portfolio): derive receipt cohort before fresh risk --- src/portfolio_truth_publish.py | 261 ++++++++++++- src/portfolio_truth_reconcile.py | 165 ++++++-- tests/test_portfolio_truth.py | 642 ++++++++++++++++++++++++++++++- 3 files changed, 1032 insertions(+), 36 deletions(-) diff --git a/src/portfolio_truth_publish.py b/src/portfolio_truth_publish.py index 036a996d..1e96e9cb 100644 --- a/src/portfolio_truth_publish.py +++ b/src/portfolio_truth_publish.py @@ -1,11 +1,16 @@ from __future__ import annotations +import fcntl +import hashlib import json +import os import tempfile -from contextlib import nullcontext +import threading +from contextlib import contextmanager, nullcontext from dataclasses import dataclass from datetime import datetime from pathlib import Path +from typing import Iterator from src.github_security_coverage import ( SecurityCoverageError, @@ -21,6 +26,7 @@ from src.portfolio_truth_types import truth_latest_path from src.producer_preflight import ProducerEvidence, verify_evidence_still_current from src.portfolio_truth_validate import ( + canonicalize_truth_snapshot_payload, validate_portfolio_report_markdown, validate_publish_targets, validate_registry_markdown, @@ -29,6 +35,9 @@ from src.project_registry import build_project_registry, load_source_paths +_PORTFOLIO_TRUTH_IN_PROCESS_LOCK = threading.Lock() + + @dataclass(frozen=True) class PortfolioTruthPublishResult: snapshot_path: Path @@ -41,10 +50,163 @@ class PortfolioTruthPublishResult: project_registry_path: Path | None = None +@dataclass(frozen=True) +class _PriorSecurityEvidence: + path: Path + content_sha256: str | None + alerts_by_full_name: dict[str, dict] + + class PortfolioTruthPublishError(RuntimeError): """Raised when publishing would corrupt or misrepresent portfolio truth.""" +def _parse_bound_datetime(value: object, *, field: str) -> datetime: + if not isinstance(value, str) or not value.strip(): + raise PortfolioTruthPublishError(f"{field} is required.") + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise PortfolioTruthPublishError(f"{field} is invalid.") from exc + if parsed.tzinfo is None: + raise PortfolioTruthPublishError(f"{field} must include a timezone.") + return parsed + + +def _load_prior_security_alerts( + latest_path: Path, + *, + current_security_metadata: dict[str, object], + security_max_age_hours: int, +) -> _PriorSecurityEvidence: + """Load validated prior receipt evidence for independent cohort derivation.""" + try: + content = latest_path.read_bytes() + except FileNotFoundError: + return _PriorSecurityEvidence( + path=latest_path, + content_sha256=None, + alerts_by_full_name={}, + ) + except OSError as exc: + raise PortfolioTruthPublishError( + f"Prior PortfolioTruth cannot authorize security cohort derivation: {exc}" + ) from exc + + try: + payload = json.loads(content) + if not isinstance(payload, dict): + raise ValueError("snapshot must be an object") + canonical = canonicalize_truth_snapshot_payload( + payload, + security_max_age_hours=security_max_age_hours, + ) + except (json.JSONDecodeError, ValueError) as exc: + raise PortfolioTruthPublishError( + f"Prior PortfolioTruth cannot authorize security cohort derivation: {exc}" + ) from exc + + projects = canonical.get("projects") or [] + receipt_projects = [ + project + for project in projects + if (project.get("security") or {}).get("receipt_schema_version") + ] + if receipt_projects: + github_security = (canonical.get("inputs") or {}).get("github_security") or {} + if not github_security.get("receipt_id") or not github_security.get( + "content_sha256" + ): + raise PortfolioTruthPublishError( + "Prior PortfolioTruth security evidence is not immutably bound." + ) + + prior_generated_at = _parse_bound_datetime( + canonical.get("generated_at"), + field="Prior PortfolioTruth generated_at", + ) + current_produced_at = _parse_bound_datetime( + current_security_metadata.get("produced_at"), + field="Current security receipt produced_at", + ) + if prior_generated_at > current_produced_at: + raise PortfolioTruthPublishError( + "Prior PortfolioTruth was generated after the current security receipt." + ) + + alerts: dict[str, dict] = {} + for project in receipt_projects: + identity = project.get("identity") or {} + security = project.get("security") or {} + repository = str(identity.get("repo_full_name") or "").strip() + if not repository or security.get("cohort_member") is not True: + continue + if repository in alerts: + raise PortfolioTruthPublishError( + "Prior PortfolioTruth security cohort contains duplicate repository " + f"identity: {repository}." + ) + remote = (project.get("repository_state") or {}).get( + "remote_default_branch" + ) or {} + alerts[repository] = { + **security, + "repo_full_name": repository, + "repository": remote, + } + return _PriorSecurityEvidence( + path=latest_path, + content_sha256=hashlib.sha256(content).hexdigest(), + alerts_by_full_name=alerts, + ) + + +def _verify_prior_security_evidence_current( + evidence: _PriorSecurityEvidence, +) -> None: + """Fail closed if the canonical truth pointer changed after candidate derivation.""" + try: + content = evidence.path.read_bytes() + except FileNotFoundError: + if evidence.content_sha256 is None: + return + raise PortfolioTruthPublishError( + "Prior PortfolioTruth disappeared after it authorized security cohort " + "derivation." + ) from None + except OSError as exc: + raise PortfolioTruthPublishError( + "Prior PortfolioTruth could not be revalidated before publication: " + f"{exc}" + ) from exc + + observed_sha256 = hashlib.sha256(content).hexdigest() + if evidence.content_sha256 is None or observed_sha256 != evidence.content_sha256: + raise PortfolioTruthPublishError( + "Prior PortfolioTruth changed after it authorized security cohort " + "derivation." + ) + + +@contextmanager +def _portfolio_truth_publication_lock(latest_path: Path) -> Iterator[None]: + """Serialize complete truth builds and replacement across local publishers.""" + lock_path = latest_path.with_name(f".{latest_path.name}.lock") + with _PORTFOLIO_TRUTH_IN_PROCESS_LOCK: + try: + descriptor = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600) + except OSError as exc: + raise PortfolioTruthPublishError( + f"PortfolioTruth publication lock is unavailable: {lock_path}: {exc}" + ) from exc + try: + fcntl.flock(descriptor, fcntl.LOCK_EX) + yield + finally: + fcntl.flock(descriptor, fcntl.LOCK_UN) + os.close(descriptor) + + _REPO_ROOT = Path(__file__).resolve().parents[1] _CONFIG_DIR = _REPO_ROOT / "config" @@ -141,10 +303,94 @@ def publish_portfolio_truth( portfolio_report_output=portfolio_report_output, ) latest_path = truth_latest_path(output_dir) + with _portfolio_truth_publication_lock(latest_path): + return _publish_portfolio_truth_locked( + workspace_root=workspace_root, + output_dir=output_dir, + registry_output=registry_output, + portfolio_report_output=portfolio_report_output, + catalog_path=catalog_path, + legacy_registry_path=legacy_registry_path, + include_notion=include_notion, + allow_empty_notion=allow_empty_notion, + release_count_by_name=release_count_by_name, + security_alerts_by_name=security_alerts_by_name, + security_coverage_metadata=security_coverage_metadata, + security_receipt_binding=security_receipt_binding, + repo_status_by_name=repo_status_by_name, + producer_evidence=producer_evidence, + producer_repo_root=producer_repo_root, + require_producer_evidence=require_producer_evidence, + now=now, + ) + + +def _publish_portfolio_truth_locked( + *, + workspace_root: Path, + output_dir: Path, + registry_output: Path, + portfolio_report_output: Path, + catalog_path: Path | None = None, + legacy_registry_path: Path | None = None, + include_notion: bool = True, + allow_empty_notion: bool = False, + release_count_by_name: dict[str, int] | None = None, + security_alerts_by_name: dict[str, dict] | None = None, + security_coverage_metadata: dict[str, object] | None = None, + security_receipt_binding: SecurityCoverageReceiptBinding | None = None, + repo_status_by_name: dict[str, dict] | None = None, + producer_evidence: ProducerEvidence | None = None, + producer_repo_root: Path | None = None, + require_producer_evidence: bool = False, + now: datetime | None = None, +) -> PortfolioTruthPublishResult: + if ( + security_coverage_metadata is not None + or security_receipt_binding is not None + ) and now is None: + raise PortfolioTruthPublishError( + "Receipt-backed security publication requires an explicit evaluation clock." + ) + if require_producer_evidence and producer_evidence is None: + raise PortfolioTruthPublishError( + "Canonical publication requires validated producer evidence." + ) + _validate_security_receipt_binding( + metadata=security_coverage_metadata, + binding=security_receipt_binding, + required=require_producer_evidence + and ( + security_alerts_by_name is not None + or security_coverage_metadata is not None + ), + ) + validate_publish_targets( + workspace_root=workspace_root, + output_dir=output_dir, + registry_output=registry_output, + portfolio_report_output=portfolio_report_output, + ) + security_max_age_hours = ( + security_receipt_binding.max_age_hours + if security_receipt_binding is not None + else 24 + ) + latest_path = truth_latest_path(output_dir) notion_context_fallback = ( load_prior_notion_context(latest_path) if allow_empty_notion else None ) prior_notion_generated_at = resolve_notion_origin(latest_path) + prior_security_evidence = ( + _load_prior_security_alerts( + latest_path, + current_security_metadata=security_coverage_metadata, + security_max_age_hours=security_max_age_hours, + ) + if security_coverage_metadata is not None + and security_alerts_by_name is not None + else None + ) build_result = build_portfolio_truth_snapshot( workspace_root=workspace_root, catalog_path=catalog_path, @@ -154,6 +400,11 @@ def publish_portfolio_truth( release_count_by_name=release_count_by_name, security_alerts_by_name=security_alerts_by_name, security_coverage_metadata=security_coverage_metadata, + prior_security_alerts_by_name=( + prior_security_evidence.alerts_by_full_name + if prior_security_evidence is not None + else None + ), repo_status_by_name=repo_status_by_name, producer=producer_evidence.to_dict() if producer_evidence else {}, prior_notion_generated_at=prior_notion_generated_at, @@ -161,11 +412,7 @@ def publish_portfolio_truth( ) validate_truth_snapshot( build_result.snapshot, - security_max_age_hours=( - security_receipt_binding.max_age_hours - if security_receipt_binding is not None - else 24 - ), + security_max_age_hours=security_max_age_hours, ) snapshot_stamp = build_result.snapshot.generated_at.strftime("%Y-%m-%dT%H%M%SZ") @@ -232,6 +479,8 @@ def publish_portfolio_truth( raise PortfolioTruthPublishError( "Security receipt normalized evidence changed after it was loaded." ) + if prior_security_evidence is not None: + _verify_prior_security_evidence_current(prior_security_evidence) for path, staged in temp_files.items(): if path in {registry_output, portfolio_report_output} and not changed[path]: staged.unlink(missing_ok=True) diff --git a/src/portfolio_truth_reconcile.py b/src/portfolio_truth_reconcile.py index ca9abd92..16e7a5a6 100644 --- a/src/portfolio_truth_reconcile.py +++ b/src/portfolio_truth_reconcile.py @@ -197,11 +197,16 @@ def _validate_security_receipt_cohort_identity( *, projects: list[PortfolioTruthProject], security_alerts_by_name: dict[str, dict], + candidate_projects: list[PortfolioTruthProject] | None = None, + prior_security_alerts_by_name: dict[str, dict] | None = None, ) -> None: - """Require receipt membership to match freshly derived default attention.""" + """Bind receipt membership to attention derived before the new receipt.""" + if candidate_projects is None: + candidate_projects = projects + prior_security_alerts_by_name = prior_security_alerts_by_name or {} receipt_repositories = tuple(sorted(security_alerts_by_name, key=str.lower)) try: - derived_repositories = derive_default_attention_cohort( + candidate_repositories = derive_default_attention_cohort( { "projects": [ { @@ -213,7 +218,7 @@ def _validate_security_receipt_cohort_identity( "attention_state": project.derived.attention_state, }, } - for project in projects + for project in candidate_projects ] }, expected_count=len(receipt_repositories), @@ -224,21 +229,117 @@ def _validate_security_receipt_cohort_identity( f"derived default attention: {exc}." ) from exc - if receipt_repositories == derived_repositories: - return - receipt_only = sorted( - set(receipt_repositories) - set(derived_repositories), + set(receipt_repositories) - set(candidate_repositories), key=str.lower, ) derived_only = sorted( - set(derived_repositories) - set(receipt_repositories), + set(candidate_repositories) - set(receipt_repositories), key=str.lower, ) - raise ValueError( - "PortfolioTruth GitHub security receipt cohort differs from freshly derived " - f"default attention: receipt_only={receipt_only}; derived_only={derived_only}." + if receipt_only or derived_only: + raise ValueError( + "PortfolioTruth GitHub security receipt cohort differs from freshly " + "derived pre-security default attention: " + f"receipt_only={receipt_only}; derived_only={derived_only}." + ) + + final_repositories = _derive_project_security_cohort(projects) + final_only = sorted( + set(final_repositories) - set(receipt_repositories), + key=str.lower, ) + if final_only: + raise ValueError( + "PortfolioTruth post-receipt attention contains repositories outside " + f"the collected security cohort: {final_only}." + ) + + departed = sorted( + set(receipt_repositories) - set(final_repositories), + key=str.lower, + ) + unresolved_departures = [ + repository + for repository in departed + if not _is_verified_security_cohort_departure( + prior_security_alerts_by_name.get(repository), + security_alerts_by_name.get(repository), + ) + ] + if unresolved_departures: + raise ValueError( + "PortfolioTruth receipt members left default attention without fresh " + "observed Dependabot resolution or repository archive evidence: " + f"{unresolved_departures}." + ) + + +def _derive_project_security_cohort( + projects: list[PortfolioTruthProject], +) -> tuple[str, ...]: + expected_count = sum( + project.derived.attention_state in DEFAULT_ATTENTION_STATES + and not project.identity.project_key.startswith("supp:") + for project in projects + ) + try: + return derive_default_attention_cohort( + { + "projects": [ + { + "identity": { + "project_key": project.identity.project_key, + "repo_full_name": project.identity.repo_full_name, + }, + "derived": { + "attention_state": project.derived.attention_state, + }, + } + for project in projects + ] + }, + expected_count=expected_count, + ) + except SecurityCoverageError as exc: + raise ValueError( + f"PortfolioTruth post-receipt security cohort is invalid: {exc}." + ) from exc + + +def _is_verified_security_cohort_departure( + prior_entry: dict[str, Any] | None, + current_entry: dict[str, Any] | None, +) -> bool: + prior = dict(prior_entry or {}) + current = dict(current_entry or {}) + prior_dependabot = dict((prior.get("providers") or {}).get("dependabot") or {}) + current_dependabot = dict((current.get("providers") or {}).get("dependabot") or {}) + current_repository = dict(current.get("repository") or {}) + prior_counts = dict(prior_dependabot.get("counts") or {}) + current_counts = dict(current_dependabot.get("counts") or {}) + observed_resolution = ( + prior_dependabot.get("state") == "observed" + and sum( + value + for value in ( + prior_counts.get("high"), + prior_counts.get("critical"), + ) + if isinstance(value, int) and not isinstance(value, bool) + ) + > 0 + and current.get("receipt_state") == "fresh" + and current_dependabot.get("state") == "observed" + and current_counts.get("high") == 0 + and current_counts.get("critical") == 0 + ) + observed_archive = ( + current.get("receipt_state") == "fresh" + and current_repository.get("state") == "observed" + and current_repository.get("archived") is True + ) + return observed_resolution or observed_archive def build_portfolio_truth_snapshot( @@ -252,6 +353,7 @@ def build_portfolio_truth_snapshot( release_count_by_name: dict[str, int] | None = None, security_alerts_by_name: dict[str, dict] | None = None, security_coverage_metadata: dict[str, Any] | None = None, + prior_security_alerts_by_name: dict[str, dict] | None = None, repo_status_by_name: dict[str, dict] | None = None, producer: dict[str, Any] | None = None, prior_notion_generated_at: str | None = None, @@ -293,23 +395,38 @@ def build_portfolio_truth_snapshot( now=now, ), ) - projects = [ - _build_truth_project( - raw_project, - catalog_data=catalog_data, - legacy_rows=legacy_rows, - notion_context=notion_context, - 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 - ] + + def materialize_projects( + security_lookup: dict[str, dict] | None, + ) -> list[PortfolioTruthProject]: + return [ + _build_truth_project( + raw_project, + catalog_data=catalog_data, + legacy_rows=legacy_rows, + notion_context=notion_context, + now=now, + release_count_by_name=release_count_by_name, + security_alerts_by_name=security_lookup, + repo_status_by_name=repo_status_by_name, + ) + for raw_project in workspace_projects + ] + + prior_security_alerts = prior_security_alerts_by_name or {} + candidate_projects = ( + materialize_projects(prior_security_alerts) + if security_coverage_metadata is not None + and security_alerts_by_name is not None + else None + ) + projects = materialize_projects(security_alerts_by_name) if security_coverage_metadata is not None and security_alerts_by_name is not None: _validate_security_receipt_cohort_identity( projects=projects, + candidate_projects=candidate_projects, security_alerts_by_name=security_alerts_by_name, + prior_security_alerts_by_name=prior_security_alerts, ) projects.sort( key=lambda item: ( diff --git a/tests/test_portfolio_truth.py b/tests/test_portfolio_truth.py index 6510b813..f41d1ea5 100644 --- a/tests/test_portfolio_truth.py +++ b/tests/test_portfolio_truth.py @@ -4,6 +4,7 @@ import json import os import subprocess +import threading import time from datetime import datetime, timedelta, timezone from pathlib import Path @@ -1758,7 +1759,7 @@ def test_security_receipt_rejects_same_count_identity_rollover(tmp_path: Path) - with pytest.raises( ValueError, match=( - "receipt cohort differs from freshly derived default attention: " + "receipt cohort differs from freshly derived pre-security default attention: " "receipt_only=\\['d/Old'\\]; derived_only=\\['d/New'\\]" ), ): @@ -1769,6 +1770,7 @@ def test_security_receipt_rejects_same_count_identity_rollover(tmp_path: Path) - now=now, security_alerts_by_name=security, security_coverage_metadata=metadata, + prior_security_alerts_by_name=security, ) @@ -2033,7 +2035,7 @@ def test_security_cohort_identity_rejects_missing_repository_name() -> None: ) -def test_security_cohort_identity_accepts_receipt_risk_driven_attention( +def test_security_cohort_identity_rejects_receipt_self_promotion_without_prior( tmp_path: Path, ) -> None: workspace = tmp_path / "workspace" @@ -2091,19 +2093,464 @@ def test_security_cohort_identity_accepts_receipt_risk_driven_attention( "path": "/evidence/github-security-coverage-latest.json", } + with pytest.raises(ValueError, match="expected 1, observed 0"): + build_portfolio_truth_snapshot( + workspace_root=workspace, + catalog_path=catalog_path, + include_notion=False, + now=now, + security_alerts_by_name=security, + security_coverage_metadata=metadata, + ) + + +def test_security_cohort_identity_accepts_observed_resolution_from_prior_risk( + tmp_path: Path, +) -> None: + workspace = tmp_path / "workspace" + project = workspace / "Manual" + project.mkdir(parents=True) + _write(project / "README.md", "# Manual\n\nRisk resolution fixture.\n") + subprocess.run(["git", "init"], cwd=project, capture_output=True, check=True) + subprocess.run( + ["git", "remote", "add", "origin", "https://github.com/d/Manual.git"], + cwd=project, + capture_output=True, + check=True, + ) + catalog_path = tmp_path / "portfolio-catalog.yaml" + catalog_path.write_text( + """ +repos: + Manual: + owner: d + lifecycle_state: manual-only + review_cadence: weekly + operating_path: maintain + category: infrastructure +""" + ) + now = datetime(2026, 8, 4, 12, tzinfo=timezone.utc) + + def receipt_entry(*, high: int, state: str = "observed") -> dict: + counts = ( + {"critical": 0, "high": high, "medium": 0, "low": 0} + if state == "observed" + else None + ) + return { + "repo_full_name": "d/Manual", + "cohort_member": True, + "cohort_policy": "portfolio-default-attention-v1", + "receipt_schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "receipt_state": "fresh", + "source_produced_at": now.isoformat(), + "providers": { + "dependabot": { + "state": state, + "observed_at": now.isoformat(), + "pagination_complete": state == "observed", + "counts": counts, + } + }, + } + + prior_security = {"d/Manual": receipt_entry(high=1)} + current_security = {"d/Manual": receipt_entry(high=0)} + metadata = { + "source_id": "github-security-coverage-receipt", + "schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "produced_at": now.isoformat(), + "state": "fresh", + "age_hours": 0.0, + "producer_commit": "a" * 40, + "cohort_policy": "portfolio-default-attention-v1", + "cohort_repository_count": 1, + "path": "/evidence/github-security-coverage-latest.json", + } + result = build_portfolio_truth_snapshot( workspace_root=workspace, catalog_path=catalog_path, include_notion=False, now=now, - security_alerts_by_name=security, + security_alerts_by_name=current_security, security_coverage_metadata=metadata, + prior_security_alerts_by_name=prior_security, ) manual = result.snapshot.projects[0] - assert manual.declared.lifecycle_state == "manual-only" - assert manual.security.dependabot_high == 1 - assert manual.derived.attention_state == "decision-needed" + assert manual.security.cohort_member is True + assert manual.security.dependabot_high == 0 + assert manual.derived.attention_state == "manual-only" + assert ( + derive_default_attention_cohort(result.snapshot.to_dict(), expected_count=0) + == () + ) + + unavailable_security = {"d/Manual": receipt_entry(high=0, state="not_requested")} + with pytest.raises(ValueError, match="without fresh observed Dependabot"): + build_portfolio_truth_snapshot( + workspace_root=workspace, + catalog_path=catalog_path, + include_notion=False, + now=now, + security_alerts_by_name=unavailable_security, + security_coverage_metadata=metadata, + prior_security_alerts_by_name=prior_security, + ) + + +def test_security_cohort_uses_prior_archive_state_and_allows_observed_exit( + tmp_path: Path, +) -> None: + workspace = tmp_path / "workspace" + project = workspace / "Active" + project.mkdir(parents=True) + _write(project / "README.md", "# Active\n\nArchive transition fixture.\n") + subprocess.run(["git", "init"], cwd=project, capture_output=True, check=True) + subprocess.run( + ["git", "remote", "add", "origin", "https://github.com/d/Active.git"], + cwd=project, + capture_output=True, + check=True, + ) + catalog_path = tmp_path / "portfolio-catalog.yaml" + catalog_path.write_text( + """ +repos: + Active: + owner: d + lifecycle_state: active + review_cadence: weekly + operating_path: maintain + category: infrastructure +""" + ) + now = datetime(2026, 8, 4, 12, tzinfo=timezone.utc) + + def receipt_entry(*, archived: bool) -> dict: + return { + "repo_full_name": "d/Active", + "cohort_member": True, + "cohort_policy": "portfolio-default-attention-v1", + "receipt_schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "receipt_state": "fresh", + "source_produced_at": now.isoformat(), + "repository": _remote_repository_result( + state="observed", + observed_at=now.isoformat(), + default_branch="main", + head_sha="b" * 40, + archived=archived, + ), + "providers": { + "dependabot": _provider_result( + "dependabot", + state="observed", + observed_at=now.isoformat(), + http_status=200, + pagination_complete=True, + counts={"critical": 0, "high": 0, "medium": 0, "low": 0}, + ) + }, + } + + metadata = { + "source_id": "github-security-coverage-receipt", + "schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "produced_at": now.isoformat(), + "state": "fresh", + "age_hours": 0.0, + "producer_commit": "a" * 40, + "cohort_policy": "portfolio-default-attention-v1", + "cohort_repository_count": 1, + "path": "/evidence/github-security-coverage-latest.json", + } + + with pytest.raises(ValueError, match="expected 1, observed 0"): + build_portfolio_truth_snapshot( + workspace_root=workspace, + catalog_path=catalog_path, + include_notion=False, + now=now, + security_alerts_by_name={"d/Active": receipt_entry(archived=False)}, + security_coverage_metadata=metadata, + repo_status_by_name={ + "Active": {"source": "audit_report", "archived": True} + }, + ) + + result = build_portfolio_truth_snapshot( + workspace_root=workspace, + catalog_path=catalog_path, + include_notion=False, + now=now, + security_alerts_by_name={"d/Active": receipt_entry(archived=True)}, + security_coverage_metadata=metadata, + prior_security_alerts_by_name={ + "d/Active": receipt_entry(archived=False) + }, + ) + + active = result.snapshot.projects[0] + assert active.security.cohort_member is True + assert active.derived.attention_state == "archived" + assert ( + derive_default_attention_cohort(result.snapshot.to_dict(), expected_count=0) + == () + ) + + +def test_receipt_publication_uses_bound_prior_risk_for_resolution( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + workspace = tmp_path / "workspace" + project = workspace / "Manual" + project.mkdir(parents=True) + _write(project / "README.md", "# Manual\n\nTwo-cycle risk fixture.\n") + subprocess.run(["git", "init"], cwd=project, capture_output=True, check=True) + subprocess.run( + ["git", "remote", "add", "origin", "https://github.com/d/Manual.git"], + cwd=project, + capture_output=True, + check=True, + ) + catalog_path = tmp_path / "portfolio-catalog.yaml" + + def write_catalog(lifecycle_state: str) -> None: + catalog_path.write_text( + f""" +repos: + Manual: + owner: d + lifecycle_state: {lifecycle_state} + review_cadence: weekly + operating_path: maintain + category: infrastructure +""" + ) + + def security_entry(*, high: int, observed_at: datetime) -> dict: + zero_counts = { + "code_scanning": { + "critical": 0, + "high": 0, + "warning": 0, + "note": 0, + }, + "secret_scanning": {"open": 0}, + } + return { + "repo_full_name": "d/Manual", + "cohort_member": True, + "cohort_policy": "portfolio-default-attention-v1", + "receipt_schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "receipt_state": "fresh", + "source_produced_at": observed_at.isoformat(), + "repository": _remote_repository_result( + state="observed", + observed_at=observed_at.isoformat(), + default_branch="main", + head_sha="b" * 40, + archived=False, + ), + "providers": { + "dependabot": _provider_result( + "dependabot", + state="observed", + observed_at=observed_at.isoformat(), + http_status=200, + pagination_complete=True, + counts={ + "critical": 0, + "high": high, + "medium": 0, + "low": 0, + }, + ), + **{ + provider: _provider_result( + provider, + state="observed", + observed_at=observed_at.isoformat(), + http_status=200, + pagination_complete=True, + counts=counts, + ) + for provider, counts in zero_counts.items() + }, + }, + } + + def metadata(*, observed_at: datetime, marker: str, count: int = 1) -> dict: + return { + "source_id": "github-security-coverage-receipt", + "schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "produced_at": observed_at.isoformat(), + "state": "fresh", + "age_hours": 0.0, + "producer_commit": "a" * 40, + "cohort_policy": "portfolio-default-attention-v1", + "cohort_repository_count": count, + "path": "/evidence/github-security-coverage-latest.json", + "receipt_id": "sha256:" + marker * 64, + "content_sha256": marker * 64, + } + + output_dir = tmp_path / "output" + registry_output = workspace / "project-registry.md" + report_output = workspace / "PORTFOLIO-AUDIT-REPORT.md" + first_at = datetime(2026, 8, 4, 11, tzinfo=timezone.utc) + write_catalog("active") + publish_portfolio_truth( + workspace_root=workspace, + output_dir=output_dir, + registry_output=registry_output, + portfolio_report_output=report_output, + catalog_path=catalog_path, + include_notion=False, + now=first_at, + security_alerts_by_name={ + "d/Manual": security_entry(high=1, observed_at=first_at) + }, + security_coverage_metadata=metadata(observed_at=first_at, marker="a"), + ) + + second_at = first_at + timedelta(hours=1) + write_catalog("manual-only") + from contextlib import contextmanager + from src import portfolio_truth_publish as publish_module + + original_lock = publish_module._portfolio_truth_publication_lock + original_verify = publish_module._verify_prior_security_evidence_current + first_recheck = threading.Event() + second_lock_attempt = threading.Event() + second_recheck = threading.Event() + release_first = threading.Event() + counter_lock = threading.Lock() + lock_attempts = 0 + verification_count = 0 + + @contextmanager + def observed_publication_lock(latest_path: Path): + nonlocal lock_attempts + with counter_lock: + lock_attempts += 1 + attempt = lock_attempts + if attempt == 2: + second_lock_attempt.set() + with original_lock(latest_path): + yield + + def gate_first_after_prior_recheck(evidence) -> None: + nonlocal verification_count + original_verify(evidence) + with counter_lock: + verification_count += 1 + verification = verification_count + if verification == 1: + first_recheck.set() + if not release_first.wait(timeout=5): + raise AssertionError("timed out waiting to release first publisher") + else: + second_recheck.set() + + monkeypatch.setattr( + publish_module, + "_portfolio_truth_publication_lock", + observed_publication_lock, + ) + monkeypatch.setattr( + publish_module, + "_verify_prior_security_evidence_current", + gate_first_after_prior_recheck, + ) + + results: dict[str, object] = {} + errors: dict[str, Exception] = {} + + def publish_resolution(label: str, observed_at: datetime, marker: str) -> None: + try: + results[label] = publish_portfolio_truth( + workspace_root=workspace, + output_dir=output_dir, + registry_output=registry_output, + portfolio_report_output=report_output, + catalog_path=catalog_path, + include_notion=False, + now=observed_at, + security_alerts_by_name={ + "d/Manual": security_entry(high=0, observed_at=observed_at) + }, + security_coverage_metadata=metadata( + observed_at=observed_at, + marker=marker, + ), + ) + except Exception as exc: + errors[label] = exc + + first_publisher = threading.Thread( + target=publish_resolution, + args=("first", second_at, "b"), + daemon=True, + ) + competing_publisher = threading.Thread( + target=publish_resolution, + args=("competing", second_at + timedelta(minutes=1), "c"), + daemon=True, + ) + first_publisher.start() + assert first_recheck.wait(timeout=5) + competing_publisher.start() + try: + assert second_lock_attempt.wait(timeout=5) + assert second_recheck.wait(timeout=0.2) is False + finally: + release_first.set() + first_publisher.join(timeout=5) + competing_publisher.join(timeout=5) + + assert first_publisher.is_alive() is False + assert competing_publisher.is_alive() is False + assert "first" not in errors + assert isinstance(errors.get("competing"), ValueError) + assert "expected 1, observed 0" in str(errors["competing"]) + assert second_recheck.is_set() is False + + resolved = results["first"] + assert hasattr(resolved, "latest_path") + payload = json.loads(resolved.latest_path.read_text()) + manual = payload["projects"][0] + + assert manual["security"]["cohort_member"] is True + assert manual["security"]["dependabot_high"] == 0 + assert manual["derived"]["attention_state"] == "manual-only" + assert derive_default_attention_cohort(payload, expected_count=0) == () + + payload["inputs"]["github_security"].pop("receipt_id") + resolved.latest_path.write_text(json.dumps(payload)) + with pytest.raises( + PortfolioTruthPublishError, + match="requires both receipt_id and content_sha256", + ): + publish_portfolio_truth( + workspace_root=workspace, + output_dir=output_dir, + registry_output=registry_output, + portfolio_report_output=report_output, + catalog_path=catalog_path, + include_notion=False, + now=second_at + timedelta(hours=1), + security_alerts_by_name={}, + security_coverage_metadata=metadata( + observed_at=second_at + timedelta(hours=1), + marker="d", + count=0, + ), + ) def test_security_overlay_absent_leaves_repos_unscanned( @@ -3041,6 +3488,41 @@ def test_publish_uses_bound_security_max_age_for_remote_evidence( ) assert alpha_payload["repository_state"]["remote_default_branch"]["state"] == "observed" + second_at = now + timedelta(hours=1) + second_security = json.loads(json.dumps(security)) + second_security["d/Alpha"]["source_produced_at"] = second_at.isoformat() + second_binding = SecurityCoverageReceiptBinding( + source_path=str(tmp_path / "security-next.json"), + receipt_id="sha256:" + "c" * 64, + content_sha256="d" * 64, + receipt_state="fresh", + max_age_hours=48, + expected_cohort_count=1, + expected_producer_commit=None, + ) + second_metadata = { + **metadata, + "produced_at": second_at.isoformat(), + "age_hours": 0.0, + "path": second_binding.source_path, + "receipt_id": second_binding.receipt_id, + "content_sha256": second_binding.content_sha256, + } + republished = publish_portfolio_truth( + workspace_root=portfolio_workspace, + output_dir=tmp_path / "max-age-output", + registry_output=portfolio_workspace / "max-age-registry.md", + portfolio_report_output=portfolio_workspace / "max-age-report.md", + catalog_path=portfolio_catalog, + legacy_registry_path=legacy_registry, + include_notion=False, + security_alerts_by_name=second_security, + security_coverage_metadata=second_metadata, + security_receipt_binding=second_binding, + now=second_at, + ) + assert republished.latest_path.is_file() + def test_generated_registry_notes_do_not_accumulate_purpose_prefix( portfolio_workspace: Path, @@ -3379,6 +3861,115 @@ def stage_then_replace_receipt(target: Path, content: str) -> Path: assert not list(output_dir.glob("portfolio-truth-*.json")) +def test_publish_refuses_prior_truth_pointer_replacement_after_load( + portfolio_workspace: Path, + portfolio_catalog: Path, + legacy_registry: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + now = datetime.now(timezone.utc).replace(microsecond=0) + alpha_path = portfolio_workspace / "Alpha" + subprocess.run(["git", "init"], cwd=alpha_path, capture_output=True, check=True) + subprocess.run( + ["git", "remote", "add", "origin", "https://github.com/d/Alpha.git"], + cwd=alpha_path, + capture_output=True, + check=True, + ) + output_dir = tmp_path / "output" + registry_output = portfolio_workspace / "project-registry.md" + report_output = portfolio_workspace / "PORTFOLIO-AUDIT-REPORT.md" + first = publish_portfolio_truth( + workspace_root=portfolio_workspace, + output_dir=output_dir, + registry_output=registry_output, + portfolio_report_output=report_output, + catalog_path=portfolio_catalog, + legacy_registry_path=legacy_registry, + include_notion=False, + now=now - timedelta(minutes=1), + ) + registry_before = registry_output.read_text() + report_before = report_output.read_text() + concurrent_truth = "concurrent PortfolioTruth replacement\n" + + security = { + "d/Alpha": { + "repo_full_name": "d/Alpha", + "receipt_schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "receipt_state": "fresh", + "source_produced_at": now.isoformat(), + "cohort_member": True, + "cohort_policy": "portfolio-default-attention-v1", + "providers": { + name: _provider_result( + name, + state="not_requested", + reason="fixture_not_requested", + ) + for name in ("dependabot", "code_scanning", "secret_scanning") + }, + "repository": _remote_repository_result( + state="observed", + observed_at=now.isoformat(), + default_branch="main", + head_sha="b" * 40, + archived=False, + ), + } + } + metadata = { + "source_id": "github-security-coverage-receipt", + "schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "produced_at": now.isoformat(), + "state": "fresh", + "age_hours": 0.0, + "producer_commit": "a" * 40, + "cohort_policy": "portfolio-default-attention-v1", + "cohort_repository_count": 1, + "path": "/evidence/github-security-coverage-latest.json", + } + + from src import portfolio_truth_publish as publish_module + + original_stage = publish_module._stage_text + replaced = False + + def stage_then_replace_prior(target: Path, content: str) -> Path: + nonlocal replaced + staged = original_stage(target, content) + if not replaced: + replaced = True + first.latest_path.write_text(concurrent_truth) + return staged + + monkeypatch.setattr(publish_module, "_stage_text", stage_then_replace_prior) + + with pytest.raises( + PortfolioTruthPublishError, + match="changed after it authorized security cohort derivation", + ): + publish_portfolio_truth( + workspace_root=portfolio_workspace, + output_dir=output_dir, + registry_output=registry_output, + portfolio_report_output=report_output, + catalog_path=portfolio_catalog, + legacy_registry_path=legacy_registry, + include_notion=False, + security_alerts_by_name=security, + security_coverage_metadata=metadata, + now=now, + ) + + assert replaced is True + assert first.latest_path.read_text() == concurrent_truth + assert registry_output.read_text() == registry_before + assert report_output.read_text() == report_before + assert not list(output_dir.glob("*.tmp")) + + def test_publish_refuses_nested_evidence_that_expires_after_snapshot( portfolio_workspace: Path, portfolio_catalog: Path, @@ -3614,6 +4205,45 @@ def test_publish_requires_producer_evidence_before_touching_outputs( assert report_output.read_text() == "sentinel-report\n" assert not output_dir.exists() + mismatched_output = tmp_path / "mismatched-output" + mismatched_registry = portfolio_workspace / "mismatched-registry.md" + mismatched_report = portfolio_workspace / "mismatched-report.md" + binding = SecurityCoverageReceiptBinding( + source_path=str(tmp_path / "security.json"), + receipt_id="sha256:" + "a" * 64, + content_sha256="b" * 64, + receipt_state="fresh", + max_age_hours=24, + expected_cohort_count=0, + expected_producer_commit=None, + ) + security_metadata = { + "receipt_id": "sha256:" + "c" * 64, + "content_sha256": binding.content_sha256, + "path": binding.source_path, + } + with pytest.raises( + PortfolioTruthPublishError, + match="receipt_id metadata does not match", + ): + publish_portfolio_truth( + workspace_root=portfolio_workspace, + output_dir=mismatched_output, + registry_output=mismatched_registry, + portfolio_report_output=mismatched_report, + catalog_path=portfolio_catalog, + legacy_registry_path=legacy_registry, + include_notion=False, + security_alerts_by_name={}, + security_coverage_metadata=security_metadata, + security_receipt_binding=binding, + now=datetime(2026, 8, 4, 12, tzinfo=timezone.utc), + ) + + assert not mismatched_output.exists() + assert not mismatched_registry.exists() + assert not mismatched_report.exists() + def test_publish_refuses_to_drop_existing_notion_context( portfolio_workspace: Path, From b840e5c7b12f2451f670447eedbab3fa2fb39984 Mon Sep 17 00:00:00 2001 From: saagpatel Date: Tue, 4 Aug 2026 10:18:44 -0700 Subject: [PATCH 3/4] fix(portfolio): bind archive transitions to final state --- src/portfolio_truth_reconcile.py | 38 ++++++++- tests/test_portfolio_truth.py | 133 +++++++++++++++++++++++++++++-- 2 files changed, 160 insertions(+), 11 deletions(-) diff --git a/src/portfolio_truth_reconcile.py b/src/portfolio_truth_reconcile.py index 16e7a5a6..2d2dc6f3 100644 --- a/src/portfolio_truth_reconcile.py +++ b/src/portfolio_truth_reconcile.py @@ -259,12 +259,22 @@ def _validate_security_receipt_cohort_identity( set(receipt_repositories) - set(final_repositories), key=str.lower, ) + + def final_project_is_archived(repository: str) -> bool: + matches = [ + project + for project in projects + if project.identity.repo_full_name == repository + ] + return len(matches) == 1 and matches[0].derived.archived is True + unresolved_departures = [ repository for repository in departed if not _is_verified_security_cohort_departure( prior_security_alerts_by_name.get(repository), security_alerts_by_name.get(repository), + final_project_archived=final_project_is_archived(repository), ) ] if unresolved_departures: @@ -310,6 +320,8 @@ def _derive_project_security_cohort( def _is_verified_security_cohort_departure( prior_entry: dict[str, Any] | None, current_entry: dict[str, Any] | None, + *, + final_project_archived: bool, ) -> bool: prior = dict(prior_entry or {}) current = dict(current_entry or {}) @@ -335,7 +347,8 @@ def _is_verified_security_cohort_departure( and current_counts.get("critical") == 0 ) observed_archive = ( - current.get("receipt_state") == "fresh" + final_project_archived + and current.get("receipt_state") == "fresh" and current_repository.get("state") == "observed" and current_repository.get("archived") is True ) @@ -398,6 +411,8 @@ def build_portfolio_truth_snapshot( def materialize_projects( security_lookup: dict[str, dict] | None, + *, + repo_status_lookup: dict[str, dict] | None, ) -> list[PortfolioTruthProject]: return [ _build_truth_project( @@ -408,19 +423,34 @@ def materialize_projects( now=now, release_count_by_name=release_count_by_name, security_alerts_by_name=security_lookup, - repo_status_by_name=repo_status_by_name, + repo_status_by_name=repo_status_lookup, ) for raw_project in workspace_projects ] prior_security_alerts = prior_security_alerts_by_name or {} + candidate_repo_status_by_name = { + name: status + for name, status in (repo_status_by_name or {}).items() + if status.get("source") == "github_api" and status.get("archived") is False + } candidate_projects = ( - materialize_projects(prior_security_alerts) + materialize_projects( + prior_security_alerts, + # Current status may only expand candidate membership. A fresh GitHub + # unarchive therefore forces receipt coverage, while archive status is + # applied only to the final projects and must be corroborated by the + # receipt before it can authorize a departure. + repo_status_lookup=candidate_repo_status_by_name, + ) if security_coverage_metadata is not None and security_alerts_by_name is not None else None ) - projects = materialize_projects(security_alerts_by_name) + projects = materialize_projects( + security_alerts_by_name, + repo_status_lookup=repo_status_by_name, + ) if security_coverage_metadata is not None and security_alerts_by_name is not None: _validate_security_receipt_cohort_identity( projects=projects, diff --git a/tests/test_portfolio_truth.py b/tests/test_portfolio_truth.py index f41d1ea5..d7259b02 100644 --- a/tests/test_portfolio_truth.py +++ b/tests/test_portfolio_truth.py @@ -1773,6 +1773,46 @@ def test_security_receipt_rejects_same_count_identity_rollover(tmp_path: Path) - prior_security_alerts_by_name=security, ) + catalog_path.write_text( + """ +repos: + Old: + owner: d + lifecycle_state: active + review_cadence: weekly + operating_path: maintain + category: infrastructure + New: + owner: d + lifecycle_state: manual-only + review_cadence: weekly + operating_path: finish + category: vanity +""" + ) + replacement_security = { + "d/New": {**security["d/Old"], "repo_full_name": "d/New"} + } + with pytest.raises( + ValueError, + match=( + "receipt cohort differs from freshly derived pre-security default attention: " + "receipt_only=\\['d/New'\\]; derived_only=\\['d/Old'\\]" + ), + ): + build_portfolio_truth_snapshot( + workspace_root=workspace, + catalog_path=catalog_path, + include_notion=False, + now=now, + security_alerts_by_name=replacement_security, + security_coverage_metadata=metadata, + prior_security_alerts_by_name=security, + repo_status_by_name={ + "Old": {"source": "github_api", "archived": True} + }, + ) + def test_security_cohort_identity_skips_repo_less_supplementary() -> None: from types import SimpleNamespace @@ -2200,6 +2240,28 @@ def receipt_entry(*, high: int, state: str = "observed") -> dict: prior_security_alerts_by_name=prior_security, ) + contradictory_archive = receipt_entry(high=0, state="not_requested") + contradictory_archive["repository"] = _remote_repository_result( + state="observed", + observed_at=now.isoformat(), + default_branch="main", + head_sha="b" * 40, + archived=True, + ) + with pytest.raises(ValueError, match="without fresh observed Dependabot"): + build_portfolio_truth_snapshot( + workspace_root=workspace, + catalog_path=catalog_path, + include_notion=False, + now=now, + security_alerts_by_name={"d/Manual": contradictory_archive}, + security_coverage_metadata=metadata, + prior_security_alerts_by_name=prior_security, + repo_status_by_name={ + "Manual": {"source": "github_api", "archived": False} + }, + ) + def test_security_cohort_uses_prior_archive_state_and_allows_observed_exit( tmp_path: Path, @@ -2267,8 +2329,12 @@ def receipt_entry(*, archived: bool) -> dict: "cohort_repository_count": 1, "path": "/evidence/github-security-coverage-latest.json", } + live_archived_status = { + "Active": {"source": "github_api", "archived": True} + } + prior_active_security = {"d/Active": receipt_entry(archived=False)} - with pytest.raises(ValueError, match="expected 1, observed 0"): + with pytest.raises(ValueError, match="without fresh observed Dependabot"): build_portfolio_truth_snapshot( workspace_root=workspace, catalog_path=catalog_path, @@ -2276,9 +2342,8 @@ def receipt_entry(*, archived: bool) -> dict: now=now, security_alerts_by_name={"d/Active": receipt_entry(archived=False)}, security_coverage_metadata=metadata, - repo_status_by_name={ - "Active": {"source": "audit_report", "archived": True} - }, + prior_security_alerts_by_name=prior_active_security, + repo_status_by_name=live_archived_status, ) result = build_portfolio_truth_snapshot( @@ -2288,19 +2353,73 @@ def receipt_entry(*, archived: bool) -> dict: now=now, security_alerts_by_name={"d/Active": receipt_entry(archived=True)}, security_coverage_metadata=metadata, - prior_security_alerts_by_name={ - "d/Active": receipt_entry(archived=False) - }, + prior_security_alerts_by_name=prior_active_security, + repo_status_by_name=live_archived_status, ) active = result.snapshot.projects[0] assert active.security.cohort_member is True assert active.derived.attention_state == "archived" + assert active.provenance["github.archived"] == { + "source": "github_api", + "detail": "true", + } assert ( derive_default_attention_cohort(result.snapshot.to_dict(), expected_count=0) == () ) + unarchived = build_portfolio_truth_snapshot( + workspace_root=workspace, + catalog_path=catalog_path, + include_notion=False, + now=now, + security_alerts_by_name={"d/Active": receipt_entry(archived=False)}, + security_coverage_metadata=metadata, + prior_security_alerts_by_name={ + "d/Active": receipt_entry(archived=True) + }, + repo_status_by_name={ + "Active": {"source": "github_api", "archived": False} + }, + ) + reactivated = unarchived.snapshot.projects[0] + assert reactivated.derived.archived is False + assert reactivated.derived.attention_state == "active-infra" + + empty_metadata = {**metadata, "cohort_repository_count": 0} + with pytest.raises(ValueError, match="expected 0, observed 1"): + build_portfolio_truth_snapshot( + workspace_root=workspace, + catalog_path=catalog_path, + include_notion=False, + now=now, + security_alerts_by_name={}, + security_coverage_metadata=empty_metadata, + prior_security_alerts_by_name={ + "d/Active": receipt_entry(archived=True) + }, + repo_status_by_name={ + "Active": {"source": "github_api", "archived": False} + }, + ) + + with pytest.raises(ValueError, match="post-receipt attention contains"): + build_portfolio_truth_snapshot( + workspace_root=workspace, + catalog_path=catalog_path, + include_notion=False, + now=now, + security_alerts_by_name={}, + security_coverage_metadata=empty_metadata, + prior_security_alerts_by_name={ + "d/Active": receipt_entry(archived=True) + }, + repo_status_by_name={ + "Active": {"source": "audit_report", "archived": False} + }, + ) + def test_receipt_publication_uses_bound_prior_risk_for_resolution( tmp_path: Path, From 50c7df30ae18a14063bfad76546e53e4360a854b Mon Sep 17 00:00:00 2001 From: saagpatel Date: Tue, 4 Aug 2026 10:35:57 -0700 Subject: [PATCH 4/4] fix(portfolio): preserve prior final archive lineage --- src/portfolio_truth_publish.py | 29 +++++- src/portfolio_truth_reconcile.py | 30 ++++++- tests/test_portfolio_truth.py | 147 +++++++++++++++++++++++++++++++ 3 files changed, 204 insertions(+), 2 deletions(-) diff --git a/src/portfolio_truth_publish.py b/src/portfolio_truth_publish.py index 1e96e9cb..25b94e1d 100644 --- a/src/portfolio_truth_publish.py +++ b/src/portfolio_truth_publish.py @@ -13,8 +13,10 @@ from typing import Iterator from src.github_security_coverage import ( + DEFAULT_ATTENTION_STATES, SecurityCoverageError, SecurityCoverageReceiptBinding, + derive_default_attention_cohort, verified_security_coverage_receipt_binding, ) from src.portfolio_truth_reconcile import ( @@ -55,6 +57,7 @@ class _PriorSecurityEvidence: path: Path content_sha256: str | None alerts_by_full_name: dict[str, dict] + final_cohort_repositories: tuple[str, ...] | None class PortfolioTruthPublishError(RuntimeError): @@ -79,7 +82,7 @@ def _load_prior_security_alerts( current_security_metadata: dict[str, object], security_max_age_hours: int, ) -> _PriorSecurityEvidence: - """Load validated prior receipt evidence for independent cohort derivation.""" + """Load validated prior receipt and final-cohort evidence.""" try: content = latest_path.read_bytes() except FileNotFoundError: @@ -87,6 +90,7 @@ def _load_prior_security_alerts( path=latest_path, content_sha256=None, alerts_by_full_name={}, + final_cohort_repositories=None, ) except OSError as exc: raise PortfolioTruthPublishError( @@ -107,6 +111,23 @@ def _load_prior_security_alerts( ) from exc projects = canonical.get("projects") or [] + prior_final_cohort_count = sum( + (project.get("derived") or {}).get("attention_state") + in DEFAULT_ATTENTION_STATES + and not str((project.get("identity") or {}).get("project_key") or "").startswith( + "supp:" + ) + for project in projects + ) + try: + final_cohort_repositories = derive_default_attention_cohort( + canonical, + expected_count=prior_final_cohort_count, + ) + except SecurityCoverageError as exc: + raise PortfolioTruthPublishError( + f"Prior PortfolioTruth final security cohort is invalid: {exc}" + ) from exc receipt_projects = [ project for project in projects @@ -158,6 +179,7 @@ def _load_prior_security_alerts( path=latest_path, content_sha256=hashlib.sha256(content).hexdigest(), alerts_by_full_name=alerts, + final_cohort_repositories=final_cohort_repositories, ) @@ -405,6 +427,11 @@ def _publish_portfolio_truth_locked( if prior_security_evidence is not None else None ), + prior_security_cohort_repositories=( + prior_security_evidence.final_cohort_repositories + if prior_security_evidence is not None + else None + ), repo_status_by_name=repo_status_by_name, producer=producer_evidence.to_dict() if producer_evidence else {}, prior_notion_generated_at=prior_notion_generated_at, diff --git a/src/portfolio_truth_reconcile.py b/src/portfolio_truth_reconcile.py index 2d2dc6f3..23565497 100644 --- a/src/portfolio_truth_reconcile.py +++ b/src/portfolio_truth_reconcile.py @@ -355,6 +355,29 @@ def _is_verified_security_cohort_departure( return observed_resolution or observed_archive +def _candidate_prior_security_alerts( + prior_security_alerts_by_name: dict[str, dict], + prior_security_cohort_repositories: tuple[str, ...] | None, +) -> dict[str, dict]: + """Keep prior final members from being removed by contradicted archive evidence.""" + if prior_security_cohort_repositories is None: + return prior_security_alerts_by_name + + prior_final_members = set(prior_security_cohort_repositories) + candidate_alerts: dict[str, dict] = {} + for repository, entry in prior_security_alerts_by_name.items(): + candidate_entry = dict(entry) + remote_repository = dict(candidate_entry.get("repository") or {}) + if ( + repository in prior_final_members + and remote_repository.get("archived") is True + ): + remote_repository.pop("archived") + candidate_entry["repository"] = remote_repository + candidate_alerts[repository] = candidate_entry + return candidate_alerts + + def build_portfolio_truth_snapshot( *, workspace_root: Path, @@ -367,6 +390,7 @@ def build_portfolio_truth_snapshot( security_alerts_by_name: dict[str, dict] | None = None, security_coverage_metadata: dict[str, Any] | None = None, prior_security_alerts_by_name: dict[str, dict] | None = None, + prior_security_cohort_repositories: tuple[str, ...] | None = None, repo_status_by_name: dict[str, dict] | None = None, producer: dict[str, Any] | None = None, prior_notion_generated_at: str | None = None, @@ -429,6 +453,10 @@ def materialize_projects( ] prior_security_alerts = prior_security_alerts_by_name or {} + candidate_prior_security_alerts = _candidate_prior_security_alerts( + prior_security_alerts, + prior_security_cohort_repositories, + ) candidate_repo_status_by_name = { name: status for name, status in (repo_status_by_name or {}).items() @@ -436,7 +464,7 @@ def materialize_projects( } candidate_projects = ( materialize_projects( - prior_security_alerts, + candidate_prior_security_alerts, # Current status may only expand candidate membership. A fresh GitHub # unarchive therefore forces receipt coverage, while archive status is # applied only to the final projects and must be corroborated by the diff --git a/tests/test_portfolio_truth.py b/tests/test_portfolio_truth.py index d7259b02..bd247f01 100644 --- a/tests/test_portfolio_truth.py +++ b/tests/test_portfolio_truth.py @@ -1771,6 +1771,7 @@ def test_security_receipt_rejects_same_count_identity_rollover(tmp_path: Path) - security_alerts_by_name=security, security_coverage_metadata=metadata, prior_security_alerts_by_name=security, + prior_security_cohort_repositories=("d/Old",), ) catalog_path.write_text( @@ -1808,6 +1809,7 @@ def test_security_receipt_rejects_same_count_identity_rollover(tmp_path: Path) - security_alerts_by_name=replacement_security, security_coverage_metadata=metadata, prior_security_alerts_by_name=security, + prior_security_cohort_repositories=("d/Old",), repo_status_by_name={ "Old": {"source": "github_api", "archived": True} }, @@ -2420,6 +2422,151 @@ def receipt_entry(*, archived: bool) -> dict: }, ) + confirmed_archive_without_live_status = build_portfolio_truth_snapshot( + workspace_root=workspace, + catalog_path=catalog_path, + include_notion=False, + now=now, + security_alerts_by_name={"d/Active": receipt_entry(archived=True)}, + security_coverage_metadata=metadata, + prior_security_alerts_by_name={ + "d/Active": receipt_entry(archived=True) + }, + # The prior canonical truth kept this identity in its final cohort because + # live GitHub status contradicted the receipt's archive claim. + prior_security_cohort_repositories=("d/Active",), + ) + confirmed = confirmed_archive_without_live_status.snapshot.projects[0] + assert confirmed.derived.archived is True + assert confirmed.derived.attention_state == "archived" + + +def test_receipt_publication_preserves_prior_final_membership_when_live_status_drops( + tmp_path: Path, +) -> None: + workspace = tmp_path / "workspace" + project = workspace / "Active" + project.mkdir(parents=True) + _write(project / "README.md", "# Active\n\nPrior final cohort fixture.\n") + subprocess.run(["git", "init"], cwd=project, capture_output=True, check=True) + subprocess.run( + ["git", "remote", "add", "origin", "https://github.com/d/Active.git"], + cwd=project, + capture_output=True, + check=True, + ) + catalog_path = tmp_path / "portfolio-catalog.yaml" + catalog_path.write_text( + """ +repos: + Active: + owner: d + lifecycle_state: active + review_cadence: weekly + operating_path: maintain + category: infrastructure +""" + ) + + def security_entry(observed_at: datetime) -> dict: + return { + "repo_full_name": "d/Active", + "cohort_member": True, + "cohort_policy": "portfolio-default-attention-v1", + "receipt_schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "receipt_state": "fresh", + "source_produced_at": observed_at.isoformat(), + "repository": _remote_repository_result( + state="observed", + observed_at=observed_at.isoformat(), + default_branch="main", + head_sha="b" * 40, + archived=True, + ), + "providers": { + "dependabot": _provider_result( + "dependabot", + state="observed", + observed_at=observed_at.isoformat(), + http_status=200, + pagination_complete=True, + counts={"critical": 0, "high": 0, "medium": 0, "low": 0}, + ), + "code_scanning": _provider_result( + "code_scanning", + state="observed", + observed_at=observed_at.isoformat(), + http_status=200, + pagination_complete=True, + counts={"critical": 0, "high": 0, "warning": 0, "note": 0}, + ), + "secret_scanning": _provider_result( + "secret_scanning", + state="observed", + observed_at=observed_at.isoformat(), + http_status=200, + pagination_complete=True, + counts={"open": 0}, + ), + }, + } + + def metadata(observed_at: datetime, marker: str) -> dict: + return { + "source_id": "github-security-coverage-receipt", + "schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "produced_at": observed_at.isoformat(), + "state": "fresh", + "age_hours": 0.0, + "producer_commit": "a" * 40, + "cohort_policy": "portfolio-default-attention-v1", + "cohort_repository_count": 1, + "path": f"/evidence/security-{marker}.json", + "receipt_id": "sha256:" + marker * 64, + "content_sha256": marker * 64, + } + + output_dir = tmp_path / "output" + registry_output = workspace / "project-registry.md" + report_output = workspace / "PORTFOLIO-AUDIT-REPORT.md" + first_at = datetime(2026, 8, 4, 12, tzinfo=timezone.utc) + first = publish_portfolio_truth( + workspace_root=workspace, + output_dir=output_dir, + registry_output=registry_output, + portfolio_report_output=report_output, + catalog_path=catalog_path, + include_notion=False, + now=first_at, + security_alerts_by_name={"d/Active": security_entry(first_at)}, + security_coverage_metadata=metadata(first_at, "a"), + repo_status_by_name={ + "Active": {"source": "github_api", "archived": False} + }, + ) + first_payload = json.loads(first.latest_path.read_text()) + first_project = first_payload["projects"][0] + assert first_project["derived"]["archived"] is False + assert first_project["derived"]["attention_state"] == "active-infra" + assert first_project["repository_state"]["remote_default_branch"]["archived"] is True + + second_at = first_at + timedelta(hours=1) + second = publish_portfolio_truth( + workspace_root=workspace, + output_dir=output_dir, + registry_output=registry_output, + portfolio_report_output=report_output, + catalog_path=catalog_path, + include_notion=False, + now=second_at, + security_alerts_by_name={"d/Active": security_entry(second_at)}, + security_coverage_metadata=metadata(second_at, "b"), + ) + second_payload = json.loads(second.latest_path.read_text()) + second_project = second_payload["projects"][0] + assert second_project["derived"]["archived"] is True + assert second_project["derived"]["attention_state"] == "archived" + def test_receipt_publication_uses_bound_prior_risk_for_resolution( tmp_path: Path,