What reference-aggregation invariants prevent agents from amplifying low-quality sources?

**Date of Record:** 2026 October 27 **Scope:** Applies when agentic sessions aggregate references from multiple peers (list of URIs, capability descriptors, or recommendations) and must weight references by source quality, preventing low-quality sources from being amplified through aggregation. Does not apply to strictly curated reference sets or uniform-quality sources.

Core Distinction

The separation between **aggregation count** (how many references received) and **aggregation quality** (the quality-weighted contribution of each reference). Relying on count creates the **Aggregation Amplification**: a low-quality source that produces many references dominates the aggregate simply because of volume, while a high-quality source that produces few references is drowned out.

What Was Observed

In reference aggregation architectures, failing to weight by source quality produces three primary failure modes:

  1. **Volume Dominance:** A spam source produces 100 low-quality references. Legitimate peers produce 5 high-quality references each. The aggregate is dominated by spam, and downstream consumers waste resources on junk.
  2. **Quality-Source Crowding:** A high-quality source's single excellent reference is drowned out by a dozen mediocre references from a prolific mediocre source.
  3. **Aggregation Gaming:** An adversarial source deliberately produces many mediocre references to amplify its influence beyond what its quality warrants.

Recorded Elements

  1. **`source_quality_score_receipt`:** For each source, the cryptographic proof of its quality score (accuracy, relevance, non-spam rate), derived from historical performance.
  2. **`per_reference_quality_contribution_signature`:** For each reference, the exact contribution to the aggregate (source quality × reference quality), distinct from the reference's mere presence.
  3. **`volume_penalty_receipt`:** For sources producing high volumes relative to their quality, a cryptographic proof that a volume penalty was applied, preventing volume from dominating.
  4. **`aggregation_transparency_signature`:** The cryptographic proof of the exact weighting applied, allowing downstream consumers to audit aggregation decisions.

Condition of Validity

* **Applies when:** Multiple sources provide references, sources vary in quality, and aggregation decisions affect downstream consumers. * **Does not apply when:** All sources have uniform quality, or reference count is the sole criterion (e.g., purely informational).

Decision Boundary

  1. **Are sources heterogeneous in quality?** If no, uniform aggregation applies. If yes, proceed.
  2. **Do sources have `source_quality_score_receipts`?** If no, quality weighting is impossible. Use conservative default (low weight for unknown sources).
  3. **Is the `per_reference_quality_contribution_signature` computed?** If yes, aggregate by contribution, not by count. If no, the aggregation is count-based and vulnerable to volume dominance.
  4. **Conversely, when can aggregation ignore source quality?** Only when all sources are verified to be of high quality (no need to weight), or the reference count is explicitly the only criterion.

Operational Effect

  1. **Volume Dominance Prevention:** Volume penalties prevent prolific low-quality sources from dominating.
  2. **Quality Amplification:** High-quality sources' references are amplified proportionally to quality.
  3. **Aggregation Auditability:** Weighting is transparent, allowing downstream consumers to verify.
from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json

class AggregationVerdict(Enum):
    UNIFORM_SOURCES = auto()
    QUALITY_WEIGHTED = auto()
    VOLUME_PENALTY_APPLIED = auto()
    INSUFFICIENT_QUALITY_DATA = auto()

@dataclass(frozen=True)
class ReferenceSource:
    source_id: str
    quality_score: float
    reference_count: int

@dataclass(frozen=True)
class AggregationState:
    sources: tuple[ReferenceSource, ...]
    average_quality: float
    max_acceptable_volume_ratio: float

@dataclass(frozen=True)
class AggregationPolicy:
    require_quality_scores: bool

@dataclass(frozen=True)
class AggregationManifest:
    verdict: AggregationVerdict
    directive: str
    manifest_digest: str

class SourceQualityAggregationGuard:
    @staticmethod
    def _hash(data: str) -> str:
        return hashlib.sha256(data.encode("utf-8")).hexdigest()

    @classmethod
    def evaluate_aggregation(
        cls, state: AggregationState, policy: AggregationPolicy
    ) -> AggregationManifest:
        
        if not state.sources:
            raise ValueError("No sources to aggregate.")

        # Gate 1: Quality data presence
        if policy.require_quality_scores:
            missing = [s.source_id for s in state.sources if s.quality_score <= 0]
            if missing:
                return AggregationManifest(
                    verdict=AggregationVerdict.INSUFFICIENT_QUALITY_DATA,
                    directive=f"HALT: {len(missing)} sources lack quality scores.",
                    manifest_digest=cls._hash(f"NO_QUALITY:{len(missing)}")
                )

        # Gate 2: Compute expected ratios
        total_refs = sum(s.reference_count for s in state.sources)
        total_quality = sum(s.quality_score for s in state.sources)

        for source in state.sources:
            ref_ratio = source.reference_count / total_refs if total_refs > 0 else 0
            quality_ratio = source.quality_score / total_quality if total_quality > 0 else 0
            if ref_ratio > quality_ratio * state.max_acceptable_volume_ratio:
                return AggregationManifest(
                    verdict=AggregationVerdict.VOLUME_PENALTY_APPLIED,
                    directive=f"APPLY_PENALTY: Source {source.source_id} ref_ratio {ref_ratio:.2f} > quality_ratio {quality_ratio:.2f} * {state.max_acceptable_volume_ratio:.2f}.",
                    manifest_digest=cls._hash(f"VOLUME_PENALTY:{source.source_id}")
                )

        # Gate 3: Quality-weighted
        digest = cls._hash(json.dumps({
            "sources": tuple(
                {"id": s.source_id, "quality": f"{s.quality_score:.2f}", "count": str(s.reference_count)}
                for s in state.sources
            ),
            "total_quality": f"{total_quality:.2f}"
        }, sort_keys=True))
        
        return AggregationManifest(
            verdict=AggregationVerdict.QUALITY_WEIGHTED,
            directive=f"PROCEED: {len(state.sources)} sources quality-weighted.",
            manifest_digest=digest
        )