ACGP-1006: Certified Source Registry Specification¶

Status: Draft
Last Updated: 2026-01-08
Spec ID: ACGP-1006
Normative Keywords: MUST, SHOULD, MAY (per RFC 2119)

Abstract¶

This document specifies the Certified Source Registry, a curated, cryptographically signed catalog of authoritative data sources used for knowledge grounding validation in ACGP. The registry ensures that AI agents base their reasoning on reliable, verifiable, and contextually appropriate information sources. This specification defines the registry structure, source metadata schema, trust scoring mechanisms, Multi-Party Authorization (MPA) requirements, API interfaces, and cryptographic integrity requirements. The registry serves as a critical component in preventing hallucination, misinformation, and unauthorized information usage.

Table of Contents¶

Introduction
Registry Architecture
Source Metadata Schema
Trust and Authority Scoring
Multi-Party Authorization (MPA)
Registry Operations
Query and Validation API
Cryptographic Requirements
Synchronization and Distribution
Performance and Scaling
Federation Model
Conformance Requirements
References

1. Introduction¶

AI agents must ground their reasoning in reliable sources to maintain accuracy and trustworthiness. The Certified Source Registry provides a standardized mechanism for validating information sources, ensuring that agents operate on verified, authoritative data appropriate for their domain and risk level.

1.1 Registry Objectives¶

Prevent Hallucination: Ensure agents reference real, verified sources
Establish Authority: Rank sources by expertise and reliability
Enforce Compliance: Meet regulatory requirements for data sourcing
Enable Auditability: Trace all decisions to authoritative sources
Maintain Currency: Track source updates and deprecation

1.2 Requirements Language¶

The key words "MUST", "MUST NOT", "REQUIRED", "SHALL", "SHALL NOT", "SHOULD", "SHOULD NOT", "RECOMMENDED", "MAY", and "OPTIONAL" in this document are to be interpreted as described in RFC 2119.

2. Registry Architecture¶

2.1 System Components¶

graph TB
    subgraph "Registry Core"
        RS[(Registry Store)]
        VI[Validation Engine]
        TS[Trust Scorer]
        MPA[MPA Controller]
    end

    subgraph "Data Sources"
        REG[Regulatory<br/>Documents]
        PEER[Peer-Reviewed<br/>Publications]
        NEWS[News Outlets]
        CORP[Corporate<br/>Data]
        TECH[Technical<br/>Documentation]
    end

    subgraph "Consumers"
        GS[Governance Stewards]
        PA[Operating Agents]
        AUDIT[Audit Systems]
    end

    subgraph "Management"
        ADMIN[Registry Admins]
        REVIEW[Review Board]
        API[Management API]
    end

    REG --> VI
    PEER --> VI
    NEWS --> VI
    CORP --> VI
    TECH --> VI

    VI --> TS
    TS --> RS
    MPA --> RS

    RS --> GS
    RS --> PA
    RS --> AUDIT

    ADMIN --> API
    REVIEW --> MPA
    API --> MPA

    style RS fill:#e8f5e9
    style MPA fill:#fff3e0
    style VI fill:#e3f2fd

2.2 Registry Layers¶

Layer	Function	Components
Storage Layer	Persistent source metadata	Distributed database, IPFS, CDN
Validation Layer	Source verification	URL validator, content hasher, schema checker
Trust Layer	Authority scoring	ML models, heuristics, human review
Governance Layer	Change control	MPA system, audit log, versioning
API Layer	External access	REST/GraphQL, streaming, webhooks

2.3 Registry States¶

stateDiagram-v2
    [*] --> Submitted: New Source
    Submitted --> UnderReview: Validation Started
    UnderReview --> Rejected: Validation Failed
    UnderReview --> PendingApproval: Validation Passed

    PendingApproval --> Rejected: MPA Denied
    PendingApproval --> Active: MPA Approved

    Active --> Suspended: Issue Detected
    Active --> Deprecated: Source Outdated
    Active --> Updated: New Version

    Suspended --> Active: Issue Resolved
    Suspended --> Revoked: Permanent Removal

    Updated --> Active: Auto-transition
    Deprecated --> Archived: After Grace Period

    Rejected --> [*]
    Revoked --> [*]
    Archived --> [*]

