open/portfolio-optimiser
1
0
Fork 0
Code Issues Releases 3 Activity

feat(s2c): debatten navigerer basen i stedet for aa faa den utlevert [skip-docs]

Browse source 
MAJOR-3/S7a-3 gjorde utforskningen billig og lot pipelinen staa. Maalt paa K2
(630 konsepter, S7bs eget instrument, kjent-positiv-kontrollen reprodusert
eksakt FOER bruk): okf.bundle_context er 648 962 o200k-tokens og rir i TRE
kopier = 1 947 342 = 99,1 % av en kjoerings prompt-tokens.

Et premiss i maaledokumentet ble presisert foerst: de tre kopiene er tre
DEBATT-turer (proposer x2, checker x1), mens genererings-prompten er 156
tokens, fordi gen_context = debate_output or context. Det avgjorde formen -
generering trengte ingen egen soem, for aa binde `context` binder
siste-utvei-fallbacken ved konstruksjon.

run_project sender naa en PEKER (fast tekst + erklaert bundle_id + antall
konseptdokumenter i scope + stigen, O(1) i korpuset) og gir debatten de SAMME
fire verktoeyene utforskningen bruker - explore.navigator_tools gjenbrukt,
aldri en andre kopi av policyen.

Etter: 753 tokens like-for-like (samme manus, samme fire prompter, -99,96 %)
og 8 942 med en debatt som faktisk gaar stigen (-99,5 %), mot operatoerens
terskel 195 000 = 4,6 % av taket. Validert besparelse og validatorens dom er
UENDRET (850 000 NOK av 3 852 500, 2 av 5 felt paa stage 4 og 5, samme
dom-noekkel), og utforskningens 18 355 er uendret til tokenet.

§4.1a maatte flytte, ikke forsvinne: dimensjonsfilteret bodde i renderingen og
bor naa i VERKTOEYENE, paa begge trinn - en listing som skjuler et fremmed
dokument mens read_file serverer det paa sti er et filter i navnet alene.
okf.in_dimension er eneste predikat.

Sporet er kaller-eid (ExplorationToolRecorder -> RunResult.debate_tool_calls ->
{run_id}-debate.json fra en finally) og skrives ogsaa TOMT: en debatt som
navigerer ingenting ER S2c-regresjonen, saa den maa kunne leses.

Load-bearing MAALT: aatte mutasjoner roede mot HELE suiten, groenn kontroll
1306/5 (fra 1295/5), golden demo-transcript.stdout BYTE-UENDRET
(shasum -a 1 av innholdet = ea8c534773acdbe41ae68f2c55724d69aaf8be4f).
M7 falsifiserte seg selv, ikke gaten - staar som maalt.

Maaling: docs/2026-09-04-s2c-debatt-k2.md

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kjell Tore Guttormsen 2026-09-04 18:04:21 +02:00
commit da5f10f140
13 changed files with 1044 additions and 98 deletions
Download patch file Download diff file Expand all files Collapse all files

7
tests/test_a5_per_approach_artifacts_loadbearing.py
View file

