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

feat(provenance): a run records which external service it actually called

Browse source 
The egress declaration (Trekk B3) says what a run MAY contact. It cannot say what
it DID: after the run, nothing distinguished "the agents queried the price
register" from "the agents ignored it", and a proposal resting on an external
service should be traceable to it.

ToolCallRecorder(FunctionMiddleware) mirrors BudgetMiddleware(ChatMiddleware) one
layer down — that one observes the debate's chat calls, this one its tool calls.
It observes only: call_next is always awaited, so a trace can never alter the run
it traces. The record lands on ProvenanceStamp.external_calls, read AFTER the
debate so it is a record rather than an intention.

MEASURED, not assumed, before any of it was written: FunctionMiddleware fires for
a tool served over a REAL MCP stdio subprocess, and context.function.name carries
the BARE tool name with no server prefix. That measurement decided the design —
MAF cannot tell us which server a tool came from, so attribution comes from our own
config, and a name allowed by two servers is recorded UNATTRIBUTED (server="")
rather than credited to the first match. Naming a service that may never have been
contacted is the one place a guess must not go.

Only CONFIGURED tools are recorded. The middleware fires for every function the
agents invoke, including the in-process retrieve_cost_docs on the road path;
logging those would turn the record into a false egress claim. An empty list is a
positive statement — nothing outside this process was contacted — which is why it
is always serialized rather than omitted.

Honesty limit, written on ExternalCall itself: this is the call and its source. It
is NOT evidence that the service's answer reached the proposal, nor a verified
rendering of that answer.

One finding, and it is the reason for measuring rather than trusting green: the
road-path negative test was VACUOUS. Its scripted tool call named an argument the
tool does not declare (code vs query), MAF rejected the call before invocation, and
the test asserted an empty record against a run where no tool ran at all — green
under the exact mutation it existed to catch. It now spies on the recorder and
asserts the invocation genuinely reached it before asserting it was not recorded.
This is last session's lesson again: a scenario that cannot distinguish two
implementations proves nothing.

The tool-call double is registered in test_scripted_client_consolidation.py's
_DELEGATING_OVERRIDES — it cannot live in the reply_selector seam, which returns a
reply STRING, and a response that is not text is its whole subject.

Load-bearing MEASURED (tests/test_b4_mcp_call_trace_loadbearing.py) against the
whole 755-test suite, four mutations all red: detach the recorder from the debate
middleware · record every function invocation · attribute an ambiguous name to the
first server · stop reading the recorder into provenance. Control: a run with no
configured servers records nothing, so the empty record is a real answer and not
the only one the seam can produce.

Ran it, not just tested it: the real recorder against a real MCP server subprocess
returns ExternalCall(server='prisregister', tool='lookup_unit_price'), and a
scripted CLI run's outbox artefact carries the empty list.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VtRd8y1PDPGwkrRXFhubqr
This commit is contained in:
Kjell Tore Guttormsen 2026-08-05 21:37:29 +02:00
commit 991131be3f
6 changed files with 402 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

21
docs/bestille-en-kjoring.md
View file

@ -176,3 +176,24 @@ Tre ting er verdt å vite som bestiller:
og få svar med feil tilgang.
Uten `--mcp-config` gjøres ingen nettverkskall i det hele tatt.
### Etterpå: hva ble faktisk kalt
Kunngjøringen er en **tillatelse** — den sier hva kjøringen *kan* kontakte. Den sier ikke om
agentene brukte tjenesten eller lot den ligge. Derfor fører kjøringen også et **kall-spor**, som
følger med artefaktene under `provenance.external_calls`:
```json
"external_calls": [{"server": "prisregister", "tool": "lookup_unit_price"}]
```
En tom liste er en påstand, ikke et hull: ingenting utenfor prosessen ble kontaktet. Verktøy som
kjører lokalt (som dokumentsøket) telles ikke med — de er ikke kontakt med noen.
**Hva sporet ikke sier.** Det viser at verktøyet ble kalt, og hvilken server det tilhører. Det er
ikke bevis for at tjenestens svar er det tallet som havnet i forslaget, og det er ingen kontrollert
gjengivelse av hva tjenesten svarte. Skal svaret etterprøves, må det gjøres mot tjenesten selv.
En sjelden detalj, tatt med fordi den ellers ville sett ut som en feil: hvis to servere tilbyr
verktøy med **samme navn**, står `server` tomt. Rammeverket oppgir bare verktøynavnet, så de to lar
seg ikke skille — og da er «vet ikke» riktigere enn å gjette på den ene.

