BlockSiFr · blocksifrdev · Sep 15, 2026 · Sep 15, 2026
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -28,7 +28,7 @@ jobs:
           test -f docs/ops/persistent-storage.md
 
       - name: Compile Python SDK
-        run: python -m py_compile sdk/python/client.py sdk/python/agt.py
+        run: python -m py_compile sdk/python/client.py sdk/python/agt.py sdk/python/aggregate.py
 
       - name: Test Python SDK
         run: python -m unittest discover -s sdk/python -p 'test_*.py' -v

diff --git a/docs/security.md b/docs/security.md
@@ -33,7 +33,7 @@ When correctly deployed with conformant implementations:
 
 8. **Stateless verifiability** — Verifiers can validate tokens without contacting the Trust Authority, eliminating dependency on TA availability at verification time.
 
-9. **Single-issuer resistance** — The aggregation algorithm's issuer weight cap prevents any single issuer from dominating a trust score.
+9. **Single-issuer resistance** — The aggregation algorithm's issuer weight cap prevents any single issuer from dominating a trust score. *(Holds as of aggregation algorithm **v1.1**. See the correction note in §3.)*
 
 ### 1.2 What TTP Does NOT Guarantee
 
@@ -116,6 +116,11 @@ Compromise at a higher level in this hierarchy has broader impact.
 
 **Residual risk:** If an issuer's private key is compromised, the attacker can generate receipts up to the issuer weight cap (40%). Multi-issuer requirements for sensitive domains mitigate this.
 
+> Under aggregation algorithm **v1.0 this bound did not hold**: capped weight was
+> re-normalized back across all issuers, so a compromised high-volume issuer could reach
+> roughly 87% of the aggregate weight while the others stayed light. Fixed in v1.1 — see
+> the correction note below.
+
 ---
 
 ### T4: Receipt Replay
@@ -135,11 +140,21 @@ Compromise at a higher level in this hierarchy has broader impact.
 **Description:** An attacker controls or compromises a single high-volume issuer and uses it to inflate an agent's trust score.
 
 **Mitigations:**
-- The aggregation algorithm caps any single issuer's contribution at `max_issuer_weight` (default: 40%).
-- Even if one issuer submits 1000 receipts, it cannot exceed 40% of the aggregate weight.
+- The aggregation algorithm caps any single issuer's contribution at `effective_cap = max(max_issuer_weight, 1 / issuer_count)` (default `max_issuer_weight`: 40%), and redistributes the excess to the **uncapped** issuers.
+- Even if one issuer submits 1000 receipts, it cannot exceed that cap of the aggregate weight.
 - Multi-issuer requirements (`min_issuer_count ≥ 2`) prevent single-issuer tokens entirely for sensitive domains.
 
-**Residual risk:** If only one issuer is deployed, that issuer's compromise fully controls agent trust. Deploy at least 2 independent issuers per domain.
+**Residual risk:** If only one issuer is deployed, that issuer's compromise fully controls agent trust — the cap cannot bind when there is nobody to redistribute to (`effective_cap = 1/1`). Deploy at least 2 independent issuers per domain.
+
+> **Correction — aggregation v1.1 (2026-09-15).** The second mitigation above was not true
+> of v1.0. That version capped a dominant issuer's *fraction* and then re-normalized
+> across all issuers, which returned most of the capped weight to it whenever the others
+> were light: 50 receipts from one issuer against one each from two others left the
+> "capped" issuer holding **87%** of the weight, and the aggregate at 0.90 rather than
+> 0.52. The guarantee stated in §1.1 item 9 and in this section holds from v1.1 onward.
+> Anyone who deployed against v1.0 and relied on the 40% bound should re-evaluate;
+> `protocol/test-vectors/aggregation-vectors.json` case `agg-009` pins the corrected
+> behaviour.
 
 ---
 

diff --git a/scripts/check-agt-parity.mjs b/scripts/check-agt-parity.mjs
@@ -1,13 +1,15 @@
 #!/usr/bin/env node