3. Source Metadata Schema¶

3.1 Core Metadata Fields¶

{
  "source_id": "uuid-v4",
  "source_type": "regulatory|peer_reviewed|news|corporate|technical|reference",
  "identifiers": {
    "doi": "10.1234/example",
    "isbn": "978-0-123456-78-9",
    "url": "https://example.com/document",
    "hash": "sha256:abcd1234...",
    "custom": {}
  },
  "metadata": {
    "title": "Source Title",
    "authors": ["Author Name"],
    "publisher": "Publisher Name",
    "publication_date": "2024-01-15",
    "last_updated": "2025-01-15T10:00:00Z",
    "language": "en",
    "jurisdiction": ["US", "EU"],
    "domain": ["finance", "healthcare"],
    "tags": ["regulation", "compliance"]
  },
  "authority": {
    "trust_score": 0.95,
    "expertise_domains": ["financial_regulation"],
    "verification_level": "cryptographic|institutional|community|automated",
    "endorsements": [
      {
        "endorser": "regulatory_body_id",
        "timestamp": "2025-01-15T10:00:00Z",
        "signature": "ES256:..."
      }
    ]
  },
  "access": {
    "availability": "public|restricted|licensed",
    "license": "CC-BY-4.0",
    "cost": 0.00,
    "authentication_required": false,
    "rate_limits": {
      "requests_per_minute": 100
    }
  },
  "validity": {
    "effective_date": "2024-01-01",
    "expiration_date": "2026-12-31",
    "supersedes": ["old-source-id"],
    "superseded_by": null,
    "revision": 3
  },
  "governance": {
    "submitted_by": "user-uuid",
    "approved_by": ["approver1-uuid", "approver2-uuid"],
    "approval_date": "2024-01-20T10:00:00Z",
    "mpa_requirement": 2,
    "review_frequency": "quarterly"
  }
}

3.2 Source Type Specifications¶

Type	Description	Authority Weight	Examples
Regulatory	Government/regulatory documents	1.0	SEC filings, FDA guidance
Peer Reviewed	Academic publications	0.9	Journal articles, research papers
Corporate	Official company sources	0.7	Annual reports, press releases
Technical	Technical documentation	0.8	API docs, specifications
News	Journalistic sources	0.6	Reuters, Bloomberg
Reference	Reference materials	0.5	Wikipedia, encyclopedias

3.3 Verification Levels¶

Level	Requirements	Trust Multiplier
Cryptographic	Digital signature (ES256) from issuer	1.0
Institutional	Verified institutional source	0.9
Community	Community consensus (3+ endorsements)	0.7
Automated	Algorithmic verification only	0.5

4. Trust and Authority Scoring¶

4.1 Trust Score Calculation¶

Trust scores align with ACGP governance thresholds and are calculated to support intervention decisions.

def calculate_trust_score(source: SourceMetadata) -> float:
    """
    Calculate trust score for a source.

    Trust scores range from 0.0 to 1.0, where:
    - 1.0 = Perfect trust (no risk)
    - 0.0 = No trust (maximum risk)

    This aligns with CTQ scores (higher is better) and opposes Risk Scores.
    """
    # Base score from source type
    base_score = SOURCE_TYPE_WEIGHTS[source.source_type]

    # Apply verification level multiplier
    verification_mult = VERIFICATION_MULTIPLIERS[source.verification_level]

    # Factor in age and updates
    age_factor = calculate_age_decay(source.publication_date)
    update_factor = calculate_update_bonus(source.last_updated)

    # Consider endorsements
    endorsement_score = sum([
        validate_endorsement_es256(e) * ENDORSER_WEIGHTS.get(e.endorser, 0.1)
        for e in source.endorsements
    ])

    # Apply domain expertise alignment
    domain_alignment = calculate_domain_match(
        source.expertise_domains,
        query_context.required_domains
    )

    # Composite score
    trust_score = (
        base_score * verification_mult * age_factor * 
        update_factor * (1 + endorsement_score) * domain_alignment
    )

    return min(1.0, max(0.0, trust_score))

