Protocol planner

`protocol_planner`

`PlanOperation` `dataclass`

Describe the non-mutating plan for one relative path.

Source code in packages/axm-init/src/axm_init/core/protocol_planner.py

Python
@dataclass(frozen=True, slots=True)
class PlanOperation:
    """Describe the non-mutating plan for one relative path."""

    path: PurePosixPath
    status: PlanStatus
    content: str | None

`PlanStatus`

Bases: StrEnum

Classify one planned relative-path operation.

Source code in packages/axm-init/src/axm_init/core/protocol_planner.py

Python
class PlanStatus(StrEnum):
    """Classify one planned relative-path operation."""

    CREATE = "create"
    UPDATE = "update"
    UNCHANGED = "unchanged"
    CONFLICT = "conflict"

`ProtocolScaffoldPlan` `dataclass`

Hold ordered file operations and merged project metadata.

Source code in packages/axm-init/src/axm_init/core/protocol_planner.py

Python
@dataclass(frozen=True, slots=True)
class ProtocolScaffoldPlan:
    """Hold ordered file operations and merged project metadata."""

    operations: tuple[PlanOperation, ...]
    metadata: str

`plan_protocol_scaffold(declaration, metadata, inventory, *, unit_declaration=None)`

Compute a deterministic scaffold plan without mutating caller state.

Source code in packages/axm-init/src/axm_init/core/protocol_planner.py

Python
def plan_protocol_scaffold(
    declaration: ProtocolScaffoldDecl,
    metadata: str,
    inventory: Mapping[str, str],
    *,
    unit_declaration: ProtocolUnitDecl | None = None,
) -> ProtocolScaffoldPlan:
    """Compute a deterministic scaffold plan without mutating caller state."""
    merged_metadata = merge_protocol_metadata(metadata, declaration)
    profile_owned = _is_profile_owned(metadata, declaration)
    operations = tuple(
        _classify(planned, inventory, profile_owned=profile_owned)
        for planned in _planned_files(declaration)
    )
    if declaration.unit_ticket is not None:
        if unit_declaration is None:
            raise ValueError("unit_ticket requires a resolved shared declaration")
        content = _render_shared_protocol(declaration, unit_declaration)
        path = PurePosixPath(
            f"src/protocols_{declaration.domain}/{declaration.unit}/{declaration.action}/protocol.py"
        )
        operations = tuple(
            _classify(
                _fixed_file(str(path), content), inventory, profile_owned=profile_owned
            )
            if item.path == path
            else item
            for item in operations
        )
    return ProtocolScaffoldPlan(operations=operations, metadata=merged_metadata)

`plan_protocol_unit(declaration, metadata, inventory)`

Plan a single shared ticket and context without action-local copies.

Source code in packages/axm-init/src/axm_init/core/protocol_planner.py

Python
def plan_protocol_unit(
    declaration: ProtocolUnitDecl,
    metadata: str,
    inventory: Mapping[str, str],
) -> ProtocolScaffoldPlan:
    """Plan a single shared ticket and context without action-local copies."""
    from axm_init.core.protocol_unit import merge_protocol_unit

    root = f"src/protocols_{declaration.domain}/{declaration.unit}"
    model = next(
        item
        for item in declaration.contracts
        if item.name == declaration.ticket.input_contract
    )
    content = (
        f"{_python_header()}\n"
        "from protocols_core import TicketTypeDef\n"
        f"from protocols_{declaration.domain}.{declaration.unit}"
        f".contracts.{model.name} "
        f"import {model.model_name}\n\n"
        "__all__ = ['get_ticket_type']\n\n"
        f"TICKET_TYPE = {declaration.ticket.ticket_type!r}\n"
        f"INPUT_CONTRACT = {declaration.ticket.input_contract!r}\n"
        f"GRAPH_NAME = {declaration.graph_name!r}\n\n"
        "def get_ticket_type() -> TicketTypeDef:\n"
        '    """Return the single unit ticket definition."""\n'
        "    return TicketTypeDef(\n"
        f"        name=TICKET_TYPE, context_model={model.model_name},\n"
        "        graph=GRAPH_NAME,\n"
        "    )\n"
    )
    planned = [
        _fixed_file(
            f"src/protocols_{declaration.domain}/__init__.py", _render_initializer()
        ),
        _fixed_file(f"{root}/__init__.py", _render_initializer()),
        _fixed_file(f"{root}/contracts/__init__.py", _render_initializer()),
        _fixed_file(f"{root}/ticket.py", content),
        *[
            _fixed_file(f"{root}/contracts/{item.name}.py", _render_contract(item))
            for item in declaration.contracts
        ],
    ]
    owned = _shared_profile_owned(metadata, declaration)
    operations = tuple(
        _classify(
            item, inventory, profile_owned=owned or item.path.name == "__init__.py"
        )
        for item in sorted(planned, key=lambda item: item.path)
    )
    return ProtocolScaffoldPlan(
        operations=operations, metadata=merge_protocol_unit(metadata, declaration)
    )

Protocol planner