feat(verdicts): a cross-base candidate collision is reported, never dropped in silence

[skip-docs] — the invariant row for this plan lands in Step 13, after the mutations.

Co-Authored-By: Claude <claude-opus-5>

src/portfolio_optimiser/run.py

@@ -109,6 +109,7 @@ from portfolio_optimiser.semretrieval import (
     load_embedder_config,
 )
 from portfolio_optimiser.verdicts import (
+    VerdictCollision,
     ExpeLContextProvider,
     ProposalFeatures,
     Verdict,
@@ -534,6 +535,32 @@ def cost_baseline_notice(anchored: bool) -> str | None:
     return None if anchored else _UNANCHORED_NOTICE
 
 
+def collision_notice(collisions: tuple[VerdictCollision, ...]) -> str | None:
+    """Render which candidates two bases both described, or ``None`` when none did.
+
+    ``None`` on an empty tuple — omission, never an empty row (``mandate.announce``'s rule, as
+    ``skipped_links_notice`` and ``cost_baseline_notice`` already follow). A dispatch where every
+    candidate belonged to exactly one base has nothing to report.
+
+    ONE renderer, taking the ALREADY-RESOLVED trace rather than a store to re-scan: a renderer that
+    recomputed the collisions would be a second resolution of the same fact, free to disagree with
+    the dispatch it describes (kø-(p)).
+
+    **Surface, stated plainly:** ``MultiBaseResult`` has no production caller today, so this notice
+    is library-facing. It is written now because the field would otherwise be a value nothing can
+    display — the same principle Step 4 applies one level down: a signal that was not stored must
+    not become an asserted absent one.
+    """
+    if not collisions:
+        return None
+    lines = ["Cross-base candidates (one verdict each; the later base's was dropped):"]
+    lines.extend(
+        f"  - {c.verdict_id}: first from {c.first_bundle_id!r}, again from {c.second_bundle_id!r}"
+        for c in collisions
+    )
+    return "\n".join(lines)
+
+
 def skipped_links_notice(skipped: tuple[okf.SkippedLink, ...]) -> str | None:
     """Render what the run could NOT read, or ``None`` when every cross-link was followed.
 
@@ -1455,6 +1482,12 @@ class MultiBaseResult:
     stopped_early: bool = False
     budget_stop: BudgetStop | None = None
     unreached: tuple[ApproachOutcome, ...] = ()
+    #: Candidates two bases both described (D2). DEFAULTS to an empty tuple, which is the
+    #: ``skipped_links`` half of the required-vs-default rule and not the
+    #: ``cost_baseline_anchored`` half: an empty trace is an honest POSITIVE statement ("no
+    #: candidate was described by two bases"), whereas a missing bool would have to assert
+    #: something about an event and both assertions would sometimes be untrue.
+    collisions: tuple[VerdictCollision, ...] = ()
 
 
 async def run_mandate_across_bundles(
@@ -1538,6 +1571,10 @@ async def run_mandate_across_bundles(
     store = store if store is not None else VerdictStore(verdicts=[])
     runs: list[BundleRun] = []
     unreached: list[ApproachOutcome] = []
+    collisions: list[VerdictCollision] = []
+    #: Which base FIRST produced each verdict id — the map the store deliberately does not keep,
+    #: and the reason this accounting lives in the dispatcher rather than in ``VerdictStore.add``.
+    first_base_by_verdict: dict[str, str] = {}
     budget_stop: BudgetStop | None = None
 
     for index, (bundle_id, sub_mandate) in enumerate(routed):
@@ -1565,6 +1602,13 @@ async def run_mandate_across_bundles(
         # A second lookup for the label would be the kø-(p) duplicate free to drift from the value
         # the run was actually dispatched with.
         project_id = str(okf.load_ir_projection(bundle_dir)["project_id"])
+        # D2: the id of the verdict THIS base minted, taken from ``run_project``'s existing
+        # ``notify`` seam rather than off the returned ``RunResult``. ``notify`` fires inside the
+        # capture block, so it is called exactly when a verdict exists (F2: never when nobody
+        # reviewed the run) and it carries the minted object, id included. The slot is built FRESH
+        # each iteration — a shared accumulator would let a later base read the previous base's
+        # verdict and manufacture a collision that never happened.
+        minted_here: list[str] = []
         result = cast(
             RunResult,
             await run_project(
@@ -1572,6 +1616,7 @@ async def run_mandate_across_bundles(
                 profile,
                 docs_dir=bundle_dir,
                 bundle_dir=bundle_dir,
+                notify=lambda verdict: minted_here.append(verdict.id),
                 verdict_input=verdict_input,
                 verdict_dir=verdict_dir,
                 dimension=dimension,
@@ -1584,6 +1629,29 @@ async def run_mandate_across_bundles(
                 meter=_run_meter(None, portfolio_meter, max_rounds),
             ),
         )
+        # D2, dispatcher-side accounting. The condition is the dispatcher's OWN map and nothing
+        # else: a base that already claimed this id is the only thing that makes "a SECOND base"
+        # a statement worth making. A Step-7 inbox verdict or a bundle seed sharing the id is NOT
+        # a cross-base collision, and is excluded exactly by never appearing in this map.
+        #
+        # The plan asked for a before/after snapshot of ``{v.id for v in store.verdicts}`` beside
+        # this. MEASURED redundant and therefore left out: ``store.add`` is first-write-wins and
+        # never removes, so any id in this map is necessarily in the store when a later base runs.
+        # The snapshot could not change the outcome of a single dispatch — a conjunct no mutation
+        # can redden is dead code wearing a guard's clothes, which is the class this repo writes
+        # rows against rather than ships.
+        for minted_id in minted_here:
+            first = first_base_by_verdict.get(minted_id)
+            if first is None:
+                first_base_by_verdict[minted_id] = bundle_id
+            else:
+                collisions.append(
+                    VerdictCollision(
+                        verdict_id=minted_id,
+                        first_bundle_id=first,
+                        second_bundle_id=bundle_id,
+                    )
+                )
         runs.append(
             BundleRun(
                 bundle_id=bundle_id,
@@ -1596,6 +1664,7 @@ async def run_mandate_across_bundles(
     return MultiBaseResult(
         runs=tuple(runs),
         store=store,
+        collisions=tuple(collisions),
         stopped_early=budget_stop is not None,
         budget_stop=budget_stop,
         unreached=tuple(unreached),