Scope boundary

Owns the signed provenance, trust-tier, write-eligibility, evidence, namespace, and retention contract for memory and knowledge-base records. It classifies what downstream gates may consider but does not execute a read, write, promotion, or deletion.

AID-I-004.004: AID-M-002.004 defines whether a memory or knowledge record is eligible for promotion and which evidence must bind it; AID-I-004.004 executes the transactional quarantine-to-trusted promotion. An eligibility contract is not a successful promotion receipt, and the runtime gate must not invent provenance or trust-tier facts.

AID-H-029.002: AID-M-002.004 defines record-level provenance and write-eligibility facts; AID-H-029.002 enforces whether data use is authorized at a particular lifecycle stage. A record can be technically eligible for storage yet unauthorized for the current stage, purpose, subject, or processing path.

Implementation guidance

• AID-M-002.004-G001: Define a versioned KB or memory write eligibility contract that captures caller identity, payload shape, evidence references, validator output, and downstream enforcement fields.
• AID-M-002.004-G002: Use policy-as-code to derive and record a recommended trust tier for each KB or memory write candidate based on evidence, source trust, and validator output.
• AID-M-002.004-G003: Define trust-tiered namespace metadata and retrieval eligibility rules with explicit promotion, probation, and quarantine states.
• AID-M-002.004-G004: Generate promotion evidence by checking candidate facts against trusted facts before any KB or memory entry is marked trusted.
• AID-M-002.004-G005: Record immutable provenance for every KB or memory write candidate and promotion decision, including who proposed it, which evidence was reviewed, and which policy produced the trust tier.
• AID-M-002.004-G006: Publish source trust-state degradation evidence when verification failures spike, so Harden controls can perform runtime circuit breaking or fail-closed admission.

AID-M-002.004-G001

Concept:

This Model control defines the canonical evidence contract for memory, KB, and vector-store write candidates. It does not make the final runtime allow/deny decision. It produces a stable eligibility record that Harden controls, storage pipelines, and auditors can consume without guessing which identity, evidence, validator score, or policy version was used.

Step 1: Define the write eligibility contract

import json as _aidefend_policy_json
import math as _aidefend_policy_math
import os as _aidefend_policy_os
import stat as _aidefend_policy_stat
import subprocess as _aidefend_policy_subprocess
import tempfile as _aidefend_policy_tempfile
from pathlib import Path as _AidefendPolicyPath


def _aidefend_read_regular_bytes(path):
    descriptor = _aidefend_policy_os.open(
        path,
        _aidefend_policy_os.O_RDONLY
        | getattr(_aidefend_policy_os, "O_BINARY", 0)
        | getattr(_aidefend_policy_os, "O_NOFOLLOW", 0),
    )
    with _aidefend_policy_os.fdopen(descriptor, "rb") as handle:
        status = _aidefend_policy_os.fstat(handle.fileno())
        if not _aidefend_policy_stat.S_ISREG(status.st_mode):
            raise RuntimeError(f"signed policy input is not a regular file: {path}")
        value = handle.read()
    if not value:
        raise RuntimeError(f"signed policy input is empty: {path}")
    return value


def _aidefend_load_signed_policy(*, path_env, signature_env, key_env, schema_version):
    policy_path = _AidefendPolicyPath(_aidefend_policy_os.environ[path_env])
    signature_path = _AidefendPolicyPath(_aidefend_policy_os.environ[signature_env])
    verify_timeout = float(_aidefend_policy_os.environ["AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS"])
    if not _aidefend_policy_math.isfinite(verify_timeout) or verify_timeout <= 0:
        raise RuntimeError("policy-verification timeout from the versioned runtime profile is invalid")
    policy_bytes = _aidefend_read_regular_bytes(policy_path)
    signature_bytes = _aidefend_read_regular_bytes(signature_path)
    with _aidefend_policy_tempfile.TemporaryDirectory(prefix="aidefend-policy-") as stage_name:
        stage = _AidefendPolicyPath(stage_name)
        _aidefend_policy_os.chmod(stage, 0o700)
        policy_snapshot = stage / "policy.json"
        signature_snapshot = stage / "policy.sig"
        policy_snapshot.write_bytes(policy_bytes)
        signature_snapshot.write_bytes(signature_bytes)
        _aidefend_policy_os.chmod(policy_snapshot, 0o400)
        _aidefend_policy_os.chmod(signature_snapshot, 0o400)
        _aidefend_policy_subprocess.run(
            [
                "cosign", "verify-blob", "--key", _aidefend_policy_os.environ[key_env],
                "--bundle", str(signature_snapshot), str(policy_snapshot),
            ],
            check=True, capture_output=True, text=True, timeout=verify_timeout,
        )
        if (
            _aidefend_read_regular_bytes(policy_snapshot) != policy_bytes
            or _aidefend_read_regular_bytes(signature_snapshot) != signature_bytes
        ):
            raise RuntimeError("private policy snapshot changed during verification")
    policy = _aidefend_policy_json.loads(policy_bytes.decode("utf-8", errors="strict"))
    if (
        not isinstance(policy, dict)
        or policy.get("schema_version") != schema_version
        or not isinstance(policy.get("policy_version"), str)
        or not policy["policy_version"]
    ):
        raise RuntimeError("signature-verified policy schema or version is invalid")
    return policy


# File: kb_trust_contract/contracts.py
from pydantic import BaseModel, Field


MEMORY_WRITE_SCHEMA_POLICY = _aidefend_load_signed_policy(
    path_env="M002004_SCHEMA_POLICY_PATH",
    signature_env="M002004_SCHEMA_POLICY_SIGNATURE",
    key_env="M002004_SCHEMA_POLICY_VERIFY_KEY",
    schema_version="aidefend.memory-write-schema-policy.v1",
)
CONTENT_MIN_LENGTH = MEMORY_WRITE_SCHEMA_POLICY.get("content_min_length")
if (
    isinstance(CONTENT_MIN_LENGTH, bool)
    or not isinstance(CONTENT_MIN_LENGTH, int)
    or CONTENT_MIN_LENGTH < 1
):
    raise RuntimeError("signed memory-write content bound is invalid")


class EvidenceRef(BaseModel):
    source_uri: str
    sha256: str
    collected_by: str


class MemoryWriteEligibility(BaseModel):
    document_id: str
    content: str = Field(min_length=CONTENT_MIN_LENGTH)
    embedding: list[float]
    evidence_refs: list[EvidenceRef]
    validator_score: float = Field(ge=0.0, le=1.0)
    source_system: str
    requested_stage: str = "memory_write"
    writer_identity: str

    def validate_contract_requirements(self) -> None:
        if len(self.embedding) < 8:
            raise ValueError("embedding_too_short")
        if not self.evidence_refs:
            raise ValueError("evidence_refs_required")
        if not self.writer_identity.startswith("spiffe://corp.internal/"):
            raise ValueError("untrusted_writer_identity")

Step 2: Build a read-only eligibility record for downstream enforcement

from __future__ import annotations

# File: kb_trust_contract/build_record.py

import hashlib
import json
from uuid import uuid4

from contracts import MEMORY_WRITE_SCHEMA_POLICY, MemoryWriteEligibility


def canonical_bytes(value: dict) -> bytes:
    return json.dumps(
        value, sort_keys=True, separators=(",", ":"), allow_nan=False
    ).encode("utf-8")


def build_eligibility_record(candidate: MemoryWriteEligibility) -> dict:
    candidate.validate_contract_requirements()
    payload = candidate.model_dump(mode="json")
    payload_sha256 = hashlib.sha256(canonical_bytes(payload)).hexdigest()
    return {
        "schema_version": "aidefend.memory_write_eligibility.v1",
        "eligibility_record_id": str(uuid4()),
        "eligibility_policy_version": MEMORY_WRITE_SCHEMA_POLICY["policy_version"],
        "document_id": candidate.document_id,
        "writer_identity": candidate.writer_identity,
        "requested_stage": candidate.requested_stage,
        "source_system": candidate.source_system,
        "validator_score": candidate.validator_score,
        "evidence_ref_count": len(candidate.evidence_refs),
        "candidate_payload_sha256": payload_sha256,
        "outcome": "ready_for_harden_stage_gate",
    }

Action: Store this eligibility record with the candidate write and pass it to the lifecycle-stage gate in AID-H-029.002. The public vector store or memory API should consume only records that carry this contract, while the actual fail-closed runtime admission decision remains in Harden.

AID-M-002.004-G002

Concept:

