Skip to content

Submission contract

The SDK accepts one closed whole-event profile, doublegate.submission/1. There is no submission_version: 2 shell, legacy fallback or version negotiation. decode_submission returns original bytes, not an authorization result.

Event identity and signing

SubmissionEvent.from_mapping(document) freezes the complete eleven-member object: contract, envelope, content_digest, evidence_manifest_digest, tenant_id, team_id, principal, membership, submitted_at, review_refs, and attribution_chain. Unknown or missing members reject.

For its restricted JCS canonical bytes B:

  • artifact_id = "sha256:" + SHA256(B).hexdigest() identifies the whole event.
  • Detached Ed25519 signs b"doublegate.submission.v1\0" + B.
  • The event's own ID and detached signature are outside B.
  • content_digest is prefixed SHA-256 of raw content bytes.
  • Envelope.envelope_digest is the bare SHA-256 of the frozen envelope's canonical bytes. Both phase payloads use envelope_digest, never the future event ID. The envelope field registry and omission rules are unchanged; no old-name alias exists.

sign_submission accepts a caller-held raw Ed25519 seed. It does not generate or enroll keys. verify_submission_signature verifies cryptography and the expected event ID only: it does not verify phase signatures, signer purpose/principal relationships, historical records, selected scope, reviews or current authority. Use the separately enrolled verify_attribution API for each phase; that API also returns evidence, not authorization.

Encoding a complete submission

from doublegate_sdk.submission import SubmissionEvent, encode_submission, decode_submission


def prepare_submission(blob, document, *, signature, space, local_verdicts,
                       local_promotion, production_evidence_manifest,
                       max_content_bytes):
    event = SubmissionEvent.from_mapping(document)
    wire = encode_submission(
        blob, event.to_dict(), signature=signature, space=space,
        local_verdicts=local_verdicts, local_promotion=local_promotion,
        production_evidence_manifest=production_evidence_manifest,
        max_content_bytes=max_content_bytes,
    )
    assert decode_submission(wire, event.artifact_id,
                             max_content_bytes=max_content_bytes) == blob
    return wire  # Not proof of authentication or admission.

The transport requires document, artifact_id, signature, content, space, local_verdicts, local_promotion, and production_evidence_manifest. Only encoding is optional. Local-record container checks do not replace gate-owned signed-record validation. space remains an exact Unicode identifier, not trimmed, normalized or authorized by parsing.

Inside the event, attribution_chain holds the exact production and submission compact JWS strings. The child references SHA-256 of the root's exact ASCII bytes. Parsing checks phase order, parent and common envelope/evidence/operation/tenant/ requester/contributor/home-team bindings, final actor/principal equality, and production.issued_at <= submission.issued_at <= submitted_at. The manifest is exactly {"manifest_version": 1, "evidence_refs": [...]} with nonempty sorted unique lowercase SHA-256 references. Its digest must match the event and both statements. Existence and authenticity of retained records are not checked. Structural parsing accepts canonically encoded invalid signatures.

Byte semantics and bounds

The encoder emits standard canonical RFC 4648 base64: no whitespace, excess padding or nonzero padding bits. Arbitrary bytes survive unchanged; archives are not extracted or executed. Absent encoding means strict UTF-8 text, without replacement or normalization. Explicit "utf-8" is not supported.

Both functions require caller-chosen nonnegative integer max_content_bytes (booleans reject). There is no SDK default or global 1 MiB policy. Declared, encoded and decoded lengths are bounded; decoded bytes must match envelope content_hash and size_bytes. Zero permits empty content only. SubmissionTooLarge is a ValueError; HTTP mapping belongs to the gate. The outer and optional route/lease artifact_id must equal the whole-event ID.

Integration boundary

Gates must independently verify enrolled event and phase signatures, exact retained evidence/review/promotion and historical references, exact selected space/scope and audience, and fresh current authority before effects. Operation conflicts belong to the gate's event-independent scoped operation journal, not to SDK hashing. Signature success is neither human intent nor admission.

Reuse the frozen event and signature on retries; token renewal must not rewrite signed history. This SDK migration alone does not cut over client, Org or crawler runtimes. Preserve unresolved durable work and receipts rather than rewriting or deleting them.

Transport fixtures use real deterministic public test signatures and inert local records. The vendored reconciliation vector retains source provenance and immutable hashes. An independent Node consumer checks actual SDK-produced bytes and signature, both phase signatures, digest/parent links and tamper refusal. These tests prove bounded byte/crypto agreement, not enrolled trust or admission.

doublegate_sdk.submission

Client-to-organization byte contract, shared by sender, door and worker.

Two identities, one canonical form (ADR-0030, ADR-0064, P1-A/01 §2):

  • content_digest is the content identity: "sha256:" + envelope.content_hash, derived from the envelope and never hashed a second time. Many submissions share one.
  • artifact_id is the submission event identity: sha256 of the event document's own canonical bytes. It is not a member of the signed document; a door recomputes it from the received bytes and compares to the route.

Canonical bytes are RFC 8785 JCS narrowed to a refusal-shaped profile: UTF-8, no whitespace, members ordered by UTF-16 code unit, strings NFC (refused, not normalized), integers in [0, 2**53-1], no floats. The signed event (SubmissionEvent) closes the member set, binds the attribution chain (the federated identity of the writer) and carries a detached Ed25519 signature; phase and event signatures are not authorization. This module does not extract archives, resolve retained authority records or admit content.

Stdlib only in its base path, on purpose: the SDK declares dependencies = []; signing and verification import cryptography lazily (the identity extra).

ENCODING_INVALID module-attribute 

ENCODING_INVALID = 'SUBMISSION_ENCODING_INVALID'

