diff --git a/pyproject.toml b/pyproject.toml index a78f96a..9dbe09a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -80,6 +80,7 @@ Documentation = "https://github.com/saagpatel/GithubRepoAuditor#readme" [project.scripts] audit = "src.cli:main" operator-os-seam-linter = "src.operator_os_seam_linter:main" +operator-os-producer-preflight = "src.producer_preflight:main" [tool.setuptools.packages.find] where = ["."] diff --git a/src/operator_os_seam_linter.py b/src/operator_os_seam_linter.py index 38460ea..ea081d9 100644 --- a/src/operator_os_seam_linter.py +++ b/src/operator_os_seam_linter.py @@ -1,6 +1,7 @@ from __future__ import annotations import argparse +import hashlib import json import re import sqlite3 @@ -10,7 +11,8 @@ from typing import Any from src.portfolio_truth_render import GENERATED_MARKDOWN_PROVENANCE_MARKER -from src.portfolio_truth_types import SCHEMA_VERSION, truth_latest_path +from src.portfolio_truth_types import LEGACY_SCHEMA_VERSIONS, SCHEMA_VERSION, truth_latest_path +from src.portfolio_catalog import load_portfolio_catalog from src.project_registry import ( BRIDGE_CANONICAL_KEY_DISAGREEMENTS, DEFAULT_NOTION_PROJECTION_ONLY_ROWS, @@ -42,6 +44,7 @@ class SeamLintFinding: artifact: str violation: str detail: str + level: str = "fail" def to_dict(self) -> dict[str, str]: return { @@ -49,6 +52,7 @@ def to_dict(self) -> dict[str, str]: "artifact": self.artifact, "violation": self.violation, "detail": self.detail, + "level": self.level, } @@ -61,13 +65,24 @@ class SeamLintResult: @property def passed(self) -> bool: - return not self.findings + return not any(finding.level == "fail" for finding in self.findings) + + @property + def state(self) -> str: + levels = {finding.level for finding in self.findings} + if "fail" in levels: + return "fail" + if "warn" in levels: + return "warn" + if "unknown" in levels: + return "unknown" + return "pass" def to_dict(self) -> dict[str, Any]: return { "schema_version": "operator_os_seam_linter.v0", "generated_at": self.generated_at.isoformat(), - "state": "pass" if self.passed else "fail", + "state": self.state, "expected_schema_version": self.expected_schema_version, "max_staleness_hours": self.max_staleness_hours, "findings": [finding.to_dict() for finding in self.findings], @@ -84,6 +99,8 @@ def lint_operator_os_seams( notification_db_path: Path | None = None, notion_snapshot_path: Path | None = None, identity_since: datetime | None = None, + contract_shadow: bool = False, + catalog_path: Path | None = None, now: datetime | None = None, ) -> SeamLintResult: generated_at = _aware(now or datetime.now(UTC)) @@ -114,6 +131,15 @@ def lint_operator_os_seams( identity_since=identity_since, ) ) + if contract_shadow: + findings.extend( + _check_contract_shadow( + truth, + truth_path=truth_path, + catalog_path=catalog_path, + now=generated_at, + ) + ) findings.extend(_check_generated_markdown(markdown_paths)) return SeamLintResult( generated_at=generated_at, @@ -126,6 +152,8 @@ def lint_operator_os_seams( def build_worklist_payload(result: SeamLintResult) -> dict[str, Any]: items = [] for finding in result.findings: + if finding.level != "fail": + continue items.append( { "item_id": f"ghra_seam_linter:{finding.check}:{Path(finding.artifact).name}", @@ -146,7 +174,7 @@ def build_worklist_payload(result: SeamLintResult) -> dict[str, Any]: return { "schema_version": WORKLIST_SCHEMA_VERSION, "generated_at": result.generated_at.isoformat(), - "state": "pass" if result.passed else "fail", + "state": result.state, "source": "GithubRepoAuditor", "items": items, } @@ -175,6 +203,8 @@ def main(argv: list[str] | None = None) -> int: identity_since=_parse_datetime(args.identity_since) if args.identity_since is not None else None, + contract_shadow=args.contract_shadow, + catalog_path=args.catalog, ) payload = result.to_dict() if args.worklist_output: @@ -215,6 +245,12 @@ def _build_parser() -> argparse.ArgumentParser: ), ) parser.add_argument("--expected-schema-version", default=SCHEMA_VERSION) + parser.add_argument("--contract-shadow", action="store_true") + parser.add_argument( + "--catalog", + type=Path, + default=Path("config") / "portfolio-catalog.yaml", + ) parser.add_argument("--max-staleness-hours", type=int, default=DEFAULT_MAX_STALENESS_HOURS) parser.add_argument("--json", action="store_true") return parser @@ -313,6 +349,217 @@ def _check_artifact_freshness( return [] +def _check_contract_shadow( + truth: dict[str, Any], + *, + truth_path: Path, + catalog_path: Path | None, + now: datetime, +) -> list[SeamLintFinding]: + findings: list[SeamLintFinding] = [] + schema_version = truth.get("schema_version") + legacy_schema = schema_version in LEGACY_SCHEMA_VERSIONS + if legacy_schema: + findings.append( + SeamLintFinding( + check="CL-PROD-001", + artifact=str(truth_path), + violation="producer lineage is unavailable in a legacy artifact", + detail=f"schema_version={schema_version}", + level="unknown", + ) + ) + + producer = truth.get("producer") + if not legacy_schema and (not isinstance(producer, dict) or not producer): + findings.append( + SeamLintFinding( + check="CL-PROD-001", + artifact=str(truth_path), + violation="producer evidence is absent", + detail="Canonical claims require an identified producer; local artifacts remain usable as unknown.", + level="unknown", + ) + ) + + inputs = truth.get("inputs") + catalog_input = inputs.get("catalog") if isinstance(inputs, dict) else None + if catalog_path is not None and catalog_path.is_file(): + actual_hash = hashlib.sha256(catalog_path.read_bytes()).hexdigest() + declared_hash = ( + catalog_input.get("sha256") if isinstance(catalog_input, dict) else None + ) + if not declared_hash: + findings.append( + SeamLintFinding( + check="CL-INP-001", + artifact=str(truth_path), + violation="catalog input hash is absent", + detail=f"catalog={catalog_path}", + level="unknown", + ) + ) + elif declared_hash != actual_hash: + findings.append( + SeamLintFinding( + check="CL-INP-001", + artifact=str(truth_path), + violation="catalog input hash differs from current catalog", + detail=f"declared={declared_hash}; actual={actual_hash}", + ) + ) + findings.extend( + _check_catalog_declaration_parity( + truth, + truth_path=truth_path, + catalog_path=catalog_path, + ) + ) + else: + findings.append( + SeamLintFinding( + check="CL-INP-001", + artifact=str(truth_path), + violation="current catalog is unavailable", + detail=f"catalog={catalog_path}", + level="unknown", + ) + ) + + findings.extend(_check_rollup_integrity(truth, truth_path=truth_path)) + findings.extend(_check_carried_freshness(truth, truth_path=truth_path, now=now)) + return findings + + +def _check_catalog_declaration_parity( + truth: dict[str, Any], *, truth_path: Path, catalog_path: Path +) -> list[SeamLintFinding]: + catalog = load_portfolio_catalog(catalog_path) + repos = catalog.get("repos") if isinstance(catalog, dict) else None + if not isinstance(repos, dict): + return [] + mismatches: dict[str, list[str]] = {} + for project in truth.get("projects", []): + if not isinstance(project, dict): + continue + identity = project.get("identity") + path = identity.get("path") if isinstance(identity, dict) else None + entry = repos.get(path) if isinstance(path, str) else None + if not isinstance(entry, dict): + continue + provenance = project.get("provenance") + if not isinstance(provenance, dict): + continue + for field in ("lifecycle_state", "intended_disposition", "owner"): + if not entry.get(field): + continue + source = provenance.get(f"declared.{field}") + if not isinstance(source, dict) or source.get("source") != "catalog_repo": + mismatches.setdefault(path, []).append(f"declared.{field}") + if not mismatches: + return [] + sample = ", ".join(sorted(mismatches)[:10]) + field_count = sum(len(fields) for fields in mismatches.values()) + return [ + SeamLintFinding( + check="CL-DECL-001", + artifact=str(truth_path), + violation="explicit catalog declarations are absent from artifact provenance", + detail=( + f"project_count={len(mismatches)}; field_count={field_count}; " + f"sample={sample}" + ), + ) + ] + + +def _check_rollup_integrity( + truth: dict[str, Any], *, truth_path: Path +) -> list[SeamLintFinding]: + projects = [item for item in truth.get("projects", []) if isinstance(item, dict)] + decision_count = sum( + 1 + for project in projects + if isinstance(project.get("derived"), dict) + and project["derived"].get("attention_state") == "decision-needed" + ) + summary = truth.get("source_summary") + counts = summary.get("attention_state_counts") if isinstance(summary, dict) else None + summary_count = counts.get("decision-needed", 0) if isinstance(counts, dict) else None + rollups = truth.get("rollups") + decision = rollups.get("decision") if isinstance(rollups, dict) else None + rollup_count = decision.get("decision_needed_count") if isinstance(decision, dict) else None + if summary_count is None or rollup_count is None: + return [ + SeamLintFinding( + check="CL-COUNT-001", + artifact=str(truth_path), + violation="decision count coverage is incomplete", + detail=( + f"projects={decision_count}; source_summary={summary_count}; " + f"rollups={rollup_count}" + ), + level="unknown", + ) + ] + if summary_count == decision_count and rollup_count == decision_count: + return [] + return [ + SeamLintFinding( + check="CL-COUNT-001", + artifact=str(truth_path), + violation="decision-needed counts do not reconcile", + detail=( + f"projects={decision_count}; source_summary={summary_count}; " + f"rollups={rollup_count}" + ), + ) + ] + + +def _check_carried_freshness( + truth: dict[str, Any], *, truth_path: Path, now: datetime +) -> list[SeamLintFinding]: + inputs = truth.get("inputs") + notion = inputs.get("notion") if isinstance(inputs, dict) else None + if not isinstance(notion, dict) or notion.get("mode") != "carried-forward": + return [] + origin = notion.get("carried_from_generated_at") + if not isinstance(origin, str) or not origin: + return [ + SeamLintFinding( + check="CL-FRESH-002", + artifact=str(truth_path), + violation="carried-forward origin is unknown", + detail="Notion advisory freshness cannot be established.", + level="unknown", + ) + ] + try: + age = now - _parse_datetime(origin) + except ValueError: + return [ + SeamLintFinding( + check="CL-FRESH-002", + artifact=str(truth_path), + violation="carried-forward origin is invalid", + detail=f"carried_from_generated_at={origin}", + level="unknown", + ) + ] + if age > timedelta(hours=48): + return [ + SeamLintFinding( + check="CL-FRESH-002", + artifact=str(truth_path), + violation="carried-forward Notion advisory is stale", + detail=f"age_hours={age.total_seconds() / 3600:.2f}; warn_after=48", + level="warn", + ) + ] + return [] + + def _check_schema_pin( truth: dict[str, Any], *, @@ -329,7 +576,7 @@ def _check_schema_pin( detail="The truth artifact must declare its schema pin.", ) ] - if declared != expected_schema_version: + if declared != expected_schema_version and declared not in LEGACY_SCHEMA_VERSIONS: return [ SeamLintFinding( check="schema_pin", diff --git a/src/portfolio_truth_publish.py b/src/portfolio_truth_publish.py index 07e915d..f5c3a40 100644 --- a/src/portfolio_truth_publish.py +++ b/src/portfolio_truth_publish.py @@ -11,6 +11,7 @@ ) from src.portfolio_truth_render import render_portfolio_report_markdown, render_registry_markdown 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 ( validate_portfolio_report_markdown, validate_publish_targets, @@ -89,7 +90,23 @@ def publish_portfolio_truth( release_count_by_name: dict[str, int] | None = None, security_alerts_by_name: dict[str, dict] | None = None, repo_status_by_name: dict[str, dict] | None = None, + producer_evidence: ProducerEvidence | None = None, + producer_repo_root: Path | None = None, + require_producer_evidence: bool = False, ) -> PortfolioTruthPublishResult: + if require_producer_evidence and producer_evidence is None: + raise PortfolioTruthPublishError( + "Canonical publication requires validated producer evidence." + ) + if producer_evidence is not None: + if producer_repo_root is None: + raise PortfolioTruthPublishError( + "producer_repo_root is required with producer evidence." + ) + try: + verify_evidence_still_current(producer_repo_root, producer_evidence) + except ValueError as exc: + raise PortfolioTruthPublishError(str(exc)) from exc validate_publish_targets( workspace_root=workspace_root, output_dir=output_dir, @@ -100,6 +117,7 @@ def publish_portfolio_truth( notion_context_fallback = ( load_prior_notion_context(latest_path) if allow_empty_notion else None ) + prior_notion_generated_at = _previous_generated_at(latest_path) build_result = build_portfolio_truth_snapshot( workspace_root=workspace_root, catalog_path=catalog_path, @@ -109,8 +127,15 @@ def publish_portfolio_truth( release_count_by_name=release_count_by_name, security_alerts_by_name=security_alerts_by_name, 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, ) validate_truth_snapshot(build_result.snapshot) + if producer_evidence is not None and producer_repo_root is not None: + try: + verify_evidence_still_current(producer_repo_root, producer_evidence) + except ValueError as exc: + raise PortfolioTruthPublishError(str(exc)) from exc snapshot_stamp = build_result.snapshot.generated_at.strftime("%Y-%m-%dT%H%M%SZ") snapshot_path = output_dir / f"portfolio-truth-{snapshot_stamp}.json" @@ -206,6 +231,15 @@ def _content_changed(path: Path, content: str) -> bool: return path.read_text() != content +def _previous_generated_at(latest_path: Path) -> str | None: + try: + payload = json.loads(latest_path.read_text()) + except (FileNotFoundError, json.JSONDecodeError, OSError): + return None + value = payload.get("generated_at") if isinstance(payload, dict) else None + return value if isinstance(value, str) and value else None + + def _guard_against_notion_context_drop( source_summary: dict[str, object], *, diff --git a/src/portfolio_truth_reconcile.py b/src/portfolio_truth_reconcile.py index dc13b2f..a4876d0 100644 --- a/src/portfolio_truth_reconcile.py +++ b/src/portfolio_truth_reconcile.py @@ -3,6 +3,7 @@ import json import logging import re +import hashlib from collections import Counter from dataclasses import dataclass from datetime import datetime, timezone @@ -23,6 +24,7 @@ load_safe_notion_project_context, ) from src.portfolio_truth_types import ( + DERIVATION_POLICY_VERSION, SCHEMA_VERSION, AdvisoryFields, DeclaredFields, @@ -220,6 +222,8 @@ def build_portfolio_truth_snapshot( release_count_by_name: dict[str, int] | None = None, security_alerts_by_name: dict[str, dict] | None = None, repo_status_by_name: dict[str, dict] | None = None, + producer: dict[str, Any] | None = None, + prior_notion_generated_at: str | None = None, ) -> PortfolioTruthBuildResult: now = now or datetime.now(timezone.utc) catalog_data = load_portfolio_catalog(catalog_path) @@ -301,12 +305,68 @@ def build_portfolio_truth_snapshot( precedence_matrix=PRECEDENCE_MATRIX, warnings=warnings, projects=projects, + derivation_policy_version=DERIVATION_POLICY_VERSION, + producer=producer or {}, + inputs=_build_input_envelope( + workspace_root=workspace_root, + catalog_data=catalog_data, + now=now, + include_notion=include_notion, + notion_context_rows=len(notion_context), + notion_context_carried_forward=notion_context_carried_forward, + prior_notion_generated_at=prior_notion_generated_at, + ), ) return PortfolioTruthBuildResult( snapshot=snapshot, catalog_data=catalog_data, legacy_rows=legacy_rows ) +def _build_input_envelope( + *, + workspace_root: Path, + catalog_data: dict[str, Any], + now: datetime, + include_notion: bool, + notion_context_rows: int, + notion_context_carried_forward: bool, + prior_notion_generated_at: str | None, +) -> dict[str, Any]: + resolved_catalog = Path(str(catalog_data.get("path") or "")) + catalog_hash = ( + hashlib.sha256(resolved_catalog.read_bytes()).hexdigest() + if resolved_catalog.is_file() + else None + ) + if not include_notion or notion_context_rows == 0: + notion_mode = "unavailable" + notion_observed_at = None + elif notion_context_carried_forward: + notion_mode = "carried-forward" if prior_notion_generated_at else "unavailable" + notion_observed_at = prior_notion_generated_at + else: + notion_mode = "live" + notion_observed_at = now.isoformat() + return { + "catalog": { + "source_id": "portfolio-catalog", + "sha256": catalog_hash, + "observed_at": now.isoformat(), + }, + "workspace": { + "source_id": "projects-root", + "observed_at": now.isoformat(), + }, + "notion": { + "mode": notion_mode, + "observed_at": notion_observed_at, + "carried_from_generated_at": ( + prior_notion_generated_at if notion_context_carried_forward else None + ), + }, + } + + def load_prior_notion_context(latest_path: Path) -> dict[str, dict[str, str]]: """Reconstruct a Notion project-context map from a previously published portfolio-truth artifact, keyed identically to live Notion context diff --git a/src/portfolio_truth_types.py b/src/portfolio_truth_types.py index ac41918..734d7c2 100644 --- a/src/portfolio_truth_types.py +++ b/src/portfolio_truth_types.py @@ -6,7 +6,9 @@ from pathlib import Path from typing import Any -SCHEMA_VERSION = "0.7.0" +SCHEMA_VERSION = "0.8.0" +LEGACY_SCHEMA_VERSIONS = {"0.7.0"} +DERIVATION_POLICY_VERSION = "portfolio_attention.v2" # The published "latest" portfolio-truth artifact. The producer # (portfolio_truth_publish) writes it; every reader resolves it through @@ -294,6 +296,16 @@ class PortfolioTruthSnapshot: precedence_matrix: dict[str, list[str]] warnings: list[str] projects: list[PortfolioTruthProject] + derivation_policy_version: str = DERIVATION_POLICY_VERSION + producer: dict[str, Any] = field(default_factory=dict) + inputs: dict[str, Any] = field(default_factory=dict) + coverage: list[dict[str, Any]] = field(default_factory=list) + exclusions: dict[str, Any] = field( + default_factory=lambda: { + "policy_version": "workspace_discovery.v1", + "counts": {}, + } + ) rollups: PortfolioTruthRollups = field(init=False) def __post_init__(self) -> None: @@ -305,6 +317,11 @@ def to_dict(self) -> dict[str, Any]: return { "schema_version": self.schema_version, "generated_at": _serialize_datetime(self.generated_at), + "derivation_policy_version": self.derivation_policy_version, + "producer": dict(self.producer), + "inputs": dict(self.inputs), + "coverage": list(self.coverage), + "exclusions": dict(self.exclusions), "workspace_root": self.workspace_root, "source_summary": self.source_summary, "precedence_matrix": self.precedence_matrix, diff --git a/src/portfolio_truth_validate.py b/src/portfolio_truth_validate.py index 7747a87..d7bb52c 100644 --- a/src/portfolio_truth_validate.py +++ b/src/portfolio_truth_validate.py @@ -10,6 +10,7 @@ ) from src.portfolio_truth_render import registry_project_labels from src.portfolio_truth_types import ( + DERIVATION_POLICY_VERSION, SCHEMA_VERSION, VALID_ACTIVITY_STATUS, VALID_ATTENTION_STATES, @@ -26,6 +27,12 @@ def validate_truth_snapshot(snapshot: PortfolioTruthSnapshot) -> None: if snapshot.schema_version != SCHEMA_VERSION: raise ValueError(f"Unexpected schema version: {snapshot.schema_version}") + if snapshot.derivation_policy_version != DERIVATION_POLICY_VERSION: + raise ValueError( + "Unexpected derivation policy version: " + f"{snapshot.derivation_policy_version}" + ) + _validate_contract_envelope(snapshot) seen_keys: set[str] = set() for project in snapshot.projects: key = project.identity.project_key @@ -91,6 +98,39 @@ def validate_truth_snapshot(snapshot: PortfolioTruthSnapshot) -> None: raise ValueError(f"Invalid doctor standard for {key}: {doctor_std}") +def _validate_contract_envelope(snapshot: PortfolioTruthSnapshot) -> None: + producer = snapshot.producer + if producer: + required = { + "repository", + "commit", + "ref", + "checkout_role", + "worktree_clean", + "verified_at", + } + missing = sorted(required - producer.keys()) + if missing: + raise ValueError(f"Producer evidence is missing fields: {missing}") + commit = producer.get("commit") + if not isinstance(commit, str) or len(commit) != 40 or any( + char not in "0123456789abcdef" for char in commit + ): + raise ValueError("Producer commit must be a lowercase 40-character SHA.") + if producer.get("worktree_clean") is not True: + raise ValueError("Canonical producer evidence must declare a clean worktree.") + notion = snapshot.inputs.get("notion") if isinstance(snapshot.inputs, dict) else None + if not isinstance(notion, dict): + raise ValueError("Portfolio truth inputs.notion is required.") + mode = notion.get("mode") + if mode not in {"live", "carried-forward", "unavailable"}: + raise ValueError(f"Invalid Notion input mode: {mode}") + if mode == "carried-forward" and not notion.get("carried_from_generated_at"): + raise ValueError("Carried-forward Notion input requires an origin timestamp.") + if mode == "live" and notion.get("carried_from_generated_at") is not None: + raise ValueError("Live Notion input cannot declare a carried-forward origin.") + + def validate_publish_targets( *, workspace_root: Path, diff --git a/src/producer_preflight.py b/src/producer_preflight.py new file mode 100644 index 0000000..40bac70 --- /dev/null +++ b/src/producer_preflight.py @@ -0,0 +1,150 @@ +from __future__ import annotations + +import argparse +import json +import subprocess +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any +from urllib.parse import urlparse + + +PREFLIGHT_SCHEMA_VERSION = "ghra_producer_preflight.v1" + + +@dataclass(frozen=True) +class ProducerEvidence: + repository: str + commit: str + ref: str + checkout_role: str + worktree_clean: bool + verified_at: datetime + + def to_dict(self) -> dict[str, Any]: + return { + "repository": self.repository, + "commit": self.commit, + "ref": self.ref, + "checkout_role": self.checkout_role, + "worktree_clean": self.worktree_clean, + "verified_at": self.verified_at.isoformat(), + } + + +@dataclass(frozen=True) +class ProducerPreflightResult: + state: str + checks: dict[str, str] + evidence: ProducerEvidence | None = None + detail: str = "" + + def to_dict(self) -> dict[str, Any]: + payload: dict[str, Any] = { + "schema_version": PREFLIGHT_SCHEMA_VERSION, + "state": self.state, + "checks": dict(self.checks), + } + if self.evidence is not None: + payload.update(self.evidence.to_dict()) + if self.detail: + payload["detail"] = self.detail + return payload + + +def inspect_canonical_producer( + *, + repo_root: Path, + expected_repository: str, + expected_ref: str, + checkout_role: str, + now: datetime | None = None, +) -> ProducerPreflightResult: + checks: dict[str, str] = {} + try: + repository = _origin_repository(repo_root) + commit = _git(repo_root, "rev-parse", "HEAD") + expected_commit = _git(repo_root, "rev-parse", "--verify", expected_ref) + status = _git(repo_root, "status", "--porcelain", "--untracked-files=all") + except (OSError, subprocess.CalledProcessError) as exc: + return ProducerPreflightResult( + state="unknown", + checks={"expected_ref_available": "unknown"}, + detail=str(exc), + ) + + checks["repository_identity"] = ( + "pass" if repository == expected_repository else "fail" + ) + checks["worktree_clean"] = "pass" if not status else "fail" + checks["expected_ref_available"] = "pass" + checks["head_matches_expected_ref"] = ( + "pass" if commit == expected_commit else "fail" + ) + state = "pass" if all(value == "pass" for value in checks.values()) else "fail" + evidence = ProducerEvidence( + repository=repository, + commit=commit, + ref=expected_ref, + checkout_role=checkout_role, + worktree_clean=not status, + verified_at=now or datetime.now(UTC), + ) + return ProducerPreflightResult(state=state, checks=checks, evidence=evidence) + + +def verify_evidence_still_current(repo_root: Path, evidence: ProducerEvidence) -> None: + current = _git(repo_root, "rev-parse", "HEAD") + if current != evidence.commit: + raise ValueError( + "Producer HEAD changed after preflight: " + f"verified={evidence.commit}; current={current}" + ) + + +def _origin_repository(repo_root: Path) -> str: + remote = _git(repo_root, "remote", "get-url", "origin") + if remote.startswith("git@") and ":" in remote: + path = remote.split(":", 1)[1] + else: + path = urlparse(remote).path + return path.strip("/").removesuffix(".git") + + +def _git(repo_root: Path, *args: str) -> str: + return subprocess.run( + ["git", "-C", str(repo_root), *args], + check=True, + capture_output=True, + text=True, + ).stdout.strip() + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Validate canonical GHRA producer identity.") + parser.add_argument("--repo-root", type=Path, required=True) + parser.add_argument("--expected-repository", required=True) + parser.add_argument("--expected-ref", required=True) + parser.add_argument("--checkout-role", default="canonical-automation") + parser.add_argument("--json", action="store_true") + args = parser.parse_args(argv) + result = inspect_canonical_producer( + repo_root=args.repo_root, + expected_repository=args.expected_repository, + expected_ref=args.expected_ref, + checkout_role=args.checkout_role, + ) + if args.json: + print(json.dumps(result.to_dict(), indent=2)) + else: + print(f"GHRA producer preflight: {result.state.upper()}") + for check, state in result.checks.items(): + print(f"- {check}: {state}") + if result.detail: + print(result.detail) + return 0 if result.state == "pass" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/fixtures/contract_linter/canonical_queue_truncated.json b/tests/fixtures/contract_linter/canonical_queue_truncated.json new file mode 100644 index 0000000..1b9c326 --- /dev/null +++ b/tests/fixtures/contract_linter/canonical_queue_truncated.json @@ -0,0 +1,5 @@ +{ + "fixture_id": "canonical_queue_truncated", + "queue": {"total": 6, "emitted": 5, "omitted": 1, "truncated": true}, + "expected": {"CL-QUEUE-001": "pass"} +} diff --git a/tests/fixtures/contract_linter/historical_evidence_artifact.json b/tests/fixtures/contract_linter/historical_evidence_artifact.json new file mode 100644 index 0000000..0c8ecd7 --- /dev/null +++ b/tests/fixtures/contract_linter/historical_evidence_artifact.json @@ -0,0 +1,5 @@ +{ + "fixture_id": "historical_evidence_artifact", + "artifact_role": "evidence-history", + "expected": {"CL-PROD-001": "pass"} +} diff --git a/tests/fixtures/contract_linter/known_good_0_8.json b/tests/fixtures/contract_linter/known_good_0_8.json new file mode 100644 index 0000000..5fb168f --- /dev/null +++ b/tests/fixtures/contract_linter/known_good_0_8.json @@ -0,0 +1,16 @@ +{ + "fixture_id": "known_good_0_8", + "projects": [ + { + "project_key": "declared-project", + "catalog_declared": true, + "artifact_catalog_provenance": true, + "attention_state": "active-product" + } + ], + "expected": { + "CL-PROD-001": "pass", + "CL-INP-001": "pass", + "CL-DECL-001": "pass" + } +} diff --git a/tests/fixtures/contract_linter/legacy_0_7_without_envelope.json b/tests/fixtures/contract_linter/legacy_0_7_without_envelope.json new file mode 100644 index 0000000..43301b1 --- /dev/null +++ b/tests/fixtures/contract_linter/legacy_0_7_without_envelope.json @@ -0,0 +1,5 @@ +{ + "fixture_id": "legacy_0_7_without_envelope", + "schema_version": "0.7.0", + "expected": {"CL-PROD-001": "unknown", "CL-INP-001": "unknown"} +} diff --git a/tests/fixtures/contract_linter/mixed_stale_and_unresolved.json b/tests/fixtures/contract_linter/mixed_stale_and_unresolved.json new file mode 100644 index 0000000..233328d --- /dev/null +++ b/tests/fixtures/contract_linter/mixed_stale_and_unresolved.json @@ -0,0 +1,21 @@ +{ + "fixture_id": "mixed_stale_and_unresolved", + "projects": [ + { + "project_key": "stale-one", + "catalog_declared": true, + "artifact_catalog_provenance": false, + "attention_state": "decision-needed" + }, + { + "project_key": "unresolved-one", + "catalog_declared": false, + "artifact_catalog_provenance": false, + "attention_state": "decision-needed" + } + ], + "expected": { + "CL-DECL-001:stale-one": "fail", + "CL-DECL-001:unresolved-one": "unknown" + } +} diff --git a/tests/fixtures/contract_linter/notion_carried_fresh.json b/tests/fixtures/contract_linter/notion_carried_fresh.json new file mode 100644 index 0000000..2420d0c --- /dev/null +++ b/tests/fixtures/contract_linter/notion_carried_fresh.json @@ -0,0 +1,5 @@ +{ + "fixture_id": "notion_carried_fresh", + "source": {"mode": "carried-forward", "age_hours": 12, "origin_known": true}, + "expected": {"CL-FRESH-002": "pass"} +} diff --git a/tests/fixtures/contract_linter/notion_carried_origin_unknown.json b/tests/fixtures/contract_linter/notion_carried_origin_unknown.json new file mode 100644 index 0000000..ab9aeeb --- /dev/null +++ b/tests/fixtures/contract_linter/notion_carried_origin_unknown.json @@ -0,0 +1,5 @@ +{ + "fixture_id": "notion_carried_origin_unknown", + "source": {"mode": "unavailable", "age_hours": null, "origin_known": false}, + "expected": {"CL-FRESH-002": "unknown"} +} diff --git a/tests/fixtures/contract_linter/notion_carried_stale.json b/tests/fixtures/contract_linter/notion_carried_stale.json new file mode 100644 index 0000000..5fb3eb0 --- /dev/null +++ b/tests/fixtures/contract_linter/notion_carried_stale.json @@ -0,0 +1,5 @@ +{ + "fixture_id": "notion_carried_stale", + "source": {"mode": "carried-forward", "age_hours": 49, "origin_known": true}, + "expected": {"CL-FRESH-002": "warn"} +} diff --git a/tests/fixtures/contract_linter/notion_live.json b/tests/fixtures/contract_linter/notion_live.json new file mode 100644 index 0000000..d7cf741 --- /dev/null +++ b/tests/fixtures/contract_linter/notion_live.json @@ -0,0 +1,5 @@ +{ + "fixture_id": "notion_live", + "source": {"mode": "live", "age_hours": 0, "origin_known": true}, + "expected": {"CL-FRESH-002": "pass"} +} diff --git a/tests/fixtures/contract_linter/stale_producer_catalog_miss.json b/tests/fixtures/contract_linter/stale_producer_catalog_miss.json new file mode 100644 index 0000000..f8307cc --- /dev/null +++ b/tests/fixtures/contract_linter/stale_producer_catalog_miss.json @@ -0,0 +1,16 @@ +{ + "fixture_id": "stale_producer_catalog_miss", + "projects": [ + { + "project_key": "declared-project", + "catalog_declared": true, + "artifact_catalog_provenance": false, + "attention_state": "decision-needed" + } + ], + "expected": { + "CL-PROD-001": "fail", + "CL-INP-001": "fail", + "CL-DECL-001": "fail" + } +} diff --git a/tests/test_contract_linter_fixtures.py b/tests/test_contract_linter_fixtures.py new file mode 100644 index 0000000..cd80b3c --- /dev/null +++ b/tests/test_contract_linter_fixtures.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +import json +from pathlib import Path + + +FIXTURE_ROOT = Path(__file__).parent / "fixtures" / "contract_linter" + + +def test_contract_linter_fixtures_have_stable_expected_outcomes() -> None: + fixtures = { + path.stem: json.loads(path.read_text()) + for path in sorted(FIXTURE_ROOT.glob("*.json")) + } + assert set(fixtures) == { + "canonical_queue_truncated", + "historical_evidence_artifact", + "known_good_0_8", + "legacy_0_7_without_envelope", + "mixed_stale_and_unresolved", + "notion_carried_fresh", + "notion_carried_origin_unknown", + "notion_carried_stale", + "notion_live", + "stale_producer_catalog_miss", + } + for fixture_id, payload in fixtures.items(): + assert payload["fixture_id"] == fixture_id + assert payload["expected"] + assert all(key.startswith("CL-") for key in payload["expected"]) + + +def test_mixed_fixture_does_not_conflate_unknown_with_stale() -> None: + payload = json.loads( + (FIXTURE_ROOT / "mixed_stale_and_unresolved.json").read_text() + ) + assert payload["expected"]["CL-DECL-001:stale-one"] == "fail" + assert payload["expected"]["CL-DECL-001:unresolved-one"] == "unknown" diff --git a/tests/test_operator_os_seam_linter.py b/tests/test_operator_os_seam_linter.py index 98dbefb..678b74f 100644 --- a/tests/test_operator_os_seam_linter.py +++ b/tests/test_operator_os_seam_linter.py @@ -161,6 +161,103 @@ def test_schema_pin_passes(tmp_path: Path) -> None: assert result.passed +def test_legacy_schema_remains_readable_during_migration(tmp_path: Path) -> None: + truth, markdown = _passing_paths(tmp_path) + _write_truth(truth, schema_version="0.7.0") + + result = lint_operator_os_seams(truth_path=truth, markdown_paths=markdown, now=NOW) + + assert result.passed + + +def test_contract_shadow_marks_legacy_lineage_unknown_without_failing( + tmp_path: Path, +) -> None: + truth, markdown = _passing_paths(tmp_path) + _write_truth(truth, schema_version="0.7.0") + + result = lint_operator_os_seams( + truth_path=truth, + markdown_paths=markdown, + contract_shadow=True, + now=NOW, + ) + + assert result.passed + assert result.state == "unknown" + assert {finding.check for finding in result.findings} == { + "CL-PROD-001", + "CL-INP-001", + "CL-COUNT-001", + } + + +def test_contract_shadow_fails_unreconciled_decision_counts(tmp_path: Path) -> None: + truth, markdown = _passing_paths(tmp_path) + payload = json.loads(truth.read_text()) + payload.update( + { + "producer": {}, + "inputs": { + "notion": {"mode": "unavailable"}, + "catalog": {"sha256": None}, + }, + "source_summary": {"attention_state_counts": {"decision-needed": 1}}, + "rollups": {"decision": {"decision_needed_count": 0}}, + } + ) + truth.write_text(json.dumps(payload)) + + result = lint_operator_os_seams( + truth_path=truth, + markdown_paths=markdown, + contract_shadow=True, + catalog_path=tmp_path / "missing-catalog.yaml", + now=NOW, + ) + + assert not result.passed + assert result.state == "fail" + assert any(finding.check == "CL-COUNT-001" for finding in result.findings) + + +def test_contract_shadow_warns_for_carried_notion_older_than_48_hours( + tmp_path: Path, +) -> None: + truth, markdown = _passing_paths(tmp_path) + payload = json.loads(truth.read_text()) + payload.update( + { + "producer": {}, + "inputs": { + "notion": { + "mode": "carried-forward", + "carried_from_generated_at": "2026-07-01T00:00:00+00:00", + }, + "catalog": {"sha256": None}, + }, + "source_summary": {"attention_state_counts": {"decision-needed": 0}}, + "rollups": {"decision": {"decision_needed_count": 0}}, + } + ) + truth.write_text(json.dumps(payload)) + + result = lint_operator_os_seams( + truth_path=truth, + markdown_paths=markdown, + contract_shadow=True, + catalog_path=tmp_path / "missing-catalog.yaml", + now=NOW, + ) + + assert result.passed + assert result.state == "warn" + assert any( + finding.check == "CL-FRESH-002" and finding.level == "warn" + for finding in result.findings + ) + + def test_schema_pin_mismatch_fails(tmp_path: Path) -> None: truth, markdown = _passing_paths(tmp_path) _write_truth(truth, schema_version="0.6.0") diff --git a/tests/test_portfolio_truth.py b/tests/test_portfolio_truth.py index dec86ca..7ff162e 100644 --- a/tests/test_portfolio_truth.py +++ b/tests/test_portfolio_truth.py @@ -14,7 +14,7 @@ apply_context_recovery_plan, build_context_recovery_plan, ) -from src.portfolio_truth_publish import publish_portfolio_truth +from src.portfolio_truth_publish import PortfolioTruthPublishError, publish_portfolio_truth from src.portfolio_truth_reconcile import build_portfolio_truth_snapshot from src.portfolio_truth_render import ( render_portfolio_report_markdown, @@ -370,7 +370,10 @@ def test_truth_snapshot_respects_declared_and_derived_fields( assert gamma.identity.section_marker == "iOS Projects" assert gamma.derived.stack == ["Swift"] - assert result.snapshot.schema_version == "0.7.0" + assert result.snapshot.schema_version == "0.8.0" + assert result.snapshot.derivation_policy_version == "portfolio_attention.v2" + assert result.snapshot.inputs["catalog"]["sha256"] + assert result.snapshot.inputs["notion"]["mode"] == "unavailable" assert result.snapshot.source_summary["attention_state_counts"]["active-product"] == 1 assert result.snapshot.source_summary["attention_state_counts"]["parked"] == 1 @@ -1368,6 +1371,38 @@ def _boom(_snapshot, _latest_json_path): assert not list(output_dir.glob("*.tmp")) +def test_publish_requires_producer_evidence_before_touching_outputs( + portfolio_workspace: Path, + portfolio_catalog: Path, + legacy_registry: Path, + tmp_path: Path, +) -> None: + output_dir = tmp_path / "output" + registry_output = portfolio_workspace / "project-registry.md" + report_output = portfolio_workspace / "PORTFOLIO-AUDIT-REPORT.md" + registry_output.write_text("sentinel-registry\n") + report_output.write_text("sentinel-report\n") + + with pytest.raises( + PortfolioTruthPublishError, + match="requires validated producer evidence", + ): + 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, + require_producer_evidence=True, + ) + + assert registry_output.read_text() == "sentinel-registry\n" + assert report_output.read_text() == "sentinel-report\n" + assert not output_dir.exists() + + def test_publish_refuses_to_drop_existing_notion_context( portfolio_workspace: Path, portfolio_catalog: Path, diff --git a/tests/test_producer_preflight.py b/tests/test_producer_preflight.py new file mode 100644 index 0000000..5a7f379 --- /dev/null +++ b/tests/test_producer_preflight.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +import subprocess +from datetime import UTC, datetime +from pathlib import Path + +import pytest + +from src.producer_preflight import ( + ProducerEvidence, + inspect_canonical_producer, + verify_evidence_still_current, +) + + +def _git(repo: Path, *args: str) -> str: + return subprocess.run( + ["git", "-C", str(repo), *args], + check=True, + capture_output=True, + text=True, + ).stdout.strip() + + +def _repo(tmp_path: Path) -> tuple[Path, str]: + repo = tmp_path / "repo" + repo.mkdir() + _git(repo, "init", "-b", "main") + _git(repo, "config", "user.email", "tests@example.invalid") + _git(repo, "config", "user.name", "Tests") + (repo / "README.md").write_text("fixture\n") + _git(repo, "add", "README.md") + _git(repo, "commit", "-m", "fixture") + _git(repo, "remote", "add", "origin", "git@github.com:saagpatel/GithubRepoAuditor.git") + commit = _git(repo, "rev-parse", "HEAD") + _git(repo, "update-ref", "refs/remotes/origin/main", commit) + return repo, commit + + +def test_canonical_producer_passes_for_clean_matching_ref(tmp_path: Path) -> None: + repo, commit = _repo(tmp_path) + result = inspect_canonical_producer( + repo_root=repo, + expected_repository="saagpatel/GithubRepoAuditor", + expected_ref="refs/remotes/origin/main", + checkout_role="canonical-automation", + now=datetime(2026, 7, 10, tzinfo=UTC), + ) + assert result.state == "pass" + assert result.evidence is not None + assert result.evidence.commit == commit + assert all(state == "pass" for state in result.checks.values()) + + +def test_canonical_producer_fails_dirty_worktree(tmp_path: Path) -> None: + repo, _ = _repo(tmp_path) + (repo / "untracked.txt").write_text("dirty\n") + result = inspect_canonical_producer( + repo_root=repo, + expected_repository="saagpatel/GithubRepoAuditor", + expected_ref="refs/remotes/origin/main", + checkout_role="canonical-automation", + ) + assert result.state == "fail" + assert result.checks["worktree_clean"] == "fail" + + +def test_canonical_producer_missing_ref_is_unknown(tmp_path: Path) -> None: + repo, _ = _repo(tmp_path) + result = inspect_canonical_producer( + repo_root=repo, + expected_repository="saagpatel/GithubRepoAuditor", + expected_ref="refs/remotes/origin/missing", + checkout_role="canonical-automation", + ) + assert result.state == "unknown" + + +def test_evidence_rejects_head_change(tmp_path: Path) -> None: + repo, commit = _repo(tmp_path) + evidence = ProducerEvidence( + repository="saagpatel/GithubRepoAuditor", + commit=commit, + ref="refs/remotes/origin/main", + checkout_role="canonical-automation", + worktree_clean=True, + verified_at=datetime.now(UTC), + ) + (repo / "README.md").write_text("changed\n") + _git(repo, "add", "README.md") + _git(repo, "commit", "-m", "move head") + with pytest.raises(ValueError, match="HEAD changed after preflight"): + verify_evidence_still_current(repo, evidence)