Skip to content

Topology C (DL+PCI)

Proof-Carrying Intents (PCI)

dir_core.pci

Proof-Carrying Intent (PCI) utilities for Topology C (DL+PCI).

Evidence Hash computation and ProofChecker per Technical Annex.

ProofChecker

Generic Proof Checker for PCI verification (Topology C §4.3).

Recomputes evidence_hash using authoritative sources. Mismatch = reject. Business-rule checks remain the responsibility of the caller/DIM.

Source code in src/dir_core/pci.py
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
class ProofChecker:
    """
    Generic Proof Checker for PCI verification (Topology C §4.3).

    Recomputes evidence_hash using authoritative sources. Mismatch = reject.
    Business-rule checks remain the responsibility of the caller/DIM.
    """

    def verify(
        self,
        pci: ProofCarryingIntent,
        get_context_hash: Callable[[], str],
        get_contract_hash: Callable[[], str],
        get_proposal_params: Callable[[Dict[str, Any]], str],
    ) -> Tuple[bool, str]:
        """
        Verify PCI evidence_hash against authoritative sources.

        Args:
            pci: The Proof-Carrying Intent to verify.
            get_context_hash: Callable returning current context hash.
            get_contract_hash: Callable returning contract hash.
            get_proposal_params: Callable(intent_payload) returning canonical proposal string.

        Returns:
            (True, "OK") if hash matches, else (False, reason).
        """
        context_hash = get_context_hash()
        contract_hash = get_contract_hash()
        proposal_params = get_proposal_params(pci.intent_payload)

        expected_hash = compute_evidence_hash(
            pci.dfid, context_hash, contract_hash, proposal_params
        )

        if expected_hash != pci.evidence_hash:
            logger.warning(
                "[DFID=%s] REJECT: Evidence Invalid (hash mismatch). Zero Trust.",
                pci.dfid[:8],
            )
            return False, "Evidence Invalid"

        return True, "OK"

verify(pci, get_context_hash, get_contract_hash, get_proposal_params)

Verify PCI evidence_hash against authoritative sources.

Parameters:

Name Type Description Default
pci ProofCarryingIntent

The Proof-Carrying Intent to verify.

required
get_context_hash Callable[[], str]

Callable returning current context hash.

required
get_contract_hash Callable[[], str]

Callable returning contract hash.

required
get_proposal_params Callable[[Dict[str, Any]], str]

Callable(intent_payload) returning canonical proposal string.

required

Returns:

Type Description
Tuple[bool, str]

(True, "OK") if hash matches, else (False, reason).

Source code in src/dir_core/pci.py
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def verify(
    self,
    pci: ProofCarryingIntent,
    get_context_hash: Callable[[], str],
    get_contract_hash: Callable[[], str],
    get_proposal_params: Callable[[Dict[str, Any]], str],
) -> Tuple[bool, str]:
    """
    Verify PCI evidence_hash against authoritative sources.

    Args:
        pci: The Proof-Carrying Intent to verify.
        get_context_hash: Callable returning current context hash.
        get_contract_hash: Callable returning contract hash.
        get_proposal_params: Callable(intent_payload) returning canonical proposal string.

    Returns:
        (True, "OK") if hash matches, else (False, reason).
    """
    context_hash = get_context_hash()
    contract_hash = get_contract_hash()
    proposal_params = get_proposal_params(pci.intent_payload)

    expected_hash = compute_evidence_hash(
        pci.dfid, context_hash, contract_hash, proposal_params
    )

    if expected_hash != pci.evidence_hash:
        logger.warning(
            "[DFID=%s] REJECT: Evidence Invalid (hash mismatch). Zero Trust.",
            pci.dfid[:8],
        )
        return False, "Evidence Invalid"

    return True, "OK"

compute_evidence_hash(dfid, context_hash, contract_hash, proposal_params)

Evidence Hash formula per Topology C Technical Annex §3.2.

Evidence_Hash = SHA256(DFID || Context_Hash || Contract_Hash || Proposal_Params)

