diff --git a/.gitignore b/.gitignore index 59b3063..7e8f782 100644 --- a/.gitignore +++ b/.gitignore @@ -2,6 +2,8 @@ .pacta/ .pacta-*/ .pytest_cache/ +artifacts*/ +repos/ __pycache__/ *.py[cod] *.egg-info/ diff --git a/AGENTS.md b/AGENTS.md index 4551b92..605e5a4 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -11,3 +11,4 @@ Guidance for future Codex runs in this repository: - Do not silently lower risk ratings. A lower score must explain the failed or missing evidence. - Do not hide proof failures behind warnings. - Keep claim cards machine-readable and reports explicit about proven claims, preconditions, exclusions, trusted base, and residual risk. +- Consequence-producing commands must be policy gated. Do not build wallet or trading-agent artifacts from `R3` arithmetic evidence. diff --git a/README.md b/README.md index 0e72add..2024899 100644 --- a/README.md +++ b/README.md @@ -38,8 +38,22 @@ pacta audit --repo ./repos/dalek-ed25519-verified pacta lean-check --repo ./repos/dalek-ed25519-verified pacta report --claims claims.yaml --out report.md pacta score --claims claims.yaml +pacta agent --config examples/repos.yaml --repo-name dalek-ed25519-verified --offline-fixture --action build-library +pacta agent --config examples/repos.yaml --repo-name dalek-ed25519-verified --clone --run-axioms --action build-library --artifact-dir artifacts-live +pacta agent --claims claims.yaml --action build-wallet-demo ``` +## Consequence Engine + +`pacta agent` turns evaluation into an operational consequence. + +- `build-library` requires `R3` by default. It builds a small Rust proof-gated component capsule under `artifacts/`. The capsule embeds the claim card and exposes whether downstream automation may use the component for lower-layer cryptographic code only. +- `build-wallet-demo` requires `R4`. An `R3` Ed25519 arithmetic claim will refuse this action and write a machine-readable denial artifact instead of building a wallet. + +This is intentional. Arithmetic proof evidence can authorize a constrained lower-layer library decision, but it must not contaminate wallet, transaction, custody, or trading-agent risk scoring. + +In live mode, `--clone --run-axioms` downloads the configured repository, replays the local Lean checks, runs the axiom audit, writes `claims.yaml` and `report.md`, and only builds the capsule if the resulting score satisfies the policy threshold. Failed replay is a hard consequence: no artifact is built. + ## Truth Boundary The Ed25519 repositories should not be marketed as fully verified wallets or fully verified Ed25519 end-to-end. The strongest current claim is lower-layer and theorem-bound: diff --git a/src/pacta/agent.py b/src/pacta/agent.py new file mode 100644 index 0000000..cc1466b --- /dev/null +++ b/src/pacta/agent.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from .artifact import ArtifactBuildResult, build_proof_gated_capsule, write_denied_wallet_demo +from .risk import risk_at_least + + +@dataclass(slots=True) +class AgentDecision: + requested_action: str + allowed: bool + risk_level: str + rationale: str + consequences: list[str] = field(default_factory=list) + artifact: ArtifactBuildResult | None = None + + def to_dict(self) -> dict[str, Any]: + return { + "requested_action": self.requested_action, + "allowed": self.allowed, + "risk_level": self.risk_level, + "rationale": self.rationale, + "consequences": self.consequences, + "artifact": self.artifact.to_dict() if self.artifact else None, + } + + +def run_agent_action( + card: dict[str, Any], + action: str, + artifact_root: str | Path, + minimum_build_risk: str = "R3", + timeout: int = 120, + dry_run: bool = False, +) -> AgentDecision: + risk = card.get("risk") or {} + level = str(risk.get("level") or "R0") + if action == "build-library": + return _build_library(card, artifact_root, level, minimum_build_risk, timeout, dry_run) + if action == "build-wallet-demo": + return _build_wallet_demo(card, artifact_root, level, dry_run) + raise ValueError(f"Unsupported agent action: {action}") + + +def write_decision(decision: AgentDecision, artifact_root: str | Path, component: str) -> Path: + component_dir = component_artifact_dir(artifact_root, component) + component_dir.mkdir(parents=True, exist_ok=True) + path = component_dir / f"agent-decision-{_slug(decision.requested_action)}.json" + path.write_text(json.dumps(decision.to_dict(), indent=2, sort_keys=True) + "\n", encoding="utf-8") + return path + + +def component_artifact_dir(artifact_root: str | Path, component: str) -> Path: + return Path(artifact_root) / _slug(component) + + +def _build_library( + card: dict[str, Any], + artifact_root: str | Path, + level: str, + minimum_build_risk: str, + timeout: int, + dry_run: bool, +) -> AgentDecision: + if not risk_at_least(level, minimum_build_risk): + return AgentDecision( + requested_action="build-library", + allowed=False, + risk_level=level, + rationale=f"Risk level {level} is below the build threshold {minimum_build_risk}.", + consequences=[ + "No library artifact was built.", + "Component remains in evaluation/quarantine state.", + ], + ) + artifact = build_proof_gated_capsule(card, artifact_root, timeout=timeout, dry_run=dry_run) + return AgentDecision( + requested_action="build-library", + allowed=artifact.built if not dry_run else True, + risk_level=level, + rationale=( + f"Risk level {level} satisfies the {minimum_build_risk} threshold for a lower-layer " + "proof-gated library capsule." + ), + consequences=[ + "Generated a consumable proof-gated component capsule.", + "Capsule permits lower-layer cryptographic use only inside the recorded theorem/backend boundary.", + "Capsule explicitly does not permit wallet or trading-agent deployment.", + ], + artifact=artifact, + ) + + +def _build_wallet_demo(card: dict[str, Any], artifact_root: str | Path, level: str, dry_run: bool) -> AgentDecision: + if not risk_at_least(level, "R4"): + reason = f"Risk level {level} is below R4; wallet demos require end-to-end primitive/API coverage." + artifact = write_denied_wallet_demo(card, artifact_root, reason, dry_run=dry_run) + return AgentDecision( + requested_action="build-wallet-demo", + allowed=False, + risk_level=level, + rationale=reason, + consequences=[ + "No wallet was built.", + "A denial artifact was written so the refusal is machine-readable.", + ], + artifact=artifact, + ) + artifact = build_proof_gated_capsule(card, artifact_root, dry_run=dry_run) + return AgentDecision( + requested_action="build-wallet-demo", + allowed=artifact.built if not dry_run else True, + risk_level=level, + rationale="Risk level permits a wallet demo scaffold.", + consequences=[ + "Generated only a policy capsule; production wallet construction still requires separate controls.", + ], + artifact=artifact, + ) + + +def _slug(value: str) -> str: + return "".join(ch if ch.isalnum() or ch in "._-" else "-" for ch in value).strip("-") or "component" diff --git a/src/pacta/artifact.py b/src/pacta/artifact.py new file mode 100644 index 0000000..e1d77c9 --- /dev/null +++ b/src/pacta/artifact.py @@ -0,0 +1,217 @@ +from __future__ import annotations + +import json +import re +import shutil +import subprocess +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + + +@dataclass(slots=True) +class ArtifactBuildResult: + artifact_dir: Path + crate_dir: Path | None + built: bool + log_path: Path | None + diagnostics: list[str] = field(default_factory=list) + + def to_dict(self) -> dict[str, Any]: + return { + "artifact_dir": str(self.artifact_dir), + "crate_dir": str(self.crate_dir) if self.crate_dir else None, + "built": self.built, + "log_path": str(self.log_path) if self.log_path else None, + "diagnostics": self.diagnostics, + } + + +def build_proof_gated_capsule( + card: dict[str, Any], + artifact_root: str | Path, + timeout: int = 120, + dry_run: bool = False, +) -> ArtifactBuildResult: + component = str(card.get("component") or "unknown-component") + artifact_dir = Path(artifact_root) / _slug(component) + crate_name = f"pacta_gated_{_crate_suffix(component)}" + crate_dir = artifact_dir / crate_name + if dry_run: + return ArtifactBuildResult( + artifact_dir=artifact_dir, + crate_dir=crate_dir, + built=False, + log_path=None, + diagnostics=["dry-run: proof-gated capsule was not written or built."], + ) + + src_dir = crate_dir / "src" + src_dir.mkdir(parents=True, exist_ok=True) + claims_text = json.dumps(card, indent=2, sort_keys=True) + "\n" + (artifact_dir / "claims.json").write_text(claims_text, encoding="utf-8") + (crate_dir / "claims.json").write_text(claims_text, encoding="utf-8") + (crate_dir / "Cargo.toml").write_text(_cargo_toml(crate_name, component), encoding="utf-8") + (src_dir / "lib.rs").write_text(_lib_rs(card), encoding="utf-8") + (crate_dir / "README.md").write_text(_crate_readme(card), encoding="utf-8") + + cargo = shutil.which("cargo") + log_path = artifact_dir / "capsule-build.log" + if not cargo: + log_path.write_text("cargo was not found on PATH; capsule source was written but not built.\n", encoding="utf-8") + return ArtifactBuildResult( + artifact_dir=artifact_dir, + crate_dir=crate_dir, + built=False, + log_path=log_path, + diagnostics=["cargo was not found on PATH."], + ) + + cmd = [cargo, "build", "--release", "--manifest-path", str(crate_dir / "Cargo.toml")] + try: + completed = subprocess.run( + cmd, + check=False, + capture_output=True, + text=True, + timeout=timeout, + ) + log_path.write_text( + "$ " + " ".join(cmd) + "\n\n" + completed.stdout + completed.stderr + f"\nexit_code: {completed.returncode}\n", + encoding="utf-8", + ) + except subprocess.TimeoutExpired: + log_path.write_text("$ " + " ".join(cmd) + f"\n\nTimed out after {timeout}s\n", encoding="utf-8") + return ArtifactBuildResult( + artifact_dir=artifact_dir, + crate_dir=crate_dir, + built=False, + log_path=log_path, + diagnostics=[f"cargo build timed out after {timeout}s."], + ) + return ArtifactBuildResult( + artifact_dir=artifact_dir, + crate_dir=crate_dir, + built=completed.returncode == 0, + log_path=log_path, + diagnostics=[] if completed.returncode == 0 else [f"cargo build exited with {completed.returncode}."], + ) + + +def write_denied_wallet_demo( + card: dict[str, Any], + artifact_root: str | Path, + reason: str, + dry_run: bool = False, +) -> ArtifactBuildResult: + component = str(card.get("component") or "unknown-component") + artifact_dir = Path(artifact_root) / _slug(component) / "wallet-demo-denied" + if dry_run: + return ArtifactBuildResult( + artifact_dir=artifact_dir, + crate_dir=None, + built=False, + log_path=None, + diagnostics=["dry-run: wallet denial artifact was not written."], + ) + artifact_dir.mkdir(parents=True, exist_ok=True) + decision = { + "component": component, + "requested_action": "build-wallet-demo", + "allowed": False, + "reason": reason, + "risk": card.get("risk", {}), + "message": "PACTA refused to create a wallet demo from insufficient proof coverage.", + } + (artifact_dir / "decision.json").write_text(json.dumps(decision, indent=2, sort_keys=True) + "\n", encoding="utf-8") + (artifact_dir / "README.md").write_text( + "# Wallet Demo Denied\n\n" + "PACTA refused to build a wallet demo for this component. Field or curve arithmetic evidence " + "does not establish wallet policy, key custody, transaction construction, encoding, hashing, " + "or full signature verification safety.\n\n" + f"Reason: {reason}\n", + encoding="utf-8", + ) + return ArtifactBuildResult( + artifact_dir=artifact_dir, + crate_dir=None, + built=False, + log_path=artifact_dir / "decision.json", + diagnostics=[reason], + ) + + +def _cargo_toml(crate_name: str, component: str) -> str: + return f"""[package] +name = "{crate_name}" +version = "0.1.0" +edition = "2021" +description = "PACTA proof-gated component capsule for {component}" +license = "MIT" + +[lib] +path = "src/lib.rs" +""" + + +def _lib_rs(card: dict[str, Any]) -> str: + component = _rust_string(str(card.get("component") or "unknown-component")) + repo_url = _rust_string(str(card.get("repo_url") or "")) + risk = card.get("risk") or {} + risk_level = _rust_string(str(risk.get("level") or "R0")) + kind = _rust_string(str(card.get("kind") or "unknown")) + backend = _rust_string(str(card.get("verified_backend") or "")) + constraints = [_rust_string(str(item)) for item in risk.get("deployment_constraints") or []] + constraints_body = ",\n ".join(f'"{item}"' for item in constraints) + return f"""//! Generated by PACTA. +//! +//! This crate is a proof-gated decision capsule, not a cryptographic implementation. +//! It lets downstream automation consume the verification boundary that allowed a +//! lower-layer component build. + +pub const COMPONENT: &str = "{component}"; +pub const REPO_URL: &str = "{repo_url}"; +pub const KIND: &str = "{kind}"; +pub const VERIFIED_BACKEND: &str = "{backend}"; +pub const RISK_LEVEL: &str = "{risk_level}"; +pub const CLAIM_CARD_JSON: &str = include_str!("../claims.json"); + +pub const DEPLOYMENT_CONSTRAINTS: &[&str] = &[ + {constraints_body} +]; + +pub fn allowed_for_lower_layer_crypto() -> bool {{ + matches!(RISK_LEVEL, "R3" | "R4" | "R5") +}} + +pub fn allowed_for_wallet_or_trading_agent() -> bool {{ + matches!(RISK_LEVEL, "R4" | "R5") +}} +""" + + +def _crate_readme(card: dict[str, Any]) -> str: + risk = card.get("risk") or {} + return f"""# PACTA Gated Component Capsule + +Component: `{card.get("component")}` + +Risk level: `{risk.get("level")}` + +This generated crate is a consumable decision artifact. It does not implement cryptography and does not +certify a wallet. Downstream automation can import it to enforce that this component is only used inside +the proof boundary captured in `claims.json`. +""" + + +def _slug(value: str) -> str: + return re.sub(r"[^a-zA-Z0-9._-]+", "-", value).strip("-") or "component" + + +def _crate_suffix(value: str) -> str: + cleaned = re.sub(r"[^a-zA-Z0-9]+", "_", value).strip("_").lower() + return cleaned or "component" + + +def _rust_string(value: str) -> str: + return value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") diff --git a/src/pacta/claims.py b/src/pacta/claims.py index 406fe5c..f8bd831 100644 --- a/src/pacta/claims.py +++ b/src/pacta/claims.py @@ -76,6 +76,8 @@ def build_claim_card( }, } assessment = score_claim_card(card) + if not _has_axiom_clean_certificate(card) and assessment.level in {"R0", "R1", "R2"}: + card["guarantees"] = ["No configured certificate was replayed and axiom-clean in this run."] card["risk"] = assessment.to_dict() return card @@ -123,3 +125,10 @@ def _from_axiom_result(name: str, result: CertificateAxiomResult | None, expecte observed_axioms=result.observed_axioms, expected_axioms=result.expected_axioms, ) + + +def _has_axiom_clean_certificate(card: dict[str, Any]) -> bool: + return any( + cert.get("status") == "proven" and cert.get("axiom_status") == "clean" + for cert in card.get("certificates") or [] + ) diff --git a/src/pacta/cli.py b/src/pacta/cli.py index 4abf87b..a5cdea4 100644 --- a/src/pacta/cli.py +++ b/src/pacta/cli.py @@ -5,6 +5,7 @@ import sys from pathlib import Path from typing import Any +from .agent import component_artifact_dir, run_agent_action, write_decision from .audit import scan_hygiene from .claims import build_claim_card from .config import RepoConfig, load_config @@ -92,6 +93,23 @@ def build_parser() -> argparse.ArgumentParser: score = sub.add_parser("score", help="Score an existing claim card.") score.add_argument("--claims", required=True) score.set_defaults(func=cmd_score) + + agent = sub.add_parser("agent", help="Apply a policy-gated consequence to verification evidence.") + agent.add_argument("--claims", help="Existing claim card to act on.") + agent.add_argument("--config", help="Repository config used to generate a claim card.") + agent.add_argument("--repo-name", help="Configured repository name.") + agent.add_argument("--repo", help="Local repository path.") + agent.add_argument("--base-dir", default="repos") + agent.add_argument("--artifact-dir", default="artifacts") + agent.add_argument("--action", choices=["build-library", "build-wallet-demo"], default="build-library") + agent.add_argument("--min-risk", default="R3") + agent.add_argument("--clone", action="store_true", help="Clone/fetch the configured repository before acting.") + agent.add_argument("--offline-fixture", action="store_true", help="Use synthetic clean evidence for a showcase run.") + agent.add_argument("--run-axioms", action="store_true", help="Replay Lean and run axiom audit before acting.") + agent.add_argument("--dry-run", action="store_true") + agent.add_argument("--timeout", type=int, default=120) + agent.add_argument("--log-dir", default=".pacta") + agent.set_defaults(func=cmd_agent) return parser @@ -233,6 +251,75 @@ def cmd_score(args: argparse.Namespace) -> int: return 0 +def cmd_agent(args: argparse.Namespace) -> int: + card = _card_for_agent(args) + decision = run_agent_action( + card, + action=args.action, + artifact_root=args.artifact_dir, + minimum_build_risk=args.min_risk, + timeout=args.timeout, + dry_run=args.dry_run, + ) + component_dir = component_artifact_dir(args.artifact_dir, str(card.get("component") or "component")) + component_dir.mkdir(parents=True, exist_ok=True) + dump_data(card, component_dir / "claims.yaml") + (component_dir / "report.md").write_text(render_markdown(card), encoding="utf-8") + decision_path = write_decision(decision, args.artifact_dir, str(card.get("component") or "component")) + print(f"action: {decision.requested_action}") + print(f"allowed: {str(decision.allowed).lower()}") + print(f"risk: {decision.risk_level}") + print(f"decision: {decision_path}") + if decision.artifact: + print(f"artifact: {decision.artifact.artifact_dir}") + if decision.artifact.crate_dir: + print(f"crate: {decision.artifact.crate_dir}") + if decision.artifact.log_path: + print(f"log: {decision.artifact.log_path}") + print(decision.rationale) + return 0 if decision.allowed else 1 + + +def _card_for_agent(args: argparse.Namespace) -> dict[str, Any]: + if args.claims: + return load_data(args.claims) + if not args.config or not args.repo_name: + raise ValueError("agent requires --claims or both --config and --repo-name") + config = load_config(args.config) + repo = config.repo_named(args.repo_name) + if args.clone: + status = clone_or_fetch(repo, args.base_dir, fetch=True) + local_path = status.local_path + else: + local_path = Path(args.repo) if args.repo else status_for(repo, args.base_dir).local_path + layout = discover_layout(local_path, repo.verification_dir) if local_path.exists() else None + if layout is None and not args.offline_fixture: + raise ValueError(f"Local repo does not exist: {local_path}") + check_result = None + axiom_result = None + if args.run_axioms: + if layout is None: + raise ValueError("--run-axioms requires a local repository") + profile = get_profile(repo.kind, repo) + check_result = lean_check_files(layout.compile_order, layout.verification_dir, timeout=args.timeout, log_dir=args.log_dir) + axiom_result = run_axiom_audit( + local_path / repo.verification_dir, + profile.axiom_imports, + repo.certificates or profile.default_certificates, + profile.expected_axioms, + timeout=args.timeout, + log_dir=args.log_dir, + ) + return build_claim_card( + repo, + local_path, + layout=layout, + lean_check=check_result, + axiom_audit=axiom_result, + offline_fixture=args.offline_fixture, + ) + + def _repo_from_optional_config(config_path: str | None, repo_name: str | None, verification_dir: str) -> RepoConfig: if config_path: config = load_config(config_path) diff --git a/src/pacta/risk.py b/src/pacta/risk.py index 71e5c75..e5c9635 100644 --- a/src/pacta/risk.py +++ b/src/pacta/risk.py @@ -7,6 +7,13 @@ from typing import Any RISK_ORDER = ["R0", "R1", "R2", "R3", "R4", "R5"] +def risk_at_least(level: str, threshold: str) -> bool: + try: + return RISK_ORDER.index(level) >= RISK_ORDER.index(threshold) + except ValueError: + return False + + @dataclass(slots=True) class RiskAssessment: level: str diff --git a/tests/test_agent.py b/tests/test_agent.py new file mode 100644 index 0000000..5141c77 --- /dev/null +++ b/tests/test_agent.py @@ -0,0 +1,37 @@ +from pacta.agent import run_agent_action + + +def _card(level="R3"): + return { + "component": "dalek-ed25519-verified", + "repo_url": "https://github.com/saymrwulf/dalek-ed25519-verified.git", + "kind": "ed25519", + "verified_backend": "serial/u64", + "risk": { + "level": level, + "rationale": "test", + "deployment_constraints": ["Use verified serial/u64 backend only."], + }, + } + + +def test_agent_allows_r3_library_build_dry_run(tmp_path): + decision = run_agent_action(_card("R3"), "build-library", tmp_path, dry_run=True) + assert decision.allowed + assert decision.artifact is not None + assert "proof-gated library capsule" in decision.rationale + + +def test_agent_denies_r2_library_build(tmp_path): + decision = run_agent_action(_card("R2"), "build-library", tmp_path) + assert not decision.allowed + assert decision.artifact is None + assert "below the build threshold" in decision.rationale + + +def test_agent_denies_wallet_demo_below_r4(tmp_path): + decision = run_agent_action(_card("R3"), "build-wallet-demo", tmp_path) + assert not decision.allowed + assert decision.artifact is not None + assert (decision.artifact.artifact_dir / "decision.json").exists() + assert "below R4" in decision.rationale diff --git a/tests/test_artifact.py b/tests/test_artifact.py new file mode 100644 index 0000000..f927fb2 --- /dev/null +++ b/tests/test_artifact.py @@ -0,0 +1,24 @@ +import json + +from pacta.artifact import build_proof_gated_capsule + + +def test_capsule_source_written_when_cargo_missing(tmp_path, monkeypatch): + monkeypatch.setattr("pacta.artifact.shutil.which", lambda name: None) + card = { + "component": "dalek-ed25519-verified", + "repo_url": "https://github.com/saymrwulf/dalek-ed25519-verified.git", + "kind": "ed25519", + "verified_backend": "serial/u64", + "risk": { + "level": "R3", + "deployment_constraints": ["Use verified serial/u64 backend only."], + }, + } + result = build_proof_gated_capsule(card, tmp_path) + assert not result.built + assert result.crate_dir is not None + assert (result.crate_dir / "Cargo.toml").exists() + assert (result.crate_dir / "src" / "lib.rs").exists() + claims = json.loads((result.crate_dir / "claims.json").read_text(encoding="utf-8")) + assert claims["risk"]["level"] == "R3" diff --git a/tests/test_claim_failures.py b/tests/test_claim_failures.py new file mode 100644 index 0000000..8751ec1 --- /dev/null +++ b/tests/test_claim_failures.py @@ -0,0 +1,29 @@ +from pacta.claims import build_claim_card +from pacta.config import RepoConfig +from pacta.lean import AxiomAuditResult, CertificateAxiomResult + + +def test_failed_replay_does_not_emit_profile_guarantees(tmp_path): + repo = RepoConfig( + name="dalek-ed25519-verified", + kind="ed25519", + certificates=["CurveFieldProofs.fieldImplementation"], + ) + audit = AxiomAuditResult( + attempted=True, + ok=False, + missing_tool=None, + certificates=[ + CertificateAxiomResult( + name="CurveFieldProofs.fieldImplementation", + status="failed", + axiom_status="not_checked", + observed_axioms=[], + expected_axioms=["propext", "Classical.choice", "Quot.sound"], + ) + ], + log_path=None, + ) + card = build_claim_card(repo, tmp_path, axiom_audit=audit) + assert card["risk"]["level"] == "R2" + assert card["guarantees"] == ["No configured certificate was replayed and axiom-clean in this run."] diff --git a/tests/test_risk.py b/tests/test_risk.py index 3042f38..b88e35f 100644 --- a/tests/test_risk.py +++ b/tests/test_risk.py @@ -1,4 +1,4 @@ -from pacta.risk import score_claim_card +from pacta.risk import risk_at_least, score_claim_card def test_ed25519_field_and_edwards_clean_scores_r3(): @@ -37,3 +37,10 @@ def test_dirty_axioms_do_not_score_r3(): def test_pasta_without_aggregate_is_foundation_r2(): result = score_claim_card({"kind": "pasta_pallas", "certificates": []}) assert result.level == "R2" + + +def test_risk_ordering(): + assert risk_at_least("R3", "R2") + assert risk_at_least("R3", "R3") + assert not risk_at_least("R2", "R3") + assert not risk_at_least("RX", "R3")