diff --git a/docs/trust-evidence.sample.json b/docs/trust-evidence.sample.json new file mode 100644 index 0000000..0d4f0b5 --- /dev/null +++ b/docs/trust-evidence.sample.json @@ -0,0 +1,49 @@ +[ + { + "schema_version": "0.1.0", + "subject": {"repo": "fullsend-ai/agents", "agent_role": "review", "config_hash": "sha256:aaa", "generated_at": "2026-08-20T00:00:00Z"}, + "evidence": { + "config_health": {"status": "pass", "score": 0.94, "source": "static-analysis", "as_of": "2026-08-20T00:00:00Z"}, + "behavioral_eval": {"status": "pass", "score": 0.88, "source": "eval-suite@rev", "as_of": "2026-08-20T00:00:00Z"}, + "audit_integrity": {"status": "pass", "source": "hash-chain-verify", "as_of": "2026-08-20T00:00:00Z"}, + "track_record": {"status": "partial", "sample_size": 42, "revert_rate": 0.02, "as_of": "2026-08-20T00:00:00Z"}, + "drift": {"status": "pass", "baseline_config_hash": "sha256:aaa", "as_of": "2026-08-20T00:00:00Z"} + }, + "composition": {"model": "tiered", "target_tier": "auto-merge", "decision": "insufficient", "blocking": ["track_record"], "rationale": "track record sample size below the auto-merge threshold"} + }, + { + "schema_version": "0.1.0", + "subject": {"repo": "fullsend-ai/agents", "agent_role": "review", "config_hash": "sha256:bbb", "generated_at": "2026-08-24T00:00:00Z"}, + "evidence": { + "config_health": {"status": "pass", "score": 0.96, "source": "static-analysis", "as_of": "2026-08-24T00:00:00Z"}, + "behavioral_eval": {"status": "pass", "score": 0.90, "source": "eval-suite@rev", "as_of": "2026-08-24T00:00:00Z"}, + "audit_integrity": {"status": "pass", "source": "hash-chain-verify", "as_of": "2026-08-24T00:00:00Z"}, + "track_record": {"status": "pass", "sample_size": 210, "revert_rate": 0.01, "as_of": "2026-08-24T00:00:00Z"}, + "drift": {"status": "pass", "baseline_config_hash": "sha256:bbb", "as_of": "2026-08-24T00:00:00Z"} + }, + "composition": {"model": "tiered", "target_tier": "auto-merge", "decision": "sufficient", "blocking": [], "rationale": "all signals pass at auto-merge thresholds"} + }, + { + "schema_version": "0.1.0", + "subject": {"repo": "fullsend-ai/agents", "agent_role": "triage", "config_hash": "sha256:ccc", "generated_at": "2026-08-24T00:00:00Z"}, + "evidence": { + "config_health": {"status": "pass", "score": 0.90, "source": "static-analysis", "as_of": "2026-08-24T00:00:00Z"}, + "behavioral_eval": {"status": "pass", "score": 0.83, "source": "eval-suite@rev", "as_of": "2026-08-24T00:00:00Z"}, + "audit_integrity": {"status": "pass", "source": "hash-chain-verify", "as_of": "2026-08-24T00:00:00Z"}, + "drift": {"status": "pass", "baseline_config_hash": "sha256:ccc", "as_of": "2026-08-24T00:00:00Z"} + }, + "composition": {"model": "tiered", "target_tier": "auto-triage", "decision": "sufficient", "blocking": [], "rationale": "static and behavioral pass; track record not required at this tier"} + }, + { + "schema_version": "0.1.0", + "subject": {"repo": "fullsend-ai/fullsend", "agent_role": "review", "config_hash": "sha256:ddd", "generated_at": "2026-08-22T00:00:00Z"}, + "evidence": { + "config_health": {"status": "fail", "score": 0.40, "source": "static-analysis", "as_of": "2026-08-22T00:00:00Z"}, + "behavioral_eval": {"status": "pass", "score": 0.80, "source": "eval-suite@rev", "as_of": "2026-08-22T00:00:00Z"}, + "audit_integrity": {"status": "pass", "source": "hash-chain-verify", "as_of": "2026-08-22T00:00:00Z"}, + "track_record": {"status": "partial", "sample_size": 12, "revert_rate": 0.08, "as_of": "2026-08-22T00:00:00Z"}, + "drift": {"status": "pass", "baseline_config_hash": "sha256:ddd", "as_of": "2026-08-22T00:00:00Z"} + }, + "composition": {"model": "tiered", "target_tier": "auto-merge", "decision": "insufficient", "blocking": ["config_health", "track_record"], "rationale": "failing config health scan blocks autonomy"} + } +] diff --git a/scripts/collect-trust-rollup.py b/scripts/collect-trust-rollup.py new file mode 100644 index 0000000..cf6f7c8 --- /dev/null +++ b/scripts/collect-trust-rollup.py @@ -0,0 +1,146 @@ +#!/usr/bin/env python3 +"""Roll up trust scorecards into a per-agent CSV for fullsend-ai. + +Reads a JSON array of trust scorecards (the artifact shape proposed in +fullsend-ai/fullsend docs/problems/trustworthiness-evidence.md) and writes +docs/trust-rollup.csv, one row per (repo, agent_role): how many scorecards +were seen, the most recent composition decision, the share of evidence +signals that passed, the average of the numeric evidence scores, and the +union of signals that ever blocked a decision. + +No live trust-evidence source exists yet, so this reads a documented sample +fixture (docs/trust-evidence.sample.json) by default. Point --input at a real +feed once one exists; the rollup logic is unchanged. +""" +from __future__ import annotations + +import argparse +import csv +import json +from collections import defaultdict +from datetime import datetime, timezone +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +SAMPLE_FILE = ROOT / "docs" / "trust-evidence.sample.json" +ROLLUP_FILE = ROOT / "docs" / "trust-rollup.csv" + +ROLLUP_HEADER = [ + "repo", "agent_role", "scorecards", "latest_decision", + "signal_pass_rate", "avg_config_health", "avg_behavioral_eval", + "avg_track_record_revert_rate", "blocking_signals", +] + +# Evidence signals expected on a scorecard; unknown keys are tolerated. +SIGNALS = ("config_health", "behavioral_eval", "audit_integrity", + "track_record", "drift") + + +def load_scorecards(path: Path) -> list[dict]: + data = json.loads(path.read_text()) + if isinstance(data, dict): + # Only unwrap the documented {"scorecards": [...]} envelope. A dict + # without that key is an unexpected shape, not an empty feed — reject + # it loudly rather than silently rolling up zero scorecards. + if "scorecards" not in data: + raise ValueError( + f"{path}: object is missing the 'scorecards' key; " + "expected a JSON array or {\"scorecards\": [...]}" + ) + data = data["scorecards"] + if not isinstance(data, list): + raise ValueError(f"{path}: expected a JSON array of scorecards") + return data + + +def _instant(card: dict) -> datetime: + """Parse a scorecard's generated_at into a UTC-comparable instant. + + ISO-8601 strings with different UTC offsets do not sort correctly as raw + text (e.g. '...T10:00+02:00' is earlier than '...T09:00Z' but sorts after + it), so compare parsed instants instead. Unparseable or missing timestamps + sort oldest so they never spuriously win "latest". + """ + raw = (card.get("subject") or {}).get("generated_at") + if not isinstance(raw, str): + return datetime.min.replace(tzinfo=timezone.utc) + try: + # fromisoformat accepts a trailing 'Z' only on 3.11+; normalize it. + dt = datetime.fromisoformat(raw.replace("Z", "+00:00")) + except ValueError: + return datetime.min.replace(tzinfo=timezone.utc) + # Treat naive timestamps as UTC so they compare against aware ones. + return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) + + +def _mean(values): + vals = [v for v in values if isinstance(v, (int, float)) and not isinstance(v, bool)] + return round(sum(vals) / len(vals), 4) if vals else "" + + +def rollup(scorecards: list[dict]) -> list[dict]: + """Aggregate scorecards into one row per (repo, agent_role).""" + groups = defaultdict(list) + for card in scorecards: + subj = card.get("subject") or {} + key = (subj.get("repo") or "", subj.get("agent_role") or "") + groups[key].append(card) + + rows = [] + for (repo, role), cards in sorted(groups.items()): + # Most recent by generated_at, compared as parsed UTC instants. + latest = max(cards, key=_instant) + passed = total = 0 + config_scores, behav_scores, revert_rates = [], [], [] + blocking = set() + for card in cards: + evidence = card.get("evidence") or {} + for name in SIGNALS: + sig = evidence.get(name) + if not isinstance(sig, dict): + continue + total += 1 + if sig.get("status") == "pass": + passed += 1 + config_scores.append((evidence.get("config_health") or {}).get("score")) + behav_scores.append((evidence.get("behavioral_eval") or {}).get("score")) + revert_rates.append((evidence.get("track_record") or {}).get("revert_rate")) + blocking.update((card.get("composition") or {}).get("blocking") or []) + rows.append({ + "repo": repo, + "agent_role": role, + "scorecards": len(cards), + "latest_decision": (latest.get("composition") or {}).get("decision") or "", + "signal_pass_rate": round(passed / total, 4) if total else "", + "avg_config_health": _mean(config_scores), + "avg_behavioral_eval": _mean(behav_scores), + "avg_track_record_revert_rate": _mean(revert_rates), + "blocking_signals": ";".join(sorted(blocking)), + }) + return rows + + +def write_rollup(rows: list[dict], path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", newline="") as f: + w = csv.DictWriter(f, fieldnames=ROLLUP_HEADER) + w.writeheader() + w.writerows(rows) + + +def main(): + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--input", default=str(SAMPLE_FILE), + help="Trust scorecard JSON (default: docs/trust-evidence.sample.json)") + p.add_argument("--output", default=str(ROLLUP_FILE), + help="CSV to write (default: docs/trust-rollup.csv)") + args = p.parse_args() + + scorecards = load_scorecards(Path(args.input)) + rows = rollup(scorecards) + write_rollup(rows, Path(args.output)) + print(f"Wrote {len(rows)} rollup rows from {len(scorecards)} scorecards to {args.output}") + + +if __name__ == "__main__": + main() diff --git a/scripts/test_collect_trust_rollup.py b/scripts/test_collect_trust_rollup.py new file mode 100644 index 0000000..b06341a --- /dev/null +++ b/scripts/test_collect_trust_rollup.py @@ -0,0 +1,126 @@ +#!/usr/bin/env python3 +"""Unit tests for collect-trust-rollup helpers (no I/O).""" +from __future__ import annotations + +import importlib.util +import unittest +from pathlib import Path + +spec = importlib.util.spec_from_file_location( + "collect_trust_rollup", Path(__file__).parent / "collect-trust-rollup.py" +) +mod = importlib.util.module_from_spec(spec) +assert spec.loader is not None +spec.loader.exec_module(mod) + +rollup = mod.rollup +_mean = mod._mean +load_scorecards = mod.load_scorecards + + +def _card(repo, role, when, statuses, scores=None, blocking=None, decision="sufficient"): + scores = scores or {} + evidence = {} + for name, status in statuses.items(): + sig = {"status": status} + if name in scores: + sig["score"] = scores[name] + evidence[name] = sig + return { + "subject": {"repo": repo, "agent_role": role, "generated_at": when}, + "evidence": evidence, + "composition": {"decision": decision, "blocking": blocking or []}, + } + + +class TestMean(unittest.TestCase): + def test_ignores_non_numeric_and_bool(self): + self.assertEqual(_mean([1.0, None, "x", True, 3.0]), 2.0) + + def test_empty_is_blank(self): + self.assertEqual(_mean([None, "x"]), "") + + +class TestLoadScorecards(unittest.TestCase): + def _load(self, payload): + import json, tempfile, os + fd, path = tempfile.mkstemp(suffix=".json") + os.write(fd, json.dumps(payload).encode()) + os.close(fd) + try: + return load_scorecards(Path(path)) + finally: + os.unlink(path) + + def test_dict_wrapper(self): + self.assertEqual(self._load({"scorecards": [{"a": 1}]}), [{"a": 1}]) + + def test_bare_list(self): + self.assertEqual(self._load([{"a": 1}]), [{"a": 1}]) + + def test_dict_without_scorecards_key_rejected(self): + # An unexpected object shape must fail loudly, not roll up zero cards. + with self.assertRaises(ValueError): + self._load({"cards": [{"a": 1}]}) + + +class TestRollup(unittest.TestCase): + def setUp(self): + self.cards = [ + _card("o/a", "review", "2026-08-20T00:00:00Z", + {"config_health": "pass", "behavioral_eval": "pass", + "audit_integrity": "pass", "track_record": "partial", "drift": "pass"}, + scores={"config_health": 0.94, "behavioral_eval": 0.88}, + blocking=["track_record"], decision="insufficient"), + _card("o/a", "review", "2026-08-24T00:00:00Z", + {"config_health": "pass", "behavioral_eval": "pass", + "audit_integrity": "pass", "track_record": "pass", "drift": "pass"}, + scores={"config_health": 0.96, "behavioral_eval": 0.90}, + blocking=[], decision="sufficient"), + ] + + def test_one_row_per_group(self): + rows = rollup(self.cards) + self.assertEqual(len(rows), 1) + self.assertEqual((rows[0]["repo"], rows[0]["agent_role"]), ("o/a", "review")) + self.assertEqual(rows[0]["scorecards"], 2) + + def test_latest_decision_uses_most_recent(self): + rows = rollup(self.cards) + self.assertEqual(rows[0]["latest_decision"], "sufficient") + + def test_pass_rate_over_all_signals(self): + # 5 + 5 signals, one partial -> 9/10. + rows = rollup(self.cards) + self.assertEqual(rows[0]["signal_pass_rate"], 0.9) + + def test_avg_scores(self): + rows = rollup(self.cards) + self.assertEqual(rows[0]["avg_config_health"], 0.95) + self.assertEqual(rows[0]["avg_behavioral_eval"], 0.89) + + def test_blocking_union_sorted(self): + rows = rollup(self.cards) + self.assertEqual(rows[0]["blocking_signals"], "track_record") + + def test_latest_respects_utc_offsets(self): + # "23:00+05:00" (18:00Z) sorts after "20:00Z" as raw text but is + # actually earlier; the parsed-instant comparison must pick 20:00Z. + cards = [ + _card("o/c", "code", "2026-08-24T23:00:00+05:00", + {"config_health": "pass"}, decision="earlier"), + _card("o/c", "code", "2026-08-24T20:00:00Z", + {"config_health": "pass"}, decision="latest"), + ] + rows = rollup(cards) + self.assertEqual(rows[0]["latest_decision"], "latest") + + def test_groups_split_by_role_and_repo(self): + cards = self.cards + [_card("o/b", "triage", "2026-08-24T00:00:00Z", + {"config_health": "pass"})] + rows = rollup(cards) + self.assertEqual(len(rows), 2) + + +if __name__ == "__main__": + unittest.main()