"""S7a-3 pkt. 1 - the DECLARED id is the identity; the mount is a filesystem accident. **The measurement that forced it** (``docs/2026-09-03-syretest-s7a2-k2.md`` ยง 1). The first delivered corpus that declares its own ``bundle_id`` -- K2, 618 of 630 concept files plus the root index, all saying ``k2-trinn1-20260903`` -- arrived mounted as ``K2-bundle-20260903``. Step 10's reconciliation refused it with ``BundleIdMismatch`` at EVERY door: the base could not be opened as delivered, and the only remedy was to re-mount it under the declared name by hand, once per delivery, forever. That is a consumer refusing a producer's legitimate output over a directory name. **Operator decision (PM, 2026-09-03).** The consumer slackens. The declared id (concept frontmatter first, root ``index.md`` as fallback -- B1's order, unchanged) IS the identity; the mount's basename answers only when nothing is declared. A disagreement between declared and mount is RECORDED (``ProvenanceStamp.bundle_id_source`` / ``DryRunReport.bundle_id_source`` + one warning line), never refused. **What still refuses, and it is the REAL collision.** Two CONCEPTS inside one base declaring DIFFERENT ids is a base that cannot say what it is -- no mount name can settle that, and every artefact stamped from it would name one of two corpora at random. ``okf.assert_declared_ids_agree`` raises there, and it is called at every door that OPENS a base. **The root index NEVER participates in that check, and that is decision B1 applied twice.** Concept beats index is a PRECEDENCE rule; an index that disagrees with its concepts is the fallback losing, not two concepts colliding. Folding the index into the agreement set would newly refuse exactly the K2-shaped bases this file exists to admit. **One naming scheme, or the pipeline breaks -- a CONSEQUENCE, not scope creep.** ``explore._bundle_index`` keyed bases on ``Path(raw).name`` and ``run_mandate_across_bundles`` on ``reconcile_bundle_id(raw).id``. While a declared id that disagreed with its mount was refused outright the two could not differ; with declared-wins they can, and then ``explore()`` mints approaches naming the MOUNT while the dispatcher routes by the DECLARED id -- so an exploration's own mandate becomes unroutable. ``_bundle_index`` therefore resolves the declared id too. It stays usable on a base that cannot be read (falling back to the basename) because two of its own arms in ``test_explore_loadbearing.py`` configure directories that do not exist and expect an id error, not an I/O one -- and because a base nobody can read is refused a moment later by whichever door actually opens it. **Newly reachable, stated:** ``run_mandate_across_bundles``'s "two knowledge bases share the id" refusal used to require two mounts with the same basename. Two differently-named mounts declaring the same id now collide there -- same code, no longer a defensive branch (arm (k)). **Honesty limit, measured and left alone.** ``reconcile_bundle_id`` without a named concept answers from the root ``index.md``, so a base whose index declares X while its concepts declare Y resolves to X. Changing that would change B1's resolution ORDER, which this order did not ask for; the agreement check above is about concepts colliding with each other, not with their index. Arms: (a) a K2-shaped base opens, declared wins * (b) the mount answers only in silence * (c) two disagreeing concepts refused BY NAME * (d) the index is not in the collision set * (e) the run stamps which source answered, and COMPLETES * (f) the dry run carries it too * (g) the renderer fires on disagreement and is silent on agreement * (h) the CLI prints it * (i) explore addresses a base by its DECLARED id * (j) ``read_bundle`` refuses an intra-base collision * (k) the dispatcher's shared-id refusal is newly reachable * (l) the field has no default. """ from __future__ import annotations import json import shutil from pathlib import Path from typing import Any import pytest from conftest import SyntheticUsageChatClient from portfolio_optimiser import explore, okf, run from portfolio_optimiser.mandate import Approach, Mandate from portfolio_optimiser.provenance import Citation, ProvenanceStamp from portfolio_optimiser.retrieval import TextSpan from portfolio_optimiser.run import MandateRoutingError, bundle_id_notice, run_project _EXAMPLES = Path(__file__).resolve().parents[1] / "shared" / "examples" _BYGG = _EXAMPLES / "bygg-energi-mikro" _DATA = Path(__file__).resolve().parents[1] / "src" / "portfolio_optimiser" / "data" / "bundles" _RUNNABLE = _DATA / "bygg-energi-mikro-a" _RUNNABLE_PID = "BYGG-ENERGI-MIKRO-A" #: The K2 shape, in one line: a base that calls itself one thing and sits in a directory called #: another. Deliberately NOT a valid directory name pattern of its own -- the point is that the two #: differ, not that either is prettier. _DECLARED = "k2-shaped-corpus" _REPLY = json.dumps( { "measure": "LED-retrofit", "affected_items": [{"code": "ENERGI-TOTAL-EL", "quantity": 180000, "unit_cost": 1.0}], "claimed_saving_nok": 30000, } ) _VERDICT_INPUT = {"decision": "approved", "rationale": "expert reviewed (sim)"} @pytest.fixture(autouse=True) def _isolate_model_env(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("PORTFOLIO_MODEL_MAP", raising=False) monkeypatch.delenv("PORTFOLIO_FOUNDRY_PROJECT_ENDPOINT", raising=False) def _copy(src: Path, tmp_path: Path, name: str) -> Path: dst = tmp_path / name shutil.copytree(src, dst) return dst def _declare(path: Path, value: str) -> None: """Insert a ``bundle_id`` line into an existing frontmatter block, after the opening ``---``.""" lines = path.read_text(encoding="utf-8").splitlines(keepends=True) assert lines[0].startswith("---"), f"{path} has no frontmatter block to declare into" lines.insert(1, f"bundle_id: {value}\n") path.write_text("".join(lines), encoding="utf-8") def _concepts(base: Path) -> list[Path]: return [p for p in sorted(base.rglob("*.md")) if p.name != "index.md"] def _factory(reply: str = _REPLY) -> Any: def factory(role: str) -> Any: return SyntheticUsageChatClient(default_reply=reply) return factory # --- (a)/(b) the slacken itself ------------------------------------------------------------------- def test_a_base_whose_declared_id_disagrees_with_its_mount_opens(tmp_path: Path) -> None: """(a) The headline. This exact shape raised ``BundleIdMismatch`` before today, at every door.""" base = _copy(_BYGG, tmp_path, "a-mount-name-nobody-declared") _declare(base / "index.md", _DECLARED) resolved = okf.reconcile_bundle_id(str(base)) assert resolved.id == _DECLARED, "the declared id is the identity; the mount is an accident" assert resolved.origin == "declared-index" assert resolved.mount == "a-mount-name-nobody-declared", ( "the mount must be CARRIED, not discarded: the warning line names both, and a resolver " "that kept only the winner could not say what it overrode" ) def test_the_mount_answers_only_when_nothing_is_declared() -> None: """(b) The control. Every base shipped in this repo takes this branch, so without it arm (a) could pass against a resolver that had simply stopped consulting the mount at all.""" resolved = okf.reconcile_bundle_id(str(_BYGG)) assert (resolved.id, resolved.origin, resolved.mount) == ( _BYGG.name, "mount-derived", _BYGG.name, ) # --- (c)/(d) what still refuses, and what deliberately does not ----------------------------------- def test_two_concepts_declaring_different_ids_are_refused_by_name(tmp_path: Path) -> None: """(c) The REAL collision, and the one no mount name can settle. Both ids must be named: an operator holding a base that cannot say what it is needs to know which two answers it gave.""" base = _copy(_BYGG, tmp_path, "two-minds") first, second = _concepts(base)[:2] _declare(first, "corpus-alfa") _declare(second, "corpus-beta") with pytest.raises(okf.BundleIdMismatch) as excinfo: okf.assert_declared_ids_agree(okf.navigate_bundle(str(base))) message = str(excinfo.value) assert "corpus-alfa" in message and "corpus-beta" in message assert issubclass(okf.BundleIdMismatch, ValueError), ( "it must land on the CLI's refusal tuple and hosting's 400 arm, never the crash channel" ) def test_concepts_agreeing_with_each_other_but_not_with_the_index_is_not_a_collision( tmp_path: Path, ) -> None: """(d) B1's precedence, applied to the agreement set: an index that disagrees with its concepts is the FALLBACK LOSING, not two concepts colliding. Fold the index into the set and every K2-shaped base -- the whole point of the slacken -- newly refuses.""" base = _copy(_BYGG, tmp_path, "index-out-of-step") _declare(base / "index.md", "what-the-index-thinks") for concept in _concepts(base): _declare(concept, "what-the-concepts-think") okf.assert_declared_ids_agree(okf.navigate_bundle(str(base))) # must not raise # The control that proves the fixture COULD have collided: the two values really are different, # so a green pass above is the rule working rather than a fixture with nothing to disagree about. assert okf.reconcile_bundle_id(str(base)).id == "what-the-index-thinks" # --- (e)/(f) the run records which source answered ------------------------------------------------ async def test_the_run_stamps_which_source_answered_and_completes( tmp_path: Path, fresh_store: Any ) -> None: """(e) End-to-end: the base that used to be refused now RUNS, and the stamp says the id was declared rather than guessed from the mount. Both halves matter -- a run that refused would fail on the first assert, and a run that stamped nothing on the second.""" base = _copy(_RUNNABLE, tmp_path, "mounted-under-another-name") _declare(base / "index.md", _DECLARED) result = await run_project( _RUNNABLE_PID, "local", docs_dir=str(base), bundle_dir=str(base), verdict_input=_VERDICT_INPUT, client_factory=_factory(), store=fresh_store, ) source = result.provenance.bundle_id_source assert source is not None assert (source.id, source.origin, source.mount) == ( _DECLARED, "declared-index", "mounted-under-another-name", ) async def test_the_road_path_stamps_no_bundle_identity_at_all( docs_dir: Any, fresh_store: Any ) -> None: """(e, control) A run with no knowledge base has no bundle identity, and says so by ABSENCE rather than by inventing one. Without this arm a constant ``declared-index`` would pass above.""" result = await run_project( "FV42-GSV-E1", "local", docs_dir=docs_dir, verdict_input=_VERDICT_INPUT, client_factory=_factory( json.dumps( { "measure": "Reduce scope", "affected_items": [{"code": "05.2", "quantity": 4300.0, "unit_cost": 215.0}], "claimed_saving_nok": 200000.0, } ) ), store=fresh_store, ) assert result.provenance.bundle_id_source is None async def test_the_dry_run_carries_the_same_fact(tmp_path: Path) -> None: """(f) ``cost_baseline_anchored``'s precedent: a dry run stops before any stamp exists, so the fact has to leave on the report or that surface goes silent again.""" base = _copy(_RUNNABLE, tmp_path, "dry-run-mount") _declare(base / "index.md", _DECLARED) report = await run_project( _RUNNABLE_PID, "local", docs_dir=str(base), bundle_dir=str(base), client_factory=_factory(), live_dry_run=True, ) source = getattr(report, "bundle_id_source", None) assert source is not None and source.id == _DECLARED and source.mount == "dry-run-mount" # --- (g)/(h) the warning line --------------------------------------------------------------------- def test_the_notice_fires_on_disagreement_and_is_silent_on_agreement() -> None: """(g) ONE renderer over the ALREADY-RESOLVED value, and ``None`` when there is nothing to warn about -- omission, never an empty row (``cost_baseline_notice``'s rule). Both names must appear: a warning that says only "mismatch" leaves the operator to go and look.""" disagreeing = okf.ResolvedBundleId( id="declared-name", origin="declared-index", mount="mount-name" ) agreeing = okf.ResolvedBundleId(id="same", origin="declared-index", mount="same") line = bundle_id_notice(disagreeing) assert line is not None and "declared-name" in line and "mount-name" in line assert bundle_id_notice(agreeing) is None assert bundle_id_notice(None) is None, "a run with no knowledge base has nothing to warn about" def test_the_cli_warns_about_a_disagreeing_mount(tmp_path: Path, capsys: Any) -> None: """(h) The behavioural half: a renderer nothing calls is a line nobody reads. Driven through ``--live-dry-run`` so the arm costs one navigation and no model call.""" base = _copy(_RUNNABLE, tmp_path, "cli-mount-name") _declare(base / "index.md", _DECLARED) rc = run.main( [_RUNNABLE_PID, "--docs-dir", str(base), "--bundle-dir", str(base), "--live-dry-run"] ) out = capsys.readouterr().out assert rc == 0, "a declared id that disagrees with the mount must no longer refuse" assert _DECLARED in out and "cli-mount-name" in out # --- (i)-(k) one naming scheme across the two doors ----------------------------------------------- def test_explore_addresses_a_base_by_its_declared_id(tmp_path: Path) -> None: """(i) The consequence that keeps ``explore()`` -> ``run_mandate_across_bundles`` connected: a minted approach names the base by the id the catalogue offered, and the dispatcher routes by the declared one. Two schemes here means an exploration's own mandate is unroutable.""" base = _copy(_BYGG, tmp_path, "explore-mount-name") _declare(base / "index.md", _DECLARED) tools = {t.name: t for t in explore.navigator_tools((str(base),))} catalogue = tools["list_bundles"].func() assert [entry["id"] for entry in catalogue] == [_DECLARED] opened = tools["read_bundle"].func(bundle_id=_DECLARED) assert opened and "refused" not in opened, "the declared id must OPEN the base" # REWRITTEN, not weakened (F99-D3): the tool returns its refusal instead of raising it. Both # halves stand -- the mount name does NOT open the base, and the answer is a refusal rather # than a listing, which is what "two schemes" would have produced. refusal = tools["read_bundle"].func(bundle_id="explore-mount-name") assert "refused" in refusal and "directories" not in refusal def test_read_bundle_refuses_a_base_that_cannot_say_what_it_is(tmp_path: Path) -> None: """(j) The refusal reaches the door that OPENS a base, not just the helper. Detach the call and an exploration reads a base whose concepts name two different corpora.""" base = _copy(_BYGG, tmp_path, "two-minds-explored") first, second = _concepts(base)[:2] _declare(first, "corpus-alfa") _declare(second, "corpus-beta") tools = {t.name: t for t in explore.navigator_tools((str(base),))} # REWRITTEN, not weakened (F99-D3): the refusal is returned, and it is pinned to the KIND, so # a base whose concepts name two corpora is still told apart from one that merely does not # exist -- the distinction ``pytest.raises(okf.BundleIdMismatch)`` used to carry. refusal = tools["read_bundle"].func(bundle_id="two-minds-explored") assert refusal["refusal"] == okf.BundleIdMismatch.__name__ assert "directories" not in refusal async def test_two_mounts_declaring_one_id_collide_in_the_dispatcher(tmp_path: Path) -> None: """(k) NEWLY REACHABLE, and said out loud: this refusal used to need two mounts with the same basename. With declared-wins, two differently-named directories claiming one id collide -- the same code, no longer a defensive branch.""" first = _copy(_BYGG, tmp_path, "mount-one") second = _copy(_BYGG, tmp_path, "mount-two") _declare(first / "index.md", _DECLARED) _declare(second / "index.md", _DECLARED) with pytest.raises(MandateRoutingError, match=_DECLARED): await run.run_mandate_across_bundles( Mandate( objective="o", approaches=(Approach(id="a", label="A", bundle_id=_DECLARED),), allow_own_proposals=False, ), (str(first), str(second)), ) # --- (l) the field has no default ----------------------------------------------------------------- def test_the_stamp_field_has_no_default() -> None: """(l) ``cost_baseline_anchored``'s rule: a stamp that forgot to say which corpus it judged must not construct at all. ``None`` is a VALUE here (the road path), which is precisely why the absence of the field cannot be allowed to mean it.""" with pytest.raises(Exception): ProvenanceStamp( # type: ignore[call-arg] citations=[ Citation(file="f", snippet="s", locator=TextSpan(start_index=0, end_index=1)) ], model="m", role="proposer", validator_decision="validated", token_usage=0, cost_baseline_anchored=True, )