feat(pm): venv-union auto-bisect — fail-alone disabled, incumbent wins

When the plugin union fails to resolve (update rebuild, plugin
install, any sync), resolve_union() bisects instead of leaving the
venv broken (settled 2026-09-02 plan, task 6):

- try the full union first — clean resolve passes through untouched;
- phase 1: each member alone against core — a plugin that conflicts
  with core pins is disabled with the resolver's message;
- phase 2: reduced union retry; mutual conflicts resolve incumbent-
  wins — the newest-enabled member (last in the discovery list) is
  disabled, the longer-standing setup survives; retries until the
  union resolves or every member is dropped.

Venv.apply routes the union through resolve_union and logs each
disable decision with its resolver reason (receipt surfacing lands
with the universal-receipts task). Pure and unit-tested with stubbed
lock_and_sync: pass-through, fail-alone, incumbent-wins tiebreak.

tests/pm: 165 passed, 0 failed.
This commit is contained in:
ethernet
2026-09-02 19:55:44 -04:00
parent 02eb115f60
commit 1e97613206
3 changed files with 120 additions and 2 deletions

View File

@@ -331,8 +331,23 @@ class Venv(StatePackage):
"(remove the plugin or enable lazy installs)",
)
# Plugin deps union into the venv through the generated
# workspace root (one lock, conflict = loud refusal).
lock_and_sync(member_dirs, extras, venv_dir=self.venv_dir())
# workspace root (one lock, conflict = loud refusal). On
# conflict, resolve_union bisects: fail-alone plugins and
# mutual-conflict losers (incumbent wins) are dropped with
# their resolver reasons, and the union retries.
from pm.workspace import resolve_union
import logging
survivors, decisions = resolve_union(
member_dirs, extras, venv_dir=self.venv_dir()
)
for decision in decisions:
logging.getLogger(__name__).warning(
"plugin %s disabled by the venv union: %s",
decision["plugin"],
decision["reason"],
)
return
uv_bin, env = pm_uv(venv=self.venv_dir())

View File

@@ -259,6 +259,64 @@ def lock_and_sync(
)
def resolve_union(
plugin_dirs: list[Path],
extras: Optional[list[str]] = None,
*,
venv_dir: Optional[Path] = None,
) -> tuple[list[Path], list[dict]]:
"""Try the full union; on failure, bisect the member set.
Returns (surviving_members, decisions). Every decision is
{plugin, action: kept|disabled, reason}. Fail-alone plugins are
disabled with the resolver's message; mutually-conflicting plugins
are resolved incumbent-wins (the most-recently-enabled member of a
conflicting pair is the one disabled — order = plugin_dirs list,
LAST wins the tiebreak because the caller passes
newest-last). Retries the union until it resolves or every member
is disabled; never raises for a resolution failure (a uv binary
absence still raises InstallError)."""
from pm.package import InstallError
survivors = list(plugin_dirs)
decisions: list[dict] = []
def _try(members: list[Path]) -> Optional[str]:
"""None on success, else the failure reason."""
try:
lock_and_sync(members, extras, venv_dir=venv_dir)
return None
except InstallError as exc:
return str(exc)
reason = _try(survivors)
if reason is None:
return survivors, decisions
# Phase 1: each member alone against core. Fail-alone = disabled.
alone_ok: dict[Path, Optional[str]] = {}
for member in list(survivors):
fail = _try([member])
alone_ok[member] = fail
if fail is not None:
decisions.append(
{"plugin": member.name, "action": "disabled", "reason": fail}
)
survivors.remove(member)
# Phase 2: retry the reduced union; if it still fails, the remaining
# set conflicts mutually — incumbent wins, newest (last) disabled.
while survivors:
reason = _try(survivors)
if reason is None:
return survivors, decisions
loser = survivors.pop() # newest-enabled = last in the list
decisions.append(
{"plugin": loser.name, "action": "disabled", "reason": reason}
)
return survivors, decisions
def _default_venv_dir() -> Path:
from pm.packages import Venv

View File

@@ -94,6 +94,51 @@ def test_zero_plugins_still_builds_a_root_with_no_members(layout):
assert "[tool.uv.workspace]" not in text or "members = []" in text
def test_resolve_union_passes_through_on_clean_resolve(monkeypatch):
a, b = Path("/x/a"), Path("/x/b")
calls = []
monkeypatch.setattr(
ws, "lock_and_sync", lambda members, extras=None, **k: calls.append(members)
)
survivors, decisions = ws.resolve_union([a, b], ["web"])
assert survivors == [a, b]
assert decisions == []
assert calls == [[a, b]]
def test_resolve_union_disables_fail_alone_plugins(monkeypatch):
a, b, bad = Path("/x/a"), Path("/x/b"), Path("/x/bad")
from pm.package import InstallError
def fake_lock(members, extras=None, **k):
if any(m == bad for m in members):
raise InstallError("venv", "bad conflicts with core pin")
return None
monkeypatch.setattr(ws, "lock_and_sync", fake_lock)
survivors, decisions = ws.resolve_union([a, bad, b])
assert bad not in survivors
assert [d["plugin"] for d in decisions] == ["bad"]
assert "core pin" in decisions[0]["reason"]
def test_resolve_union_incumbent_wins_on_mutual_conflict(monkeypatch):
old, new = Path("/x/old-plug"), Path("/x/new-plug")
from pm.package import InstallError
def fake_lock(members, extras=None, **k):
if old in members and new in members:
raise InstallError("venv", "old-plug and new-plug are incompatible")
return None
monkeypatch.setattr(ws, "lock_and_sync", fake_lock)
# newest-enabled LAST (the tiebreak contract)
survivors, decisions = ws.resolve_union([old, new])
assert survivors == [old] # incumbent survives
assert [d["plugin"] for d in decisions] == ["new-plug"]
assert "incompatible" in decisions[0]["reason"]
def test_member_stamp_hash_changes_with_plugin_set(layout):
_, _, plug_a, _ = layout
stamp_empty = ws.members_stamp([])