Route-index parsing

akms.task_context.routes

Parsing and graph-bound validation for task route indexes.

parse_route_index

parse_route_index(
    source: TaskRouteIndex | Mapping[str, Any] | str | Path,
    *,
    graph: Any | None = None,
) -> TaskRouteIndex

Parse, canonicalize, and optionally graph-validate a route index.

Source code in packages/akms/src/akms/task_context/routes.py

def parse_route_index(
    source: TaskRouteIndex | Mapping[str, Any] | str | Path,
    *,
    graph: Any | None = None,
) -> TaskRouteIndex:
    """Parse, canonicalize, and optionally graph-validate a route index."""

    if isinstance(source, TaskRouteIndex):
        index = source
    else:
        data = _load_route_source(source)
        raw_issues = _inspect_raw_routes(data)
        if raw_issues:
            raise RouteIndexValidationError(raw_issues)
        try:
            index = TaskRouteIndex.model_validate(data)
        except ValidationError as error:
            raise RouteIndexValidationError(_pydantic_issues(error)) from error

    if graph is not None:
        validate_route_index_nodes(index, graph)
    return index

validate_route_index_nodes

validate_route_index_nodes(
    index: TaskRouteIndex, graph: Any
) -> TaskRouteIndex

Fail closed when a route references a node absent from the graph.

Source code in packages/akms/src/akms/task_context/routes.py

def validate_route_index_nodes(
    index: TaskRouteIndex,
    graph: Any,
) -> TaskRouteIndex:
    """Fail closed when a route references a node absent from the graph."""

    nodes = _node_membership(graph)
    issues: list[RouteValidationIssue] = []
    for field in ("by_path", "by_symbol"):
        routes = getattr(index, field)
        for route_key, records in routes.items():
            for record_index, record in enumerate(records):
                if record.node_id not in nodes:
                    issues.append(
                        RouteValidationIssue(
                            code="missing_graph_node",
                            location=(
                                field,
                                route_key,
                                record_index,
                                "node_id",
                            ),
                            message=(
                                "Route references nonexistent graph node "
                                f"'{record.node_id}'"
                            ),
                            node_id=record.node_id,
                        )
                    )
    if issues:
        raise RouteIndexValidationError(issues)
    return index