4.2 Authority Ranking¶

Sources are ranked by composite authority:

Authority_Rank = Trust_Score × Relevance × Recency × Accessibility

where:
  - Trust_Score ∈ [0, 1]
  - Relevance ∈ [0, 1] based on domain/tag match
  - Recency ∈ [0, 1] with exponential decay
  - Accessibility ∈ [0.5, 1] based on availability

4.3 Trust Score Integration with Governance¶

Trust scores integrate with ACGP governance decisions:

trust_score_integration:
  # Map trust scores to risk assessment
  high_trust: # Trust Score >= 0.8
    - Reduces CTQ risk penalties
    - Allows lower verification requirements
    - Supports higher ACL tiers

  medium_trust: # Trust Score 0.5-0.8
    - Standard CTQ evaluation
    - Normal verification requirements
    - Standard ACL tier support

  low_trust: # Trust Score < 0.5
    - Increases CTQ risk penalties
    - Requires additional verification
    - May trigger escalation for high ACL tiers

4.4 Domain-Specific Weights¶

domain_weights:
  finance:
    regulatory: 1.0
    peer_reviewed: 0.8
    corporate: 0.7
    news: 0.6

  healthcare:
    peer_reviewed: 1.0
    regulatory: 0.9
    technical: 0.7
    corporate: 0.5

  technology:
    technical: 1.0
    peer_reviewed: 0.8
    corporate: 0.7
    news: 0.5

5. Multi-Party Authorization (MPA)¶

5.1 MPA Requirements by Operation¶

Operation	Min Approvers	Approver Roles	Time Window
Add Source (Low Risk)	1	Any admin	24 hours
Add Source (High Risk)	2	Admin + Domain expert	48 hours
Modify Authority Score	2	Admin + Security	24 hours
Deprecate Source	1	Any admin	12 hours
Revoke Source	3	Admin + Security + Legal	6 hours
Emergency Revoke	2	Security + Any	Immediate

5.2 MPA Workflow¶

sequenceDiagram
    participant Req as Requestor
    participant MPA as MPA Controller
    participant A1 as Approver 1
    participant A2 as Approver 2
    participant REG as Registry
    participant LOG as Audit Log

    Req->>MPA: Submit Change Request
    MPA->>MPA: Validate Request
    MPA->>MPA: Determine Required Approvers

    MPA->>A1: Request Approval
    MPA->>A2: Request Approval

    A1->>MPA: Approve + Sign (ES256)
    A2->>MPA: Approve + Sign (ES256)

    MPA->>MPA: Verify ES256 Signatures
    MPA->>MPA: Check Quorum

    alt Approved
        MPA->>REG: Apply Change
        REG->>LOG: Log Change
        MPA->>Req: Change Applied
    else Denied
        MPA->>LOG: Log Denial
        MPA->>Req: Change Denied
    end

5.3 Approval Record¶

{
  "approval_id": "uuid-v4",
  "request": {
    "type": "add_source|modify|revoke",
    "source_id": "uuid-v4",
    "changes": {},
    "requestor": "user-uuid",
    "timestamp": "2025-01-15T10:00:00Z",
    "justification": "Reason for change"
  },
  "approvals": [
    {
      "approver_id": "approver1-uuid",
      "role": "admin",
      "decision": "approve|deny",
      "timestamp": "2025-01-15T11:00:00Z",
      "signature": {
        "algorithm": "ES256",
        "value": "base64_encoded_signature"
      },
      "comments": "Optional feedback"
    }
  ],
  "result": {
    "status": "approved|denied|timeout",
    "applied_at": "2025-01-15T12:00:00Z",
    "applied_by": "system",
    "rollback_available": true
  }
}

6. Registry Operations¶

6.1 Source Lifecycle Operations¶

6.1.1 Source Submission¶

