Skip to content

geno_lewm.encoder.runtime_identity

runtime_identity

Closed identity contract for a cache-building Carbon encoder runtime.

EncoderRuntimeIdentity dataclass 

EncoderRuntimeIdentity(model_id: str, revision: str, state_contract_version: str, runtime_hash: str, weights_hash: str | None = None, schema_version: str = ENCODER_RUNTIME_IDENTITY_SCHEMA_VERSION)

Pinned model and byte identities required for cache construction.

cache_identity_hash property 

cache_identity_hash: str

Return the identity committed into cache keys for this state contract.

to_dict 

to_dict() -> dict[str, object]

Return the canonical JSON-native contract payload.

Source code in geno_lewm/encoder/runtime_identity.py 
80
81
82
83
84
85
86
87
88
89
90
91
def to_dict(self) -> dict[str, object]:
    """Return the canonical JSON-native contract payload."""
    payload: dict[str, object] = {
        "schema_version": self.schema_version,
        "model_id": self.model_id,
        "revision": self.revision,
        "state_contract_version": self.state_contract_version,
        "runtime_hash": self.runtime_hash,
    }
    if self.weights_hash is not None:
        payload["weights_hash"] = self.weights_hash
    return payload

parse_encoder_runtime_identity_bytes 

parse_encoder_runtime_identity_bytes(body: bytes, *, source: str) -> EncoderRuntimeIdentity

Parse one duplicate-key-free, closed runtime identity JSON object.

Source code in geno_lewm/encoder/runtime_identity.py 
 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
def parse_encoder_runtime_identity_bytes(
    body: bytes,
    *,
    source: str,
) -> EncoderRuntimeIdentity:
    """Parse one duplicate-key-free, closed runtime identity JSON object."""
    try:
        payload = json.loads(body, object_pairs_hook=_reject_duplicate_keys)
    except (UnicodeDecodeError, json.JSONDecodeError, InputError) as exc:
        raise InputError(
            "encoder runtime identity is invalid JSON",
            details={"source": source, "error": str(exc)},
        ) from exc
    if type(payload) is not dict:
        raise InputError("encoder runtime identity must contain one JSON object")
    observed = frozenset(payload)
    if not _REQUIRED_KEYS.issubset(observed) or not observed.issubset(
        _REQUIRED_KEYS | _OPTIONAL_KEYS
    ):
        raise InputError(
            "encoder runtime identity has an invalid closed schema",
            details={
                "required": sorted(_REQUIRED_KEYS),
                "optional": sorted(_OPTIONAL_KEYS),
                "observed": sorted(observed),
            },
        )
    return EncoderRuntimeIdentity(
        schema_version=payload["schema_version"],
        model_id=payload["model_id"],
        revision=payload["revision"],
        state_contract_version=payload["state_contract_version"],
        runtime_hash=payload["runtime_hash"],
        weights_hash=payload.get("weights_hash"),
    )

encoder_runtime_identity_from_mapping 

encoder_runtime_identity_from_mapping(raw: object) -> EncoderRuntimeIdentity

Validate a JSON-native mapping through the same closed parser.

Source code in geno_lewm/encoder/runtime_identity.py 
131
132
133
134
135
136
137
def encoder_runtime_identity_from_mapping(raw: object) -> EncoderRuntimeIdentity:
    """Validate a JSON-native mapping through the same closed parser."""
    try:
        body = json.dumps(raw, sort_keys=True, allow_nan=False).encode("utf-8")
    except (TypeError, ValueError) as exc:
        raise InputError("encoder_runtime_identity must be finite JSON-native data") from exc
    return parse_encoder_runtime_identity_bytes(body, source="encoder_runtime_identity")