Python First Admission and Historical Trust

Python First Admission and Historical Trust

The Python AdmissionService layers authenticated First Admission and Historical Trust above the six-stage Signed Document verifier (MWP-ADM-005). The controlling language-independent flow is the local First Admission and Historical Trust runtime page, and the exact requirements are the local First-Admission and Historical-Trust clauses.

Public types

Public API	Responsibility
AdmissionCurrentKeyResolver.resolve_current	Return complete Organization-wide Registry evidence that the deployment asserts is current for a new first-admission decision.
TrustedAdmissionContext.issue	Issue the trusted record ID, trusted acceptance instant, and accepting service only after authoritative absence.
AdmissionLog.lookup	Return AdmissionLookupStatus.FOUND with authenticated bytes or AdmissionLookupStatus.AUTHORITATIVE_ABSENCE. Transport failure, cache miss, indeterminate state, and unauthenticated absence fail closed under MWP-ADM-003.
AdmissionLog.append_or_return_existing	Atomically append candidate bytes or return the concurrent authoritative winner.
AuthenticatedAdmissionRecord	Bind returned record bytes to the accepting service identity authenticated by the adapter for MWP-ADM-009 validation.
AdmissionService.prepare_first_admission	Create and validate candidate bytes from an already verified document and trusted context. It does not append or imply admission.
AdmissionService.admit_first	Implement the current-evidence, authoritative-absence, atomic append, and returned-record flow in MWP-ADM-006.
AdmissionService.verify_historical_admission	Rerun verification with historical Registry evidence and require a found record without issuing context or appending, as required by MWP-ADM-008.

The adapter result is typed evidence. A caller-provided trust boolean is not an authoritative lookup outcome and must never select the success path.

First-admission call

Informative example

admitted = AdmissionService().admit_first(
    SignedDocumentKind.COMMAND,
    document_bytes,
    current_registry,
    admission_log,
    trusted_context,
)

admit_first verifies before calling the log. When lookup returns AdmissionLookupStatus.AUTHORITATIVE_ABSENCE, it calls prepare_first_admission, invokes append_or_return_existing, then validates the record actually returned. A concurrent winner is acceptable only when its bytes, authenticated service, document binding, and trusted time all validate (MWP-ADM-009, MWP-ADM-010).

Historical replay call

Informative example

historical = AdmissionService().verify_historical_admission(
    SignedDocumentKind.COMMAND,
    document_bytes,
    historical_registry,
    admission_log,
)

Historical replay uses retained Registry history, requires a found Admission record, and never calls TrustedAdmissionContext.issue or AdmissionLog.append_or_return_existing. Matching record IDs in the example demonstrate that replay recovered the first authoritative record; they do not by themselves prove Command freshness, signer authorization, state-machine acceptance, or portable log-proof verification. Command freshness and signer authorization remain separate requirements under MWP-ADM-013 and MWP-ADM-014.

Runnable exact-pinned example

Informative example

The example uses only public SDK APIs and the declared cryptography runtime dependency. It deterministically creates an Ed25519 key, a complete Organization-wide Registry, a signed Command, and typed in-memory adapters. The fixed private seed is demonstration material and must never be used outside the example.

"""Run First Admission and Historical Trust with typed in-memory adapters."""

from __future__ import annotations

import base64
import hashlib
import json
from dataclasses import dataclass
from threading import Lock

from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from missionweaveprotocol import (
    AdmissionContextValue,
    AdmissionCurrentKeyResolver,
    AdmissionLog,
    AdmissionLookup,
    AdmissionLookupStatus,
    AdmissionService,
    AuthenticatedAdmissionRecord,
    KeyRegistryCompleteness,
    KeyRegistrySnapshot,
    KeyResolutionRequest,
    KeyResolver,
    PrincipalEvidence,
    SignedDocumentCodec,
    SignedDocumentKind,
    SigningKey,
    TrustedAdmissionContext,
)

AGENT_ID = "urn:missionweaveprotocol:agent:website-example"
KEY_ID = "urn:missionweaveprotocol:key:website-example"
ORGANIZATION_ID = "urn:missionweaveprotocol:organization:website-example"


class DeterministicSigningKey:
    """Demonstration-only Ed25519 signing adapter with a fixed seed."""

    algorithm = "Ed25519"

    def __init__(self, key_id: str) -> None:
        self.key_id = key_id
        self._private_key = Ed25519PrivateKey.from_private_bytes(bytes(range(1, 33)))
        self.public_key_bytes = self._private_key.public_key().public_bytes(
            encoding=serialization.Encoding.Raw,
            format=serialization.PublicFormat.Raw,
        )

    def sign(self, message: bytes) -> bytes:
        return self._private_key.sign(message)


@dataclass(frozen=True, slots=True)
class CurrentRegistryFixture:
    """Expose one complete Registry as both current and historical evidence."""

    registry_bytes: bytes

    def _snapshot(self) -> KeyRegistrySnapshot:
        return KeyRegistrySnapshot(
            completeness=KeyRegistryCompleteness.ORGANIZATION_WIDE,
            registry_bytes=self.registry_bytes,
        )

    def resolve_current(self, request: KeyResolutionRequest) -> KeyRegistrySnapshot:
        del request
        return self._snapshot()

    def resolve(self, request: KeyResolutionRequest) -> KeyRegistrySnapshot:
        del request
        return self._snapshot()


