portfolio-optimiser/tests/test_tool_call_path_loadbearing.py

"""S7a-3 pkt. 3 - the artefact says WHICH documents the navigator opened, not just which base.

**The measured silence** (``docs/2026-09-03-syretest-s7a2-k2.md``). ``ExplorationToolRecorder``
(MAJOR-1, session 67) records a call's NAME and its ``bundle_id``, in invocation order. Over the
three example bases that was enough: 39 concepts, and a run that opened two of them left a trace an
operator could reason about. Over K2 it is not: **629 concepts**, and ``tool_calls`` reads
``read_file`` twice with a base id, so **which two of 629 the navigator actually opened cannot be
read out of the delivered artefact at all**. Sessions 77 and 81 both had to instrument the run by
hand to answer it.

``path`` is the argument that answers it, and it is recorded for the two tools that take one -
``read_file`` and (since pkt. 2) ``read_dir``. The RESULT is still never recorded, and that is the
same decision MAJOR-1 made rather than an omission: the result is the base's content, which is
precisely the thing measured at 89 % of every prompt token, and a trace carrying it would be a
second copy of the context rather than a record of the run.

``""`` for a tool that takes no path, exactly as ``bundle_id`` is ``""`` for ``list_bundles``: a
value invented for an argument nobody passed is the false attribution ``mcp_tools.ToolCallRecorder``
refuses for unconfigured tools. ONE argument reader serves both fields - two copies of "read this
key out of either shape ``FunctionInvocationContext`` allows" would be the ko-(p) drift.

Arms: (a) the path is recorded for both tools that take one, in call order * (b) ``""`` for the
tools that take none, and for a non-string * (c) the RESULT is still absent * (d) the artefact
carries it * (e) end-to-end: which concepts a scripted navigator opened is readable from
``{run_id}-exploration.json``.
"""

from __future__ import annotations

import asyncio
import json
from pathlib import Path
from types import SimpleNamespace
from typing import Any

from portfolio_optimiser import explore as ex


class _FakeContext:
    """The two attributes the middleware reads, and nothing else."""

    def __init__(self, name: str, arguments: Any) -> None:
        self.function = SimpleNamespace(name=name)
        self.arguments = arguments


async def _invoke(recorder: ex.ExplorationToolRecorder, context: Any) -> None:
    async def _next() -> None:
        return None

    await recorder.process(context, _next)


def _record(*calls: tuple[str, Any]) -> list[ex.ToolCall]:
    sink: list[ex.ToolCall] = []
    recorder = ex.ExplorationToolRecorder(sink)
    for name, arguments in calls:
        asyncio.run(_invoke(recorder, _FakeContext(name, arguments)))
    return sink


# --- (a)/(b) what is recorded, and what is deliberately not ---------------------------------------


def test_the_path_is_recorded_for_both_tools_that_take_one() -> None:
    """(a) The headline. ``read_dir`` is in here because pkt. 2 made it the rung the navigator
    descends through: a trace that named the documents but not the directories would say where a
    run ENDED without saying how it got there."""
    sink = _record(
        ("read_dir", {"bundle_id": "k2", "path": "del-ii-bilag-7-prisskjema"}),
        ("read_file", {"bundle_id": "k2", "path": "del-ii-bilag-7-prisskjema/sheet-1.md"}),
    )

    assert [(c.name, c.bundle_id, c.path) for c in sink] == [
        ("read_dir", "k2", "del-ii-bilag-7-prisskjema"),
        ("read_file", "k2", "del-ii-bilag-7-prisskjema/sheet-1.md"),
    ]


def test_a_tool_that_takes_no_path_records_an_empty_one() -> None:
    """(b) ``bundle_id``'s own rule, applied to the second field: a value invented for an argument
    nobody passed is false attribution. Both argument shapes are exercised, because
    ``FunctionInvocationContext.arguments`` is ``BaseModel | Mapping`` and a reader that assumed one
    would be blind on the other - and a non-string yields ``""`` rather than a coerced label."""
    sink = _record(
        ("list_bundles", {}),
        ("read_bundle", {"bundle_id": "k2"}),
        ("read_file", SimpleNamespace(bundle_id="k2", path="modelled.md")),
        ("read_file", {"bundle_id": "k2", "path": 7}),
    )

    assert [c.path for c in sink] == ["", "", "modelled.md", ""]


