Protocols

protocols

ProtocolRecord dataclass

Static category inventory entry; declaration and validity stay separate.

Source code in packages/axm-init/src/axm_init/checks/protocols.py

@dataclass
class ProtocolRecord:
    """Static category inventory entry; declaration and validity stay separate."""

    graph_name: str
    state: str
    location: str
    factory: str
    family: str | None = None
    claims: list[tuple[str, str, str]] = dataclass_field(default_factory=list)
    registration_findings: list[str] = dataclass_field(default_factory=list)
    skeleton_findings: list[str] = dataclass_field(default_factory=list)

check_author_grammar(project)

Validate import-time forms through syntax and resolved imports.

Source code in packages/axm-init/src/axm_init/checks/protocols.py

def check_author_grammar(project: Path) -> CheckResult:
    """Validate import-time forms through syntax and resolved imports."""
    context = _protocol_package(project)
    details = [] if context is None else _author_findings(project, context[2])
    return _content_result("author_grammar", details, applicable=context is not None)

check_protocol_assembly(project)

Validate assembly factories and graph identities statically.

Source code in packages/axm-init/src/axm_init/checks/protocols.py

def check_protocol_assembly(project: Path) -> CheckResult:
    """Validate assembly factories and graph identities statically."""
    context = _protocol_package(project)
    details = (
        []
        if context is None
        else _assembly_findings(project, context[0], context[1], context[2])
    )
    return _content_result("protocol_assembly", details, applicable=context is not None)

check_protocol_components(project)

Validate principal components without importing inspected code.

Source code in packages/axm-init/src/axm_init/checks/protocols.py

def check_protocol_components(project: Path) -> CheckResult:
    """Validate principal components without importing inspected code."""
    context = _protocol_package(project)
    details = [] if context is None else _component_findings(project, context[2])
    return _content_result(
        "protocol_components", details, applicable=context is not None
    )

check_protocol_draft(project)

Keep drafts informative and reject ready protocols containing skeletons.

Source code in packages/axm-init/src/axm_init/checks/protocols.py

def check_protocol_draft(project: Path) -> CheckResult:
    """Keep drafts informative and reject ready protocols containing skeletons."""
    records = protocol_inventory(project)
    details = [finding for record in records for finding in record.skeleton_findings]
    return _content_result(
        "protocol_draft", details, applicable=any(r.state == "ready" for r in records)
    )

check_protocol_registration(project)

Check declared state, local factory bindings and inspected collisions.

Source code in packages/axm-init/src/axm_init/checks/protocols.py

def check_protocol_registration(project: Path) -> CheckResult:
    """Check declared state, local factory bindings and inspected collisions."""
    records = protocol_inventory(project)
    details = [
        finding for record in records for finding in record.registration_findings
    ]
    details.extend(protocol_collisions([(project.name, records)]))
    return _content_result("protocol_registration", details, applicable=bool(records))

check_protocol_ticket(project)

Validate ticket presence and references without importing inspected code.

Source code in packages/axm-init/src/axm_init/checks/protocols.py

def check_protocol_ticket(project: Path) -> CheckResult:
    """Validate ticket presence and references without importing inspected code."""
    context = _protocol_package(project)
    details = [] if context is None else _ticket_findings(project, *context)
    if context is not None:
        data = cast(
            "dict[str, object]",
            tomllib.loads((project / "pyproject.toml").read_text(encoding="utf-8")),
        )
        profile = _nested(data, "tool", "axm-init", "protocols") or {}
        entries = {
            group: _nested(data, "project", "entry-points", group) or {}
            for group in (
                *GRAPH_GROUPS,
                "axm.ticket_types",
                "axm.tribunal.ticket_types",
            )
        }
        details.extend(
            shared_ticket_findings(
                project, context[0], _tables(profile.get("units")) or [], entries
            )
        )
    return _content_result("protocol_ticket", details, applicable=context is not None)

check_protocols_profile(project)

Validate a declared protocol profile without importing inspected code.

Source code in packages/axm-init/src/axm_init/checks/protocols.py