66
src/portfolio_optimiser/mcp_tools.py
View file

@ -32,12 +32,20 @@ Every refusal below exists because its absence is a live hazard:
from __future__ import annotations
import os
from collections.abc import Awaitable, Callable
from pathlib import Path
from typing import Any, Literal
from agent_framework import MCPStdioTool, MCPStreamableHTTPTool
from agent_framework import (
FunctionInvocationContext,
FunctionMiddleware,
MCPStdioTool,
MCPStreamableHTTPTool,
)
from pydantic import BaseModel, ConfigDict, Field, model_validator
from portfolio_optimiser.provenance import ExternalCall
class McpServerConfig(BaseModel):
"""One external MCP server a run is permitted to contact.
@ -170,6 +178,62 @@ def build_mcp_tools(configs: tuple[McpServerConfig, ...]) -> list[Any]:
return tools
def tool_server_index(configs: tuple[McpServerConfig, ...]) -> dict[str, str]:
"""Map each allowed tool name to the server that serves it — the attribution table B4 needs.
A name allowed by MORE THAN ONE configured server maps to ``""`` (unattributed) rather than to
one of them. MEASURED: MAF hands function middleware the BARE tool name, with no server prefix,
so the two really are indistinguishable at that seam. Picking the first match would be a guess
written into a provenance record, which is the one place a guess must never go — validation,
never repair, applied to attribution.
"""
index: dict[str, str] = {}
for config in configs:
for name in config.allowed_tools:
index[name] = "" if name in index else config.name
return index
class ToolCallRecorder(FunctionMiddleware):
"""Function middleware that records WHICH configured external tool a run actually called (B4).
Mirrors ``budget.BudgetMiddleware(ChatMiddleware)`` one layer down: that one observes the
debate's chat calls, this one its tool calls. It observes only — ``call_next`` is always
awaited, and nothing here can block, alter or short-circuit an invocation. A trace that changed
the run it traces would not be a trace.
**Only CONFIGURED tools are recorded.** The middleware fires for every function the agents
invoke, including the in-process ``retrieve_cost_docs`` on the road path. Logging those would
turn the record into a false egress claim — the whole value of the record is that its entries
mean "something outside this process was contacted".
Recording is idempotent per ``(server, tool)`` and returned SORTED: the record answers *what was
contacted*, and it is stamped into a byte-deterministic artefact, so it must not vary with how
many times an agent happened to ask.
"""
def __init__(self, index: dict[str, str]) -> None:
self._index = index
self._seen: set[tuple[str, str]] = set()
def note(self, tool_name: str) -> None:
"""Record one invocation by tool name. Unconfigured names are IGNORED, not recorded as
unattributed calls — an in-process tool is not an external service with a missing label."""
if tool_name in self._index:
self._seen.add((self._index[tool_name], tool_name))
def calls(self) -> list[ExternalCall]:
return [ExternalCall(server=s, tool=t) for s, t in sorted(self._seen)]
async def process(
self, context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]]
) -> None:
name = getattr(getattr(context, "function", None), "name", None)
if isinstance(name, str):
self.note(name)
await call_next()
def service_labels(configs: tuple[McpServerConfig, ...]) -> tuple[str, ...]:
"""The egress declaration, one label per server: what will be contacted and which tools are
permitted. Feeds ``mandate.announce(external_services=...)``, so an operator sees the full

23
src/portfolio_optimiser/provenance.py
View file

@ -29,6 +29,26 @@ class Citation(BaseModel):
snippet: str
class ExternalCall(BaseModel):
"""One external service call a run actually made (Trekk B4).
**What this is evidence of, and what it is not.** It records that ``tool`` was invoked and which
configured ``server`` it belongs to. It is NOT evidence that the service's answer reached the
proposal, and it is not a verified rendering of what the service returned — the framework hands
the answer to the agent, and what the agent does with it is the agent's. Reading this as "the
figure came from the price register" would claim more than the record supports.
``server`` is ``""`` when the tool name cannot be attributed to exactly one configured server.
MEASURED against a real MCP stdio subprocess: MAF passes the BARE tool name to function
middleware, with no server prefix, so two servers exposing one tool name are indistinguishable
at this seam. Unattributed is the honest answer there; naming the first match would put a
service in the record that may never have been contacted.
"""
server: str
tool: str
class ProvenanceStamp(BaseModel):
"""Authoritative provenance for one proposal — at least one citation is mandatory."""
@ -37,6 +57,9 @@ class ProvenanceStamp(BaseModel):
role: str
validator_decision: Literal["validated", "rejected"]
token_usage: int
#: External service calls the run made (B4). EMPTY is a positive statement — "nothing outside
#: this process was contacted" — not an absent field, which is why it is always serialized.
external_calls: list[ExternalCall] = Field(default_factory=list)
def to_annotations(self) -> list[Annotation]:
"""Map to MAF ``Annotation`` dicts for display only (NOT the source of truth)."""