The external caller submits only immutable references to a quarantined candidate and its producer, validator, and evidence records. Trusted admission middleware verifies those bytes and derives the policy facts locally; the caller cannot assert a trust tier, signature verdict, source status, score, evidence count, or risk flags. Runtime enforcement remains with the Harden stage gate and retrieval controls that consume the recommendation.

Agent-produced content is never its own evidence. A trusted writer identity proves who submitted the candidate, not that the fact is true. Content that a model or agent generated, summarized, translated, or relayed from another agent is a derived origin: it can only reach trusted on the strength of independent roots (an external source or a human input) recorded in its evidence manifest, and a re-send, summary, or cross-agent relay of one root still counts as that one root. Without such a root the policy returns quarantined no matter how trusted the writer identity is, which closes the loop where an agent stores its own earlier output and later cites it back as confirmation. The candidate declares its origin and the manifest declares each reference's root; the admission client verifies both against signed bytes and derives the counts itself.

Step 1: Accept references, not verdict fields

# File: kb_trust_contract/candidate_request.json
{
  "candidate_ref": "candidates/7f4e/candidate.json",
  "candidate_signature_ref": "candidates/7f4e/candidate.sig",
  "validator_receipt_ref": "validators/7f4e/result.json",
  "validator_signature_ref": "validators/7f4e/result.sig",
  "evidence_manifest_ref": "evidence/7f4e/manifest.json",
  "evidence_signature_ref": "evidence/7f4e/manifest.sig"
}

Step 2: Evaluate a Rego policy that returns both tier and reasons

Load the rule module and threshold data only from the same signature-verified, versioned OPA bundle. Its signed data document supplies the trusted and probation evidence-count, independent-root-count, and validator-score thresholds; missing, mistyped, or inconsistent policy data leaves the default quarantine decision in force. The trusted rule compares the deduplicated independent-root count against trusted_min_independent_root_count; the raw reference count is only a lower bound on citations, never a substitute, so one external document cited three times still fails a policy that requires two independent roots.

# File: policy/kb_trust_tier.rego
package kb.trust_tier

import rego.v1

default decision := {"tier": "quarantined", "reasons": ["default_quarantine"]}

valid_policy_data if {
    is_number(data.policy.trusted_min_evidence_ref_count)
    data.policy.trusted_min_evidence_ref_count >= 1
    data.policy.trusted_min_evidence_ref_count == floor(data.policy.trusted_min_evidence_ref_count)
    is_number(data.policy.probation_min_evidence_ref_count)
    data.policy.probation_min_evidence_ref_count >= 1
    data.policy.probation_min_evidence_ref_count == floor(data.policy.probation_min_evidence_ref_count)
    data.policy.trusted_min_evidence_ref_count >= data.policy.probation_min_evidence_ref_count
    # Independent-root thresholds count deduplicated external or human roots,
    # never raw references: three citations of one document are one root.
    is_number(data.policy.trusted_min_independent_root_count)
    data.policy.trusted_min_independent_root_count >= 1
    data.policy.trusted_min_independent_root_count == floor(data.policy.trusted_min_independent_root_count)
    is_number(data.policy.probation_min_independent_root_count)
    data.policy.probation_min_independent_root_count >= 0
    data.policy.probation_min_independent_root_count == floor(data.policy.probation_min_independent_root_count)
    data.policy.trusted_min_independent_root_count >= data.policy.probation_min_independent_root_count
    is_number(data.policy.trusted_min_validator_score)
    is_number(data.policy.probation_min_validator_score)
    data.policy.probation_min_validator_score >= 0
    data.policy.trusted_min_validator_score <= 1
    data.policy.trusted_min_validator_score >= data.policy.probation_min_validator_score
}

valid_input if {
    is_string(input.identity)
    input.identity != ""
    is_number(input.validator_score)
    input.validator_score >= 0
    input.validator_score <= 1
    is_number(input.evidence_ref_count)
    input.evidence_ref_count >= 0
    input.evidence_ref_count == floor(input.evidence_ref_count)
    is_boolean(input.content_signature_verified)
    is_boolean(input.source_registry_trusted)
    is_array(input.high_risk_flags)
    input.origin_kind in {"external_source", "human_input", "model_output", "agent_summary", "agent_relay", "translation", "link_prefill"}
    is_number(input.independent_root_count)
    input.independent_root_count >= 0
    input.independent_root_count == floor(input.independent_root_count)
    is_boolean(input.self_referential_evidence)
}

derived_origin if input.origin_kind in {"model_output", "agent_summary", "agent_relay", "translation", "link_prefill"}

# Its own earlier output, summarized or relayed, is never independent evidence.
not_self_referential if input.self_referential_evidence == false

# Derived content needs at least one root that is not agent output even on
# probation; original content may reach probation on the policy minimum alone.
probation_grounded if {
    not derived_origin
    input.independent_root_count >= data.policy.probation_min_independent_root_count
}

probation_grounded if {
    derived_origin
    input.independent_root_count >= 1
    input.independent_root_count >= data.policy.probation_min_independent_root_count
}

decision := {"tier": "trusted", "reasons": ["signature_verified", "sufficient_independent_evidence", "high_validator_score"]} if {
    valid_policy_data
    valid_input
    not_self_referential
    input.content_signature_verified == true
    input.source_registry_trusted == true
    input.evidence_ref_count >= data.policy.trusted_min_evidence_ref_count
    input.independent_root_count >= data.policy.trusted_min_independent_root_count
    input.validator_score >= data.policy.trusted_min_validator_score
    count(input.high_risk_flags) == 0
}

decision := {"tier": "probation", "reasons": ["manual_promotion_required"]} if {
    valid_policy_data
    valid_input
    not_self_referential
    probation_grounded
    input.content_signature_verified == false
    input.source_registry_trusted == true
    input.evidence_ref_count >= data.policy.probation_min_evidence_ref_count
    input.validator_score >= data.policy.probation_min_validator_score
    count(input.high_risk_flags) == 0
}

Step 3: Derive policy input inside the trusted admission client

import math as _aidefend_math
import os as _aidefend_os

# Admission injects these required values from one signature-verified, versioned runtime profile.
def _aidefend_required(name):
    value = _aidefend_os.environ.get(name)
    if value is None or not value.strip():
        raise RuntimeError(f"required runtime-profile field is absent: {name}")
    return value.strip()


