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

test(verdict-gate): read_file skal nekte verdict-laget - fire armer roede FOER impl

Browse source 
Iron Law-steget. Testene foerst; produksjonskoden her er KUN den nakne
exception-deklarasjonen, saa hver arm kan observeres roed for seg i stedet for
aa skjules bak en collection-feil. Ingen gate er bygget.

MAALT foer noe ble skrevet: read_file(bygg-energi-mikro, verdict-led-fro.md)
returnerer alle 2 883 tegn av dommen, og i en debattkjoering der proposeren ber
om den ved sti naar kroppen prompt 1 og 3.

Roedt observert paa fire armer:
  (1) utforskningsveien nekter ikke        DID NOT RAISE
  (2) debattveien nekter ikke              DID NOT RAISE
  (4) kroppen lekker til prompt [1, 3]     - lekkasje-halvdelen, ikke sporet
  (5) ulenket dom nektes ikke              DID NOT RAISE

Groent, som forventet, paa de to bevaringsarmene: (3) den gatede ExpeL-ruten
naar fortsatt hypotese-prompten, og (6) index.md leses fortsatt hel. De er
ikke vakuoese - de er dét som skiller en gate fra en vegg, og de maa staa
groenne baade foer og etter.

Sporet er maalt separat: en nekt fanges av MAF og kjoeringen fortsetter, mens
recorderen registrerer FOER call_next - saa mutasjonen "registrer etter
call_next" toemmer sporet for nettopp et nektet kall (maalt: trace = ()).
Det gjoer arm 4s spor-halvdel til en egen gate, ikke en passasjer paa nekten.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kjell Tore Guttormsen 2026-09-04 20:10:29 +02:00
commit e7d1a31ecf
2 changed files with 349 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

345
tests/test_verdict_layer_refusal_loadbearing.py Normal file
View file

