add policy-gated agent consequences

This commit is contained in:
saymrwulf 2026-07-03 11:05:06 +02:00
parent 5d10f20283
commit be5bd182d0
12 changed files with 562 additions and 1 deletions

2
.gitignore vendored
View file

@ -2,6 +2,8 @@
.pacta/
.pacta-*/
.pytest_cache/
artifacts*/
repos/
__pycache__/
*.py[cod]
*.egg-info/

View file

@ -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.

View file

@ -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:

127
src/pacta/agent.py Normal file
View file

@ -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"

217
src/pacta/artifact.py Normal file
View file

@ -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")

View file

@ -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 []
)

View file

@ -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)

View file

@ -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

37
tests/test_agent.py Normal file
View file

@ -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

24
tests/test_artifact.py Normal file
View file

@ -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"

View file

@ -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."]

View file

@ -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")