Skip to content

Protocol unit

protocol_unit

Shared unit declarations, resolved before any scaffold mutation.

merge_protocol_unit(metadata, declaration)

Persist one shared binding without changing existing actions.

Source code in packages/axm-init/src/axm_init/core/protocol_unit.py
Python
def merge_protocol_unit(metadata: str, declaration: ProtocolUnitDecl) -> str:
    """Persist one shared binding without changing existing actions."""
    from tomlkit.items import Table

    document = parse(metadata)
    profile = table_at(table_at(table_at(document, "tool"), "axm-init"), "protocols")
    profile["schema_version"] = 2
    profile["domain"] = declaration.domain
    units = profile.get("units")
    if units is None:
        units = aot()
        profile["units"] = units
    if not isinstance(units, AoT):
        raise ValueError("protocol units must be an array of tables")
    matching = _unit_table(units, declaration.unit)
    if not isinstance(matching, Table):
        raise ValueError("protocol unit must be a table")
    matching["contracts"] = [item.name for item in declaration.contracts]
    matching["ticket_type"] = declaration.ticket.ticket_type
    matching["input_contract"] = declaration.ticket.input_contract
    matching["entry_action"] = declaration.ticket.entry_action
    return dumps(document)

resolve_protocol_unit(metadata, declarations, supplied)

Resolve only this unit; reject changed bindings and identity collisions.

Source code in packages/axm-init/src/axm_init/core/protocol_unit.py
Python
def resolve_protocol_unit(
    metadata: str,
    declarations: tuple[ProtocolScaffoldDecl, ...],
    supplied: ProtocolUnitDecl | None,
) -> ProtocolUnitDecl | None:
    """Resolve only this unit; reject changed bindings and identity collisions."""
    if not declarations:
        raise ValueError("protocols must contain at least one declaration")
    first = declarations[0]
    domain, units = _units(metadata)
    if domain and domain != first.domain:
        raise ValueError("protocol domain conflicts with existing profile")
    unit = _requested_unit(units, first.unit)
    shared = _resolved(first, _persisted(first.domain, unit), supplied)
    _validate_ownership(units, declarations, shared, first.domain)
    _validate_references(declarations, unit, shared)
    return shared