Skip to content

symfonic.capabilities.memory.rendered_manifest

rendered_manifest

Strict sealed rendered-selection manifest codec, never an authority issuer.

decode 

decode(value, scope, run_id)

Validate exact typed turn ownership and unique publication identities.

Source code in src/symfonic/capabilities/memory/rendered_manifest.py 
 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
 99
100
101
def decode(value, scope, run_id):
    """Validate exact typed turn ownership and unique publication identities."""
    if type(value) is not str or not 1 <= len(value) <= 131072:
        _refuse()
    try:
        document = json.loads(value)
        if (type(document) is not dict or set(document) != _FIELDS
                or encode(document) != value
                or document["schema"] not in ("symfonic.memory.rendered-selection.v1",
                                              "symfonic.memory.rendered-selection.v2")
                or document["run"] != run_id or document["scope"] != scope.path
                or type(document["session"]) is not str
                or type(document["round"]) is not int or document["round"] < 1
                or document["mode"] not in ("invoke", "stream")
                or not _digest(document["wire_digest"])
                or type(document["records"]) is not list
                or not 1 <= len(document["records"]) <= 256):
            _refuse()
        seen, records = set(), []
        sources = set()
        for item in document["records"]:
            if (type(item) is not dict or set(item) not in (_RECORD, _RECORD | {"edge"})
                    or any(type(item[key]) is not int or item[key] < 0
                           for key in ("message", "block", "start", "end"))
                    or item["end"] <= item["start"] or not _digest(item["line_digest"])):
                _refuse()
            publication = PublishedRecallRef(item["id"], _owner(item["owner"]),
                                             item["generation"])
            if (not publication.owner.covers(scope) or publication.record_id in seen):
                _refuse()
            if "edge" in item:
                sources.add(_linked(item, publication, scope))
            seen.add(publication.record_id)
            records.append((publication, item))
        if document["schema"].endswith(".v1") and any("edge" in item for _, item in records):
            _refuse()
        if not sources.issubset({ref for ref, item in records if "edge" not in item}):
            _refuse()
        return tuple(records)
    except MemoryContractError:
        raise
    except (ValueError, TypeError, KeyError):
        _refuse()