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_publish.py b/src/portfolio_truth_publish.py index 036a996d..25b94e1d 100644 --- a/src/portfolio_truth_publish.py +++ b/src/portfolio_truth_publish.py @@ -1,15 +1,22 @@ 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 ( + DEFAULT_ATTENTION_STATES, SecurityCoverageError, SecurityCoverageReceiptBinding, + derive_default_attention_cohort, verified_security_coverage_receipt_binding, ) from src.portfolio_truth_reconcile import ( @@ -21,6 +28,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 +37,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 +52,183 @@ 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] + final_cohort_repositories: tuple[str, ...] | None + + 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 and final-cohort evidence.""" + try: + content = latest_path.read_bytes() + except FileNotFoundError: + return _PriorSecurityEvidence( + path=latest_path, + content_sha256=None, + alerts_by_full_name={}, + final_cohort_repositories=None, + ) + 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 [] + 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 + 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, + final_cohort_repositories=final_cohort_repositories, + ) + + +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 +325,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 +422,16 @@ 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 + ), + 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, @@ -161,11 +439,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 +506,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 7f042c29..23565497 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,191 @@ 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], + candidate_projects: list[PortfolioTruthProject] | None = None, + prior_security_alerts_by_name: dict[str, dict] | None = None, +) -> None: + """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: + candidate_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 candidate_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 + + receipt_only = sorted( + set(receipt_repositories) - set(candidate_repositories), + key=str.lower, + ) + derived_only = sorted( + set(candidate_repositories) - set(receipt_repositories), + key=str.lower, + ) + 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, + ) + + 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: + 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, + *, + final_project_archived: bool, +) -> 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 = ( + final_project_archived + and 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 _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, @@ -199,6 +389,8 @@ 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, + 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, @@ -240,19 +432,60 @@ 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, + + def materialize_projects( + security_lookup: dict[str, dict] | None, + *, + repo_status_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_lookup, + ) + for raw_project in workspace_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() + if status.get("source") == "github_api" and status.get("archived") is False + } + candidate_projects = ( + materialize_projects( + 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 + # 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, + 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, + candidate_projects=candidate_projects, security_alerts_by_name=security_alerts_by_name, - repo_status_by_name=repo_status_by_name, + prior_security_alerts_by_name=prior_security_alerts, ) - for raw_project in workspace_projects - ] projects.sort( key=lambda item: ( item.identity.section_marker.lower(), @@ -774,11 +1007,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..bd247f01 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 @@ -17,6 +18,7 @@ _provider_result, _remote_repository_result, collect_security_coverage, + derive_default_attention_cohort, load_security_coverage_receipt, write_security_coverage_receipt, ) @@ -704,6 +706,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 +913,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 ): @@ -1501,70 +1622,1201 @@ def test_bound_security_identity_and_high_findings_reach_decision_queue( "archived": False, }, "providers": { - "dependabot": { - "state": "observed", - "observed_at": observed_at, - "pagination_complete": True, - "counts": {"critical": 0, "high": 2, "medium": 1, "low": 0}, - }, - "code_scanning": { - "state": "observed", - "observed_at": observed_at, - "pagination_complete": True, - "counts": {"critical": 0, "high": 0, "warning": 0, "note": 0}, - }, - "secret_scanning": { - "state": "observed", - "observed_at": observed_at, - "pagination_complete": True, - "counts": {"open": 0}, + "dependabot": { + "state": "observed", + "observed_at": observed_at, + "pagination_complete": True, + "counts": {"critical": 0, "high": 2, "medium": 1, "low": 0}, + }, + "code_scanning": { + "state": "observed", + "observed_at": observed_at, + "pagination_complete": True, + "counts": {"critical": 0, "high": 0, "warning": 0, "note": 0}, + }, + "secret_scanning": { + "state": "observed", + "observed_at": observed_at, + "pagination_complete": True, + "counts": {"open": 0}, + }, + }, + } + } + metadata = { + "source_id": "github-security-coverage-receipt", + "schema_version": "GitHubSecurityCoverageReceiptV1", + "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", + "receipt_id": "sha256:" + "c" * 64, + "content_sha256": "d" * 64, + } + + result = build_portfolio_truth_snapshot( + workspace_root=portfolio_workspace, + catalog_path=portfolio_catalog, + legacy_registry_path=legacy_registry, + include_notion=False, + now=now, + security_alerts_by_name=security, + security_coverage_metadata=metadata, + ) + truth = result.snapshot.to_dict() + alpha = next( + project + for project in truth["projects"] + if project["identity"]["display_name"] == "Alpha" + ) + decision = next( + item for item in build_decision_queue(truth) if item["project"] == "Alpha" + ) + + assert truth["inputs"]["github_security"] == metadata + assert alpha["risk"]["security_risk"] is True + assert alpha["security"]["dependabot_high"] == 2 + assert alpha["security"]["dependabot_medium"] == 1 + assert decision["decision_type"] == "security follow-up" + assert "critical=0, high=2" in decision["evidence"][0] + + result.snapshot.inputs["github_security"].pop("content_sha256") + with pytest.raises(ValueError, match="requires both receipt_id"): + 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 pre-security 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, + prior_security_alerts_by_name=security, + prior_security_cohort_repositories=("d/Old",), + ) + + 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, + prior_security_cohort_repositories=("d/Old",), + repo_status_by_name={ + "Old": {"source": "github_api", "archived": True} + }, + ) + + +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_rejects_receipt_self_promotion_without_prior( + 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", + } + + 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=current_security, + security_coverage_metadata=metadata, + prior_security_alerts_by_name=prior_security, + ) + manual = result.snapshot.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(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, + ) + + 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, +) -> 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", + } + live_archived_status = { + "Active": {"source": "github_api", "archived": True} + } + prior_active_security = {"d/Active": receipt_entry(archived=False)} + + 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/Active": receipt_entry(archived=False)}, + security_coverage_metadata=metadata, + prior_security_alerts_by_name=prior_active_security, + repo_status_by_name=live_archived_status, + ) + + 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=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} + }, + ) + + 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, + 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() }, }, } - } - metadata = { - "source_id": "github-security-coverage-receipt", - "schema_version": "GitHubSecurityCoverageReceiptV1", - "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", - "receipt_id": "sha256:" + "c" * 64, - "content_sha256": "d" * 64, - } - result = build_portfolio_truth_snapshot( - workspace_root=portfolio_workspace, - catalog_path=portfolio_catalog, - legacy_registry_path=legacy_registry, + 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=now, - security_alerts_by_name=security, - security_coverage_metadata=metadata, + 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"), ) - truth = result.snapshot.to_dict() - alpha = next( - project - for project in truth["projects"] - if project["identity"]["display_name"] == "Alpha" + + 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, ) - decision = next( - item for item in build_decision_queue(truth) if item["project"] == "Alpha" + monkeypatch.setattr( + publish_module, + "_verify_prior_security_evidence_current", + gate_first_after_prior_recheck, ) - assert truth["inputs"]["github_security"] == metadata - assert alpha["risk"]["security_risk"] is True - assert alpha["security"]["dependabot_high"] == 2 - assert alpha["security"]["dependabot_medium"] == 1 - assert decision["decision_type"] == "security follow-up" - assert "critical=0, high=2" in decision["evidence"][0] + results: dict[str, object] = {} + errors: dict[str, Exception] = {} - result.snapshot.inputs["github_security"].pop("content_sha256") - with pytest.raises(ValueError, match="requires both receipt_id"): - validate_truth_snapshot(result.snapshot) + 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( @@ -2502,6 +3754,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, @@ -2733,6 +4020,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" @@ -2832,6 +4127,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, @@ -2842,6 +4246,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() @@ -3059,6 +4471,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, 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)