@dataclass(frozen=True, slots=True)
class FixedTrustedAdmissionContext:
    """Issue trusted acceptance metadata after authoritative log absence."""

    trusted_accepted_at: str
    accepted_by: PrincipalEvidence

    def issue(self, organization_id: str, signing_hash: str) -> AdmissionContextValue:
        logical_key_digest = hashlib.sha256(
            f"{organization_id}\0{signing_hash}".encode()
        ).hexdigest()
        return AdmissionContextValue(
            admission_record_id=(f"urn:missionweaveprotocol:admission-record:{logical_key_digest}"),
            trusted_accepted_at=self.trusted_accepted_at,
            accepted_by=self.accepted_by,
        )


class InMemoryAdmissionLog:
    """Model authenticated lookup and atomic append-or-return-existing behavior."""

    def __init__(self, authenticated_service: PrincipalEvidence) -> None:
        self._authenticated_service = authenticated_service
        self._records: dict[tuple[str, str], AuthenticatedAdmissionRecord] = {}
        self._lock = Lock()
        self.append_count = 0

    def lookup(self, organization_id: str, signing_hash: str) -> AdmissionLookup:
        with self._lock:
            record = self._records.get((organization_id, signing_hash))
        if record is None:
            return AdmissionLookup(AdmissionLookupStatus.AUTHORITATIVE_ABSENCE)
        return AdmissionLookup(AdmissionLookupStatus.FOUND, record)

    def append_or_return_existing(
        self,
        organization_id: str,
        signing_hash: str,
        candidate_bytes: bytes,
    ) -> AuthenticatedAdmissionRecord:
        key = (organization_id, signing_hash)
        with self._lock:
            existing = self._records.get(key)
            if existing is not None:
                return existing

            candidate = AuthenticatedAdmissionRecord(
                record_bytes=candidate_bytes,
                authenticated_service=self._authenticated_service,
            )
            self._records[key] = candidate
            self.append_count += 1
            return candidate


def main() -> None:
    signing_key: SigningKey = DeterministicSigningKey(KEY_ID)
    registry_bytes = json.dumps(
        {
            "organizationId": ORGANIZATION_ID,
            "bindings": [
                {
                    "keyId": KEY_ID,
                    "principal": {"type": "agent", "id": AGENT_ID},
                    "algorithm": "Ed25519",
                    "publicKey": base64.urlsafe_b64encode(signing_key.public_key_bytes)
                    .rstrip(b"=")
                    .decode("ascii"),
                    "validFrom": "2026-07-01T00:00:00Z",
                    "validityHistory": [],
                }
            ],
        },
        separators=(",", ":"),
    ).encode("utf-8")
    registry = CurrentRegistryFixture(registry_bytes)
    current_registry: AdmissionCurrentKeyResolver = registry
    historical_registry: KeyResolver = registry

    document_bytes = (
        SignedDocumentCodec()
        .sign(
            SignedDocumentKind.COMMAND,
            {
                "protocolVersion": "0.1",
                "actionId": "urn:uuid:11111111-2222-4333-8444-555555555555",
                "actor": {"type": "agent", "id": AGENT_ID},
                "sessionEpoch": 1,
                "membershipEpoch": 1,
                "groupId": "urn:missionweaveprotocol:group:website-example",
                "kind": "message.post",
                "correlationId": "urn:uuid:aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee",
                "issuedAt": "2026-07-15T00:00:00Z",
                "payload": {"text": "Admission example"},
            },
            signing_key,
        )
        .canonical_document_bytes
    )

    admission_service = PrincipalEvidence(
        type="service",
        id="urn:missionweaveprotocol:service:admission",
    )
    trusted_context: TrustedAdmissionContext = FixedTrustedAdmissionContext(
        trusted_accepted_at="2026-07-15T00:05:00Z",
        accepted_by=admission_service,
    )
    admission_log_impl = InMemoryAdmissionLog(admission_service)
    admission_log: AdmissionLog = admission_log_impl
    service = AdmissionService()

    first = service.admit_first(
        SignedDocumentKind.COMMAND,
        document_bytes,
        current_registry,
        admission_log,
        trusted_context,
    )
    historical = service.verify_historical_admission(
        SignedDocumentKind.COMMAND,
        document_bytes,
        historical_registry,
        admission_log,
    )

    assert first.record.admission_record_id == historical.record.admission_record_id
    assert first.verified.signing_hash == historical.verified.signing_hash
    assert first.record.raw_bytes == historical.record.raw_bytes
    assert admission_log_impl.append_count == 1
    print(f"first admission: {first.record.admission_record_id}")
    print(f"historical replay: {historical.record.admission_record_id}")


if __name__ == "__main__":
    main()

Run it from an exact checkout after installing all SDK extras:

uv sync --all-extras
uv run python /absolute/path/to/examples/sdk/python/admission.py

Expected output contains the same First-Admission Record ID for first admission and historical replay. The program also asserts matching signing hashes and record bytes, and that the log appended exactly once.

Implementation note

Replace the example adapters in a deployment:

• CurrentRegistryFixture with an authenticated current-Registry resolver;
• InMemoryAdmissionLog with an authenticated, append-only, atomic Admission Log; and
• FixedTrustedAdmissionContext with an Organization-controlled trusted-time and accepting-service issuer.

Replace DeterministicSigningKey with a deployment-owned signing service or key adapter. Never embed a private signing seed in production code.

Do not collapse those boundaries into one availability or trust flag.

Post-cryptographic Admission rejection remains protected stage admission and wire code AUTH_INVALID_SIGNATURE (MWP-ADM-012).