def submit_source(source_data: dict, submitter: User) -> SourceSubmission:
    # Validate schema
    if not validate_schema(source_data, SOURCE_SCHEMA):
        raise ValidationError("Invalid source schema")

    # Check for duplicates
    if find_duplicate(source_data):
        raise DuplicateError("Source already exists")

    # Assess risk level
    risk_level = assess_source_risk(source_data)

    # Create submission
    submission = SourceSubmission(
        source_data=source_data,
        submitter=submitter,
        risk_level=risk_level,
        status="submitted",
        mpa_requirement=determine_mpa_requirement(risk_level)
    )

    # Start validation workflow
    start_validation_workflow(submission)

    return submission

6.1.2 Source Validation¶

async def validate_source(submission: SourceSubmission) -> ValidationResult:
    results = []

    # URL validation
    if submission.source_data.get("url"):
        results.append(await validate_url(submission.source_data["url"]))

    # Content verification
    content_hash = await fetch_and_hash(submission.source_data)
    results.append(verify_content_hash(content_hash, submission.source_data))

    # Metadata validation
    results.append(validate_metadata(submission.source_data))

    # Authority verification
    if submission.source_data.get("endorsements"):
        for endorsement in submission.source_data["endorsements"]:
            results.append(await verify_endorsement_es256(endorsement))

    return ValidationResult(
        passed=all(r.passed for r in results),
        details=results,
        trust_score=calculate_initial_trust(results)
    )

6.1.3 Source Updates¶

def update_source(source_id: str, updates: dict, requestor: User) -> UpdateResult:
    source = get_source(source_id)

    # Check if update requires MPA
    if requires_mpa(source, updates):
        return initiate_mpa_workflow(source, updates, requestor)

    # Apply updates
    updated_source = apply_updates(source, updates)

    # Re-calculate trust score
    updated_source.trust_score = calculate_trust_score(updated_source)

    # Version the change
    updated_source.revision += 1
    updated_source.last_updated = datetime.utcnow()

    # Store and propagate
    store_source(updated_source)
    propagate_update(updated_source)

    return UpdateResult(success=True, new_revision=updated_source.revision)

6.2 Bulk Operations¶

6.2.1 Batch Import¶

batch_import:
  format: jsonl | csv | xml
  validation: parallel
  max_batch_size: 1000
  error_handling: 
    strategy: continue_on_error
    error_threshold: 10%
  mpa_requirement:
    bulk_add: 3 approvers
    review_window: 72 hours

6.2.2 Periodic Review¶

async def periodic_review():
    # Review sources by age
    old_sources = query_sources(
        last_updated_before=datetime.utcnow() - timedelta(days=90)
    )

    for source in old_sources:
        # Check if still valid
        if not await check_source_validity(source):
            await mark_for_review(source)

        # Check for newer versions
        if newer_version := await find_newer_version(source):
            await suggest_upgrade(source, newer_version)

    # Review trust scores
    await recalculate_all_trust_scores()

    # Generate review report
    return generate_review_report()

7. Query and Validation API¶

7.1 Query Interface¶

7.1.1 REST API Endpoints¶

endpoints:
  # Query sources
  GET /sources:
    parameters:
      - domain: string[]
      - type: string[]
      - min_trust: float
      - jurisdiction: string[]
      - tags: string[]
      - limit: integer
      - offset: integer
    response: SourceList

  # Get specific source
  GET /sources/{source_id}:
    response: SourceMetadata

  # Validate source reference
  POST /validate:
    body:
      source_ref: string
      context: object
      min_trust_required: float  # Based on ACL tier
    response: ValidationResult

  # Batch validation
  POST /validate/batch:
    body:
      source_refs: string[]
      context: object
    response: ValidationResult[]

7.1.2 GraphQL Schema¶

type Source {
  id: ID!
  type: SourceType!
  title: String!
  trustScore: Float!
  metadata: SourceMetadata!
  authority: AuthorityInfo!
  validity: ValidityInfo!
}

