feat(trust-rollup): add trust scorecard rollup collector

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"}
+  }
+]

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

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