ARTIFACT_ID_MISMATCH module-attribute 

ARTIFACT_ID_MISMATCH = 'SUBMISSION_ARTIFACT_ID_MISMATCH'

MAX_SAFE_INTEGER module-attribute 

MAX_SAFE_INTEGER = 2 ** 53 - 1

SHA256_ID module-attribute 

SHA256_ID = re.compile('^sha256:[0-9a-f]{64}$')

BARE_SHA256 module-attribute 

BARE_SHA256 = re.compile('^[0-9a-f]{64}$')

DOMAIN module-attribute 

DOMAIN = b'doublegate.submission.v1\x00'

EVENT_FIELDS module-attribute 

EVENT_FIELDS = frozenset({'contract', 'envelope', 'content_digest', 'evidence_manifest_digest', 'tenant_id', 'team_id', 'principal', 'membership', 'submitted_at', 'review_refs', 'attribution_chain'})

CanonicalizationRefused

Bases: ValueError

The document breaks the §2 profile. Carries the contract's code.

Source code in src/doublegate_sdk/submission.py 
88
89
90
91
92
93
94
class CanonicalizationRefused(ValueError):
    """The document breaks the §2 profile. Carries the contract's code."""

    code = ENCODING_INVALID

    def __init__(self, message: str) -> None:
        super().__init__(message)

code class-attribute instance-attribute 

code = ENCODING_INVALID

RouteIdMismatch

Bases: ValueError

The route id is not sha256 of the received canonical bytes.

Source code in src/doublegate_sdk/submission.py 
 97
 98
 99
100
101
102
103
class RouteIdMismatch(ValueError):
    """The route id is not ``sha256`` of the received canonical bytes."""

    code = ARTIFACT_ID_MISMATCH

    def __init__(self, message: str) -> None:
        super().__init__(message)

code class-attribute instance-attribute 

code = ARTIFACT_ID_MISMATCH

SubmissionEvent dataclass

One immutable whole-document commitment. Parsing is NOT authorization.