12
src/portfolio_optimiser/run.py
View file

@ -67,9 +67,11 @@ from portfolio_optimiser.mandate import (
)
from portfolio_optimiser.mcp_tools import (
McpServerConfig,
ToolCallRecorder,
build_mcp_tools,
load_mcp_config,
service_labels,
tool_server_index,
)
from portfolio_optimiser.provenance import ProvenanceStamp
from portfolio_optimiser.reference_domain import Project, load_reference_projects
@ -536,12 +538,16 @@ async def run_project(
)
factory = client_factory if client_factory is not None else _default_factory(profile)
budget_mw = BudgetMiddleware(meter)
# Trekk B4: the egress DECLARATION says what a run may contact; this records what it actually
# called. Attached only when servers are configured — with none there is nothing to attribute a
# call to, and the middleware list stays exactly what it was before Trekk B.
call_recorder = ToolCallRecorder(tool_server_index(mcp_servers)) if mcp_servers else None
debate = fresh_workflow(
factory,
max_rounds=max_rounds,
enable_layer1_hitl=enable_layer1_hitl,
tools=debate_tools,
middleware=[budget_mw],
middleware=[budget_mw] if call_recorder is None else [budget_mw, call_recorder],
)
# S4.2 cut (comparison protocol §4 pkt 2/3): everything above is offline — contracts, budget, and
# the EAGER client build (fresh_workflow constructs the proposer+checker clients, workflow.py:64).
@ -661,6 +667,10 @@ async def run_project(
"validated" if isinstance(validator_outcome, ValidatedProposal) else "rejected"
),
token_usage=meter.tokens,
# B4: which external service the debate actually called. Read AFTER the debate, so it is a
# record rather than an intention. The honesty limit lives on ``ExternalCall`` itself: this
# is the call and its source, not a verified rendering of the service's answer.
external_calls=call_recorder.calls() if call_recorder is not None else [],
)
# 6b. Step 3/4 checker gate (målbilde §2/§6): the validator falsifies the numbers, the checker

278
tests/test_b4_mcp_call_trace_loadbearing.py Normal file
View file