@ -0,0 +1,345 @@
"""The verdict layer is REFUSED by ``read_file``, however the path was found.
The gap S2c stated as an honesty limit and left open (``docs/2026-09-04-s2c-debatt-k2.md``, closing
section): ``read_file`` is path-addressable to the ``type: verdict`` layer. No listing names it —
``Bundle.context_files`` drops it at every level, so neither ``read_bundle`` nor ``read_dir`` nor
``bundle_context`` ever mentions a verdict — but a GUESSED path reached one, and reaching one that
way bypasses the gated ExpeL fold that is the only sanctioned route from a prior expert judgement
into a hypothesis (målbilde §4/§5). MEASURED before this work: ``read_file`` on the fixture base's
``verdict-led-fro.md`` returned all 2 883 characters of it.
The property was inherited from S7a-3 and, since S2c handed the DEBATE the navigator's four tools,
became reachable from the pipeline as well — which is why the rule lives in ONE place, the tool,
rather than in two callers. Rendering cannot hold it: S2c measured that a filter in the listing
while ``read_file`` still serves the bytes is "a filter in name only", and prompt text cannot hold
it either, because a model-chosen path is untrusted input by construction.
Arms, each with a named detach point:
* **(1) the exploration path refuses.** ``navigator_tools`` called directly, ``dimension=None`` —
the exploration's own call. CONTROL: an ordinary concept file in the SAME base still reads, so
the refusal is caused by the layer and not by an unreachable fixture.
* **(2) the debate path refuses.** The same tool as ``run_project`` builds it, driven through the
real seam, plus a BEHAVIOURAL run whose proposer manuscript asks for the verdict by path.
* **(3) the gated route still works.** The seed verdict's realization signal still reaches the
hypothesis prompt through the ExpeL fold — the refusal closes the ungated door, not the door.
* **(4) a refused read is RECORDED, never silent.** The call stands in ``debate_tool_calls`` and in
``{run_id}-debate.json`` with its path, while the verdict's body reaches NO prompt. Recording is
gated on its own: MEASURED, moving the recorder's append after ``call_next`` empties the trace
for exactly this call, because a refused invocation never returns.
* **(5) the rule reads the DOCUMENT, not the walk.** A verdict file no index links to — never
navigated, therefore absent from ``Bundle.verdicts`` — is refused too. "However the path was
found" is the whole point: gating on the navigation would leave the guessed path to an unlinked
judgement open, which is the defect in its own disguise.
* **(6) the ladder is intact.** ``index.md`` and ordinary concept files still read whole.
"""
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.explore import VerdictLayerRefused, navigator_tools
from portfolio_optimiser.run import run_project
from portfolio_optimiser.simulation import ScriptedChatClient, scripted_factory
from portfolio_optimiser.validator import ValidatedProposal
from portfolio_optimiser.verdicts import seed_store_from_bundle
BUNDLE_DIR = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro"
_PID = "BYGG-KONTOR-NORD"
_VERDICT_FILE = "verdict-led-fro.md"
_CONCEPT_FILE = "metode-ipmvp-a.md"
_VERDICT_INPUT = {"decision": "approved", "rationale": "expert reviewed (sim)"}
_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"
#: Written into a COPY of the base, into the verdict body alone. A sentinel of our own rather than
#: a phrase lifted out of the shipped file: the ExpeL fold legitimately renders parts of a verdict
#: into the generation prompt, so a probe made of the file's own prose could not tell the ungated
#: read from the gated one.
_VERDICT_SENTINEL = "VERDICT-LEAK-SENTINEL-q4w5e6"
#: The exact ``_verdict_rationale`` fewshot string (test_step1_expel_loadbearing's marker).
_REALIZATION_SIGNAL = "realiseringsgrad=0.82"
def _tools(bundle_dir: str, dimension: str | None = 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 (S2c's helper: ``invoke`` returns ``[Content]``, and 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))
def _base_with_a_marked_verdict(tmp_path: Path) -> str:
"""A copy of the fixture base whose verdict body carries ``_VERDICT_SENTINEL``."""
copy = tmp_path / "bundle"
shutil.copytree(BUNDLE_DIR, copy)
verdict = copy / _VERDICT_FILE
verdict.write_text(
verdict.read_text(encoding="utf-8") + f"\n{_VERDICT_SENTINEL}\n", encoding="utf-8"
)
return str(copy)
def _base_with_an_unlinked_verdict(tmp_path: Path) -> str:
"""A copy of the fixture base plus a ``type: verdict`` file NO index links to, so navigation
never reaches it and it is absent from ``Bundle.verdicts``."""
copy = tmp_path / "unlinked"
shutil.copytree(BUNDLE_DIR, copy)
(copy / "verdict-orphan.md").write_text(
"---\ntype: verdict\ntitle: Orphan judgement\ndecision: approved\n---\n\n"
f"{_VERDICT_SENTINEL}\n",
encoding="utf-8",
)
return str(copy)
def _recording_factory(
sink: list[str], *, script: dict[str, Any] | None = None
) -> Callable[[str], BaseChatClient]:
"""S2c's recording factory: every prompt blob (text PLUS function call/result contents) lands
in ``sink``. ``Message.text`` alone measures a context-bearing prompt at a few characters."""
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]
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))
sink.append("\n".join(parts))
return original(messages=messages, options=options, stream=stream, **kwargs)
client._inner_get_response = recording # type: ignore[attr-defined,method-assign]
return client
return factory
# ------------------------------------------------------------------- (1) the exploration path
async def test_the_exploration_path_refuses_a_verdict_document(tmp_path) -> None:
"""LOAD-BEARING (1): the exploration's own call — ``dimension=None``, which admits every
dimension — still refuses the verdict layer.
Detach point: remove the gate from ``read_file`` → RED."""
bundle_dir = _base_with_a_marked_verdict(tmp_path)
bundle_id = okf.reconcile_bundle_id(bundle_dir).id
tools = _tools(bundle_dir)
with pytest.raises(VerdictLayerRefused):
await _invoke(tools["read_file"], bundle_id=bundle_id, path=_VERDICT_FILE)
async def test_an_ordinary_document_in_the_same_base_still_reads(tmp_path) -> None:
"""CAUSALITY CONTROL for (1): a concept file in the SAME base reads whole, so the refusal above
is caused by the layer rather than by a fixture nothing can open."""
bundle_dir = _base_with_a_marked_verdict(tmp_path)
bundle_id = okf.reconcile_bundle_id(bundle_dir).id
body = await _invoke(_tools(bundle_dir)["read_file"], bundle_id=bundle_id, path=_CONCEPT_FILE)
assert "IPMVP" in body, f"the control document did not read: {body[:120]!r}"
async def test_no_listing_names_the_verdict_either(tmp_path) -> None:
"""The OTHER half of "no listing names it": ``read_bundle`` still hides the layer, so the two
rungs agree. Without this the gate could be green while the listing advertised the document it
then refuses — the disagreement S2c's ``in_dimension`` rule exists to prevent."""
bundle_dir = _base_with_a_marked_verdict(tmp_path)
bundle_id = okf.reconcile_bundle_id(bundle_dir).id
listing = await _invoke(_tools(bundle_dir)["read_bundle"], bundle_id=bundle_id)
assert _VERDICT_FILE not in listing
# ------------------------------------------------------------------------ (2) the debate path
async def test_the_debate_path_refuses_a_verdict_document(tmp_path, monkeypatch) -> None:
"""LOAD-BEARING (2): the tool ``run_project`` hands the debate refuses the same document.
Built through the REAL seam rather than by calling ``navigator_tools`` again: the two callers
share one construction, and a test that rebuilt the tools itself would prove nothing about what
the pipeline actually gave its agents (S2c arm (e)'s own form).
Detach point: remove the gate from ``read_file`` → RED."""
import portfolio_optimiser.run as run_module
bundle_dir = _base_with_a_marked_verdict(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,
verdict_input=_VERDICT_INPUT,
client_factory=_recording_factory(sink),
)
assert captured, "the debate was never built"
read_file = next(t for t in captured[0] if getattr(t, "name", "") == "read_file")
with pytest.raises(VerdictLayerRefused):
await _invoke(read_file, bundle_id=bundle_id, path=_VERDICT_FILE)
# --------------------------------------------------------------- (3) the gated route survives
async def test_the_gated_expel_route_still_reaches_the_hypothesis(
make_recording_client_factory,
) -> None:
"""LOAD-BEARING (3): the sanctioned door is untouched — the seed verdict's realization signal
still reaches the generation prompt through the Step-1 ExpeL fold.
This is what separates a gate from a wall. A refusal that also closed the fold would look
identical on arms (1), (2) and (4) and would have removed the loop's whole learning path."""
store = seed_store_from_bundle(str(BUNDLE_DIR))
assert store.verdicts, "precondition: the bundle seeds exactly one verdict"
factory, recorded = make_recording_client_factory(_VALID_REPLY)
result = await run_project(
_PID,
"local",
docs_dir=str(BUNDLE_DIR),
bundle_dir=str(BUNDLE_DIR),
verdict_input=_VERDICT_INPUT,
store=store,
client_factory=factory,
)
assert isinstance(result.outcome, ValidatedProposal)
gen_prompts = [p for p in recorded if "SavingsProposal" in p]
assert gen_prompts, "the generation call must have happened"
assert any(_REALIZATION_SIGNAL in p for p in gen_prompts), (
"the prior verdict no longer reaches the hypothesis prompt — the refusal closed the GATED "
"route as well, which is a wall rather than a gate"
)
# ------------------------------------------------------------------ (4) the refusal is traced
async def test_a_refused_read_is_recorded_and_leaks_nothing(tmp_path) -> None:
"""LOAD-BEARING (4): the debate asks for the verdict by path; the read is refused, the CALL is
in the trace with its path, and the verdict's body reaches no prompt.
Both halves are needed and each has its own detach point. Without the recording an operator
reading ``{run_id}-debate.json`` after a paid run cannot tell a run that tried the ungated door
from one that never did — MEASURED: moving the recorder's append after ``call_next`` empties
the trace for exactly a refused call, because the invocation never returns. Without the leak
probe the gate could be recording refusals while the bytes left anyway."""
bundle_dir = _base_with_a_marked_verdict(tmp_path)
bundle_id = okf.reconcile_bundle_id(bundle_dir).id
outbox_dir = tmp_path / "outbox"
sink: list[str] = []
result = await run_project(
_PID,
"local",
docs_dir=bundle_dir,
bundle_dir=bundle_dir,
verdict_input=_VERDICT_INPUT,
outbox_dir=str(outbox_dir),
run_id="verdict-gate",
client_factory=_recording_factory(
sink,
script={
"proposer": [
{"call": "read_file", "args": {"bundle_id": bundle_id, "path": _VERDICT_FILE}},
_VALID_REPLY,
],
"checker": _CHECKER_REPLY,
},
),
)
leaking = [i for i, prompt in enumerate(sink) if _VERDICT_SENTINEL in prompt]
assert not leaking, (
f"the verdict body reached prompt(s) {leaking} — a guessed path still walks around the "
"gated ExpeL fold"
)
observed = [(c.name, c.bundle_id, c.path) for c in result.debate_tool_calls]
assert ("read_file", bundle_id, _VERDICT_FILE) in observed, (
f"the refused read left no trace: {observed} — a refusal nobody can see is a refusal "
"nobody can audit"
)
payload = json.loads((outbox_dir / "verdict-gate-debate.json").read_text(encoding="utf-8"))
assert {"name": "read_file", "bundle_id": bundle_id, "path": _VERDICT_FILE} in payload[
"tool_calls"
], f"the artefact does not carry the refused call: {payload['tool_calls']}"
# ------------------------------------------------------ (5) the document, not the walk, decides
async def test_an_unlinked_verdict_is_refused_too(tmp_path) -> None:
"""LOAD-BEARING (5): a verdict file navigation never reached is refused as well.
"However the path was found" is the order's own wording and the reason the rule reads the
RESOLVED DOCUMENT's frontmatter rather than looking the path up among the navigated files. An
implementation gated on ``Bundle.verdicts`` passes arms (1), (2) and (4) and still serves this
one — the defect wearing the fix's clothes.
Detach point: gate on the walk instead of the document → RED here and nowhere else."""
bundle_dir = _base_with_an_unlinked_verdict(tmp_path)
bundle_id = okf.reconcile_bundle_id(bundle_dir).id
assert not any(f.name == "verdict-orphan.md" for f in okf.navigate_bundle(bundle_dir).files), (
"precondition: the orphan must be OUTSIDE the walk, or this arm proves nothing"
)
with pytest.raises(VerdictLayerRefused):
await _invoke(
_tools(bundle_dir)["read_file"], bundle_id=bundle_id, path="verdict-orphan.md"
)
# ------------------------------------------------------------------------ (6) ladder intact
async def test_the_index_still_reads_whole(tmp_path) -> None:
"""The ladder's top rung is untouched: ``index.md`` is navigation, never a judgement, and
``read_file(id, 'index.md')`` is the disclosure level ``list_bundles``' excerpt points at.
Detach point: gate on the complement of ``context_files`` (which drops index files too) → RED."""
bundle_dir = _base_with_a_marked_verdict(tmp_path)
bundle_id = okf.reconcile_bundle_id(bundle_dir).id
whole = await _invoke(_tools(bundle_dir)["read_file"], bundle_id=bundle_id, path="index.md")
assert "progressiv disclosure" in whole.lower()