def check_protocols_profile(project: Path) -> CheckResult:
    """Validate a declared protocol profile without importing inspected code."""
    metadata = project / "pyproject.toml"
    try:
        text = metadata.read_text(encoding="utf-8")
        data = cast("dict[str, object]", tomllib.loads(text))
    except (OSError, tomllib.TOMLDecodeError) as exc:
        return CheckResult(
            name="protocols.profile",
            category=_CATEGORY,
            passed=False,
            weight=4,
            message="Protocol metadata is unreadable",
            details=[f"pyproject.toml:1: {exc}. Correction: fix the TOML metadata"],
            fix="Fix pyproject.toml before validating the protocol profile.",
        )

    profile = _nested(data, "tool", "axm-init", "protocols")
    if profile is None:
        return CheckResult(
            name="protocols.profile",
            category=_CATEGORY,
            passed=True,
            weight=0,
            message="Protocol profile not declared",
            details=[],
            fix="",
        )

    details: list[str] = []
    schema_version = profile.get("schema_version")
    if schema_version not in (_SCHEMA_VERSION, 2):
        details.append(
            _finding(
                text,
                "schema_version",
                f"unsupported schema_version {schema_version!r}",
                f"set schema_version = {_SCHEMA_VERSION}",
            )
        )

    domain = _validate_identity(data, profile, text, details)
    units = _validate_units(text, profile, details)
    if schema_version != 2 and any("ticket_type" in unit for unit in units):
        details.append(
            _finding(
                text,
                "schema_version",
                "shared units require schema_version = 2",
                "upgrade the shared profile to schema_version = 2",
            )
        )
    if domain is not None:
        _validate_wheel_inclusion(data, text, domain, details)
        _validate_protocol_tree(project, text, domain, units, details)

    passed = not details
    return CheckResult(
        name="protocols.profile",
        category=_CATEGORY,
        passed=passed,
        weight=4,
        message=(
            "Protocol profile is statically coherent"
            if passed
            else f"Protocol profile has {len(details)} finding(s)"
        ),
        details=details,
        fix="" if passed else "Apply every correction listed in the findings.",
    )

check_protocols_resources(project)

Validate declared prompt files and their distribution configuration.

Source code in packages/axm-init/src/axm_init/checks/protocols.py

def check_protocols_resources(project: Path) -> CheckResult:
    """Validate declared prompt files and their distribution configuration."""
    metadata = project / "pyproject.toml"
    try:
        text = metadata.read_text(encoding="utf-8")
        data = cast("dict[str, object]", tomllib.loads(text))
    except (OSError, tomllib.TOMLDecodeError):
        return CheckResult(
            name="protocols.protocols_resources",
            category=_CATEGORY,
            passed=True,
            weight=0,
            message="Protocol prompt resources are not applicable",
            details=[],
            fix="",
        )

    profile = _nested(data, "tool", "axm-init", "protocols")
    if profile is None:
        return CheckResult(
            name="protocols.protocols_resources",
            category=_CATEGORY,
            passed=True,
            weight=0,
            message="Protocol profile not declared",
            details=[],
            fix="",
        )

    prompt_paths = _declared_prompt_paths(project, profile)
    details = _missing_prompt_findings(project, text, prompt_paths)
    details.extend(
        _excluded_prompt_findings(project, text, data, profile, prompt_paths)
    )

    passed = not details
    return CheckResult(
        name="protocols.protocols_resources",
        category=_CATEGORY,
        passed=passed,
        weight=2,
        message=(
            "Protocol prompt resources are present and distributable"
            if passed
            else f"Protocol prompt resources have {len(details)} finding(s)"
        ),
        details=details,
        fix="" if passed else "Apply every correction listed in the findings.",
    )

protocol_collisions(inventories)

Report collisions only among inspected declarations, per registry.

Source code in packages/axm-init/src/axm_init/checks/protocols.py

def protocol_collisions(
    inventories: list[tuple[str, list[ProtocolRecord]]],
) -> list[str]:
    """Report collisions only among inspected declarations, per registry."""
    claims: dict[tuple[str, str], set[str]] = {}
    for member, records in inventories:
        for record in records:
            for registry, identity, location in record.claims:
                claims.setdefault((registry, identity), set()).add(
                    f"{member}: {location}"
                )
    return [
        f"{registry} identifier {identity!r} collides: {', '.join(sorted(locations))}. "
        "Correction: assign distinct identifiers in this registry."
        for (registry, identity), locations in sorted(claims.items())
        if len(locations) > 1
    ]

protocol_inventory(project)

Inspect declared protocols without importing their modules or registries.

Source code in packages/axm-init/src/axm_init/checks/protocols.py

def protocol_inventory(project: Path) -> list[ProtocolRecord]:
    """Inspect declared protocols without importing their modules or registries."""
    context = _protocol_package(project)
    if context is None:
        return []
    domain, root, package = context
    text = (project / "pyproject.toml").read_text(encoding="utf-8")
    data = cast("dict[str, object]", tomllib.loads(text))
    profile = _nested(data, "tool", "axm-init", "protocols") or {}
    groups = {
        group: _nested(data, "project", "entry-points", group) or {}
        for group in GRAPH_GROUPS
    }
    modules = _module_index(root, package)
    records = []
    for unit, action, declaration in _declared_actions(
        _tables(profile.get("units")) or []
    ):
        qualified = f"protocols_{domain}.{unit}.{action}.protocol"
        record = ProtocolRecord(
            graph_name=f"{domain}.{unit}.{action}",
            state=str(declaration.get("state", "draft")),
            location=_state_location(text, unit, action),
            factory=f"{qualified}:build_protocol",
        )
        _inspect_record(
            project,
            root / unit / action,
            package,
            record,
            partial(
                _groups_registering,
                factory=record.factory,
                groups=groups,
                modules=modules,
            ),
        )
        record.registration_findings = _registration_findings(
            record, groups, modules, text
        )
        if declaration.get("unit_ticket") is not None:
            _inspect_shared_record(project, root / unit, package, record)
        records.append(record)
    return records