Source code in src/doublegate_sdk/submission.py 
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
@dataclass(frozen=True, slots=True)
class SubmissionEvent:
    """One immutable whole-document commitment. Parsing is NOT authorization."""

    _canonical: bytes

    @classmethod
    def from_mapping(cls, document: dict[str, Any]) -> SubmissionEvent:
        _closed(document, EVENT_FIELDS)
        if document['contract'] != 'doublegate.submission/1':
            raise ValueError('unsupported submission contract')
        canonical = canonical_submission(document)
        if not isinstance(document['envelope'], dict):
            raise ValueError('envelope must be an object')
        env = Envelope.from_mapping(document['envelope'])
        if env.to_dict() != document['envelope']:
            raise ValueError('envelope optional fields must use canonical omission')
        for field in ('content_digest', 'evidence_manifest_digest'):
            _prefixed_digest(document[field])
        if document['content_digest'] != 'sha256:'+env.content_hash:
            raise ValueError('content_digest disagrees with envelope content_hash')
        for field in ('tenant_id', 'team_id'):
            _identifier(document[field])
        principal = document['principal']
        _closed(principal, {'kind', 'id', 'sponsor'})
        if principal['kind'] not in ('person', 'workload'):
            raise ValueError('principal kind must be person or workload')
        _identifier(principal['id'])
        if principal['sponsor'] is not None:
            _identifier(principal['sponsor'])
            if principal['kind'] == 'person':
                raise ValueError('person sponsor must be null')
        membership = document['membership']
        _closed(membership, {'issuer', 'revision', 'assertion_digest'})
        _identifier(membership['issuer'])
        if not re.match(r'^https?://[^/\s]+', membership['issuer']):
            raise ValueError('membership issuer must be a URL')
        if type(membership['revision']) is not int or not 0 <= membership['revision'] <= 9007199254740991:
            raise ValueError('membership revision must be a safe integer')
        _digest(membership['assertion_digest'])
        timestamp = document['submitted_at']
        if not isinstance(timestamp, str) or not re.fullmatch(r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z', timestamp):
            raise ValueError('submitted_at must be UTC with millisecond precision')
        submitted = datetime.strptime(timestamp, '%Y-%m-%dT%H:%M:%S.%fZ').replace(tzinfo=timezone.utc)
        refs = document['review_refs']
        if not isinstance(refs, list):
            raise ValueError('review_refs must be an array')
        for ref in refs:
            _closed(ref, {'stage', 'record_digest'})
            _identifier(ref['stage'])
            _digest(ref['record_digest'])
        chain = decode_attribution_chain(document['attribution_chain'],
            expected_envelope_digest=env.envelope_digest,
            expected_evidence_digest=document['evidence_manifest_digest'][7:])
        production, submission = [s.payload.to_dict() for s in chain.statements]
        if not production['issued_at'] <= submission['issued_at'] <= submitted.timestamp():
            raise ValueError('phase/submission chronology is invalid')
        for field, expected in (('tenant_id', document['tenant_id']),
                                ('home_team_id', document['team_id']),
                                ('actor_id', principal['id'])):
            if submission[field] != expected:
                raise ValueError(f'submission phase {field} disagrees with event')
        return cls(canonical)

    def to_dict(self) -> dict[str, Any]:
        return json.loads(self._canonical)

    def canonical_bytes(self) -> bytes:
        return self._canonical

    @property
    def artifact_id(self) -> str:
        return 'sha256:'+hashlib.sha256(self._canonical).hexdigest()

    @property
    def content_digest(self) -> str:
        return self.to_dict()['content_digest']

    @property
    def signing_input(self) -> bytes:
        return DOMAIN+self._canonical

artifact_id property 

artifact_id: str

content_digest property 

content_digest: str

signing_input property 

signing_input: bytes

from_mapping classmethod 

from_mapping(document: dict[str, Any]) -> SubmissionEvent
Source code in src/doublegate_sdk/submission.py 
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
@classmethod
def from_mapping(cls, document: dict[str, Any]) -> SubmissionEvent:
    _closed(document, EVENT_FIELDS)
    if document['contract'] != 'doublegate.submission/1':
        raise ValueError('unsupported submission contract')
    canonical = canonical_submission(document)
    if not isinstance(document['envelope'], dict):
        raise ValueError('envelope must be an object')
    env = Envelope.from_mapping(document['envelope'])
    if env.to_dict() != document['envelope']:
        raise ValueError('envelope optional fields must use canonical omission')
    for field in ('content_digest', 'evidence_manifest_digest'):
        _prefixed_digest(document[field])
    if document['content_digest'] != 'sha256:'+env.content_hash:
        raise ValueError('content_digest disagrees with envelope content_hash')
    for field in ('tenant_id', 'team_id'):
        _identifier(document[field])
    principal = document['principal']
    _closed(principal, {'kind', 'id', 'sponsor'})
    if principal['kind'] not in ('person', 'workload'):
        raise ValueError('principal kind must be person or workload')
    _identifier(principal['id'])
    if principal['sponsor'] is not None:
        _identifier(principal['sponsor'])
        if principal['kind'] == 'person':
            raise ValueError('person sponsor must be null')
    membership = document['membership']
    _closed(membership, {'issuer', 'revision', 'assertion_digest'})
    _identifier(membership['issuer'])
    if not re.match(r'^https?://[^/\s]+', membership['issuer']):
        raise ValueError('membership issuer must be a URL')
    if type(membership['revision']) is not int or not 0 <= membership['revision'] <= 9007199254740991:
        raise ValueError('membership revision must be a safe integer')
    _digest(membership['assertion_digest'])
    timestamp = document['submitted_at']
    if not isinstance(timestamp, str) or not re.fullmatch(r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z', timestamp):
        raise ValueError('submitted_at must be UTC with millisecond precision')
    submitted = datetime.strptime(timestamp, '%Y-%m-%dT%H:%M:%S.%fZ').replace(tzinfo=timezone.utc)
    refs = document['review_refs']
    if not isinstance(refs, list):
        raise ValueError('review_refs must be an array')
    for ref in refs:
        _closed(ref, {'stage', 'record_digest'})
        _identifier(ref['stage'])
        _digest(ref['record_digest'])
    chain = decode_attribution_chain(document['attribution_chain'],
        expected_envelope_digest=env.envelope_digest,
        expected_evidence_digest=document['evidence_manifest_digest'][7:])
    production, submission = [s.payload.to_dict() for s in chain.statements]
    if not production['issued_at'] <= submission['issued_at'] <= submitted.timestamp():
        raise ValueError('phase/submission chronology is invalid')
    for field, expected in (('tenant_id', document['tenant_id']),
                            ('home_team_id', document['team_id']),
                            ('actor_id', principal['id'])):
        if submission[field] != expected:
            raise ValueError(f'submission phase {field} disagrees with event')
    return cls(canonical)

to_dict 

to_dict() -> dict[str, Any]
Source code in src/doublegate_sdk/submission.py 
404
405
def to_dict(self) -> dict[str, Any]:
    return json.loads(self._canonical)

canonical_bytes 

canonical_bytes() -> bytes
Source code in src/doublegate_sdk/submission.py 
407
408
def canonical_bytes(self) -> bytes:
    return self._canonical

SubmissionTooLarge

Bases: ValueError

A wire or decoded-content bound was exceeded (HTTP 413).

Source code in src/doublegate_sdk/submission.py 
515
516
class SubmissionTooLarge(ValueError):
    """A wire or decoded-content bound was exceeded (HTTP 413)."""

canonical_bytes 

canonical_bytes(doc: Mapping[str, Any]) -> bytes

The one byte string this document has under the §2 profile.

UTF-8, no BOM, no insignificant whitespace, members ordered by UTF-16 code unit. Raises :class:CanonicalizationRefused rather than normalizing.

Source code in src/doublegate_sdk/submission.py 
177
178
179
180
181
182
183
184
185
def canonical_bytes(doc: Mapping[str, Any]) -> bytes:
    """The one byte string this document has under the §2 profile.

    UTF-8, no BOM, no insignificant whitespace, members ordered by UTF-16 code
    unit. Raises :class:`CanonicalizationRefused` rather than normalizing.
    """
    if not isinstance(doc, Mapping):
        raise CanonicalizationRefused("a submission is one JSON object")
    return _object(doc, path="$").encode("utf-8")

envelope_content_hash 

envelope_content_hash(blob: bytes) -> str

The envelope member content_hash: bare 64 lowercase hex, no prefix.

This is the one byte-hashing operation the contract sanctions, and it fills an envelope member — mirroring make_envelope deriving content_hash and size_bytes from the blob so they cannot disagree with it.

It is not the submission's content_digest and its bare return value cannot be mistaken for one. To obtain the submission member, build the envelope and then call :func:content_digest_from_envelope; ADR-0068's mapping clarification forbids hashing content bytes a second time.

Source code in src/doublegate_sdk/submission.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def envelope_content_hash(blob: bytes) -> str:
    """The **envelope** member ``content_hash``: bare 64 lowercase hex, no prefix.

    This is the one byte-hashing operation the contract sanctions, and it fills
    an *envelope* member — mirroring ``make_envelope`` deriving ``content_hash``
    and ``size_bytes`` from the blob so they cannot disagree with it.

    It is **not** the submission's ``content_digest`` and its bare return value
    cannot be mistaken for one. To obtain the submission member, build the
    envelope and then call :func:`content_digest_from_envelope`; ADR-0068's
    mapping clarification forbids hashing content bytes a second time.
    """
    if not isinstance(blob, (bytes, bytearray, memoryview)):
        raise TypeError("envelope_content_hash hashes the content blob")
    return hashlib.sha256(bytes(blob)).hexdigest()

content_digest_from_content_hash 

content_digest_from_content_hash(content_hash: str) -> str

The submission member content_digest, derived from bare hex.

This is the derivation ADR-0068 mandates for the submission's content_digest::

content_digest == "sha256:" + envelope.content_hash

A total function of a value already present in the same signed document. Nothing is computed, so the two values cannot be signed while disagreeing.

Source code in src/doublegate_sdk/submission.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def content_digest_from_content_hash(content_hash: str) -> str:
    """The submission member ``content_digest``, derived from bare hex.

    **This is the derivation ADR-0068 mandates** for the submission's
    ``content_digest``::

        content_digest == "sha256:" + envelope.content_hash

    A total function of a value already present in the same signed document.
    Nothing is computed, so the two values cannot be signed while disagreeing.
    """
    if not isinstance(content_hash, str):
        raise TypeError(
            "content_digest is derived from the envelope's content_hash (bare "
            "64 lowercase hex), not from content bytes")
    if not BARE_SHA256.match(content_hash):
        raise CanonicalizationRefused(
            "envelope content_hash must be bare 64 lowercase hex; a 'sha256:' "
            "prefix is wire presentation and never the stored spelling")
    return "sha256:" + content_hash

content_digest_from_envelope 

content_digest_from_envelope(envelope: Mapping[str, Any]) -> str

The submission member content_digest, derived from the envelope.

This is the derivation ADR-0068 mandates and the one a producer should call: build the envelope first, then emit content_digest from it.

Source code in src/doublegate_sdk/submission.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
def content_digest_from_envelope(envelope: Mapping[str, Any]) -> str:
    """The submission member ``content_digest``, derived from the envelope.

    **This is the derivation ADR-0068 mandates** and the one a producer should
    call: build the envelope first, then emit ``content_digest`` from it.
    """
    if not isinstance(envelope, Mapping):
        raise CanonicalizationRefused("an envelope is one JSON object")
    if "content_hash" not in envelope:
        raise CanonicalizationRefused(
            "envelope carries no content_hash; content_digest is derived from "
            "it and is never computed over content bytes")
    return content_digest_from_content_hash(envelope["content_hash"])

content_hash_from_content_digest 

content_hash_from_content_digest(content_digest: str) -> str

Strip the wire prefix at the door; return the bare stored spelling.

Consequence 2 of the mapping clarification: the sha256: prefix is wire presentation, stripped at the door and never entering the store, so the existing content-keyed dedup index is unchanged. Lossless in both directions with :func:content_digest_from_content_hash.

Source code in src/doublegate_sdk/submission.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def content_hash_from_content_digest(content_digest: str) -> str:
    """Strip the wire prefix at the door; return the bare stored spelling.

    Consequence 2 of the mapping clarification: the ``sha256:`` prefix is wire
    presentation, stripped at the door and never entering the store, so the
    existing content-keyed dedup index is unchanged. Lossless in both
    directions with :func:`content_digest_from_content_hash`.
    """
    if not isinstance(content_digest, str):
        raise TypeError("content_digest is a string")
    if not SHA256_ID.match(content_digest):
        raise CanonicalizationRefused(
            "content_digest must be 'sha256:' + 64 lowercase hex")
    return content_digest[len("sha256:"):]

verify_content_digest 

verify_content_digest(doc: Mapping[str, Any]) -> str

Door check: an encoding check of one value, not a reconciliation of two.

Validates that the received content_digest equals "sha256:" plus the received envelope.content_hash, then returns the bare hex to store. A document in which they differ is malformed, not contested, and is refused with SUBMISSION_ENCODING_INVALID. No content blob is read: content and encoding are the transport shell outside the signed document.

Source code in src/doublegate_sdk/submission.py 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
def verify_content_digest(doc: Mapping[str, Any]) -> str:
    """Door check: an encoding check of one value, not a reconciliation of two.

    Validates that the received ``content_digest`` equals ``"sha256:"`` plus the
    received ``envelope.content_hash``, then returns the bare hex to store. A
    document in which they differ is *malformed*, not contested, and is refused
    with ``SUBMISSION_ENCODING_INVALID``. No content blob is read: ``content``
    and ``encoding`` are the transport shell outside the signed document.
    """
    if not isinstance(doc, Mapping):
        raise CanonicalizationRefused("a submission is one JSON object")
    if "content_digest" not in doc:
        raise CanonicalizationRefused("submission carries no content_digest")
    if "envelope" not in doc:
        raise CanonicalizationRefused("submission carries no envelope")
    expected = content_digest_from_envelope(doc["envelope"])
    received = doc["content_digest"]
    if received != expected:
        raise CanonicalizationRefused(
            "content_digest does not equal 'sha256:' + envelope.content_hash; "
            "the document is malformed, not contested")
    return content_hash_from_content_digest(expected)

submission_artifact_id 

submission_artifact_id(doc: Mapping[str, Any]) -> str

ADR-0068 — the submission event identity.

sha256 of this document's own canonical bytes, spelled sha256: + 64 lowercase hex. The document must not carry artifact_id: a document computed over its own bytes cannot contain its own hash, which is why the clarification took it out of the closed member set (ten members, not eleven).

Source code in src/doublegate_sdk/submission.py 
282
283
284
285
286
287
288
289
290
291
292
293
294
def submission_artifact_id(doc: Mapping[str, Any]) -> str:
    """ADR-0068 — the **submission event** identity.

    ``sha256`` of this document's own canonical bytes, spelled ``sha256:`` + 64
    lowercase hex. The document must not carry ``artifact_id``: a document
    computed over its own bytes cannot contain its own hash, which is why the
    clarification took it out of the closed member set (ten members, not eleven).
    """
    if "artifact_id" in doc:
        raise CanonicalizationRefused(
            "artifact_id is not a member of the signed document (ADR-0068 clarification); "
            "it is sha256 of the document's own canonical bytes")
    return "sha256:" + hashlib.sha256(canonical_bytes(doc)).hexdigest()

verify_route_artifact_id 

verify_route_artifact_id(route_id: str, doc: Mapping[str, Any]) -> str

Recompute the id from the received body and refuse on mismatch.

This is step 6 of the clarification's implementer sequence, and the reason the route segment needs no signature of its own: the id is a pure function of the signed bytes, so tampering with either side moves the comparison.

Source code in src/doublegate_sdk/submission.py 
297
298
299
300
301
302
303
304
305
306
307
308
309
def verify_route_artifact_id(route_id: str, doc: Mapping[str, Any]) -> str:
    """Recompute the id from the received body and refuse on mismatch.

    This is step 6 of the clarification's implementer sequence, and the reason
    the route segment needs no signature of its own: the id is a pure function
    of the signed bytes, so tampering with either side moves the comparison.
    """
    if not isinstance(route_id, str) or not SHA256_ID.match(route_id):
        raise RouteIdMismatch("route artifact_id must be sha256: + 64 lowercase hex")
    computed = submission_artifact_id(doc)
    if route_id != computed:
        raise RouteIdMismatch("route artifact_id does not equal sha256 of the received canonical bytes")
    return computed

canonical_submission 

canonical_submission(value: Any) -> bytes

Restricted JCS bytes of any JSON value, under the same profile as :func:canonical_bytes.

canonical_bytes is the document form (one object); this accepts any JSON value so a member can be canonicalized on its own. Both refuse non-NFC strings, invalid Unicode, floats and integers outside [0, 2**53-1] with a :class:CanonicalizationRefused, which is a ValueError.

Source code in src/doublegate_sdk/submission.py 
320
321
322
323
324
325
326
327
328
329
330
331
def canonical_submission(value: Any) -> bytes:
    """Restricted JCS bytes of any JSON value, under the same profile as :func:`canonical_bytes`.

    ``canonical_bytes`` is the document form (one object); this accepts any JSON value so a
    member can be canonicalized on its own. Both refuse non-NFC strings, invalid Unicode,
    floats and integers outside ``[0, 2**53-1]`` with a :class:`CanonicalizationRefused`,
    which is a ``ValueError``.
    """
    try:
        return _value(value, path="$").encode("utf-8")
    except RecursionError as exc:
        raise CanonicalizationRefused("nesting too deep") from exc

sign_submission 

sign_submission(event: SubmissionEvent, *, private_key: bytes) -> str

Sign with caller-held raw Ed25519 seed; no key generation or enrollment.

Source code in src/doublegate_sdk/submission.py 
435
436
437
438
439
440
def sign_submission(event: SubmissionEvent, *, private_key: bytes) -> str:
    """Sign with caller-held raw Ed25519 seed; no key generation or enrollment."""
    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
    event = SubmissionEvent.from_mapping(event.to_dict())
    signature = Ed25519PrivateKey.from_private_bytes(private_key).sign(event.signing_input)
    return base64.urlsafe_b64encode(signature).decode('ascii').rstrip('=')

verify_submission_signature 

verify_submission_signature(event: SubmissionEvent, signature: str, *, public_key: bytes, expected_artifact_id: str) -> None

Check route and crypto only. Caller owns key purpose/principal enrollment.

Does not verify either phase signature, historical records, local promotion, selected scope, current authority or operation uniqueness.

Source code in src/doublegate_sdk/submission.py 
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
def verify_submission_signature(event: SubmissionEvent, signature: str, *,
                                public_key: bytes, expected_artifact_id: str) -> None:
    """Check route and crypto only. Caller owns key purpose/principal enrollment.

    Does not verify either phase signature, historical records, local promotion,
    selected scope, current authority or operation uniqueness.
    """
    from cryptography.exceptions import InvalidSignature
    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
    event = SubmissionEvent.from_mapping(event.to_dict())
    if event.artifact_id != expected_artifact_id:
        raise ValueError('route names a different artifact')
    try:
        Ed25519PublicKey.from_public_bytes(public_key).verify(_signature_bytes(signature), event.signing_input)
    except (InvalidSignature, ValueError) as exc:
        raise ValueError('invalid submission signature or public key') from exc

decode_submission 

decode_submission(body: dict[str, Any], artifact_id: str | None = None, *, max_content_bytes: int) -> bytes

Return exact content bytes after strict, UNTRUSTED submission parsing.

Only doublegate.submission/1 is accepted. Parsing checks shape and byte bindings, not signatures, retained evidence, membership, grants or current authority.

Source code in src/doublegate_sdk/submission.py 
519
520
521
522
523
524
525
526
527
def decode_submission(body: dict[str, Any], artifact_id: str | None = None,
                      *, max_content_bytes: int) -> bytes:
    """Return exact content bytes after strict, UNTRUSTED submission parsing.

    Only doublegate.submission/1 is accepted. Parsing checks shape and byte bindings,
    not signatures, retained evidence, membership, grants or current authority.
    """
    return _parse_submission(body, artifact_id,
                             max_content_bytes=max_content_bytes).content

encode_submission 

encode_submission(content: bytes, document: dict[str, Any], *, max_content_bytes: int, signature: str, space: str, local_verdicts: list[dict], local_promotion: dict, production_evidence_manifest: dict) -> dict[str, Any]

Encode exact content with a frozen event and its detached signature.

No bearer or authority is stored. Reuse this body unchanged on token refresh. Structural validation only: a 64-byte invalid signature can still parse.

Source code in src/doublegate_sdk/submission.py 
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
def encode_submission(content: bytes, document: dict[str, Any], *,
                      max_content_bytes: int, signature: str, space: str,
                      local_verdicts: list[dict], local_promotion: dict,
                      production_evidence_manifest: dict) -> dict[str, Any]:
    """Encode exact content with a frozen event and its detached signature.

    No bearer or authority is stored. Reuse this body unchanged on token refresh.
    Structural validation only: a 64-byte invalid signature can still parse.
    """
    if not isinstance(content, bytes):
        raise ValueError('content must be bytes')
    if type(max_content_bytes) is not int or max_content_bytes < 0:
        raise ValueError('max_content_bytes must be a non-negative integer')
    if len(content) > max_content_bytes:
        raise SubmissionTooLarge('decoded content exceeds configured byte limit')
    body = {'document': json.loads(json.dumps(document)),
            'artifact_id': 'sha256:'+hashlib.sha256(canonical_submission(document)).hexdigest(),
            'signature': signature, 'content': base64.b64encode(content).decode('ascii'),
            'encoding': 'base64', 'space': space, 'local_verdicts': local_verdicts,
            'local_promotion': local_promotion,
            'production_evidence_manifest': production_evidence_manifest}
    decode_submission(body, max_content_bytes=max_content_bytes)
    return body

doublegate_sdk.envelope

Envelope v1 — the atomic unit everywhere downstream.

Normative source: ADR-0022 (design/docs/adr/ADR-0022-envelope-v1-frozen.md). This module IS that table. If the two ever disagree, the ADR wins and this module is the bug.

Rules that make envelope_digest a stable hash:

  • canonical JSON: sorted keys, UTF-8, no whitespace, no null values emitted (absent means absent);
  • envelope_digest is sha256 of the canonical envelope and is NOT a field — the envelope cannot contain its own hash;
  • lifecycle state (state, provisional, content_erased, verdicts, findings) lives in the store, never in the envelope, so erasure never changes an id;
  • v2 may ADD optional fields; it may not remove, rename or retype a v1 field. envelope_version selects the canonicalization rules, so v1 ids stay valid forever.

ENVELOPE_VERSION module-attribute 

ENVELOPE_VERSION = 1

TRUST_CLASSES module-attribute 

TRUST_CLASSES = frozenset({'T-0', 'T-1', 'T-2', 'T-3', 'T-4', 'T-5'})

REQUIRED_FIELDS module-attribute 

REQUIRED_FIELDS = frozenset(k for k, (req, _) in _FIELDS.items() if req)

OPTIONAL_FIELDS module-attribute 

OPTIONAL_FIELDS = frozenset(k for k, (req, _) in _FIELDS.items() if not req)

ALL_FIELDS module-attribute 

ALL_FIELDS = REQUIRED_FIELDS | OPTIONAL_FIELDS

EnvelopeError

Bases: ValueError

The envelope violates ADR-0022. Message names the field.

Source code in src/doublegate_sdk/envelope.py 
57
58
class EnvelopeError(ValueError):
    """The envelope violates ADR-0022. Message names the field."""

Envelope dataclass

An admitted artifact's immutable description.

Frozen: the envelope never changes after admission — that is what lets envelope_digest be a hash of it. Build one via :func:make_envelope or :meth:Envelope.from_mapping, never by editing attributes.

Source code in src/doublegate_sdk/envelope.py 
 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
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
@dataclass(frozen=True, slots=True)
class Envelope:
    """An admitted artifact's immutable description.

    Frozen: the envelope never changes after admission — that is what lets
    ``envelope_digest`` be a hash of it. Build one via :func:`make_envelope` or
    :meth:`Envelope.from_mapping`, never by editing attributes.
    """

    content_hash: str
    content_type: str
    trust_class: str
    source_uri: str
    writer_identity: str
    deployment_id: str
    ingest_ts: int
    size_bytes: int
    derives_from: tuple[str, ...] = field(default=())
    content_key_id: str | None = None
    envelope_version: int = ENVELOPE_VERSION

    # ---- construction ----------------------------------------------------

    @classmethod
    def from_mapping(cls, m: Mapping[str, Any]) -> Envelope:
        """Validate a mapping against ADR-0022 and build an Envelope.

        Unknown keys are rejected: a v1 reader must not silently accept fields
        it cannot hash consistently. (A v2 reader will know v2's fields.)
        """
        unknown = set(m) - ALL_FIELDS
        if unknown:
            raise EnvelopeError(f"unknown field(s): {sorted(unknown)}")
        missing = REQUIRED_FIELDS - set(m)
        if missing:
            raise EnvelopeError(f"missing required field(s): {sorted(missing)}")
        for name, (_, types) in _FIELDS.items():
            if name in m:
                v = m[name]
                if v is None:
                    raise EnvelopeError(f"{name}: null is not allowed; omit the field")
                # bool is an int subclass; reject it for int fields explicitly
                if isinstance(v, bool) and int in types:
                    raise EnvelopeError(f"{name}: expected int, got bool")
                if not isinstance(v, types):
                    raise EnvelopeError(f"{name}: expected {types[0].__name__}, got {type(v).__name__}")
        env = cls(
            content_hash=m["content_hash"],
            content_type=m["content_type"],
            trust_class=m["trust_class"],
            source_uri=m["source_uri"],
            writer_identity=m["writer_identity"],
            deployment_id=m["deployment_id"],
            ingest_ts=m["ingest_ts"],
            size_bytes=m["size_bytes"],
            derives_from=tuple(m.get("derives_from", ())),
            content_key_id=m.get("content_key_id"),
            envelope_version=m["envelope_version"],
        )
        env.validate()
        return env

    def validate(self) -> None:
        """Semantic checks beyond type: version, hash shape, class, sizes."""
        if self.envelope_version != ENVELOPE_VERSION:
            raise EnvelopeError(
                f"envelope_version: this reader knows v{ENVELOPE_VERSION}, got v{self.envelope_version}"
            )
        if not _SHA256_HEX.fullmatch(self.content_hash):
            raise EnvelopeError("content_hash: must be lowercase sha256 hex (64 chars)")
        if self.trust_class not in TRUST_CLASSES:
            raise EnvelopeError(f"trust_class: {self.trust_class!r} not in T-0..T-5")
        for s in ("content_type", "source_uri", "writer_identity", "deployment_id"):
            if not getattr(self, s):
                raise EnvelopeError(f"{s}: must be non-empty")
        if self.ingest_ts < 0:
            raise EnvelopeError("ingest_ts: must be non-negative epoch milliseconds")
        if self.size_bytes < 0:
            raise EnvelopeError("size_bytes: must be non-negative")
        for parent in self.derives_from:
            if not isinstance(parent, str) or not _SHA256_HEX.fullmatch(parent):
                raise EnvelopeError("derives_from: every entry must be an artifact_id (sha256 hex)")
        if self.content_key_id is not None and not self.content_key_id:
            raise EnvelopeError("content_key_id: must be non-empty when present")

    # ---- canonical form -----------------------------------------------------

    def to_dict(self) -> dict[str, Any]:
        """Plain mapping with NO null values and NO empty optional list.

        ``derives_from`` is emitted only when non-empty, ``content_key_id``
        only when set — "absent means absent" (ADR-0022 decision 1). An empty
        list and an absent list must hash identically, so we never emit ``[]``.
        """
        d: dict[str, Any] = {
            "envelope_version": self.envelope_version,
            "content_hash": self.content_hash,
            "content_type": self.content_type,
            "trust_class": self.trust_class,
            "source_uri": self.source_uri,
            "writer_identity": self.writer_identity,
            "deployment_id": self.deployment_id,
            "ingest_ts": self.ingest_ts,
            "size_bytes": self.size_bytes,
        }
        if self.derives_from:
            d["derives_from"] = list(self.derives_from)
        if self.content_key_id is not None:
            d["content_key_id"] = self.content_key_id
        return d

    def canonical_bytes(self) -> bytes:
        """The exact bytes that are hashed. Sorted keys, no whitespace, UTF-8."""
        return canonical_json(self.to_dict())

    @property
    def envelope_digest(self) -> str:
        """sha256 of the canonical envelope. Not a field (ADR-0022 decision 2)."""
        return hashlib.sha256(self.canonical_bytes()).hexdigest()

content_hash instance-attribute 

content_hash: str

content_type instance-attribute 

content_type: str

trust_class instance-attribute 

trust_class: str

source_uri instance-attribute 

source_uri: str

writer_identity instance-attribute 

writer_identity: str

deployment_id instance-attribute 

deployment_id: str

ingest_ts instance-attribute 

ingest_ts: int

size_bytes instance-attribute 

size_bytes: int

derives_from class-attribute instance-attribute 

derives_from: tuple[str, ...] = field(default=())

content_key_id class-attribute instance-attribute 

content_key_id: str | None = None

envelope_version class-attribute instance-attribute 

envelope_version: int = ENVELOPE_VERSION

envelope_digest property 

envelope_digest: str

sha256 of the canonical envelope. Not a field (ADR-0022 decision 2).

from_mapping classmethod 

from_mapping(m: Mapping[str, Any]) -> Envelope

Validate a mapping against ADR-0022 and build an Envelope.

Unknown keys are rejected: a v1 reader must not silently accept fields it cannot hash consistently. (A v2 reader will know v2's fields.)

Source code in src/doublegate_sdk/envelope.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
@classmethod
def from_mapping(cls, m: Mapping[str, Any]) -> Envelope:
    """Validate a mapping against ADR-0022 and build an Envelope.

    Unknown keys are rejected: a v1 reader must not silently accept fields
    it cannot hash consistently. (A v2 reader will know v2's fields.)
    """
    unknown = set(m) - ALL_FIELDS
    if unknown:
        raise EnvelopeError(f"unknown field(s): {sorted(unknown)}")
    missing = REQUIRED_FIELDS - set(m)
    if missing:
        raise EnvelopeError(f"missing required field(s): {sorted(missing)}")
    for name, (_, types) in _FIELDS.items():
        if name in m:
            v = m[name]
            if v is None:
                raise EnvelopeError(f"{name}: null is not allowed; omit the field")
            # bool is an int subclass; reject it for int fields explicitly
            if isinstance(v, bool) and int in types:
                raise EnvelopeError(f"{name}: expected int, got bool")
            if not isinstance(v, types):
                raise EnvelopeError(f"{name}: expected {types[0].__name__}, got {type(v).__name__}")
    env = cls(
        content_hash=m["content_hash"],
        content_type=m["content_type"],
        trust_class=m["trust_class"],
        source_uri=m["source_uri"],
        writer_identity=m["writer_identity"],
        deployment_id=m["deployment_id"],
        ingest_ts=m["ingest_ts"],
        size_bytes=m["size_bytes"],
        derives_from=tuple(m.get("derives_from", ())),
        content_key_id=m.get("content_key_id"),
        envelope_version=m["envelope_version"],
    )
    env.validate()
    return env

validate 

validate() -> None

Semantic checks beyond type: version, hash shape, class, sizes.

Source code in src/doublegate_sdk/envelope.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def validate(self) -> None:
    """Semantic checks beyond type: version, hash shape, class, sizes."""
    if self.envelope_version != ENVELOPE_VERSION:
        raise EnvelopeError(
            f"envelope_version: this reader knows v{ENVELOPE_VERSION}, got v{self.envelope_version}"
        )
    if not _SHA256_HEX.fullmatch(self.content_hash):
        raise EnvelopeError("content_hash: must be lowercase sha256 hex (64 chars)")
    if self.trust_class not in TRUST_CLASSES:
        raise EnvelopeError(f"trust_class: {self.trust_class!r} not in T-0..T-5")
    for s in ("content_type", "source_uri", "writer_identity", "deployment_id"):
        if not getattr(self, s):
            raise EnvelopeError(f"{s}: must be non-empty")
    if self.ingest_ts < 0:
        raise EnvelopeError("ingest_ts: must be non-negative epoch milliseconds")
    if self.size_bytes < 0:
        raise EnvelopeError("size_bytes: must be non-negative")
    for parent in self.derives_from:
        if not isinstance(parent, str) or not _SHA256_HEX.fullmatch(parent):
            raise EnvelopeError("derives_from: every entry must be an artifact_id (sha256 hex)")
    if self.content_key_id is not None and not self.content_key_id:
        raise EnvelopeError("content_key_id: must be non-empty when present")

to_dict 

to_dict() -> dict[str, Any]

Plain mapping with NO null values and NO empty optional list.

derives_from is emitted only when non-empty, content_key_id only when set — "absent means absent" (ADR-0022 decision 1). An empty list and an absent list must hash identically, so we never emit [].

Source code in src/doublegate_sdk/envelope.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def to_dict(self) -> dict[str, Any]:
    """Plain mapping with NO null values and NO empty optional list.

    ``derives_from`` is emitted only when non-empty, ``content_key_id``
    only when set — "absent means absent" (ADR-0022 decision 1). An empty
    list and an absent list must hash identically, so we never emit ``[]``.
    """
    d: dict[str, Any] = {
        "envelope_version": self.envelope_version,
        "content_hash": self.content_hash,
        "content_type": self.content_type,
        "trust_class": self.trust_class,
        "source_uri": self.source_uri,
        "writer_identity": self.writer_identity,
        "deployment_id": self.deployment_id,
        "ingest_ts": self.ingest_ts,
        "size_bytes": self.size_bytes,
    }
    if self.derives_from:
        d["derives_from"] = list(self.derives_from)
    if self.content_key_id is not None:
        d["content_key_id"] = self.content_key_id
    return d

canonical_bytes 

canonical_bytes() -> bytes

The exact bytes that are hashed. Sorted keys, no whitespace, UTF-8.

Source code in src/doublegate_sdk/envelope.py 
172
173
174
def canonical_bytes(self) -> bytes:
    """The exact bytes that are hashed. Sorted keys, no whitespace, UTF-8."""
    return canonical_json(self.to_dict())

canonical_json 

canonical_json(obj: Any) -> bytes

Canonical JSON per ADR-0022: sorted keys, no whitespace, UTF-8, no NaN.

ensure_ascii=False so non-ASCII content hashes as its UTF-8 bytes, not as \uXXXX escapes — the escaped form is not canonical across encoders.

Source code in src/doublegate_sdk/envelope.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
def canonical_json(obj: Any) -> bytes:
    """Canonical JSON per ADR-0022: sorted keys, no whitespace, UTF-8, no NaN.

    ``ensure_ascii=False`` so non-ASCII content hashes as its UTF-8 bytes, not
    as ``\\uXXXX`` escapes — the escaped form is not canonical across encoders.
    """
    return json.dumps(
        obj,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
        allow_nan=False,
    ).encode("utf-8")

content_hash 

content_hash(blob: bytes) -> str

sha256 of a content blob — the value that goes in content_hash.

Source code in src/doublegate_sdk/envelope.py 
197
198
199
def content_hash(blob: bytes) -> str:
    """sha256 of a content blob — the value that goes in ``content_hash``."""
    return hashlib.sha256(blob).hexdigest()

make_envelope 

make_envelope(*, blob: bytes, content_type: str, trust_class: str, source_uri: str, writer_identity: str, deployment_id: str, ingest_ts: int, derives_from: tuple[str, ...] | list[str] = (), content_key_id: str | None = None) -> Envelope

Build a validated v1 envelope from a content blob and its provenance.

content_hash and size_bytes are derived from blob so they cannot disagree with it.

Source code in src/doublegate_sdk/envelope.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def make_envelope(
    *,
    blob: bytes,
    content_type: str,
    trust_class: str,
    source_uri: str,
    writer_identity: str,
    deployment_id: str,
    ingest_ts: int,
    derives_from: tuple[str, ...] | list[str] = (),
    content_key_id: str | None = None,
) -> Envelope:
    """Build a validated v1 envelope from a content blob and its provenance.

    ``content_hash`` and ``size_bytes`` are derived from ``blob`` so they
    cannot disagree with it.
    """
    env = Envelope(
        content_hash=content_hash(blob),
        content_type=content_type,
        trust_class=trust_class,
        source_uri=source_uri,
        writer_identity=writer_identity,
        deployment_id=deployment_id,
        ingest_ts=ingest_ts,
        size_bytes=len(blob),
        derives_from=tuple(derives_from),
        content_key_id=content_key_id,
    )
    env.validate()
    return env