type Query {
  sources(
    filter: SourceFilter
    sort: SourceSort
    pagination: Pagination
  ): SourceConnection!

  source(id: ID!): Source

  validateSource(
    ref: String!
    context: ValidationContext
    minTrustRequired: Float
  ): ValidationResult!
}

input SourceFilter {
  domain: [String!]
  type: [SourceType!]
  minTrust: Float
  maxAge: Int
  jurisdiction: [String!]
}

7.2 Validation API¶

7.2.1 Real-time Validation¶

class RegistryValidator:
    async def validate_source_reference(
        self,
        reference: str,
        context: dict
    ) -> ValidationResult:
        # Parse reference
        source_id = self.parse_reference(reference)

        # Fetch source metadata
        source = await self.registry.get_source(source_id)

        if not source:
            return ValidationResult(
                valid=False,
                reason="Source not found in registry"
            )

        # Check validity period
        if not self.is_valid_period(source):
            return ValidationResult(
                valid=False,
                reason="Source outside validity period"
            )

        # Check domain match
        if not self.matches_domain(source, context.get("domain")):
            return ValidationResult(
                valid=False,
                reason="Source not valid for domain"
            )

        # Check trust threshold (aligned with ACGP-1005)
        required_trust = self.get_required_trust_by_acl(context.get("acl_tier", "ACL-2"))
        if source.trust_score < required_trust:
            return ValidationResult(
                valid=False,
                reason=f"Trust score {source.trust_score} below required {required_trust}"
            )

        # Check ACL tier requirements
        if not self.meets_acl_requirements(source, context.get("acl_tier")):
            return ValidationResult(
                valid=False,
                reason="Source not permitted for ACL tier"
            )

        return ValidationResult(
            valid=True,
            source=source,
            trust_score=source.trust_score,
            warnings=self.check_warnings(source)
        )

    def get_required_trust_by_acl(self, acl_tier: str) -> float:
        """
        Map ACL tiers to minimum trust score requirements.

        Trust scores align with CTQ quality (higher is better).
        These thresholds ensure sources are appropriate for agent risk level.
        """
        return {
            'ACL-0': 0.5,   # Minimal risk - moderate trust required
            'ACL-1': 0.6,   # Low risk - good trust required
            'ACL-2': 0.7,   # Medium risk - high trust required
            'ACL-3': 0.8,   # High risk - very high trust required
            'ACL-4': 0.85,  # Very high risk - excellent trust required
            'ACL-5': 0.9    # Critical risk - near-perfect trust required
        }.get(acl_tier, 0.7)

7.2.2 Batch Validation¶

async def validate_batch(
    references: List[str],
    context: dict
) -> BatchValidationResult:
    results = await asyncio.gather(*[
        validate_source_reference(ref, context)
        for ref in references
    ])

    return BatchValidationResult(
        total=len(references),
        valid=sum(1 for r in results if r.valid),
        invalid=sum(1 for r in results if not r.valid),
        average_trust=mean([r.trust_score for r in results if r.valid]),
        results=results
    )

8. Cryptographic Requirements¶

8.1 Registry Integrity¶

8.1.1 Merkle Tree Structure¶

class RegistryMerkleTree:
    def __init__(self):
        self.root = None
        self.leaves = []

    def add_source(self, source: SourceMetadata):
        # Create leaf hash (SHA-256)
        leaf = self.hash_source_sha256(source)
        self.leaves.append(leaf)

        # Rebuild tree
        self.rebuild_tree()

    def hash_source_sha256(self, source: SourceMetadata) -> str:
        # Canonical JSON serialization
        canonical = json.dumps(source, sort_keys=True, separators=(',', ':'))

        # SHA-256 hash
        return hashlib.sha256(canonical.encode()).hexdigest()

    def get_proof(self, source_id: str) -> MerkleProof:
        # Generate inclusion proof
        leaf_index = self.find_leaf_index(source_id)
        return self.generate_proof_path(leaf_index)

    def verify_proof(self, source: SourceMetadata, proof: MerkleProof) -> bool:
        # Verify source is in registry
        leaf = self.hash_source_sha256(source)
        computed_root = self.compute_root_from_proof(leaf, proof)
        return computed_root == self.root