def test_the_result_is_still_never_recorded() -> None:
    """(c) MAJOR-1's decision, restated because pkt. 3 is the moment it would be easiest to undo:
    the result is the base's CONTENT, measured at 89 % of every prompt token in a K2 run. A trace
    carrying it would be a second copy of the context rather than a record of the run."""
    sink = _record(("read_file", {"bundle_id": "k2", "path": "a.md"}))

    recorded = json.dumps([vars(c) for c in sink])
    # P19 DEL C added HOW the listing was asked for (``filter``/``offset``/``limit``). The claim
    # this arm makes is unchanged and is the one that matters: the fields are all ARGUMENTS, and
    # none of them is the result.
    assert set(vars(sink[0])) == {"name", "bundle_id", "path", "filter", "offset", "limit"}
    assert "result" not in recorded


# --- (d)/(e) it reaches the artefact --------------------------------------------------------------


def test_the_artefact_carries_the_path_beside_the_name_and_the_base() -> None:
    """(d) A field no artefact carries is a field nobody can read after the run - MAJOR-1's second
    half, and the reason ``trace_payload`` is the ONE rendering of a trace."""
    trace = ex.ExplorationTrace()
    trace.tool_calls.append(ex.ToolCall(name="read_file", bundle_id="k2", path="prisskjema.md"))

    payload = ex.trace_payload(trace, stop=None, completed=True, mandate=None)

    assert payload["tool_calls"] == [
        {
            "name": "read_file",
            "bundle_id": "k2",
            "path": "prisskjema.md",
            # P19 DEL C: always present, zero/empty for a call that did not pass them — an absent
            # key and "not narrowed" must not read the same.
            "filter": "",
            "offset": 0,
            "limit": 0,
        }
    ]
    assert json.dumps(payload), "the payload must stay plain data - the RAW layer is MAF-free"


def test_which_documents_a_scripted_navigator_opened_is_readable_from_the_artefact(
    tmp_path: Path,
) -> None:
    """(e) The end-to-end claim in the order's own words: 'which concepts the navigator opened can
    be read out of the artefact'. Driven through the SCRIPTED door (MAJOR-1's step list), because a
    constant-reply rehearsal emits no ``function_call`` at all and would leave this arm asserting
    over an empty list.

    Deliberately reuses ``test_scripted_explore_door_loadbearing``'s harness rather than rebuilding
    it: a second copy of the CLI argv assembly is the drift this repo names ko-(p).
    """
    from test_scripted_explore_door_loadbearing import (
        _BUNDLE_ID,
        _HYPOTHESIS,
        _MANAGER_STAGES,
        _PROPOSER_REPLY,
        _artefact,
        _explore_argv,
        _replies_file,
    )

    from portfolio_optimiser import run

    replies = _replies_file(
        tmp_path,
        {
            "proposer": _PROPOSER_REPLY,
            "checker": "VERDICT: APPROVE",
            "manager": list(_MANAGER_STAGES),
            "navigator": [
                {"call": "read_bundle", "args": {"bundle_id": _BUNDLE_ID}},
                {"call": "read_dir", "args": {"bundle_id": _BUNDLE_ID, "path": ""}},
                {"call": "read_file", "args": {"bundle_id": _BUNDLE_ID, "path": "index.md"}},
                "NAVIGATOR: read the index.",
            ],
            "hypothesiser": _HYPOTHESIS,
        },
    )

    rc = run.main(_explore_argv(tmp_path, replies, "scripted-paths"))
    assert rc in (0, 1), rc

    calls = _artefact(tmp_path, "scripted-paths")["tool_calls"]
    opened = [c["path"] for c in calls if c["name"] == "read_file"]

    assert opened == ["index.md"], (
        "the artefact must name WHICH document was opened; with the name and base alone, a K2 run "
        f"says 'read_file, k2' twice over 629 concepts and answers nothing. Got {calls!r}"
    )
    assert [c["name"] for c in calls] == ["read_bundle", "read_dir", "read_file"]