@ -27,7 +27,8 @@ Detach points, each RED on its own:
* stamp every per-approach artefact with the run's single verdict id -> judging one clears all.
The control (``test_without_a_mandate_the_outbox_is_byte_unchanged``) proves the addition is inert
on the no-mandate path: the same two filenames as before, carrying no ``approach_id`` key at all.
on the no-mandate path: the same un-suffixed filenames as before, carrying no ``approach_id`` key at
all.
"""
from __future__ import annotations
@ -257,8 +258,12 @@ async def test_without_a_mandate_the_outbox_is_byte_unchanged(tmp_path: Path) ->
outbox = tmp_path / "outbox"
await _run(None, outbox)
# ``-debate.json`` is S2c's tool trace, written on every bundle-path run with an outbox and
# unrelated to the per-approach keying this control is about; it is listed so the assertion
# stays an EXACT set rather than degrading into "at least these".
written = sorted(p.name for p in outbox.glob("*.json"))
assert written == [
f"{_RUN_ID}-debate.json",
f"{_RUN_ID}-outcome.json",
f"{_RUN_ID}-proposal.json",
f"{_RUN_ID}-runconfig.json",

384
tests/test_debate_navigation_cost_loadbearing.py Normal file
View file

@ -0,0 +1,384 @@
"""S2c LOAD-BEARING: the DEBATE navigates the knowledge base; it is never handed the whole of it.
MAJOR-3 and S7a-3 made the EXPLORATION cheap — 18 355 o200k tokens over the whole of K2 — and
left the pipeline's own two phases stuffing. Measured on K2 (630 concepts,
``docs/2026-09-04-syretest-s7b-k2.md`` § 3.4, re-measured with the same instrument before this
change): ``okf.bundle_context`` is **648 962 o200k tokens** and rides in **three** prompts (two
proposer turns + the checker's), so debate + generation is **1 947 342 tokens = 99,1 %** of a
run's prompt cost. Ninety-nine percent of what a run pays for is context nobody asked for twice.
The move is S7a-3's own, one seam over: ``run_project``'s bundle path stops rendering the base
into the task message and instead hands the debate the SAME four navigator tools the exploration
uses (``navigator_tools``) plus a POINTER naming the base. Generation is covered by the same
change: ``gen_context = debate_output or context``, so the last-resort fallback is bounded by
construction rather than being the whole corpus.
Arms, each with a named detach point:
* **(a) no prompt carries the base.** A sentinel that lives only in a concept body is ABSENT from
every prompt the debate and the generation call see. CONTROL: the same sentinel IS in
``bundle_context``, so its absence is caused by the seam and not by an empty fixture.
* **(b) bounded, and not vacuously so.** Every prompt is under a ceiling that lives HERE, never in
``run.py`` (the ``read_bundle``/catalogue rule: raising the constant is the regression the gate
exists to catch), while ``bundle_context`` for the same base is over FIVE times it — without the
flat control a green bound could just mean the fixture is small. And the pointer must NAME the
base, because a bounded prompt that omits the id is a debate that cannot call the tools at all.
* **(c) the tools are actually there.** The four navigator tools reach the built workflow.
* **(d) the tool trace is a first-class artefact.** ``RunResult.debate_tool_calls`` and
``{run_id}-debate.json`` carry name + bundle_id + path in CALL ORDER — S7a-3 pkt. 3's rule, on
the second surface that now opens a base: over 629 concepts a trace reading ``read_file`` twice
answers nothing about which two.
* **(e) the dimension scope survives the move.** §4.1a promised the agents read ONLY
dimension-matched bundle knowledge. That promise used to be kept by ``bundle_context``'s filter;
with navigation it has to be kept by the TOOLS, on BOTH rungs — a listing that hides a foreign
document while ``read_file`` still serves it is a filter in name only.
"""
from __future__ import annotations
import json
import shutil
from collections.abc import Callable
from pathlib import Path
from typing import Any
import pytest
from agent_framework import BaseChatClient
from portfolio_optimiser import okf
from portfolio_optimiser.dimension import Dimension
from portfolio_optimiser.explore import DimensionScopeRefused, navigator_tools
from portfolio_optimiser.run import run_project
from portfolio_optimiser.simulation import ScriptedChatClient, scripted_factory
BUNDLE_DIR = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro"
_PID = "BYGG-KONTOR-NORD"
_VERDICT_INPUT = {"decision": "approved", "rationale": "expert reviewed (sim)"}
#: The prompt budget one debate/generation call may spend on bundle context, in CHARACTERS. It
#: lives in the TEST for ``_CATALOGUE_EXCERPT_CHARS``' reason: a ceiling imported from the
#: implementation moves with it, and raising it is precisely the regression this gate catches.
#: Characters rather than o200k tokens because ``tiktoken`` is not a project dependency, and a gate
#: that skips when an optional package is missing is a gate that can be silently absent
#: (MAJOR-3's own stated deviation, same reason).
_CEILING_CHARS = 1_500
_VALID_REPLY = (
'{"measure":"LED-retrofit","affected_items":'
'[{"code":"ENERGI-TOTAL-EL","quantity":300000,"unit_cost":1.0}],'
'"claimed_saving_nok":30000}'
)
_CHECKER_REPLY = "Reasoning holds.\nVERDICT: APPROVE"
def _prompt_blob(messages: Any) -> str:
"""Everything a prompt actually carries: text PLUS ``function_call``/``function_result``
contents. ``Message.text`` alone measures a context-bearing prompt at a few characters — the
corrected S7a-2 instrument, and the reason S2c could be measured at all."""
parts: list[str] = []
for message in messages:
text = getattr(message, "text", "") or ""
if text:
parts.append(text)
for content in getattr(message, "contents", ()) or ():
for attr in ("result", "arguments"):
value = getattr(content, attr, None)
if value:
parts.append(str(value))
return "\n".join(parts)
def _recording_factory(
sink: list[str], *, script: dict[str, Any] | None = None
) -> Callable[[str], BaseChatClient]:
"""A scripted client factory whose every prompt lands in ``sink`` as the FULL blob."""
def factory(role: str) -> BaseChatClient:
if script is not None and role in script:
client: BaseChatClient = scripted_factory(script, [])(role)
else:
client = ScriptedChatClient(
_CHECKER_REPLY if role == "checker" else _VALID_REPLY, role=role
)
original = client._inner_get_response # type: ignore[attr-defined]
def recording(*, messages, options, stream=False, **kwargs): # type: ignore[no-untyped-def]
sink.append(_prompt_blob(messages))
return original(messages=messages, options=options, stream=stream, **kwargs)
client._inner_get_response = recording # type: ignore[attr-defined,method-assign]
return client
return factory
async def _run(**kwargs: Any) -> tuple[Any, list[str]]:
sink: list[str] = []
script = kwargs.pop("script", None)
result = await run_project(
_PID,
"local",
docs_dir=str(BUNDLE_DIR),
bundle_dir=str(BUNDLE_DIR),
verdict_input=_VERDICT_INPUT,
client_factory=_recording_factory(sink, script=script),
**kwargs,
)
return result, sink
def _sentinel_from_the_base() -> str:
"""A string that exists ONLY inside a concept body of the fixture base — the leak probe."""
bundle = okf.navigate_bundle(str(BUNDLE_DIR))
reference = next(f for f in bundle.context_files if f.name == "kilder-realiseringsgap.md")
line = next(row for row in reference.body.splitlines() if len(row.strip()) > 60)
return line.strip()
# ---------------------------------------------------------------- (a) nothing carries the base
async def test_no_debate_or_generation_prompt_carries_the_whole_base() -> None:
"""LOAD-BEARING (a): a sentinel living only in a concept body reaches NO prompt.
Detach point: restore ``context = okf.bundle_context(bundle, ...)`` in ``run_project``'s
bundle arm → RED (the sentinel is back in all three prompts)."""
sentinel = _sentinel_from_the_base()
_, sink = await _run()
assert sink, "the debate never ran — no prompt was captured"
leaking = [i for i, prompt in enumerate(sink) if sentinel in prompt]
assert not leaking, (
f"the whole knowledge base is still being stuffed into prompt(s) {leaking}: "
"the debate is handed the corpus instead of navigating it"
)
def test_the_sentinel_is_really_in_the_base() -> None:
"""CAUSALITY CONTROL for (a): the sentinel IS what ``bundle_context`` renders, so its absence
above is caused by the seam rather than by a fixture that never held it."""
sentinel = _sentinel_from_the_base()
assert sentinel in okf.bundle_context(okf.navigate_bundle(str(BUNDLE_DIR)))
# ------------------------------------------------------------------- (b) bounded, not vacuous
async def test_every_prompt_stays_under_the_ceiling() -> None:
"""LOAD-BEARING (b): the debate + generation prompts are bounded by the POINTER, so their cost
follows the number of bases configured (which the operator chose) and not the size of the
corpus (which they did not)."""
_, sink = await _run()
worst = max(len(prompt) for prompt in sink)
assert worst <= _CEILING_CHARS, (
f"the widest debate/generation prompt is {worst} characters, over the {_CEILING_CHARS} "
"ceiling — the corpus is riding along again"
)
def test_the_flat_form_is_far_over_the_ceiling() -> None:
"""FLAT CONTROL for (b): ``bundle_context`` on the SAME base is over five times the ceiling, so
a green bound above cannot just mean the fixture is small."""
whole = okf.bundle_context(okf.navigate_bundle(str(BUNDLE_DIR)))
assert len(whole) > 5 * _CEILING_CHARS, (
"the fixture base is too small for the bound to prove anything"
)
async def test_the_pointer_names_the_base_the_tools_take() -> None:
"""LOAD-BEARING (b, second half): a bounded prompt that does not NAME the base is a debate
that cannot call a single tool — bounded and useless is the vacuous form of this gate.
Detach point: drop the bundle id from the pointer → RED."""
_, sink = await _run()
bundle_id = okf.reconcile_bundle_id(str(BUNDLE_DIR)).id
assert any(bundle_id in prompt for prompt in sink), (
f"no prompt names the knowledge base {bundle_id!r}; the navigator tools take that id, so "
"the debate has been given a bounded prompt it cannot act on"
)
# ------------------------------------------------------------------------- (c) the tools exist
async def test_the_debate_is_given_the_navigator_tools(monkeypatch) -> None:
"""LOAD-BEARING (c): the bundle path hands the debate the SAME four tools the exploration uses.
Detach point: drop ``navigator_tools(...)`` from ``debate_tools`` → RED."""
import portfolio_optimiser.run as run_module
captured: list[list[Any]] = []
original = run_module.fresh_workflow
def spy(*args: Any, **kwargs: Any) -> Any:
captured.append(list(kwargs.get("tools") or []))
return original(*args, **kwargs)
monkeypatch.setattr(run_module, "fresh_workflow", spy)
await _run()
assert captured, "the debate was never built"
names = {getattr(t, "name", "") for t in captured[0]}
assert {"list_bundles", "read_bundle", "read_dir", "read_file"} <= names, (
f"the debate's tool list is {sorted(names)} — it cannot navigate the base it was pointed at"
)
# ------------------------------------------------------------------------- (d) the tool trace
async def test_the_debate_tool_trace_reaches_the_result_and_the_outbox(tmp_path) -> None:
"""LOAD-BEARING (d): what the debate OPENED is recorded, in call order, with the path.
The proposer is driven by a step MANUSCRIPT (MAJOR-1 b) — the only offline form that can make a
scripted role emit a ``function_call``, and therefore the only way a free rehearsal can prove
the debate opens anything at all.
Detach points, each RED on its own: drop the ``ExplorationToolRecorder`` from the debate's
middleware; stop writing ``{run_id}-debate.json``; drop ``path`` from the payload."""
bundle_id = okf.reconcile_bundle_id(str(BUNDLE_DIR)).id
outbox_dir = tmp_path / "outbox"
result, _ = await _run(
outbox_dir=str(outbox_dir),
run_id="s2c",
script={
"proposer": [
{"call": "read_bundle", "args": {"bundle_id": bundle_id}},
{
"call": "read_file",
"args": {"bundle_id": bundle_id, "path": "metode-ipmvp-a.md"},
},
_VALID_REPLY,
],
"checker": _CHECKER_REPLY,
},
)
observed = [(c.name, c.bundle_id, c.path) for c in result.debate_tool_calls]
assert observed[:2] == [
("read_bundle", bundle_id, ""),
("read_file", bundle_id, "metode-ipmvp-a.md"),
], f"the debate's tool trace is {observed} — the call sequence is not recorded as it happened"
payload = json.loads((outbox_dir / "s2c-debate.json").read_text(encoding="utf-8"))
assert payload["tool_calls"][:2] == [
{"name": "read_bundle", "bundle_id": bundle_id, "path": ""},
{"name": "read_file", "bundle_id": bundle_id, "path": "metode-ipmvp-a.md"},
], f"the artefact does not carry the call sequence: {payload['tool_calls']}"
async def test_a_debate_that_opened_nothing_says_so(tmp_path) -> None:
"""CONTROL for (d): a run whose agents called no tool leaves an EMPTY trace rather than no
artefact — "the debate never opened the base" is the S2c regression signal itself, so it must
be readable off the leaving, not inferred from a missing file."""
outbox_dir = tmp_path / "outbox"
result, _ = await _run(outbox_dir=str(outbox_dir), run_id="s2c-quiet")
assert result.debate_tool_calls == ()
payload = json.loads((outbox_dir / "s2c-quiet-debate.json").read_text(encoding="utf-8"))
assert payload["tool_calls"] == []
# ------------------------------------------------------------------ (e) the dimension survives
_ENERGY_DIM = Dimension(
id="energi", label="Energi", allowed_measure_types=frozenset({"energy_efficiency"})
)
_ASFALT_SENTINEL = "ASFALT-LEAK-SENTINEL-x7y8z9"
_ASFALT_FILE = "asfalt-dekke.md"
def _bundle_with_a_foreign_dimension(tmp_path: Path) -> str:
"""A copy of the fixture base plus ONE concept file marked ``dimension: asfalt``, linked from
the index so navigation reaches it."""
copy = tmp_path / "bundle"
shutil.copytree(BUNDLE_DIR, copy)
(copy / _ASFALT_FILE).write_text(
f"---\ntype: reference\ntitle: Asfaltdekke\ndimension: asfalt\n---\n\n{_ASFALT_SENTINEL}\n",
encoding="utf-8",
)
index = copy / "index.md"
index.write_text(
index.read_text(encoding="utf-8") + f"\n- [Asfaltdekke]({_ASFALT_FILE})\n", encoding="utf-8"
)
return str(copy)
def _tools(bundle_dir: str, dimension: str | None) -> dict[str, Any]:
return {t.name: t for t in navigator_tools([bundle_dir], dimension=dimension)}
async def _invoke(tool: Any, **arguments: Any) -> str:
"""A tool's answer as text. ``FunctionTool.invoke`` returns ``[Content]``, so a test that
stringified the list would compare object reprs and pass against anything."""
return "".join(getattr(c, "text", "") or "" for c in await tool.invoke(arguments=arguments))
async def test_a_foreign_dimension_document_is_neither_listed_nor_readable(tmp_path) -> None:
"""LOAD-BEARING (e): under a dimension the navigator can neither SEE nor READ a document from
another one — both rungs, because a listing filter alone is a filter in name only.
The tools are called DIRECTLY: a ``ScriptedChatClient`` returns text and never emits a tool
call, so a test that only drove ``run_project`` would leave the whole tool surface outside the
gate (``test_explore_loadbearing``'s own measured correction).
Detach points: drop ``dimension`` from ``directory_listing``; drop the gate in ``read_file``."""
bundle_dir = _bundle_with_a_foreign_dimension(tmp_path)
bundle_id = okf.reconcile_bundle_id(bundle_dir).id
tools = _tools(bundle_dir, "energi")
listing = await _invoke(tools["read_bundle"], bundle_id=bundle_id)
assert _ASFALT_FILE not in listing, (
"a document from another dimension is still listed to the agents"
)
with pytest.raises(DimensionScopeRefused):
await _invoke(tools["read_file"], bundle_id=bundle_id, path=_ASFALT_FILE)
async def test_without_a_dimension_the_same_document_is_listed_and_readable(tmp_path) -> None:
"""CAUSALITY CONTROL for (e): with ``dimension=None`` the SAME file is both listed and read, so
its refusal above is caused by the scope and not by the file being unreachable."""
bundle_dir = _bundle_with_a_foreign_dimension(tmp_path)
bundle_id = okf.reconcile_bundle_id(bundle_dir).id
tools = _tools(bundle_dir, None)
listing = await _invoke(tools["read_bundle"], bundle_id=bundle_id)
assert _ASFALT_FILE in listing
assert _ASFALT_SENTINEL in await _invoke(
tools["read_file"], bundle_id=bundle_id, path=_ASFALT_FILE
)
async def test_a_dimension_scoped_run_gives_the_debate_scoped_tools(tmp_path, monkeypatch) -> None:
"""LOAD-BEARING (e, wiring): ``run_project``'s ``dimension`` reaches the TOOLS, not just the
(now absent) rendered context — otherwise §4.1a's promise is kept by nothing at all.
Detach point: build the debate's navigator tools without ``dimension=`` → RED."""
import portfolio_optimiser.run as run_module
bundle_dir = _bundle_with_a_foreign_dimension(tmp_path)
bundle_id = okf.reconcile_bundle_id(bundle_dir).id
captured: list[list[Any]] = []
original = run_module.fresh_workflow
def spy(*args: Any, **kwargs: Any) -> Any:
captured.append(list(kwargs.get("tools") or []))
return original(*args, **kwargs)
monkeypatch.setattr(run_module, "fresh_workflow", spy)
sink: list[str] = []
await run_project(
_PID,
"local",
docs_dir=bundle_dir,
bundle_dir=bundle_dir,
dimension=_ENERGY_DIM,
verdict_input=_VERDICT_INPUT,
client_factory=_recording_factory(sink),
)
read_file = next(t for t in captured[0] if getattr(t, "name", "") == "read_file")
with pytest.raises(DimensionScopeRefused):
await _invoke(read_file, bundle_id=bundle_id, path=_ASFALT_FILE)

97
tests/test_dimension_loadbearing.py
View file

@ -6,10 +6,17 @@ things, each with a named detach point:
The proposal validates on the numbers, so the ONLY possible rejecter is the ``admits`` scope
gate (closes the green-but-dead trap). RED if ``admits`` is removed ⇒ the foreign candidate
slips through. Control: an in-dimension candidate passes.
- **Context scope (§4.1a):** a dimension-scoped ``run_project`` feeds ONLY dimension-matched bundle
text into the agent prompt — a sentinel from ANOTHER dimension's concept file is ABSENT from the
captured prompt. RED if the ``dimension=`` arg to ``bundle_context`` is dropped ⇒ the foreign-
dimension context leaks in. Control: ``dimension=None`` ⇒ the sentinel is present.
- **Context scope (§4.1a), as it stands after S2c:** the promise is unchanged — a dimension-scoped
run lets the agents read ONLY dimension-matched bundle knowledge — but the debate no longer
RECEIVES a rendered context, it NAVIGATES the base, so the filter had to move from
``bundle_context`` to the tools the debate is handed. These two arms gate the FIRST rung: the
listing the debate can see never names a foreign-dimension document. RED if the ``dimension=``
arg is dropped anywhere along ``run_project`` → ``navigator_tools`` → ``directory_listing``.
Control: ``dimension=None`` ⇒ the same document IS listed. The SECOND rung (``read_file``
refusing a foreign document by path) is gated in
``tests/test_debate_navigation_cost_loadbearing.py`` — a listing filter with an ungated reader
behind it is a filter in name only, so the two halves get their own arms and their own
mutations.
Patterns: ``test_checker_gate_loadbearing.py:59/87`` (gate + causality control),
``conftest.py:184`` (recording client), ``test_step8_promotion_loadbearing.py:51`` (bundle copy).
@ -21,9 +28,11 @@ import shutil
from collections.abc import Callable
from pathlib import Path
import pytest
from agent_framework import BaseChatClient
from conftest import SyntheticUsageChatClient
from portfolio_optimiser import okf
from portfolio_optimiser.dimension import Dimension
from portfolio_optimiser.run import run_project
from portfolio_optimiser.validator import Rejection, ValidatedProposal
@ -128,54 +137,72 @@ def _bundle_with_asfalt_file(tmp_path: Path) -> str:
return str(dst)
async def _run_and_capture(bundle_dir: str, dimension: Dimension | None) -> str:
"""Run the bundle path with a prompt-recording client and return the concatenated prompt text
that reached the agents."""
sink: list[str] = []
async def _listing_the_debate_can_see(bundle_dir: str, dimension: Dimension | None) -> str:
"""Run the bundle path, capture the tools the debate was built with, and return what its
``read_bundle`` rung answers — the S2c successor to reading the prompt text.
The tool is invoked DIRECTLY afterwards rather than through an agent turn, because a
``ScriptedChatClient`` returns text and never emits a ``function_call``: a test that only drove
the run would leave the entire tool surface outside the gate (``test_explore_loadbearing``'s
own measured correction)."""
import portfolio_optimiser.run as run_module
captured: list[list[object]] = []
original = run_module.fresh_workflow
def spy(*args: object, **kwargs: object) -> object:
captured.append(list(kwargs.get("tools") or [])) # type: ignore[arg-type]
return original(*args, **kwargs) # type: ignore[arg-type]
def factory(role: str) -> BaseChatClient:
client = SyntheticUsageChatClient(
return SyntheticUsageChatClient(
default_reply=_valid_reply("energy_efficiency", "ENERGI-TOTAL-EL")
)
_orig = client._inner_get_response
def _recording(*, messages, stream, options, **kwargs): # type: ignore[no-untyped-def]
sink.append(" ".join(getattr(m, "text", "") or "" for m in messages))
return _orig(messages=messages, stream=stream, options=options, **kwargs)
monkeypatch = pytest.MonkeyPatch()
try:
monkeypatch.setattr(run_module, "fresh_workflow", spy)
await run_project(
"BYGG-KONTOR-NORD",
"local",
docs_dir=bundle_dir,
bundle_dir=bundle_dir,
verdict_input=_VERDICT_INPUT,
dimension=dimension,
client_factory=factory,
)
finally:
monkeypatch.undo()
client._inner_get_response = _recording # type: ignore[method-assign]
return client
await run_project(
"BYGG-KONTOR-NORD",
"local",
docs_dir=bundle_dir,
bundle_dir=bundle_dir,
verdict_input=_VERDICT_INPUT,
dimension=dimension,
client_factory=factory,
)
return " ".join(sink)
assert captured, "the debate was never built"
read_bundle = next(t for t in captured[0] if getattr(t, "name", "") == "read_bundle")
bundle_id = okf.reconcile_bundle_id(bundle_dir).id
answer = await read_bundle.invoke(arguments={"bundle_id": bundle_id})
return "".join(getattr(c, "text", "") or "" for c in answer)
async def test_dimension_scopes_the_agent_context(tmp_path) -> None:
"""LOAD-BEARING (§4.1a): a dimension-scoped run feeds ONLY dimension-matched bundle text into
the prompt — the asfalt sentinel is ABSENT. RED if the ``dimension=`` arg to ``bundle_context``
is dropped (the foreign-dimension context leaks into the prompt)."""
"""LOAD-BEARING (§4.1a, first rung): under a dimension the listing the debate can see does NOT
name the foreign-dimension document. RED if ``dimension=`` is dropped anywhere between
``run_project`` and ``directory_listing``.
Asserts on the FILE NAME rather than the body sentinel, because a listing carries names and
sizes, never bodies — an assert on the sentinel would be green against every implementation
and would prove nothing (the vacuous form this repo keeps measuring)."""
bundle_dir = _bundle_with_asfalt_file(tmp_path)
scoped_prompt = await _run_and_capture(bundle_dir, _ENERGY_DIM)
assert _ASFALT_SENTINEL not in scoped_prompt, (
"another dimension's context leaked into the prompt — bundle_context is not dimension-scoped"
scoped = await _listing_the_debate_can_see(bundle_dir, _ENERGY_DIM)
assert "asfalt-note.md" not in scoped, (
"another dimension's document is listed to the agents — the tools are not dimension-scoped"
)
async def test_no_dimension_leaves_context_unscoped(tmp_path) -> None:
"""CAUSALITY CONTROL: with ``dimension=None`` the asfalt sentinel IS present — proving its
"""CAUSALITY CONTROL: with ``dimension=None`` the asfalt document IS listed — proving its
absence above is caused by the dimension scope, not by the file being unreachable."""
bundle_dir = _bundle_with_asfalt_file(tmp_path)
full_prompt = await _run_and_capture(bundle_dir, None)
assert _ASFALT_SENTINEL in full_prompt, (
unscoped = await _listing_the_debate_can_see(bundle_dir, None)
assert "asfalt-note.md" in unscoped, (
"the asfalt file is unreachable even without a filter — the control does not prove causality"
)

24
tests/test_mcp_run_loadbearing.py
View file

@ -4,8 +4,8 @@ opened and closed around the debate, and nothing is contacted that was not annou
Krav 3 is only met if the external service is reachable *while the run works*. Three detach points,
each RED on its own:
* drop the MCP tools from ``debate_tools`` -> the agents never get the tool (and on the bundle path
they get NO tools at all, which is what that path had before);
* drop the MCP tools from ``debate_tools`` -> the agents never get the tool (on the bundle path
they are then left with the four navigator tools alone — before S2c that path had none);
* skip the ``AsyncExitStack`` entry -> the tools are constructed but never connected, so they are
present and useless — the failure mode that looks like success;
* let a dry run enter them -> ``--live-dry-run`` would contact a third party while claiming to stop
@ -115,7 +115,8 @@ async def _run(**kwargs: Any):
async def test_configured_server_becomes_a_tool_the_agents_have(fake_tools, captured_tools) -> None:
"""On the bundle path the agents had NO tools at all; a configured server is the first one."""
"""On the bundle path a configured server is the first EXTERNAL tool; since S2c the four
in-process navigator tools sit alongside it."""
await _run(mcp_servers=(_SERVER,))
assert captured_tools, "the debate was never built"
assert any(isinstance(t, _FakeMcpTool) for t in captured_tools[0])
@ -138,10 +139,21 @@ async def test_dry_run_never_contacts_a_configured_server(fake_tools) -> None:
async def test_run_without_mcp_servers_keeps_the_tool_list_unchanged(captured_tools) -> None:
"""CONTROL: with nothing configured the bundle path still hands the agents no tools — the
pre-Trekk-B behaviour, so every assertion above rests on the configuration and not on the run."""
"""CONTROL: with nothing configured the bundle path hands the agents no EXTERNAL tool, so every
assertion above rests on the configuration and not on something the run does anyway.
Before S2c this read ``captured_tools[0] == []`` — the bundle path had no tools at all. It now
navigates its knowledge base, so the control asserts what it always meant: nothing here reaches
outside the process. Asserting the exact navigator set as well keeps it from degrading into
"some tools, whatever they are"."""
await _run()
assert captured_tools[0] == []
assert not any(isinstance(t, _FakeMcpTool) for t in captured_tools[0])
assert {getattr(t, "name", "") for t in captured_tools[0]} == {
"list_bundles",
"read_bundle",
"read_dir",
"read_file",
}
async def test_portfolio_mode_gives_every_project_the_configured_tools(

5
tests/test_parse_failure_capture_loadbearing.py
View file

@ -237,8 +237,13 @@ async def test_a_run_whose_replies_parse_writes_no_artefact(tmp_path: Path) -> N
assert not _artefact(outbox_dir, run_id).exists()
# The pre-existing artefacts are untouched — the addition is inert on the clean path.
# ``-debate.json`` (S2c) is written on EVERY bundle-path run that has an outbox,
# unlike this artefact whose PRESENCE is its signal: a debate that opened nothing is
# the S2c regression itself, so it must be readable rather than inferred from a
# file that is not there.
written = sorted(p.name for p in outbox_dir.iterdir())
assert written == [
f"{run_id}-debate.json",
f"{run_id}-outcome.json",
f"{run_id}-proposal.json",
f"{run_id}-runconfig.json",

33
tests/test_scripted_explore_door_loadbearing.py
View file

@ -442,16 +442,25 @@ def test_a_step_naming_an_unknown_key_is_refused_by_name(tmp_path, capsys) -> No
assert "result" in err, err
def test_a_step_list_is_refused_for_a_debate_role(tmp_path, capsys) -> None:
"""The list form is the EXPLORATION's, and is refused elsewhere by name rather than accepted
and quietly ignored. The debate's proposer is driven by ``generate``'s own call, not by an
agent loop that would invoke a tool between turns, so a script of calls there would describe a
rehearsal that cannot happen.
def test_a_step_list_is_accepted_for_a_debate_role(tmp_path) -> None:
"""REWRITTEN at S2c, not deleted: this arm used to pin the refusal of a step list for the
debate's roles, on the ground that the proposer answers ``generate``'s own call rather than an
agent loop that could invoke a tool between turns. Since S2c the proposer and the checker ARE
agents holding the four navigator tools, so that ground is measurably false and the refusal
would have kept the free half of the measurement ladder away from the very seam S2c builds —
the identical vacuity MAJOR-1 closed for the navigator.
What replaces it is the positive: a scripted proposer opens the base, and the debate's own
artefact names WHICH document. RED if the list form is refused for a debate role again, and RED
if the debate's ``ExplorationToolRecorder`` is detached (the run succeeds, the trace is empty).
"""
replies = _replies_file(
tmp_path,
{
"proposer": [{"call": "retrieve_cost_docs", "args": {"query": "x"}}],
"proposer": [
{"call": "read_bundle", "args": {"bundle_id": "bygg-energi-mikro"}},
_PROPOSER_REPLY,
],
"checker": "VERDICT: APPROVE",
"manager": list(_MANAGER_STAGES),
"navigator": "NAVIGATOR: read the index.",
@ -461,6 +470,12 @@ def test_a_step_list_is_refused_for_a_debate_role(tmp_path, capsys) -> None:
rc = run.main(_explore_argv(tmp_path, replies, "scripted-debate-list"))
assert rc == 1
err = capsys.readouterr().err
assert "run refused" in err and "proposer" in err, err
assert rc == 0
payload = json.loads(
(tmp_path / "outbox" / "scripted-debate-list-debate.json").read_text(encoding="utf-8")
)
assert payload["tool_calls"][0] == {
"name": "read_bundle",
"bundle_id": "bygg-energi-mikro",
"path": "",
}, payload["tool_calls"]