The reference implementation uses proposal_params (canonical JSON of intent) in place of H_r (rule-set hash) for MVP. See Technical Annex §3.2 for full spec.

The DIM recalculates this using authoritative Registry and ContextStore data. It never trusts the agent's claimed hash.

Source code in src/dir_core/pci.py
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def compute_evidence_hash(
    dfid: str,
    context_hash: str,
    contract_hash: str,
    proposal_params: str,
) -> str:
    """
    Evidence Hash formula per Topology C Technical Annex §3.2.

    Evidence_Hash = SHA256(DFID || Context_Hash || Contract_Hash || Proposal_Params)

    The reference implementation uses proposal_params (canonical JSON of intent)
    in place of H_r (rule-set hash) for MVP. See Technical Annex §3.2 for full spec.

    The DIM recalculates this using authoritative Registry and ContextStore data.
    It never trusts the agent's claimed hash.
    """
    payload = f"{dfid}{context_hash}{contract_hash}{proposal_params}"
    return hashlib.sha256(payload.encode()).hexdigest()

hash_content(obj)

SHA256 of canonical JSON.

Source code in src/dir_core/pci.py
22
23
24
def hash_content(obj: Any) -> str:
    """SHA256 of canonical JSON."""
    return hashlib.sha256(_canonical_json(obj).encode()).hexdigest()

proposal_params_for_hash(proposal)

Canonical string of proposal fields for Evidence Hash.

For domain-specific subsets, pass a dict with only the fields to include.

Source code in src/dir_core/pci.py
48
49
50
51
52
53
def proposal_params_for_hash(proposal: Dict[str, Any]) -> str:
    """Canonical string of proposal fields for Evidence Hash.

    For domain-specific subsets, pass a dict with only the fields to include.
    """
    return _canonical_json(proposal)

Decision Ledger

dir_core.ledger

Decision Ledger (Topology C §4.2) — append-only, verified decisions only.

DecisionLedger

Append-only list storing only verified decisions.

Unverified decisions must never become binding. The Ledger is the source of truth; only DIM-approved entries are appended. This prevents "Day Two" failures where hallucinated or forged agent outputs become operational.

Source code in src/dir_core/ledger.py
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
class DecisionLedger:
    """
    Append-only list storing only verified decisions.

    Unverified decisions must never become binding. The Ledger is the source
    of truth; only DIM-approved entries are appended. This prevents "Day Two"
    failures where hallucinated or forged agent outputs become operational.
    """

    def __init__(self) -> None:
        self._entries: List[Dict[str, Any]] = []

    def append(self, pci: ProofCarryingIntent) -> None:
        """Append a verified PCI. Called only by DIM after successful verification."""
        entry = {
            "dfid": pci.dfid,
            "intent_payload": pci.intent_payload,
            "evidence_hash": pci.evidence_hash,
        }
        self._entries.append(entry)
        logger.info(
            "[DFID=%s] Ledger appended entry #%d. Policy Bound.",
            pci.dfid[:8],
            len(self._entries),
        )

    def entries(self) -> List[Dict[str, Any]]:
        """Return all ledger entries (read-only copy)."""
        return list(self._entries)

    def __len__(self) -> int:
        return len(self._entries)

append(pci)

Append a verified PCI. Called only by DIM after successful verification.

Source code in src/dir_core/ledger.py
25
26
27
28
29
30
31
32
33
34
35
36
37
def append(self, pci: ProofCarryingIntent) -> None:
    """Append a verified PCI. Called only by DIM after successful verification."""
    entry = {
        "dfid": pci.dfid,
        "intent_payload": pci.intent_payload,
        "evidence_hash": pci.evidence_hash,
    }
    self._entries.append(entry)
    logger.info(
        "[DFID=%s] Ledger appended entry #%d. Policy Bound.",
        pci.dfid[:8],
        len(self._entries),
    )

entries()

Return all ledger entries (read-only copy).

Source code in src/dir_core/ledger.py
39
40
41
def entries(self) -> List[Dict[str, Any]]:
    """Return all ledger entries (read-only copy)."""
    return list(self._entries)