@ -0,0 +1,278 @@
"""Load-bearing: a run records WHICH external service it actually called (Trekk B4).
Trekk B3 made a run declare, before its first call, which servers it MAY contact. That is the
permission. It is not the record: after the run, the egress declaration still cannot tell you
whether the agents used the price register or ignored it, and a proposal that rests on an external
service should be traceable to it.
``FunctionMiddleware`` + ``FunctionInvocationContext`` were verified present in the pinned MAF 1.9.0,
and the middleware was MEASURED to fire for a tool served over a REAL MCP stdio subprocess — with
``context.function.name`` carrying the bare tool name and NO server prefix. That measurement is why
attribution comes from our own config rather than from MAF: the framework cannot tell us which
server a tool came from, so two servers exposing one tool name are indistinguishable at this seam
and must be reported as unattributed rather than guessed.
**Honesty limit, pinned in the code it describes:** this records the CALL and its SOURCE. It is not
evidence that the service's answer reached the proposal, nor a verified rendering of that answer.
Detach points, each RED on its own:
* drop the recorder from the debate middleware -> a called service leaves no trace, and the run
reports the same empty record as a run that called nothing;
* record every function invocation -> the road path's IN-PROCESS retrieval tool is reported as an
external service call, which turns the record into a false egress claim;
* attribute an ambiguous tool name to the first server that allows it -> the record names a service
that may never have been contacted.
The control (``test_a_run_without_mcp_servers_records_no_calls``) proves the empty record is a real
statement rather than the only thing the seam can produce.
"""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from pathlib import Path
from typing import Any
import pytest
from agent_framework import ChatResponse, Content, FunctionTool, Message, UsageDetails, tool
from portfolio_optimiser import run as run_module
from portfolio_optimiser.mcp_tools import McpServerConfig, ToolCallRecorder, tool_server_index
from portfolio_optimiser.provenance import ExternalCall
from portfolio_optimiser.reference_domain import load_reference_projects
from portfolio_optimiser.run import run_project
from portfolio_optimiser.simulation import ScriptedChatClient
from portfolio_optimiser.verdicts import VerdictStore
BUNDLE_DIR = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro"
_VERDICT_INPUT = {"decision": "approved", "rationale": "expert reviewed (sim)"}
_REPLY = (
'{"measure":"LED-retrofit av kontorbelysning","affected_items":'
'[{"code":"ENERGI-TOTAL-EL","quantity":300000,"unit_cost":1.0}],"claimed_saving_nok":30000}'
)
_SERVER = McpServerConfig(
name="prisregister",
transport="http",
url="https://intern.example/mcp",
allowed_tools=("lookup_unit_price",),
timeout_seconds=15,
)
@tool(name="lookup_unit_price", description="Look up a unit price from the price register.")
def _lookup_unit_price(code: str) -> str:
return f"unit price for {code}: 1.0"
def _as_context_manager(function_tool: FunctionTool) -> Any:
"""Wrap a real ``FunctionTool`` so it is ALSO an async context manager — the two halves a MAF
``MCPTool`` presents to the run path (``AsyncExitStack`` enters it; the agents invoke it).
A stand-in rather than a live server on purpose: the subject here is what OUR middleware records,
and the real MCP transport is already measured against a spawned server in
``test_ingest_golden_mcp.py``. What matters is that the agent genuinely INVOKES the tool, so the
middleware fires for real rather than being called by the test directly.
"""
cls = type(
"_ContextManagedTool",
(type(function_tool),),
{
"__aenter__": lambda self: _aenter(self),
"__aexit__": lambda self, *exc: _aexit(),
},
)
wrapped = cls.__new__(cls)
wrapped.__dict__.update(function_tool.__dict__)
return wrapped
async def _aenter(self: Any) -> Any:
return self
async def _aexit() -> bool:
return False
class _ToolCallingClient(ScriptedChatClient):
"""Emits ONE tool call on its first response, then delegates to the canonical scripted body.
Registered in ``test_scripted_client_consolidation.py``'s ``_DELEGATING_OVERRIDES``: it cannot
live in the ``reply_selector`` seam, which returns a reply STRING — this double's whole subject
is a response that is a function CALL rather than text (the same reason the two ordering/failure
probes are registered there).
"""
def __init__(self, call_tool: str, arguments: dict[str, Any], reply: str) -> None:
super().__init__(reply=reply, tokens_per_reply=8)
self._call_tool = call_tool
self._arguments = arguments
self._called = False
def _inner_get_response(
self,
*,
messages: Sequence[Any],
options: Mapping[str, Any],
stream: bool = False,
**kwargs: Any,
) -> Any:
if self._called or stream:
return super()._inner_get_response(
messages=messages, options=options, stream=stream, **kwargs
)
self._called = True
self.call_count += 1
async def _coro() -> ChatResponse:
return ChatResponse(
messages=[
Message(
role="assistant",
contents=[
Content(
"function_call",
call_id="call-1",
name=self._call_tool,
arguments=self._arguments,
)
],
)
],
response_id="synthetic",
usage_details=UsageDetails(total_token_count=8),
)
return _coro()
#: The argument name each tool actually declares. MEASURED, and load-bearing: a call whose argument
#: names do not match the tool's signature is rejected by MAF BEFORE invocation, so the middleware
#: never fires — the first version of the road-path test below passed ``code`` to
#: ``retrieve_cost_docs(query)`` and was therefore VACUOUS. It asserted an empty record against a run
#: where no tool ran at all, and stayed green under the mutation it exists to catch.
_ARGUMENTS = {
"lookup_unit_price": {"code": "ENERGI-TOTAL-EL"},
"retrieve_cost_docs": {"query": "cost saving measure"},
}
def _factory(call_tool: str):
def factory(role: str) -> ScriptedChatClient:
# Only the proposer calls a tool; the checker stays textual, so the recorded call cannot be
# an artefact of every agent calling everything.
return (
_ToolCallingClient(call_tool, _ARGUMENTS[call_tool], _REPLY)
if role == "proposer"
else ScriptedChatClient(reply="VERDICT: APPROVE", tokens_per_reply=8)
)
return factory
# --- The recorder's own contract ------------------------------------------------------------
def test_an_unconfigured_tool_name_is_not_recorded_as_egress() -> None:
"""Only a CONFIGURED MCP tool is an external call. RED when the recorder logs every function:
the in-process retrieval tool would then be reported as contact with a third party."""
recorder = ToolCallRecorder(tool_server_index((_SERVER,)))
recorder.note("retrieve_cost_docs")
assert recorder.calls() == []
def test_an_ambiguous_tool_name_is_recorded_without_a_server() -> None:
"""Two servers allowing one tool name cannot be told apart at this seam (MEASURED: MAF passes
the bare tool name). RED when the recorder picks the first match — the record would then name a
service that may never have been contacted."""
other = _SERVER.model_copy(update={"name": "reservepris"})
recorder = ToolCallRecorder(tool_server_index((_SERVER, other)))
recorder.note("lookup_unit_price")
assert recorder.calls() == [ExternalCall(server="", tool="lookup_unit_price")]
# --- The run path ---------------------------------------------------------------------------
async def test_a_called_mcp_tool_is_recorded_in_provenance(monkeypatch: pytest.MonkeyPatch) -> None:
"""The seam itself: the agents call the configured service, and the run's provenance says so.
RED when the recorder is not in the debate middleware — the run then reports the same empty
record whether it contacted the price register or ignored it.
"""
monkeypatch.setattr(
run_module, "build_mcp_tools", lambda _c: [_as_context_manager(_lookup_unit_price)]
)
result = await run_project(
"BYGG-KONTOR-NORD",
"local",
docs_dir=str(BUNDLE_DIR),
bundle_dir=str(BUNDLE_DIR),
verdict_input=_VERDICT_INPUT,
store=VerdictStore(verdicts=[]),
client_factory=_factory("lookup_unit_price"),
mcp_servers=(_SERVER,),
)
assert result.provenance.external_calls == [
ExternalCall(server="prisregister", tool="lookup_unit_price")
]
async def test_a_local_tool_call_is_not_recorded_as_an_external_call(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The road path's ``retrieve_cost_docs`` runs IN PROCESS. Calling it is not egress, and a run
that only called it must not claim it contacted the price register.
RED when the recorder logs every function invocation rather than the configured ones.
The ``observed`` spy is what keeps this test HONEST rather than merely green. An empty record is
also what a run that invoked nothing produces, so without proof that the invocation actually
reached the recorder this asserts nothing — which is exactly what happened to the first version
of it (the tool call named an argument the tool does not declare, MAF rejected it before
invocation, and the test stayed green under the very mutation it exists to catch).
"""
monkeypatch.setattr(
run_module, "build_mcp_tools", lambda _c: [_as_context_manager(_lookup_unit_price)]
)
observed: list[str] = []
original = ToolCallRecorder.note
def spy(self: ToolCallRecorder, tool_name: str) -> None:
observed.append(tool_name)
original(self, tool_name)
monkeypatch.setattr(ToolCallRecorder, "note", spy)
rv13 = {p.id: p for p in load_reference_projects()}["RV13-RAS-TP"]
result = await run_project(
"RV13-RAS-TP",
"local",
docs_dir=rv13.docs_dir,
verdict_input=rv13.verdict_input,
store=VerdictStore(verdicts=[]),
client_factory=_factory("retrieve_cost_docs"),
mcp_servers=(_SERVER,),
)
assert observed == ["retrieve_cost_docs"], "the local tool must actually have been invoked"
assert result.provenance.external_calls == []
async def test_a_run_without_mcp_servers_records_no_calls() -> None:
"""Control: no configured servers -> an empty record, on the pre-B4 path, unchanged.
Without this the assertions above could all pass against a seam that can only ever produce one
answer.
"""
result = await run_project(
"BYGG-KONTOR-NORD",
"local",
docs_dir=str(BUNDLE_DIR),
bundle_dir=str(BUNDLE_DIR),
verdict_input=_VERDICT_INPUT,
store=VerdictStore(verdicts=[]),
client_factory=_factory("lookup_unit_price"),
)
assert result.provenance.external_calls == []

4
tests/test_scripted_client_consolidation.py
View file

@ -35,6 +35,10 @@ _DELEGATING_OVERRIDES = [
# delegates. Like the ordering probe it cannot live in the reply-selector seam — that seam
# returns a reply string, and this double's whole subject is the absence of one.
"tests/test_portfolio_failure_accounting_loadbearing.py",
# B4 tool-call probe: its first response is a function CALL rather than text, then it delegates.
# It cannot live in the reply-selector seam either — that seam returns a reply STRING, and a
# response that is not text is precisely this double's subject.
"tests/test_b4_mcp_call_trace_loadbearing.py",
]
# Doubles in a DIFFERENT lineage (``spikes._harness.FakeChatClient``). There is no canonical