8.1.2 Digital Signatures¶

signature_requirements:
  algorithm: ES256  # ECDSA with P-256 and SHA-256 (standardized)
  key_size: 256 bits

  signing_operations:
    - source_submission
    - source_approval
    - trust_score_update
    - source_revocation
    - endorsements

  signature_format:
    header:
      alg: ES256
      typ: JWT
      kid: key_identifier
    payload:
      operation: string
      data_hash: sha256_hash
      timestamp: unix_timestamp
      nonce: random_string
    signature: base64url_encoded

8.2 Source Content Verification¶

class ContentVerifier:
    def verify_source_content(
        self,
        source: SourceMetadata,
        content: bytes
    ) -> VerificationResult:
        # Calculate content hash (SHA-256)
        content_hash = hashlib.sha256(content).hexdigest()

        # Compare with registered hash
        if content_hash != source.identifiers.get("hash"):
            return VerificationResult(
                verified=False,
                reason="Content hash mismatch"
            )

        # Verify ES256 digital signature if present
        if signature := source.metadata.get("signature"):
            if not self.verify_signature_es256(content, signature):
                return VerificationResult(
                    verified=False,
                    reason="Invalid ES256 signature"
                )

        return VerificationResult(
            verified=True,
            content_hash=content_hash,
            timestamp=datetime.utcnow()
        )

9. Synchronization and Distribution¶

9.1 Registry Replication¶

graph TB
    subgraph "Primary Registry"
        PR[(Primary Node)]
        PW[Write Endpoint]
    end

    subgraph "Regional Replicas"
        R1[(US-East)]
        R2[(EU-West)]
        R3[(AP-South)]
    end

    subgraph "Edge Caches"
        E1[CDN 1]
        E2[CDN 2]
        E3[CDN 3]
    end

    PW --> PR
    PR -->|Sync| R1
    PR -->|Sync| R2
    PR -->|Sync| R3

    R1 --> E1
    R2 --> E2
    R3 --> E3

    style PR fill:#4caf50
    style R1 fill:#81c784
    style R2 fill:#81c784
    style R3 fill:#81c784

9.2 Synchronization Protocol¶

class RegistrySync:
    async def sync_with_primary(self):
        # Get local state
        local_version = self.get_local_version()
        local_root = self.get_merkle_root()

        # Query primary for updates
        updates = await self.primary.get_updates_since(
            version=local_version,
            root=local_root
        )

        # Validate updates (ES256 signatures)
        for update in updates:
            if not self.validate_update_es256(update):
                raise SyncError(f"Invalid update signature: {update.id}")

        # Apply updates in order
        for update in updates:
            await self.apply_update(update)

        # Verify final state
        if not await self.verify_consistency():
            await self.trigger_full_resync()

9.3 Distribution Strategy¶

Component	Update Frequency	Propagation Time	Cache TTL
Source Metadata	On change	<60 seconds	1 hour
Trust Scores	Hourly	<5 minutes	30 minutes
Merkle Root	Every update	<30 seconds	5 minutes
Revocations	Immediate	<10 seconds	No cache

10. Performance and Scaling¶

10.1 Performance Requirements¶

Operation	P50	P95	P99	SLA
Single Query	10ms	50ms	100ms	99.9%
Batch Query (100)	50ms	200ms	500ms	99.5%
Validation	5ms	25ms	50ms	99.99%
Update Propagation	1s	30s	60s	99%
MPA Approval	1min	1hr	24hr	95%

10.2 Scaling Architecture¶

scaling_strategy:
  storage:
    primary_db: PostgreSQL
    sharding: by_domain
    replicas: 3 per region

  caching:
    l1_cache:
      type: in-memory
      size: 1GB
      ttl: 60s
    l2_cache:
      type: Redis
      size: 10GB
      ttl: 300s
    l3_cache:
      type: CDN
      ttl: 3600s

  query_optimization:
    indexes:
      - domain
      - trust_score
      - publication_date
      - source_type
    materialized_views:
      - high_trust_sources
      - recent_sources
      - domain_authorities