def _aidefend_positive_float(name):
    try:
        value = float(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not numeric: {name}") from error
    if not _aidefend_math.isfinite(value) or value <= 0:
        raise RuntimeError(f"runtime-profile field must be finite and positive: {name}")
    return value


def _aidefend_positive_int(name):
    value = _aidefend_required(name)
    if not value.isdecimal() or int(value) < 1:
        raise RuntimeError(f"runtime-profile field must be a positive integer: {name}")
    return int(value)


RUNTIME_PROFILE_VERSION = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_VERSION")
RUNTIME_PROFILE_SHA256 = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
if (len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")
        or RUNTIME_PROFILE_SHA256 == "0" * 64):
    raise RuntimeError("runtime-profile digest must be lowercase SHA-256")
TRUST_POLICY_REQUEST_TIMEOUT_SECONDS = _aidefend_positive_float("M002004_TRUST_POLICY_REQUEST_TIMEOUT_SECONDS")
TRUST_POLICY_MAX_RESPONSE_BYTES = _aidefend_positive_int("M002004_TRUST_POLICY_MAX_RESPONSE_BYTES")
if TRUST_POLICY_MAX_RESPONSE_BYTES < 2:
    raise RuntimeError("trust-policy response byte bound must permit a JSON object")

# File: kb_trust_contract/policy_client.py
import hashlib
import json
import stat
import subprocess
import tempfile
from pathlib import Path
from types import MappingProxyType
from urllib.parse import quote, urlsplit

import requests


class TrustPolicyUnavailable(RuntimeError):
    pass


class TrustFactsUnavailable(RuntimeError):
    pass


def _reject_duplicate_object_keys(pairs):
    result = {}
    for key, item in pairs:
        if key in result:
            raise ValueError(f"duplicate JSON key: {key}")
        result[key] = item
    return result


def _reject_nonfinite_json_constant(token):
    raise ValueError(f"non-finite JSON number: {token}")


VALID_TRUST_TIERS = frozenset({"trusted", "probation", "quarantined"})
INDEPENDENT_ORIGIN_KINDS = frozenset({"external_source", "human_input"})
DERIVED_ORIGIN_KINDS = frozenset({"model_output", "agent_summary", "agent_relay", "translation", "link_prefill"})
TRUST_POLICY_URL = _aidefend_required("M002004_TRUST_POLICY_URL")
TRUST_POLICY_CA_BUNDLE = _aidefend_required("M002004_TRUST_POLICY_CA_BUNDLE")
TRUST_POLICY_CLIENT_CERT = _aidefend_required("M002004_TRUST_POLICY_CLIENT_CERT")
TRUST_POLICY_CLIENT_KEY = _aidefend_required("M002004_TRUST_POLICY_CLIENT_KEY")
if TRUST_POLICY_URL != TRUST_POLICY_URL.strip():
    raise RuntimeError("trust-policy URL contains surrounding whitespace")
try:
    _trust_policy_endpoint = urlsplit(TRUST_POLICY_URL)
    _trust_policy_endpoint.port
except ValueError as error:
    raise RuntimeError("trust-policy URL is invalid") from error
if (
    _trust_policy_endpoint.scheme.lower() != "https"
    or not _trust_policy_endpoint.hostname
    or _trust_policy_endpoint.username is not None
    or _trust_policy_endpoint.password is not None
    or _trust_policy_endpoint.query
    or _trust_policy_endpoint.fragment
):
    raise RuntimeError("trust-policy URL must be authenticated HTTPS without embedded credentials")


_VERIFIED_FACTS_SEAL = object()


class VerifiedTrustFacts:
    __slots__ = ("_payload", "bindings")

    def __init__(self, payload: dict, bindings: dict, seal: object):
        if seal is not _VERIFIED_FACTS_SEAL:
            raise TypeError("VerifiedTrustFacts can only be created by local verification")
        self._payload = MappingProxyType(dict(payload))
        self.bindings = MappingProxyType(dict(bindings))

    def policy_input(self) -> dict:
        return dict(self._payload)


_CAPTURED_BYTES: dict[Path, bytes] = {}


def _regular_bytes(path: Path) -> bytes:
    descriptor = _aidefend_os.open(
        path,
        _aidefend_os.O_RDONLY
        | getattr(_aidefend_os, "O_BINARY", 0)
        | getattr(_aidefend_os, "O_NOFOLLOW", 0),
    )
    with _aidefend_os.fdopen(descriptor, "rb") as handle:
        status = _aidefend_os.fstat(handle.fileno())
        if not stat.S_ISREG(status.st_mode):
            raise TrustFactsUnavailable(f"signed input is not a regular file: {path}")
        value = handle.read()
    if not value:
        raise TrustFactsUnavailable(f"signed input is empty: {path}")
    return value


def _captured_bytes(path: Path) -> bytes:
    if path not in _CAPTURED_BYTES:
        _CAPTURED_BYTES[path] = _regular_bytes(path)
    return _CAPTURED_BYTES[path]


def _sha256(path: Path) -> str:
    return hashlib.sha256(_captured_bytes(path)).hexdigest()


def _resolve_ref(spool: Path, value: object) -> Path:
    if not isinstance(value, str) or not value or Path(value).is_absolute():
        raise TrustFactsUnavailable("artifact reference must be a non-empty relative path")
    try:
        resolved = (spool / value).resolve(strict=True)
    except (OSError, RuntimeError) as error:
        raise TrustFactsUnavailable("artifact reference is unavailable or unsafe") from error
    if not resolved.is_relative_to(spool) or not resolved.is_file() or resolved.is_symlink():
        raise TrustFactsUnavailable("artifact reference escapes the verified spool or is unsafe")
    return resolved


def _verify_blob(path: Path, signature: Path, key_env: str) -> None:
    payload = _regular_bytes(path)
    signature_bytes = _regular_bytes(signature)
    try:
        with tempfile.TemporaryDirectory(prefix="aidefend-trust-facts-") as stage_name:
            stage = Path(stage_name)
            _aidefend_os.chmod(stage, 0o700)
            payload_snapshot = stage / "payload"
            signature_snapshot = stage / "payload.sig"
            payload_snapshot.write_bytes(payload)
            signature_snapshot.write_bytes(signature_bytes)
            _aidefend_os.chmod(payload_snapshot, 0o400)
            _aidefend_os.chmod(signature_snapshot, 0o400)
            subprocess.run(
                [
                    "cosign", "verify-blob", "--key", _aidefend_required(key_env),
                    "--bundle", str(signature_snapshot), str(payload_snapshot),
                ],
                check=True, capture_output=True, text=True,
                timeout=TRUST_POLICY_REQUEST_TIMEOUT_SECONDS,
            )
            if (
                _regular_bytes(payload_snapshot) != payload
                or _regular_bytes(signature_snapshot) != signature_bytes
            ):
                raise TrustFactsUnavailable(
                    "private signed-input snapshot changed during verification"
                )
    except (OSError, subprocess.SubprocessError) as error:
        raise TrustFactsUnavailable(f"signature verification failed for {path.name}") from error
    _CAPTURED_BYTES[path] = payload


def _load_signed_json(
    spool: Path,
    request: dict,
    value_field: str,
    signature_field: str,
    key_env: str,
) -> tuple[Path, dict]:
    path = _resolve_ref(spool, request[value_field])
    signature = _resolve_ref(spool, request[signature_field])
    _verify_blob(path, signature, key_env)
    try:
        value = json.loads(
            _CAPTURED_BYTES[path].decode("utf-8", errors="strict"),
            object_pairs_hook=_reject_duplicate_object_keys,
            parse_constant=_reject_nonfinite_json_constant,
        )
    except (OSError, UnicodeError, ValueError) as error:
        raise TrustFactsUnavailable(f"signed JSON is unreadable: {path.name}") from error
    if not isinstance(value, dict):
        raise TrustFactsUnavailable(f"signed JSON is not an object: {path.name}")
    return path, value


def _read_bounded_json_response(response) -> object:
    try:
        status = response.status_code
        if (
            isinstance(status, bool)
            or not isinstance(status, int)
            or not 200 <= status < 300
        ):
            raise ValueError("HTTP response status is not successful")
        content_type = response.headers.get("Content-Type")
        if (
            not isinstance(content_type, str)
            or content_type.split(";", 1)[0].strip().lower() != "application/json"
        ):
            raise ValueError("HTTP response media type is not application/json")
        content_encoding = response.headers.get("Content-Encoding")
        if (
            content_encoding is not None
            and content_encoding.strip().lower() not in {"", "identity"}
        ):
            raise ValueError("HTTP response encoding is unsupported")
        length_header = response.headers.get("Content-Length")
        expected_length = None
        if length_header is not None:
            length_value = length_header.strip()
            if (
                not length_value
                or any(character not in "0123456789" for character in length_value)
            ):
                raise ValueError("HTTP Content-Length is invalid")
            expected_length = int(length_value)
            if expected_length > TRUST_POLICY_MAX_RESPONSE_BYTES:
                raise ValueError("HTTP response exceeds the runtime-profile limit")
        payload = bytearray()
        for chunk in response.iter_content(
            chunk_size=min(64 * 1024, TRUST_POLICY_MAX_RESPONSE_BYTES)
        ):
            if not isinstance(chunk, bytes):
                raise ValueError("HTTP response yielded non-byte content")
            payload.extend(chunk)
            if len(payload) > TRUST_POLICY_MAX_RESPONSE_BYTES:
                raise ValueError("HTTP response exceeds the runtime-profile limit")
        if expected_length is not None and len(payload) != expected_length:
            raise ValueError("HTTP response length differs from Content-Length")
        return json.loads(
            payload.decode("utf-8"),
            object_pairs_hook=_reject_duplicate_object_keys,
            parse_constant=_reject_nonfinite_json_constant,
        )
    finally:
        response.close()


def _source_registry_readback(source_id: str, writer_identity: str) -> tuple[bool, dict]:
    base = _aidefend_required("M002004_SOURCE_REGISTRY_URL")
    if base != base.strip():
        raise TrustFactsUnavailable("source-registry origin contains surrounding whitespace")
    try:
        endpoint = urlsplit(base)
        endpoint.port
    except ValueError as error:
        raise TrustFactsUnavailable("source-registry origin is invalid") from error
    if (
        endpoint.scheme.lower() != "https"
        or not endpoint.hostname
        or endpoint.username is not None
        or endpoint.password is not None
        or endpoint.path not in {"", "/"}
        or endpoint.query
        or endpoint.fragment
    ):
        raise TrustFactsUnavailable("source-registry URL must be authenticated HTTPS")
    try:
        response = requests.get(
            base.rstrip("/") + "/v1/sources/" + quote(source_id, safe=""),
            timeout=TRUST_POLICY_REQUEST_TIMEOUT_SECONDS,
            verify=_aidefend_required("M002004_SOURCE_REGISTRY_CA_BUNDLE"),
            cert=(
                _aidefend_required("M002004_SOURCE_REGISTRY_CLIENT_CERT"),
                _aidefend_required("M002004_SOURCE_REGISTRY_CLIENT_KEY"),
            ),
            headers={"Accept": "application/json", "Accept-Encoding": "identity"},
            allow_redirects=False,
            stream=True,
        )
        record = _read_bounded_json_response(response)
    except (requests.RequestException, TypeError, ValueError) as error:
        raise TrustFactsUnavailable("authoritative source-registry readback failed") from error
    expected = {"source_id", "status", "allowed_writer_identities", "registry_version"}
    if not isinstance(record, dict) or set(record) != expected:
        raise TrustFactsUnavailable("source-registry response schema differs")
    identities = record["allowed_writer_identities"]
    if (
        record["source_id"] != source_id
        or record["status"] not in {"trusted", "untrusted", "suspended"}
        or not isinstance(identities, list)
        or any(not isinstance(item, str) or not item for item in identities)
        or len(set(identities)) != len(identities)
        or not isinstance(record["registry_version"], str)
        or not record["registry_version"]
    ):
        raise TrustFactsUnavailable("source-registry record is invalid")
    return record["status"] == "trusted" and writer_identity in identities, record


def derive_verified_trust_facts(request: dict) -> VerifiedTrustFacts:
    required_request = {
        "candidate_ref", "candidate_signature_ref",
        "validator_receipt_ref", "validator_signature_ref",
        "evidence_manifest_ref", "evidence_signature_ref",
    }
    if not isinstance(request, dict) or set(request) != required_request:
        raise TrustFactsUnavailable("candidate request schema differs")
    spool = Path(_aidefend_required("M002004_VERIFIED_SPOOL_ROOT")).resolve(strict=True)
    if not spool.is_dir() or spool.is_symlink():
        raise TrustFactsUnavailable("verified spool root is unsafe")

    candidate_path, candidate = _load_signed_json(
        spool, request, "candidate_ref", "candidate_signature_ref",
        "M002004_CANDIDATE_VERIFY_KEY",
    )
    if set(candidate) != {"schema_version", "source_id", "writer_identity", "content", "origin"}:
        raise TrustFactsUnavailable("candidate schema differs")
    origin = candidate["origin"]
    if (
        candidate["schema_version"] != "aidefend.memory-write-candidate.v2"
        or not isinstance(candidate["source_id"], str) or not candidate["source_id"]
        or not isinstance(candidate["writer_identity"], str)
        or not candidate["writer_identity"].startswith("spiffe://")
        or not isinstance(candidate["content"], str) or not candidate["content"]
        or not isinstance(origin, dict)
        or set(origin) != {"kind", "producer_identity"}
        or origin["kind"] not in INDEPENDENT_ORIGIN_KINDS | DERIVED_ORIGIN_KINDS
        or not isinstance(origin["producer_identity"], str) or not origin["producer_identity"]
    ):
        raise TrustFactsUnavailable("candidate fields are invalid")
    candidate_sha256 = _sha256(candidate_path)

    validator_path, validator = _load_signed_json(
        spool, request, "validator_receipt_ref", "validator_signature_ref",
        "M002004_VALIDATOR_VERIFY_KEY",
    )
    if set(validator) != {
        "schema_version", "candidate_sha256", "validator_score",
        "high_risk_flags", "validator_version",
    }:
        raise TrustFactsUnavailable("validator receipt schema differs")
    score = validator["validator_score"]
    flags = validator["high_risk_flags"]
    if (
        validator["schema_version"] != "aidefend.memory-validator-result.v1"
        or validator["candidate_sha256"] != candidate_sha256
        or isinstance(score, bool) or not isinstance(score, (int, float))
        or not _aidefend_math.isfinite(score) or not 0 <= score <= 1
        or not isinstance(flags, list)
        or any(not isinstance(flag, str) or not flag for flag in flags)
        or len(set(flags)) != len(flags)
        or not isinstance(validator["validator_version"], str)
        or not validator["validator_version"]
    ):
        raise TrustFactsUnavailable("validator receipt is invalid or bound to different bytes")

    evidence_path, evidence = _load_signed_json(
        spool, request, "evidence_manifest_ref", "evidence_signature_ref",
        "M002004_EVIDENCE_VERIFY_KEY",
    )
    if set(evidence) != {"schema_version", "candidate_sha256", "evidence_refs"}:
        raise TrustFactsUnavailable("evidence manifest schema differs")
    refs = evidence["evidence_refs"]
    if (
        evidence["schema_version"] != "aidefend.memory-evidence-manifest.v1"
        or evidence["candidate_sha256"] != candidate_sha256
        or not isinstance(refs, list)
    ):
        raise TrustFactsUnavailable("evidence manifest is invalid or bound to different bytes")
    seen = set()
    roots: dict[str, str] = {}
    self_referential = False
    for ref in refs:
        if (
            not isinstance(ref, dict)
            or set(ref) != {"source_uri", "sha256", "root_kind", "root_id", "root_producer_identity"}
            or not isinstance(ref["source_uri"], str) or not ref["source_uri"]
            or not isinstance(ref["sha256"], str) or len(ref["sha256"]) != 64
            or set(ref["sha256"]) - set("0123456789abcdef")
            or ref["sha256"] == "0" * 64
            or ref["source_uri"] in seen
            or ref["root_kind"] not in INDEPENDENT_ORIGIN_KINDS | DERIVED_ORIGIN_KINDS
            or not isinstance(ref["root_id"], str) or not ref["root_id"]
            or not isinstance(ref["root_producer_identity"], str) or not ref["root_producer_identity"]
        ):
            raise TrustFactsUnavailable("evidence reference is invalid or duplicated")
        seen.add(ref["source_uri"])
        # A summary, translation, or relay of one root is still that root: count root_id once.
        if roots.setdefault(ref["root_id"], ref["root_kind"]) != ref["root_kind"]:
            raise TrustFactsUnavailable("evidence root declared with conflicting kinds")
        if ref["root_kind"] in DERIVED_ORIGIN_KINDS and ref["root_producer_identity"] == candidate["writer_identity"]:
            self_referential = True
    independent_root_count = sum(1 for kind in roots.values() if kind in INDEPENDENT_ORIGIN_KINDS)

    source_trusted, registry = _source_registry_readback(
        candidate["source_id"], candidate["writer_identity"]
    )
    payload = {
        "identity": candidate["writer_identity"],
        "validator_score": float(score),
        "evidence_ref_count": len(refs),
        "content_signature_verified": True,
        "source_registry_trusted": source_trusted,
        "high_risk_flags": sorted(flags),
        "origin_kind": origin["kind"],
        "independent_root_count": independent_root_count,
        "self_referential_evidence": self_referential,
    }
    bindings = {
        "candidate_sha256": candidate_sha256,
        "validator_receipt_sha256": _sha256(validator_path),
        "evidence_manifest_sha256": _sha256(evidence_path),
        "source_id": candidate["source_id"],
        "source_registry_version": registry["registry_version"],
        "runtime_profile_version": RUNTIME_PROFILE_VERSION,
        "runtime_profile_sha256": RUNTIME_PROFILE_SHA256,
        "max_json_response_bytes": TRUST_POLICY_MAX_RESPONSE_BYTES,
        "request_timeout_seconds": TRUST_POLICY_REQUEST_TIMEOUT_SECONDS,
        "policy_input_sha256": hashlib.sha256(
            json.dumps(payload, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")
        ).hexdigest(),
    }
    return VerifiedTrustFacts(payload, bindings, _VERIFIED_FACTS_SEAL)


def evaluate_trust_tier(facts: VerifiedTrustFacts) -> dict:
    if not isinstance(facts, VerifiedTrustFacts):
        raise TypeError("policy evaluation requires locally verified trust facts")
    try:
        response = requests.post(
            TRUST_POLICY_URL,
            json={"input": facts.policy_input()},
            timeout=TRUST_POLICY_REQUEST_TIMEOUT_SECONDS,
            verify=TRUST_POLICY_CA_BUNDLE,
            cert=(TRUST_POLICY_CLIENT_CERT, TRUST_POLICY_CLIENT_KEY),
            headers={"Accept": "application/json", "Accept-Encoding": "identity"},
            allow_redirects=False,
            stream=True,
        )
        body = _read_bounded_json_response(response)
    except requests.RequestException as error:
        raise TrustPolicyUnavailable("trust_policy_transport_or_http_failure") from error
    except (TypeError, ValueError) as error:
        raise TrustPolicyUnavailable("trust_policy_invalid_or_oversized_response") from error

    if not isinstance(body, dict):
        raise TrustPolicyUnavailable("trust_policy_invalid_response_envelope")
    result = body.get("result")
    if not isinstance(result, dict) or set(result) != {"tier", "reasons"}:
        raise TrustPolicyUnavailable("trust_policy_invalid_result_schema")
    tier = result["tier"]
    reasons = result["reasons"]
    if not isinstance(tier, str) or tier not in VALID_TRUST_TIERS:
        raise TrustPolicyUnavailable("trust_policy_invalid_tier")
    if (
        not isinstance(reasons, list)
        or not reasons
        or any(not isinstance(reason, str) or not reason.strip() for reason in reasons)
    ):
        raise TrustPolicyUnavailable("trust_policy_invalid_reasons")
    normalized_reasons = [reason.strip() for reason in reasons]
    if len(set(normalized_reasons)) != len(normalized_reasons):
        raise TrustPolicyUnavailable("trust_policy_duplicate_reasons")
    return {
        "tier": tier,
        "reasons": normalized_reasons,
        "bindings": dict(facts.bindings),
    }

Action: Expose only the reference-based handler that calls derive_verified_trust_facts and then evaluate_trust_tier; never expose the OPA input shape as a caller-writable API. Persist the tier, reasons, exact byte digests, authoritative registry version, runtime-profile identity, origin kind, independent-root count, and policy-input digest with the write-eligibility record. Test the loop explicitly: an agent writing a summary of its own earlier memory, two agents relaying one claim to each other, and one external document cited under three URLs must each resolve to zero or one independent root and never to trusted. Signature, registry, validator, evidence, transport, mTLS, or result-schema failure yields INSUFFICIENT_DATA for trusted promotion—never trusted—and AID-H-029.002 owns the runtime fail-closed decision.

External authorship survives user submission

The controlled ingress in AID-H-002.002-G009 records link_prefill in the signed origin and ancestry records. It is included among non-independent/derived origins and never counts as a human_input evidence root merely because a user clicked Submit, edited a draft, or relayed it through an agent. Preserve the original external root through revisions; an independently obtained supporting source is a separate evidence root. In this contract, human_input means an explicitly attested original human contribution recorded by the trusted collector, not every message with role=user. Unknown authorship remains non-independent. A valid independent evidence root supports a factual claim; it does not authorize changes to source trust, future behavior, or tool permissions. The memory_write gate in AID-H-029.002 applies that separate authority check.

AID-M-002.004-G003

Concept:

Tier assignment needs a storage contract that downstream systems can enforce consistently. This sub-technique defines the namespace, metadata, and promotion-state contract; runtime query filtering and entitlement checks are owned by Harden controls such as AID-H-020 and AID-H-029.002.

Step 1: Persist records with tier-specific namespace metadata

# File: kb_trust_contract/persistence.py
import os
from pinecone import Pinecone

pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
index = pc.Index(host=os.environ["PINECONE_INDEX_HOST"])


def upsert_memory_record(write_request: dict, decision: dict) -> None:
    metadata = {
        "trust_tier": decision["tier"],
        "decision_reasons": ",".join(decision["reasons"]),
        "source_system": write_request["source_system"],
        "evidence_refs": ",".join(
            evidence["source_uri"] for evidence in write_request["evidence_refs"]
        ),
    }
    index.upsert(
        namespace=decision["tier"],
        vectors=[
            {
                "id": write_request["document_id"],
                "values": write_request["embedding"],
                "metadata": metadata,
            }
        ],
    )

Step 2: Publish a retrieval eligibility manifest for Harden controls

# File: kb_trust_contract/retrieval_eligibility.yaml
schema_version: aidefend.retrieval_eligibility.v1
default_retrieval_scope:
  include_namespaces:
    - trusted
  exclude_namespaces:
    - quarantined
probation_retrieval:
  requires:
    - purpose_authorization
    - fresh_reverification
    - request_trace_link
owner_controls:
  runtime_stage_gate: AID-H-029.002
  rag_retrieval_authorization: AID-H-020.003
  rag_chunk_integrity: AID-H-020.001

Step 3: Define promotion and quarantine handling as explicit operational rules

# File: kb_trust_contract/trust_tier_rules.yaml
trusted:
  readable_by_default: true
probation:
  readable_by_default: false
  promotion_requires:
    - second_evidence_ref
    - analyst_review
quarantined:
  readable_by_default: false
  retention_days: 30

Action: Keep namespace names, promotion requirements, and retrieval eligibility in versioned metadata. Application code should not infer these rules from naming conventions; it should consume this contract through the Harden controls that own runtime enforcement.

AID-M-002.004-G004

Concept:

A write candidate can have valid identity and evidence yet still contradict already trusted knowledge. Before a candidate is eligible for trusted promotion, compare it against the existing trusted fact set for the same entity and predicate. Contradictions should produce promotion evidence that downstream controls can use for probation, quarantine, or human review.

Step 1: Normalize claims into comparable keys

# File: kb_trust_contract/claims.py
from pydantic import BaseModel, ConfigDict, Field


class Claim(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)
    entity_id: str = Field(min_length=1)
    predicate: str = Field(min_length=1)
    value: str = Field(min_length=1)
    confidence: float = Field(ge=0, le=1, allow_inf_nan=False)
    evidence_refs: list[str] = Field(min_length=1)

Step 2: Return promotion evidence when a contradictory fact already exists

from __future__ import annotations

# File: kb_trust_contract/contradictions.py

from claims import Claim


def contradiction_key(claim: Claim) -> tuple[str, str]:
    return claim.entity_id, claim.predicate


def detect_contradiction(candidate: Claim, trusted_claims: list[Claim]) -> Claim | None:
    for trusted in trusted_claims:
        if contradiction_key(trusted) != contradiction_key(candidate):
            continue
        if trusted.value != candidate.value:
            return trusted
    return None

Step 3: Resolve evidence to independent roots and refuse circular evidence

A contradiction check only compares a candidate with trusted facts. It cannot see that the candidate's "evidence" is the agent's own earlier output, a summary of it, or a relay through a second agent. Resolve every evidence reference through the verified ancestry map (the provenance records from AID-M-002.004-G005) to its root, count each root once, and treat a root produced by a model or agent as derived. A candidate whose roots include the writer's own derived output is self-referential; a candidate with no external or human root has nothing independent behind it. Both are promotion evidence for probation, never trusted eligibility.

# File: kb_trust_contract/evidence_roots.py
from __future__ import annotations

INDEPENDENT_ORIGIN_KINDS = frozenset({"external_source", "human_input"})
DERIVED_ORIGIN_KINDS = frozenset({"model_output", "agent_summary", "agent_relay", "translation", "link_prefill"})
ANCESTRY_FIELDS = frozenset({"origin_kind", "root_id", "producer_identity", "parent_refs"})
# Bound the traversal so a hostile or runaway ancestry map fails closed
# instead of exhausting the process; take the value from signed policy.
MAX_ANCESTRY_RECORDS = 4096


def _validated_record(ref: str, ancestry: dict[str, dict]) -> dict:
    record = ancestry.get(ref)
    if not isinstance(record, dict) or set(record) != ANCESTRY_FIELDS:
        raise ValueError(f"evidence ancestry is missing or malformed for {ref}")
    if record["origin_kind"] not in INDEPENDENT_ORIGIN_KINDS | DERIVED_ORIGIN_KINDS:
        raise ValueError(f"unknown origin kind for {ref}")
    parents = record["parent_refs"]
    if not isinstance(parents, list) or any(not isinstance(p, str) or not p for p in parents):
        raise ValueError(f"invalid parent references for {ref}")
    if len(parents) != len(set(parents)):
        raise ValueError(f"duplicate parent references for {ref}")
    return record


def resolve_evidence_roots(evidence_refs: list[str], ancestry: dict[str, dict]) -> list[dict]:
    """Follow every evidence reference through the verified ancestry map to its roots.

    A summary, translation, re-send, or cross-agent relay of one source is still
    that one source, and two paths that reach the same root (a diamond) are one
    root. The walk is iterative with an explicit stack: a record is "active"
    while its parents are being resolved and "done" afterwards. Reaching an
    active record again is a cycle, whether it has one parent or several.
    Missing ancestry is an error, never an independent root."""
    roots: dict[str, dict] = {}
    state: dict[str, str] = {}
    for ref in evidence_refs:
        stack = [ref]
        while stack:
            current = stack[-1]
            if state.get(current) == "done":
                stack.pop()
                continue
            record = _validated_record(current, ancestry)
            parents = record["parent_refs"]
            if state.get(current) != "active":
                if len(state) >= MAX_ANCESTRY_RECORDS:
                    raise ValueError("evidence ancestry exceeds the bounded record budget")
                state[current] = "active"
                if not parents:
                    root_id = record["root_id"]
                    if not isinstance(root_id, str) or not root_id:
                        raise ValueError(f"root without identity at {current}")
                    roots.setdefault(root_id, {
                        "root_id": root_id,
                        "origin_kind": record["origin_kind"],
                        "producer_identity": record["producer_identity"],
                    })
                    state[current] = "done"
                    stack.pop()
                    continue
            for parent in parents:
                if state.get(parent) == "active":
                    raise ValueError(f"evidence ancestry cycle at {parent}")
            pending = [parent for parent in parents if state.get(parent) != "done"]
            if pending:
                # Resolve every parent first; a synthesized summary cannot hide
                # its inputs because each parent is walked on its own.
                stack.extend(pending)
                continue
            state[current] = "done"
            stack.pop()
    return sorted(roots.values(), key=lambda item: item["root_id"])


def circular_evidence_finding(candidate_writer: str, roots: list[dict]) -> str | None:
    """Return the reason a candidate cannot be promoted on this evidence, or None."""
    independent = [r for r in roots if r["origin_kind"] in INDEPENDENT_ORIGIN_KINDS]
    if any(r["producer_identity"] == candidate_writer for r in roots if r["origin_kind"] in DERIVED_ORIGIN_KINDS):
        return "self_referential_evidence"
    if not independent:
        return "no_independent_evidence_root"
    return None

The traversal is iterative with an explicit stack and a record budget taken from policy: two paths to one root (a diamond) count once, a record reached again while it is still being resolved is a cycle whatever its parent count, and an ancestry map larger than the budget fails closed instead of exhausting the process.

# File: tests/test_evidence_roots.py
import unittest
from evidence_roots import circular_evidence_finding, resolve_evidence_roots


def record(kind, root_id, producer, parents):
    return {"origin_kind": kind, "root_id": root_id, "producer_identity": producer, "parent_refs": parents}


class EvidenceRootTests(unittest.TestCase):
    def test_relays_and_summaries_of_one_source_are_one_root(self):
        ancestry = {
            "doc": record("external_source", "doc-1", "crawler", []),
            "summary": record("agent_summary", "", "spiffe://agent-a", ["doc"]),
            "relay": record("agent_relay", "", "spiffe://agent-b", ["summary"]),
            "translation": record("translation", "", "spiffe://agent-c", ["relay"]),
        }
        roots = resolve_evidence_roots(["summary", "relay", "translation"], ancestry)
        self.assertEqual([r["root_id"] for r in roots], ["doc-1"])
        self.assertIsNone(circular_evidence_finding("spiffe://agent-z", roots))

    def test_diamond_counts_the_shared_root_once(self):
        ancestry = {
            "doc": record("external_source", "doc-1", "crawler", []),
            "left": record("agent_summary", "", "spiffe://agent-a", ["doc"]),
            "right": record("translation", "", "spiffe://agent-b", ["doc"]),
            "merged": record("agent_summary", "", "spiffe://agent-c", ["left", "right"]),
        }
        self.assertEqual([r["root_id"] for r in resolve_evidence_roots(["merged"], ancestry)], ["doc-1"])

    def test_cycles_fail_closed_with_one_or_several_parents(self):
        single = {
            "a": record("agent_summary", "", "spiffe://agent-a", ["b"]),
            "b": record("agent_relay", "", "spiffe://agent-b", ["a"]),
        }
        multi = {
            "a": record("agent_summary", "", "spiffe://agent-a", ["b", "c"]),
            "b": record("agent_relay", "", "spiffe://agent-b", ["a"]),
            "c": record("external_source", "doc-c", "crawler", []),
        }
        for ancestry in (single, multi):
            with self.assertRaisesRegex(ValueError, "cycle"):
                resolve_evidence_roots(["a"], ancestry)

    def test_self_referential_and_missing_roots_are_findings(self):
        ancestry = {
            "own": record("model_output", "mem-7", "spiffe://agent-a", []),
            "doc": record("external_source", "doc-1", "crawler", []),
        }
        roots = resolve_evidence_roots(["own", "doc"], ancestry)
        self.assertEqual(circular_evidence_finding("spiffe://agent-a", roots), "self_referential_evidence")
        only_own = resolve_evidence_roots(["own"], ancestry)
        self.assertEqual(circular_evidence_finding("spiffe://agent-b", only_own), "no_independent_evidence_root")
        with self.assertRaisesRegex(ValueError, "missing or malformed"):
            resolve_evidence_roots(["unknown"], ancestry)

    def test_record_budget_is_enforced(self):
        chain = {f"n{i}": record("agent_relay", "", "spiffe://agent-a", [f"n{i + 1}"]) for i in range(5000)}
        chain["n5000"] = record("external_source", "doc-1", "crawler", [])
        with self.assertRaisesRegex(ValueError, "budget"):
            resolve_evidence_roots(["n0"], chain)

Step 4: Produce a promotion decision record instead of overwriting trusted state

# File: kb_trust_contract/promotion.py
from claims import Claim
from contradictions import detect_contradiction
from evidence_roots import circular_evidence_finding, resolve_evidence_roots
import hashlib
import json


def build_promotion_evidence(
    candidate: Claim, trusted_claims: list[Claim], *, trusted_version: str,
    expected_trusted_sha256: str, population_complete: bool,
    candidate_writer: str, evidence_ancestry: dict,
) -> dict:
    # Call only after authenticating the immutable trusted-fact manifest and the
    # signed ancestry records; a missing ancestry entry raises, it never passes.
    if not trusted_version or population_complete is not True:
        raise ValueError("trusted fact version and complete population are required")
    if not isinstance(candidate_writer, str) or not candidate_writer:
        raise ValueError("candidate writer identity is required")
    canonical = lambda value: json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False).encode("utf-8")
    candidate_record = candidate.model_dump(mode="json")
    trusted_records = [claim.model_dump(mode="json") for claim in trusted_claims]
    for record in [candidate_record, *trusted_records]:
        if any(not ref.strip() for ref in record["evidence_refs"]):
            raise ValueError("evidence references must be nonempty")
    trusted_sha256 = hashlib.sha256(canonical(trusted_records)).hexdigest()
    if trusted_sha256 != expected_trusted_sha256:
        raise ValueError("trusted fact population differs from verified manifest")
    conflict = detect_contradiction(candidate, trusted_claims)
    roots = resolve_evidence_roots(candidate.evidence_refs, evidence_ancestry)
    root_finding = circular_evidence_finding(candidate_writer, roots)
    if conflict:
        reason = "contradicts_trusted_fact"
    elif root_finding:
        reason = root_finding
    else:
        reason = "no_conflicting_trusted_fact"
    return {
        "schema_version": "aidefend.kb-promotion-evidence.v2",
        "promotion_state": "probation_required" if (conflict or root_finding) else "trusted_eligible",
        "reason": reason,
        "evidence_roots": roots,
        "independent_root_count": sum(1 for r in roots if r["origin_kind"] in {"external_source", "human_input"}),
        "candidate": candidate_record,
        "candidate_sha256": hashlib.sha256(canonical(candidate_record)).hexdigest(),
        "conflicting_trusted_fact": conflict.model_dump(mode="json") if conflict else None,
        "trusted_version": trusted_version,
        "trusted_population_sha256": trusted_sha256,
        "population_complete": True,
    }

Action: Do not let new writes silently replace contradictory trusted facts. Preserve both the candidate and the conflicting trusted fact in the promotion evidence so analysts and Harden controls can see exactly what changed and why trusted promotion was not eligible.

Pass the version and digest from the authenticated immutable fact-set manifest, covering the complete applicable entity/predicate population in its signed order. Store this evidence with the proposed write and its references; Harden promotion enforcement consumes it independently. A missing fact population or digest mismatch is ERROR/INSUFFICIENT_DATA, not trusted eligibility. A legitimate changed fact can require review; this contradiction signal does not prove attacker intent. Keep the ancestry map authoritative: it is read from signed provenance records, never from the candidate payload, so a writer cannot invent an external root for its own output.

Include link-prefill-only ancestry and a relay of that ancestry in the promotion regression suite: both must have zero independent roots. A submit-confirmation receipt is authority to submit exact text, not independent corroboration that its assertions are true. A copied source and its summary are one root, including when a person pasted or confirmed them.

AID-M-002.004-G005

Concept:

Once a write candidate is evaluated or promoted, its provenance record becomes part of the control evidence for later audits, investigations, and restore decisions. The record should be append-only and should capture the workload identity, evidence references, validator output, trust-tier recommendation, promotion state, and final namespace metadata.

Step 1: Build a provenance event payload at the trust-contract layer

# File: kb_trust_contract/provenance_event.json
{
  "document_id": "cust-4421-status",
  "writer_identity": "spiffe://corp.internal/agents/rag-ingestor",
  "recommended_trust_tier": "trusted",
  "promotion_state": "trusted_eligible",
  "validator_score": 0.93,
  "evidence_refs": [
    "s3://evidence/customer-4421/ticket-19.pdf"
  ],
  "policy_version": "kb-write-v7",
  "recorded_at": "2026-04-08T21:12:00Z"
}

Step 2: Append the record to an immutable provenance service

from __future__ import annotations

import math as _aidefend_math
import os as _aidefend_os

# Admission injects these required values from one signature-verified, versioned runtime profile.
def _aidefend_required(name):
    value = _aidefend_os.environ.get(name)
    if value is None or not value.strip():
        raise RuntimeError(f"required runtime-profile field is absent: {name}")
    return value.strip()


def _aidefend_positive_float(name):
    try:
        value = float(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not numeric: {name}") from error
    if not _aidefend_math.isfinite(value) or value <= 0:
        raise RuntimeError(f"runtime-profile field must be finite and positive: {name}")
    return value


def _aidefend_positive_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 1:
        raise RuntimeError(f"runtime-profile field must be positive: {name}")
    return value


def _aidefend_nonnegative_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 0:
        raise RuntimeError(f"runtime-profile field cannot be negative: {name}")
    return value


def _aidefend_fraction(name):
    value = _aidefend_positive_float(name)
    if value > 1:
        raise RuntimeError(f"runtime-profile field must be in (0, 1]: {name}")
    return value


RUNTIME_PROFILE_VERSION = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_VERSION")
RUNTIME_PROFILE_SHA256 = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
if (len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")):
    raise RuntimeError("runtime-profile digest must be lowercase SHA-256")
PROVENANCE_APPEND_TIMEOUT_SECONDS = _aidefend_positive_float("M002004_PROVENANCE_APPEND_TIMEOUT_SECONDS")
PROVENANCE_APPEND_MAX_RESPONSE_BYTES = _aidefend_positive_int("M002004_PROVENANCE_APPEND_MAX_RESPONSE_BYTES")

# File: kb_trust_contract/provenance_client.py
import hashlib
import json
from urllib.parse import urlsplit

import requests


def _reject_duplicate_keys(pairs):
    result = {}
    for key, item in pairs:
        if key in result:
            raise ValueError(f"duplicate JSON key: {key}")
        result[key] = item
    return result


def _reject_nonfinite(token):
    raise ValueError(f"non-finite JSON number: {token}")


def _https_origin() -> str:
    raw = _aidefend_required("M002004_PROVENANCE_ORIGIN")
    try:
        parsed = urlsplit(raw)
        parsed.port
    except ValueError as error:
        raise RuntimeError("provenance append origin is invalid") from error
    if (
        parsed.scheme.lower() != "https"
        or not parsed.hostname
        or parsed.username is not None
        or parsed.password is not None
        or parsed.path not in {"", "/"}
        or parsed.query
        or parsed.fragment
    ):
        raise RuntimeError("provenance append URL must be an exact HTTPS origin")
    return raw.rstrip("/")


def _read_bounded_json(response) -> object:
    try:
        status = response.status_code
        if isinstance(status, bool) or not isinstance(status, int) or not 200 <= status < 300:
            raise ValueError("HTTP status is not successful")
        content_type = response.headers.get("Content-Type")
        if (
            not isinstance(content_type, str)
            or content_type.split(";", 1)[0].strip().lower() != "application/json"
        ):
            raise ValueError("provenance append response media type is not application/json")
        encoding = response.headers.get("Content-Encoding")
        if encoding is not None and encoding.strip().lower() not in {"", "identity"}:
            raise ValueError("HTTP response encoding is unsupported")
        declared = response.headers.get("Content-Length")
        expected = None
        if declared is not None:
            text = declared.strip()
            if not text or any(character not in "0123456789" for character in text):
                raise ValueError("HTTP Content-Length is invalid")
            expected = int(text)
            if expected > PROVENANCE_APPEND_MAX_RESPONSE_BYTES:
                raise ValueError("HTTP response exceeds the signed byte bound")
        body = bytearray()
        for chunk in response.iter_content(chunk_size=min(64 * 1024, PROVENANCE_APPEND_MAX_RESPONSE_BYTES)):
            if not isinstance(chunk, bytes):
                raise ValueError("HTTP response yielded non-byte content")
            body.extend(chunk)
            if len(body) > PROVENANCE_APPEND_MAX_RESPONSE_BYTES:
                raise ValueError("HTTP response exceeds the signed byte bound")
        if expected is not None and len(body) != expected:
            raise ValueError("HTTP response length differs from Content-Length")
        return json.loads(
            body.decode("utf-8"),
            object_pairs_hook=_reject_duplicate_keys,
            parse_constant=_reject_nonfinite,
        )
    finally:
        response.close()


PROVENANCE_ORIGIN = _https_origin()
PROVENANCE_CA = _aidefend_required("M002004_PROVENANCE_CA_BUNDLE")
PROVENANCE_CERT = _aidefend_required("M002004_PROVENANCE_CLIENT_CERT")
PROVENANCE_KEY = _aidefend_required("M002004_PROVENANCE_CLIENT_KEY")


def append_provenance(event: dict) -> dict:
    canonical = json.dumps(
        event, sort_keys=True, separators=(",", ":"), allow_nan=False
    ).encode("utf-8")
    payload_sha256 = hashlib.sha256(canonical).hexdigest()
    try:
        with requests.Session() as session:
            with session.post(
                PROVENANCE_ORIGIN + "/v1/append",
                data=canonical,
                timeout=PROVENANCE_APPEND_TIMEOUT_SECONDS,
                verify=PROVENANCE_CA,
                cert=(PROVENANCE_CERT, PROVENANCE_KEY),
                headers={
                    "Content-Type": "application/json",
                    "Accept": "application/json",
                    "Accept-Encoding": "identity",
                },
                allow_redirects=False,
                stream=True,
            ) as response:
                receipt = _read_bounded_json(response)
    except (requests.RequestException, TypeError, ValueError) as error:
        raise RuntimeError("provenance append write failed closed") from error
    required = {"schema_version", "record_id", "event_sha256", "provenance_version"}
    if (
        not isinstance(receipt, dict)
        or set(receipt) != required
        or receipt["schema_version"] != "aidefend.provenance-append-receipt.v1"
        or not isinstance(receipt["record_id"], str)
        or not receipt["record_id"]
        or receipt["event_sha256"] != payload_sha256
        or not isinstance(receipt["provenance_version"], str)
        or not receipt["provenance_version"]
    ):
        raise RuntimeError("provenance append receipt is invalid or misbound")
    return receipt

Action: Write provenance records before trusted promotion is considered complete. The write candidate and its provenance must share the same stable document_id so later evidence collection can prove who introduced the fact, which policy recommended the trust tier, and which downstream Harden decision consumed the evidence.

AID-M-002.004-G006

Concept:

When verification failures spike for a tenant, source connector, or workload identity, the Model layer should publish precise source trust-state evidence instead of silently enforcing its own circuit breaker. Harden controls such as AID-H-029.002 can then decide whether to fail closed, route to quarantine, require step-up review, or disable trusted promotion.

Step 1: Track failure counts per source and tenant as evidence

from __future__ import annotations

# File: kb_trust_contract/source_trust_signal.py

from dataclasses import dataclass

import redis

r = redis.Redis(host="redis", port=6379, decode_responses=True)


@dataclass(frozen=True)
class TrustSignalPolicy:
    window_seconds: int
    review_threshold: int


def record_verification_failure(
    tenant_id: str,
    source_system: str,
    policy: TrustSignalPolicy,
) -> dict:
    key = f"kb-trust-failures:{tenant_id}:{source_system}"
    count = r.incr(key)
    if count == 1:
        r.expire(key, policy.window_seconds)
    return {
        "schema_version": "aidefend.source_trust_signal.v1",
        "tenant_id": tenant_id,
        "source_system": source_system,
        "failure_count": count,
        "window_seconds": policy.window_seconds,
        "state": (
            "degraded_review_required"
            if count >= policy.review_threshold
            else "normal"
        ),
    }

Step 2: Publish the finding for Harden enforcement

import math as _aidefend_math
import os as _aidefend_os

# Admission injects these required values from one signature-verified, versioned runtime profile.
def _aidefend_required(name):
    value = _aidefend_os.environ.get(name)
    if value is None or not value.strip():
        raise RuntimeError(f"required runtime-profile field is absent: {name}")
    return value.strip()


def _aidefend_positive_float(name):
    try:
        value = float(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not numeric: {name}") from error
    if not _aidefend_math.isfinite(value) or value <= 0:
        raise RuntimeError(f"runtime-profile field must be finite and positive: {name}")
    return value


def _aidefend_positive_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 1:
        raise RuntimeError(f"runtime-profile field must be positive: {name}")
    return value


def _aidefend_nonnegative_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 0:
        raise RuntimeError(f"runtime-profile field cannot be negative: {name}")
    return value


def _aidefend_fraction(name):
    value = _aidefend_positive_float(name)
    if value > 1:
        raise RuntimeError(f"runtime-profile field must be in (0, 1]: {name}")
    return value


RUNTIME_PROFILE_VERSION = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_VERSION")
RUNTIME_PROFILE_SHA256 = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
if (len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")):
    raise RuntimeError("runtime-profile digest must be lowercase SHA-256")
TRUST_SIGNAL_REQUEST_TIMEOUT_SECONDS = _aidefend_positive_float("M002004_TRUST_SIGNAL_REQUEST_TIMEOUT_SECONDS")
TRUST_SIGNAL_MAX_RESPONSE_BYTES = _aidefend_positive_int("M002004_TRUST_SIGNAL_MAX_RESPONSE_BYTES")

# File: kb_trust_contract/publish_signal.py
import hashlib
import json
from urllib.parse import urlsplit

import requests


def _reject_duplicate_keys(pairs):
    result = {}
    for key, item in pairs:
        if key in result:
            raise ValueError(f"duplicate JSON key: {key}")
        result[key] = item
    return result


def _reject_nonfinite(token):
    raise ValueError(f"non-finite JSON number: {token}")


def _https_origin() -> str:
    raw = _aidefend_required("M002004_PROVENANCE_ORIGIN")
    try:
        parsed = urlsplit(raw)
        parsed.port
    except ValueError as error:
        raise RuntimeError("source-trust signal origin is invalid") from error
    if (
        parsed.scheme.lower() != "https"
        or not parsed.hostname
        or parsed.username is not None
        or parsed.password is not None
        or parsed.path not in {"", "/"}
        or parsed.query
        or parsed.fragment
    ):
        raise RuntimeError("source-trust signal URL must be an exact HTTPS origin")
    return raw.rstrip("/")


def _read_bounded_json(response) -> object:
    try:
        status = response.status_code
        if isinstance(status, bool) or not isinstance(status, int) or not 200 <= status < 300:
            raise ValueError("HTTP status is not successful")
        content_type = response.headers.get("Content-Type")
        if (
            not isinstance(content_type, str)
            or content_type.split(";", 1)[0].strip().lower() != "application/json"
        ):
            raise ValueError("source-trust signal response media type is not application/json")
        encoding = response.headers.get("Content-Encoding")
        if encoding is not None and encoding.strip().lower() not in {"", "identity"}:
            raise ValueError("HTTP response encoding is unsupported")
        declared = response.headers.get("Content-Length")
        expected = None
        if declared is not None:
            text = declared.strip()
            if not text or any(character not in "0123456789" for character in text):
                raise ValueError("HTTP Content-Length is invalid")
            expected = int(text)
            if expected > TRUST_SIGNAL_MAX_RESPONSE_BYTES:
                raise ValueError("HTTP response exceeds the signed byte bound")
        body = bytearray()
        for chunk in response.iter_content(chunk_size=min(64 * 1024, TRUST_SIGNAL_MAX_RESPONSE_BYTES)):
            if not isinstance(chunk, bytes):
                raise ValueError("HTTP response yielded non-byte content")
            body.extend(chunk)
            if len(body) > TRUST_SIGNAL_MAX_RESPONSE_BYTES:
                raise ValueError("HTTP response exceeds the signed byte bound")
        if expected is not None and len(body) != expected:
            raise ValueError("HTTP response length differs from Content-Length")
        return json.loads(
            body.decode("utf-8"),
            object_pairs_hook=_reject_duplicate_keys,
            parse_constant=_reject_nonfinite,
        )
    finally:
        response.close()


PROVENANCE_ORIGIN = _https_origin()
PROVENANCE_CA = _aidefend_required("M002004_PROVENANCE_CA_BUNDLE")
PROVENANCE_CERT = _aidefend_required("M002004_PROVENANCE_CLIENT_CERT")
PROVENANCE_KEY = _aidefend_required("M002004_PROVENANCE_CLIENT_KEY")


def publish_source_trust_signal(signal: dict) -> dict:
    canonical = json.dumps(
        signal, sort_keys=True, separators=(",", ":"), allow_nan=False
    ).encode("utf-8")
    payload_sha256 = hashlib.sha256(canonical).hexdigest()
    try:
        with requests.Session() as session:
            with session.post(
                PROVENANCE_ORIGIN + "/v1/source-trust-signals",
                data=canonical,
                timeout=TRUST_SIGNAL_REQUEST_TIMEOUT_SECONDS,
                verify=PROVENANCE_CA,
                cert=(PROVENANCE_CERT, PROVENANCE_KEY),
                headers={
                    "Content-Type": "application/json",
                    "Accept": "application/json",
                    "Accept-Encoding": "identity",
                },
                allow_redirects=False,
                stream=True,
            ) as response:
                receipt = _read_bounded_json(response)
    except (requests.RequestException, TypeError, ValueError) as error:
        raise RuntimeError("source-trust signal write failed closed") from error
    required = {"schema_version", "signal_id", "signal_sha256", "provenance_version"}
    if (
        not isinstance(receipt, dict)
        or set(receipt) != required
        or receipt["schema_version"] != "aidefend.source-trust-signal-receipt.v1"
        or not isinstance(receipt["signal_id"], str)
        or not receipt["signal_id"]
        or receipt["signal_sha256"] != payload_sha256
        or not isinstance(receipt["provenance_version"], str)
        or not receipt["provenance_version"]
    ):
        raise RuntimeError("source-trust signal receipt is invalid or misbound")
    return receipt

Action: Persist the source trust-state finding with the policy version and affected tenant/source key. The runtime action — circuit breaking, fail-closed memory write admission, quarantine-only routing, or step-up review — is performed by AID-H-029.002 or the relevant Harden/RAG enforcement control, not by this Model provenance control.

Open-source tools

• SPIRE
• Envoy
• Open Policy Agent (OPA)
• Kyverno
• Milvus
• Weaviate
• Temporal
• Cosign

Commercial / hosted tools

• Pinecone
• Databricks Unity Catalog
• Temporal Cloud
• ServiceNow IT Service Management

Threat mappings

MITRE ATLAS

• AML.T0070 RAG Poisoning
• AML.T0071 False RAG Entry Injection
• AML.T0080 AI Agent Context Poisoning
• AML.T0080.000 AI Agent Context Poisoning: Memory
• AML.T0131 Crafted AI Assistant Links (link-prefill ancestry cannot count as an independent human evidence root for memory eligibility)

MAESTRO

• Data Tampering (L2)
• Compromised RAG Pipelines (L2)

OWASP LLM Top 10 2026

• LLM05:2026 Data and Model Poisoning (the trust-tier contract supplies provenance and contradiction evidence to downstream write admission; it does not itself enforce the write decision)

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI06:2026 Memory & Context Poisoning

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-6.1.1 Knowledge Base Poisoning (version-bound candidate/trusted-fact contradiction evidence flags runtime knowledge poisoning)

Google Secure AI Framework 2.0 - Risks

• N/A

Databricks AI Security Framework 3.0

• Agents - Core 13.1: Memory Poisoning