diff --git a/meshterm/ui/pathgraph.py b/meshterm/ui/pathgraph.py index 48f8ed3..e0343ae 100644 --- a/meshterm/ui/pathgraph.py +++ b/meshterm/ui/pathgraph.py @@ -454,13 +454,56 @@ def _balanced_x(ordered_nodes: list[str], edges: set[tuple[str, str]]) -> dict[s resolves to one x. Because every edge steps the from-origin distance up and the to-us distance down, the fraction rises strictly along each path: edges only ever run left to right. + + A pair walked in *both* directions is a 2-cycle in the edge set, and the longest-path + relaxation would loop through it, inflating every downstream depth toward the node-count + cap and jamming other relays hard against the ends. So the rank runs over the graph with + each such pair (transitively) merged to one representative — the acyclic flow the picture + really is; the pair, drawn as a single vertical, rightly shares an x anyway. """ - up = _longest_paths(ordered_nodes, edges) - down = _longest_paths(ordered_nodes, {(v, u) for u, v in edges}) - return { - node: (up[node] / (up[node] + down[node])) if (up[node] + down[node]) else 0.0 - for node in ordered_nodes + rep = _merge_bidir_pairs(ordered_nodes, edges) + reps: list[str] = [] + seen: set[str] = set() + for node in ordered_nodes: + if rep[node] not in seen: + seen.add(rep[node]) + reps.append(rep[node]) + rep_edges = {(rep[u], rep[v]) for u, v in edges if rep[u] != rep[v]} + up = _longest_paths(reps, rep_edges) + down = _longest_paths(reps, {(v, u) for u, v in rep_edges}) + frac = { + r: (up[r] / (up[r] + down[r])) if (up[r] + down[r]) else 0.0 for r in reps } + return {node: frac[rep[node]] for node in ordered_nodes} + + +def _merge_bidir_pairs( + ordered_nodes: list[str], edges: set[tuple[str, str]] +) -> dict[str, str]: + """Map each node to a representative, uniting any two joined by a both-ways edge. + + Two nodes walked in both directions form a 2-cycle; uniting them — transitively, so a + chain of such pairs folds into one group — lets the balanced rank treat the flow as the + acyclic run it otherwise is. A node in no such pair maps to itself. The choice of which + member is the representative doesn't matter: every member is assigned the group's one + fraction, and the group's rank is structural. + """ + parent = {node: node for node in ordered_nodes} + + def find(node: str) -> str: + root = node + while parent[root] != root: + root = parent[root] + while parent[node] != root: # path-compress + parent[node], node = root, parent[node] + return root + + for u, v in edges: + if (v, u) in edges: + ru, rv = find(u), find(v) + if ru != rv: + parent[ru] = rv + return {node: find(node) for node in ordered_nodes} def _longest_paths(ordered_nodes: list[str], edges: set[tuple[str, str]]) -> dict[str, int]: diff --git a/tests/test_pathgraph.py b/tests/test_pathgraph.py index 212fd89..5c1e635 100644 --- a/tests/test_pathgraph.py +++ b/tests/test_pathgraph.py @@ -4,10 +4,13 @@ from __future__ import annotations import re +import pytest + from meshterm.ui.pathgraph import ( DST_NODE, SRC_NODE, PathLayer, + _balanced_x, _coalesce_prefixes, _route, render_path_graph, @@ -203,6 +206,30 @@ def test_route_bidirectional_is_a_single_straight_segment() -> None: assert _route("u", "v", {"u": (20, 1), "v": (24, 30)}, bidir=True) == [(20, 1), (24, 30)] +def test_bidir_pair_does_not_jam_a_sibling_relay_against_the_origin() -> None: + """A both-ways pair's 2-cycle must not inflate the rank and squash another relay left. + + ``aa`` and ``bb`` are walked in both directions (the vertical-drawn pair); ``cc`` is a + plain relay one hop off the origin feeding ``bb``. Ranked over the raw edge set the cycle + would drive ``cc`` toward ``0``; collapsing the pair leaves ``cc`` at its natural third of + the way across, and the pair shares one x (they draw as a single vertical). + """ + layers = [ + PathLayer(("bb",), WHITE, 3), + PathLayer(("aa", "bb"), GREY, 2), + PathLayer(("bb", "aa"), GREY, 2), # the reverse — makes aa<->bb a 2-cycle + PathLayer(("cc", "bb"), GREY, 1), + ] + seqs = [(SRC_NODE, *layer.hops, DST_NODE) for layer in layers] + ordered = list(dict.fromkeys(n for seq in seqs for n in seq)) + edges = {pair for seq in seqs for pair in zip(seq, seq[1:])} + x = _balanced_x(ordered, edges) + assert x[SRC_NODE] == 0.0 and x[DST_NODE] == 1.0 + assert x["cc"] == pytest.approx(1 / 3) # its natural spot, not jammed toward 0 + assert x["aa"] == x["bb"] # the pair shares one x — it draws as a single vertical + assert x["cc"] < x["aa"] # and still sits left of the pair it feeds + + def test_labels_ellipsize_only_when_wider_than_the_canvas() -> None: """A name shortens only when it physically cannot fit — wider than the whole canvas.""" name = "x" * 30