10.3 Capacity Planning¶

def estimate_capacity(config: RegistryConfig) -> CapacityEstimate:
    # Storage requirements
    avg_source_size = 10_000  # bytes
    total_sources = config.expected_sources
    metadata_storage = total_sources * avg_source_size

    # Index overhead (30% of data)
    index_storage = metadata_storage * 0.3

    # Audit log (10x metadata over time)
    audit_storage = metadata_storage * 10

    # Query load
    qps = config.expected_qps
    cache_hit_ratio = 0.8
    db_qps = qps * (1 - cache_hit_ratio)

    return CapacityEstimate(
        storage_required=metadata_storage + index_storage + audit_storage,
        cache_size_required=metadata_storage * 0.1,
        db_connections=db_qps / 10,  # 10 queries per connection
        bandwidth_required=qps * avg_source_size * 0.1  # 10% return full source
    )

11. Federation Model¶

11.1 Federation Overview¶

Registry federation enables multiple independent registries to interoperate while maintaining autonomy and trust boundaries.

┌─────────────────────────────────────────────────────────────────┐
│                    FEDERATED REGISTRY MODEL                      │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌─────────────┐    trust    ┌─────────────┐                    │
│  │  Primary    │◄───────────►│  Partner    │                    │
│  │  Registry   │             │  Registry   │                    │
│  │  (org A)    │             │  (org B)    │                    │
│  └──────┬──────┘             └──────┬──────┘                    │
│         │                           │                           │
│    ┌────┴────┐                 ┌────┴────┐                     │
│    │ Sources │                 │ Sources │                     │
│    │ A1, A2  │                 │ B1, B2  │                     │
│    └─────────┘                 └─────────┘                     │
│                                                                  │
│  ┌─────────────────────────────────────────────────────────┐    │
│  │                  FEDERATION PROTOCOL                     │    │
│  │                                                          │    │
│  │  1. Trust establishment (mutual TLS + certificate)       │    │
│  │  2. Capability negotiation (supported source types)      │    │
│  │  3. Query forwarding (cross-registry lookups)            │    │
│  │  4. Trust score translation (normalize between systems)  │    │
│  │  5. Audit trail synchronization (cross-registry refs)    │    │
│  │                                                          │    │
│  └─────────────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────────┘

11.2 Trust Bootstrapping¶

Establishing trust between registries:

federation:
  trust_bootstrapping:
    # Step 1: Certificate exchange
    method: mutual_tls
    certificate_requirements:
      issuer: approved_ca_list
      validity_min_days: 30
      key_algorithm: ECDSA-P256

    # Step 2: Capability declaration
    capabilities:
      - source_types: [regulatory, peer_reviewed, technical]
      - trust_model: acgp_standard
      - api_version: "1.0"

    # Step 3: Trust level assignment
    initial_trust_level: provisional
    graduation_criteria:
      - successful_validations: 100
      - time_in_provisional: 30_days
      - audit_passed: true

    # Trust levels
    trust_levels:
      provisional: 0.5    # Limited trust, queries verified
      standard: 0.8       # Normal operations
      trusted: 1.0        # Full trust, no verification

11.3 Cross-Registry Queries¶

class FederatedRegistry:
    """Federated registry query handler."""

    def __init__(self):
        self.local_registry = LocalRegistry()
        self.federation_peers = []

    async def query(self, source_ref: str) -> SourceRecord:
        """
        Query source across federation.

        1. Check local registry first
        2. If not found, query federation peers
        3. Apply trust translation for foreign sources
        """
        # Try local first
        local_result = self.local_registry.query(source_ref)
        if local_result:
            return local_result

        # Query federation peers
        for peer in self.federation_peers:
            if not peer.supports_source_type(source_ref):
                continue

            try:
                remote_result = await peer.query(source_ref)
                if remote_result:
                    # Translate trust score based on peer trust level
                    translated = self.translate_trust(
                        remote_result,
                        peer.trust_level
                    )
                    return translated
            except FederationError as e:
                self.log_federation_failure(peer, e)
                continue

        return None

    def translate_trust(self, record: SourceRecord, peer_trust: float) -> SourceRecord:
        """
        Adjust trust score based on peer trust level.

        Formula: effective_trust = source_trust * peer_trust * 0.9
        The 0.9 factor accounts for federation overhead.
        """
        record.trust_score = record.trust_score * peer_trust * 0.9
        record.provenance = f"federated:{record.provenance}"
        return record