-// The PCTR <-> AGT bridge exists twice: packages/pctr/src/agt.mjs for JavaScript and
-// sdk/python/agt.py for Python, because AGT is Python-first. Two implementations agree
-// only as long as something checks, so this runs both over the same inputs and fails on
-// any divergence. Run: node scripts/check-agt-parity.mjs
+// PCTR exists twice: JavaScript in packages/pctr/src, Python in sdk/python, because AGT
+// is Python-first and the aggregation algorithm is normative for both. Two
+// implementations agree only as long as something checks, so this runs both over the same
+// inputs — AGT bridge and trust aggregation — and fails on any divergence.
+// Run: node scripts/check-agt-parity.mjs
 import { execFileSync } from 'node:child_process';
 import path from 'node:path';
 import { fileURLToPath } from 'node:url';
 
 import { classifyAction } from '../packages/pctr/src/consequences.mjs';
+import { aggregateTrust } from '../packages/pctr/src/aggregate.mjs';
 import { trustTier, toAgtScore, ringForSeverity, domainFor, normalizeAgtEvent } from '../packages/pctr/src/agt.mjs';
 
 const root = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..');
@@ -21,6 +23,29 @@ const ACTIONS = [
 const SCORES = [0, 0.1, 0.29, 0.3, 0.6, 0.849, 0.85, 0.9178, 1];
 const SEVERITIES = ['CRITICAL', 'HIGH', 'MEDIUM', 'LOW', '?'];
 const CONSEQUENCES = ['MONEY_MOVED', 'DATA_DELETED', 'SECRET_EXPOSED', 'SOMETHING_NEW'];
+// Aggregation is implemented in both languages against the same normative spec, so the
+// two must agree on scores, not merely each pass the vectors alone.
+const NOW = 1_700_000_000_000;
+const AGGREGATION = {
+  'single-perfect': [{ receipt_id: 'a', issuer_id: 'A', score: 1.0, timestamp: NOW }],
+  'single-zero': [{ receipt_id: 'a', issuer_id: 'A', score: 0.0, timestamp: NOW }],
+  'two-issuers-divergent': [
+    { receipt_id: 'a', issuer_id: 'A', score: 1.0, timestamp: NOW },
+    { receipt_id: 'b', issuer_id: 'B', score: 0.0, timestamp: NOW }],
+  'dominant-issuer-capped': [
+    ...Array.from({ length: 50 }, (_, i) => ({ receipt_id: `l${i}`, issuer_id: 'LOUD', score: 1.0, timestamp: NOW })),
+    { receipt_id: 'b', issuer_id: 'B', score: 0.2, timestamp: NOW },
+    { receipt_id: 'c', issuer_id: 'C', score: 0.2, timestamp: NOW }],
+  'decayed': [
+    { receipt_id: 'fresh', issuer_id: 'A', score: 1.0, timestamp: NOW },
+    { receipt_id: 'old', issuer_id: 'A', score: 0.0, timestamp: NOW - 120_000 }],
+  'outside-window': [{ receipt_id: 'a', issuer_id: 'A', score: 1.0, timestamp: NOW - 400_000 }],
+  'three-issuers': [
+    { receipt_id: 'a', issuer_id: 'A', score: 0.95, timestamp: NOW },
+    { receipt_id: 'b', issuer_id: 'B', score: 0.92, timestamp: NOW - 10_000 },
+    { receipt_id: 'c', issuer_id: 'C', score: 0.88, timestamp: NOW - 20_000 }]
+};
+
 const EVENTS = [
   ['policy-allow', { allowed: true, action: 'allow', agentId: 'f', operation: 'reports.read', approvers: [], rateLimited: false }],
   ['policy-warn', { allowed: true, action: 'warn', agentId: 'f', operation: 'reports.read', approvers: [], rateLimited: false }],
@@ -42,7 +67,7 @@ const EVENTS = [
 const key = (score) => (Number.isInteger(score) ? String(score) : String(score));
 
 function javascript() {
-  const out = { classify: {}, tiers: {}, scores: {}, rings: {}, domains: {}, events: {} };
+  const out = { classify: {}, tiers: {}, scores: {}, rings: {}, domains: {}, events: {}, aggregation: {} };
   for (const [action, params] of ACTIONS) {
     const c = classifyAction(action, params);
     out.classify[`${action}|${JSON.stringify(params)}`] = [c.consequence, c.severity, c.reversible];
@@ -54,21 +79,28 @@ function javascript() {
     const r = normalizeAgtEvent(event);
     out.events[name] = r ? [r.event, r.subject ?? null] : null;
   }
+  for (const [name, receipts] of Object.entries(AGGREGATION)) {
+    const r = aggregateTrust(receipts, NOW);
+    out.aggregation[name] = r.error ? [r.error] : [Number(r.score.toFixed(6)), r.contributing_issuers];
+  }
   return out;
 }
 
 const PY = `
 import json, sys
 sys.path.insert(0, ${JSON.stringify(path.join(root, 'sdk', 'python'))})
 from agt import classify_action, trust_tier, to_agt_score, ring_for_severity, domain_for, normalize_agt_event
+from aggregate import aggregate_trust
 
 actions = json.loads(${JSON.stringify(JSON.stringify(ACTIONS))})
 scores = json.loads(${JSON.stringify(JSON.stringify(SCORES))})
 severities = json.loads(${JSON.stringify(JSON.stringify(SEVERITIES))})
 consequences = json.loads(${JSON.stringify(JSON.stringify(CONSEQUENCES))})
 events = json.loads(${JSON.stringify(JSON.stringify(EVENTS))})
+aggregation = json.loads(${JSON.stringify(JSON.stringify(AGGREGATION))})
+now = ${NOW}
 
-out = {'classify': {}, 'tiers': {}, 'scores': {}, 'rings': {}, 'domains': {}, 'events': {}}
+out = {'classify': {}, 'tiers': {}, 'scores': {}, 'rings': {}, 'domains': {}, 'events': {}, 'aggregation': {}}
 for action, params in actions:
     c = classify_action(action, params)
     out['classify'][action + '|' + json.dumps(params, separators=(',', ':'))] = [c['consequence'], c['severity'], c['reversible']]
@@ -83,6 +115,9 @@ for consequence in consequences:
 for name, event in events:
     r = normalize_agt_event(event)
     out['events'][name] = [r['event'], r.get('subject')] if r else None
+for name, receipts in aggregation.items():
+    r = aggregate_trust(receipts, now)
+    out['aggregation'][name] = [r['error']] if r.get('error') else [round(r['score'], 6), r['contributing_issuers']]
 print(json.dumps(out))
 `;
 
@@ -107,9 +142,9 @@ for (const section of Object.keys(js)) {
 }
 
 if (differences.length) {
-  console.error(`PCTR/AGT bridge parity FAILED — ${differences.length} of ${checks} checks differ:\n`);
+  console.error(`PCTR parity FAILED — ${differences.length} of ${checks} checks differ:\n`);
   console.error(differences.join('\n'));
-  console.error('\npackages/pctr/src/agt.mjs and sdk/python/agt.py must agree. Fix both.');
+  console.error('\nThe JavaScript and Python implementations must agree. Fix both.');
   process.exit(1);
 }
-console.log(`PCTR/AGT bridge parity OK — JavaScript and Python agree across ${checks} checks.`);
+console.log(`PCTR parity OK — JavaScript and Python agree across ${checks} checks (AGT bridge + trust aggregation).`);
diff --git a/sdk/python/README.md b/sdk/python/README.md
@@ -63,7 +63,25 @@ verified 0.85 — `to_agt_trust_score()` returns AGT's `TrustScore {overall, dim
 tier}`. The 0-1000 integer scale (`to_agt_score()`) is only for downstream consumers that
 ask for it.
 
+## Trust aggregation (`aggregate.py`)
+
+The normative algorithm from
+[`protocol/aggregation-spec.md`](../../protocol/aggregation-spec.md) **v1.1** — time
+decay, negative-signal amplification, and issuer weight capping that redistributes to the
+uncapped issuers. The same eleven test vectors run against both bindings.
+
+```python
+from aggregate import aggregate_trust, score_label
+
+result = aggregate_trust(receipts, current_time_ms)
+print(result["score"], score_label(result["score"]))
+```
+
+An empty receipt window returns `INSUFFICIENT_TRUST_DATA` with `score: None` — absent
+evidence is never a score.
+
 The two implementations are held in step by `scripts/check-agt-parity.mjs`, which runs
-both over the same corpus in CI and fails on any divergence.
+both over the same corpus in CI — AGT bridge and trust aggregation, 59 checks — and fails
+on any divergence.
 
 Tests: `python3 -m unittest discover -s sdk/python -p 'test_*.py'`
diff --git a/sdk/python/aggregate.py b/sdk/python/aggregate.py
@@ -0,0 +1,119 @@
+"""TTP trust score aggregation — the normative algorithm.
+
+Implements protocol/aggregation-spec.md v1.1 step for step, mirroring
+packages/pctr/src/aggregate.mjs. The same eleven test vectors run against both
+bindings (scripts/check-agt-parity.mjs), so they cannot drift apart.
+
+Two properties, both deliberate:
+  - Negative signals weigh more (default 1.5x), so an agent behaving well most of the
+    time cannot average away a few dangerous actions.
+  - No single issuer may contribute more than effective_cap of the score, and the excess
+    goes to the *uncapped* issuers. v1.0 re-normalized across everyone, which handed the
+    excess straight back whenever the other issuers were light.
+"""
+
+from __future__ import annotations
+
+import math
+from typing import Any, Dict, List, Mapping, Optional
+
+__all__ = ["DEFAULT_PARAMS", "INSUFFICIENT_TRUST_DATA", "aggregate_trust", "score_label"]
+
+DEFAULT_PARAMS: Dict[str, float] = {
+    "receipt_window_s": 300,
+    "max_issuer_weight": 0.40,
+    "negative_weight_multiplier": 1.5,
+    "decay_half_life_s": 120,
+}
+
+INSUFFICIENT_TRUST_DATA = "INSUFFICIENT_TRUST_DATA"
+
+
+def aggregate_trust(receipts: Optional[List[Mapping[str, Any]]] = None,
+                    current_time_ms: Optional[int] = None,
+                    params: Optional[Mapping[str, float]] = None) -> Dict[str, Any]:
+    """Aggregate behavioural receipts into a trust score in [0.0, 1.0]."""
+    p = {**DEFAULT_PARAMS, **(params or {})}
+    receipts = list(receipts or [])
+
+    # Step 1 — filter to the receipt window.
+    window = [r for r in receipts
+              if current_time_ms - r["timestamp"] <= p["receipt_window_s"] * 1000]
+    if not window:
+        return {"error": INSUFFICIENT_TRUST_DATA, "score": None,
+                "contributing_receipts": 0, "contributing_issuers": 0}
+
+    # Steps 2 and 3 — time decay, then negative signal amplification.
+    weighted = []
+    for r in window:
+        age_s = (current_time_ms - r["timestamp"]) / 1000
+        decay = math.exp(-math.log(2) * age_s / p["decay_half_life_s"])
+        negative = r["score"] < 0.5
+        weighted.append({**r, "age_s": age_s,
+                         "signal_weight": decay * p["negative_weight_multiplier"] if negative else decay})
+
+    # Step 4 — per-issuer weighted score.
+    by_issuer: Dict[str, Dict[str, float]] = {}
+    for r in weighted:
+        e = by_issuer.setdefault(r["issuer_id"], {"weighted_sum": 0.0, "total_weight": 0.0})
+        e["weighted_sum"] += r["score"] * r["signal_weight"]
+        e["total_weight"] += r["signal_weight"]
+
+    issuers = [{"issuer_id": k, "issuer_score": e["weighted_sum"] / e["total_weight"],
+                "issuer_raw_weight": e["total_weight"]}
+               for k, e in by_issuer.items()]
+
+    # Step 5 — cap, then water-fill the excess onto the uncapped issuers. A cap below
+    # 1/n is infeasible, so that is the floor on the cap actually applied.
+    total_raw = sum(i["issuer_raw_weight"] for i in issuers)
+    effective_cap = max(p["max_issuer_weight"], 1 / len(issuers))
+    for i in issuers:
+        i["weight"] = i["issuer_raw_weight"] / total_raw
+        i["capped"] = False
+
+    for _ in range(len(issuers) + 1):
+        over = [i for i in issuers if not i["capped"] and i["weight"] > effective_cap + 1e-12]
+        if not over:
+            break
+        excess = 0.0
+        for i in over:
+            excess += i["weight"] - effective_cap
+            i["weight"] = effective_cap
+            i["capped"] = True
+        free = [i for i in issuers if not i["capped"]]
+        free_total = sum(i["weight"] for i in free)
+        if not free or free_total == 0:
+            break
+        for i in free:
+            i["weight"] += excess * (i["weight"] / free_total)
+
+    # Step 6 — combine, and clamp for floating point.
+    raw_score = sum(i["issuer_score"] * i["weight"] for i in issuers)
+
+    return {
+        "score": max(0.0, min(1.0, raw_score)),
+        "contributing_receipts": len(window),
+        "contributing_issuers": len(issuers),
+        "oldest_receipt_age_s": round(max(r["age_s"] for r in weighted)),
+        "issuers": [{"issuer_id": i["issuer_id"],
+                     "issuer_score": round(i["issuer_score"], 6),
+                     "weight": round(i["weight"], 6),
+                     "capped": i["capped"]} for i in issuers],
+    }
+
+
+# The scale from protocol/scoring-semantics.md, so a score can be read in words.
+def score_label(score: Optional[float]) -> str:
+    if score is None:
+        return "Unknown"
+    if score >= 0.90:
+        return "Excellent"
+    if score >= 0.70:
+        return "Good"
+    if score >= 0.50:
+        return "Marginal"
+    if score >= 0.30:
+        return "Poor"
+    if score >= 0.10:
+        return "Bad"
+    return "Critical"
diff --git a/sdk/python/test_aggregate.py b/sdk/python/test_aggregate.py
@@ -0,0 +1,66 @@
+"""The normative aggregation vectors, run against the Python binding.
+
+The same vectors run against the JavaScript binding in
+packages/pctr/tests/aggregation.test.mjs. Both must agree with the spec and each other.
+"""
+import json
+import os
+import unittest
+
+from aggregate import DEFAULT_PARAMS, INSUFFICIENT_TRUST_DATA, aggregate_trust, score_label
+
+VECTORS = json.load(open(os.path.join(os.path.dirname(__file__), "..", "..",
+                                      "protocol", "test-vectors", "aggregation-vectors.json")))
+
+
+class ConformanceTests(unittest.TestCase):
+    def test_defaults_match_the_spec(self):
+        self.assertEqual(DEFAULT_PARAMS, VECTORS["params"])
+
+    def test_every_vector(self):
+        for case in VECTORS["cases"]:
+            with self.subTest(case=case["id"]):
+                result = aggregate_trust(case["receipts"], case["current_time_ms"], VECTORS["params"])
+                if case["expected"].get("error"):
+                    self.assertEqual(result["error"], case["expected"]["error"])
+                    continue
+                self.assertAlmostEqual(result["score"], case["expected"]["score"], delta=0.001,
+                                       msg=f'{case["id"]}: {case["description"]}')
+                if "contributing_receipts" in case["expected"]:
+                    self.assertEqual(result["contributing_receipts"], case["expected"]["contributing_receipts"])
+                if "contributing_issuers" in case["expected"]:
+                    self.assertEqual(result["contributing_issuers"], case["expected"]["contributing_issuers"])
+
+
+class PropertyTests(unittest.TestCase):
+    def test_empty_window_is_insufficient_data_not_zero(self):
+        result = aggregate_trust([{"receipt_id": "r", "issuer_id": "A", "score": 1.0, "timestamp": 0}], 10_000_000)
+        self.assertEqual(result["error"], INSUFFICIENT_TRUST_DATA)
+        self.assertIsNone(result["score"])
+
+    def test_a_capped_issuer_holds_its_cap_and_no_more(self):
+        now = 1_700_000_000_000
+        receipts = [{"receipt_id": f"loud{i}", "issuer_id": "LOUD", "score": 1.0, "timestamp": now} for i in range(50)]
+        receipts += [{"receipt_id": "b", "issuer_id": "B", "score": 0.2, "timestamp": now},
+                     {"receipt_id": "c", "issuer_id": "C", "score": 0.2, "timestamp": now}]
+        result = aggregate_trust(receipts, now)
+        loud = next(i for i in result["issuers"] if i["issuer_id"] == "LOUD")
+        self.assertTrue(loud["capped"])
+        self.assertAlmostEqual(loud["weight"], 0.4, delta=0.001)
+        self.assertLess(result["score"], 0.55)
+        self.assertAlmostEqual(sum(i["weight"] for i in result["issuers"]), 1.0, delta=0.001)
+
+    def test_danger_cannot_be_averaged_away(self):
+        now = 1_700_000_000_000
+        receipts = [{"receipt_id": f"g{i}", "issuer_id": "A", "score": 1.0, "timestamp": now} for i in range(9)]
+        receipts.append({"receipt_id": "bad", "issuer_id": "A", "score": 0.0, "timestamp": now})
+        self.assertLess(aggregate_trust(receipts, now)["score"], 0.87)
+
+    def test_labels(self):
+        self.assertEqual(score_label(0.95), "Excellent")
+        self.assertEqual(score_label(0.05), "Critical")
+        self.assertEqual(score_label(None), "Unknown")
+
+
+if __name__ == "__main__":
+    unittest.main()