11.4 Federation Governance¶

federation_governance:
  admission:
    requirements:
      - signed_participation_agreement
      - security_audit_passed
      - operational_capacity_verified
    approval: majority_vote  # Existing federation members

  operations:
    audit_frequency: quarterly
    sync_interval: 1_hour
    failure_threshold: 5  # Consecutive failures before suspension

  dispute_resolution:
    process: binding_arbitration
    neutral_party: acgp_foundation

  exit:
    notice_period: 90_days
    data_handling: source_references_preserved

11.5 Anti-Patterns¶

Registry Capture - One organization controlling majority of sources - Mitigation: Diversity requirements for trusted registries

Trust Score Inflation - Registries artificially inflating scores for partners - Mitigation: Cross-validation requirements, audit trails

Circular References - Registry A trusts B, B trusts C, C trusts A with amplification - Mitigation: Trust decay per hop, cycle detection

Single Point of Failure - All queries routing through one registry - Mitigation: Peer-to-peer queries, no hub dependency

12. Conformance Requirements¶

12.1 Registry Implementation Requirements¶

A conformant Certified Source Registry MUST:

12.1.1 Core Functionality¶

Implement the complete source metadata schema
Support all source lifecycle states
Calculate trust scores using the defined algorithm
Enforce MPA requirements for changes
Maintain cryptographic integrity with Merkle trees

12.1.2 API Requirements¶

Provide REST or GraphQL query interface
Support single and batch validation
Implement rate limiting and authentication
Return results within performance SLAs
Support pagination for large result sets

12.1.3 Security Requirements¶

Sign all source records with ES256 (standardized)
Verify ES256 signatures on submission
Maintain audit log of all changes
Support TLS 1.3 for all connections
Implement role-based access control

12.1.4 Operational Requirements¶

Achieve 99.9% availability for queries
Propagate updates within 60 seconds
Support at least 1000 sources
Handle 100 QPS minimum
Retain audit logs for 7 years

12.1.5 Integration Requirements¶

Provide trust scores aligned with ACGP-1005 thresholds
Support ACL-tier based trust requirements
Integrate with governance decision flow
Support trust score to risk score mapping

12.2 Client Integration Requirements¶

Clients using the registry MUST: - Cache query results respecting TTL - Validate source references before use - Handle validation failures gracefully - Report invalid sources to registry - Implement exponential backoff for retries - Apply ACL-tier appropriate trust thresholds

12.3 Testing Requirements¶

Implementations MUST pass: - Schema validation test suite - Trust score calculation tests - MPA workflow tests - Performance benchmark suite - Security penetration tests - ES256 signature verification tests

13. References¶

Normative References¶

ACGP-1000: Core Protocol Specification
ACGP-1001: Terminology and Definitions
ACGP-1002: Architecture Specification
ACGP-1003: Message Formats & Wire Protocol
ACGP-1005: ARS-CTQ-ACL Integration Framework
ACGP-1007: Security Considerations
RFC 2119: Key words for use in RFCs
RFC 7515: JSON Web Signature (JWS)

Informative References¶

DOI Specification: Digital Object Identifier System
ISBN Standard: International Standard Book Number
Dublin Core: Metadata standards
IPFS: InterPlanetary File System for distributed storage
Merkle Trees: R.C. Merkle, "A Digital Signature Based on a Conventional Encryption Function"

Cryptographic Standards¶

FIPS 186-4: Digital Signature Standard
NIST SP 800-56A: Pair-Wise Key Establishment Schemes
RFC 7518: JSON Web Algorithms (JWA)

End of ACGP-1006