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

refactor(fase2a): konsolider fire skriptede klienter til én kanonisk (S2.5)

Browse source
This commit is contained in:
Kjell Tore Guttormsen 2026-07-15 07:37:43 +02:00
commit 620d5cfb83
3 changed files with 136 additions and 108 deletions
Download patch file Download diff file Expand all files Collapse all files

51
src/portfolio_optimiser/simulation.py
View file

@ -67,21 +67,48 @@ _CHECKER_APPROVE = "Tallene er innenfor feasibelt område og resonnementet holde
class ScriptedChatClient(OpenAIChatCompletionClient):
"""A network-free, SCRIPTED chat client: it returns a fixed ``reply`` and records every prompt
it receives into a shared ``sink`` (the observation probe). It is a stand-in for a real model —
it proves the loop's plumbing, NOT model behaviour.
"""The ONE canonical network-free scripted chat client (S2.5 consolidation): a single
``_inner_get_response`` body shared by the simulation's fixed-reply client AND conftest's three
test doubles (which subclass it). Parametrized by a ``reply_selector`` over
``(prompt_blob, role)`` plus an optional ``sink`` recording every prompt — the four
previously-divergent ``_inner_get_response`` bodies collapse to this one.
Subclasses the LAYERED ``OpenAIChatCompletionClient`` (not the minimal ``BaseChatClient``) so the
always-attached ``BudgetMiddleware`` is not silently no-op'd (verified). Construction is offline
(loopback ``base_url`` + dummy key); ``_inner_get_response`` intercepts before any HTTP."""
(loopback ``base_url`` + dummy key); ``_inner_get_response`` intercepts before any HTTP.
Back-compat constructors are preserved (divergent PUBLIC surfaces the external test call-sites
depend on): ``ScriptedChatClient(reply, sink)`` (POSITIONAL — used by ``scripted_factory``) is
sugar for a constant selector; the ``call_count`` attribute + ``model``/OTEL ``"synthetic"`` are
always present; subclasses pass ``reply_selector=`` / ``default_reply=`` for scripted-list,
prompt-scan, or record-only behaviour."""
OTEL_PROVIDER_NAME = "synthetic"
def __init__(self, reply: str, sink: list[str], *, tokens_per_reply: int = 8) -> None:
def __init__(
self,
reply: str | None = None,
sink: list[str] | None = None,
*,
reply_selector: Callable[[str, str], str] | None = None,
role: str = "",
default_reply: str = "ok",
tokens_per_reply: int = 8,
) -> None:
super().__init__(model="synthetic", api_key="synthetic", base_url="http://127.0.0.1:9/v1")
self._reply = reply
self._sink = sink
self._role = role
self._default = default_reply
# The reply-selector over (prompt_blob, role). A positional ``reply`` is sugar for a constant
# selector (scripted_factory back-compat); with neither, the constant is ``default_reply``.
if reply_selector is not None:
self._select: Callable[[str, str], str] = reply_selector
elif reply is not None:
self._select = lambda _prompt, _role: reply
else:
self._select = lambda _prompt, _role: self._default
self._tokens = tokens_per_reply
self.call_count = 0
def _inner_get_response(
self,
@ -91,7 +118,11 @@ class ScriptedChatClient(OpenAIChatCompletionClient):
stream: bool = False,
**kwargs: Any,
) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]:
self._sink.append(" ".join(getattr(m, "text", "") or "" for m in messages))
blob = " ".join(getattr(m, "text", "") or "" for m in messages)
if self._sink is not None:
self._sink.append(blob)
self.call_count += 1
reply = self._select(blob, self._role)
usage = UsageDetails(total_token_count=self._tokens)
if stream:
@ -99,14 +130,14 @@ class ScriptedChatClient(OpenAIChatCompletionClient):
# The framework accepts a {"type": "text", ...} dict here (its types under-specify it).
yield ChatResponseUpdate(
role="assistant",
contents=[{"type": "text", "text": self._reply}], # type: ignore[list-item]
contents=[{"type": "text", "text": reply}], # type: ignore[list-item]
)
return self._build_response_stream(_agen())
async def _coro() -> ChatResponse:
return ChatResponse(
messages=[Message(role="assistant", contents=[self._reply])],
messages=[Message(role="assistant", contents=[reply])],
response_id="synthetic",
usage_details=usage,
)
@ -121,7 +152,7 @@ def scripted_factory(replies: dict[str, str], sink: list[str]) -> Callable[[str]
per-turn counter); the shared ``sink`` spans the debate turns and the generation call."""
def factory(role: str) -> BaseChatClient:
return ScriptedChatClient(replies[role], sink)
return ScriptedChatClient(replies[role], sink, role=role)
return factory

129
tests/conftest.py
View file

@ -9,32 +9,20 @@ the gated live arm (Step 14).
from __future__ import annotations
from collections.abc import Callable, Sequence
from typing import Any
import pytest
from agent_framework import (
BaseChatClient,
ChatResponse,
ChatResponseUpdate,
Message,
UsageDetails,
)
from agent_framework_openai import OpenAIChatCompletionClient
from agent_framework import BaseChatClient
from portfolio_optimiser.simulation import ScriptedChatClient
from portfolio_optimiser.verdicts import VerdictStore, seed_store
class SyntheticUsageChatClient(OpenAIChatCompletionClient):
"""Network-free chat client returning scripted/default replies WITH a synthetic
``UsageDetails`` (``total_token_count``), so strict usage accounting does not hard-fail.
Subclasses the LAYERED ``OpenAIChatCompletionClient`` (not the minimal ``BaseChatClient``)
so it inherits the ``ChatMiddlewareLayer`` — a ``ChatMiddleware`` attached to an agent
backed by the minimal base would silently no-op (verified). Construction is offline
(loopback ``base_url``, dummy key); the ``_inner_get_response`` override intercepts the
raw call before any HTTP, so no network is touched."""
OTEL_PROVIDER_NAME = "synthetic"
class SyntheticUsageChatClient(ScriptedChatClient):
"""The scripted-list-then-default test double — now a THIN subclass of the canonical
``ScriptedChatClient`` (S2.5 consolidation). It keeps its full PUBLIC surface (the
``default_reply=`` kwarg constructor, the ``call_count`` attribute, ``model``/OTEL
``"synthetic"``) but delegates the shared ``_inner_get_response`` body to the canonical — the
scripted-list-then-default behaviour lives in its selector."""
def __init__(
self,
@ -43,41 +31,17 @@ class SyntheticUsageChatClient(OpenAIChatCompletionClient):
default_reply: str = "ok",
tokens_per_reply: int = 8,
) -> None:
super().__init__(model="synthetic", api_key="synthetic", base_url="http://127.0.0.1:9/v1")
self._scripted = list(scripted or [])
self._idx = 0
self._default = default_reply
self._tokens = tokens_per_reply
self.call_count = 0
scripted_list = list(scripted or [])
counter = {"i": 0}
def _next_reply(self) -> str:
reply = self._scripted[self._idx] if self._idx < len(self._scripted) else self._default
self._idx += 1
self.call_count += 1
return reply
def _select(_blob: str, _role: str) -> str:
i = counter["i"]
counter["i"] = i + 1
return scripted_list[i] if i < len(scripted_list) else default_reply
def _inner_get_response(
self, *, messages: Sequence[Message], stream: bool, options: Any, **kwargs: Any
) -> Any:
reply = self._next_reply()
usage = UsageDetails(total_token_count=self._tokens)
if stream:
async def _agen() -> Any:
yield ChatResponseUpdate(
role="assistant", contents=[{"type": "text", "text": reply}]
)
return self._build_response_stream(_agen())
async def _coro() -> ChatResponse:
return ChatResponse(
messages=[Message(role="assistant", contents=[reply])],
response_id="synthetic",
usage_details=usage,
)
return _coro()
super().__init__(
reply_selector=_select, default_reply=default_reply, tokens_per_reply=tokens_per_reply
)
@pytest.fixture()
@ -102,43 +66,24 @@ _PORTFOLIO_DEFAULT_REPLY = (
)
class _ProjectAwareUsageChatClient(SyntheticUsageChatClient):
"""A ``SyntheticUsageChatClient`` that selects its reply by scanning the incoming prompt for
a known ``project_id`` substring (the prompt embeds ``project.id`` at run.py:162 and
generate.py:48), falling back to a default valid proposal. This keeps ``run_portfolio``'s
single ``client_factory`` production-shaped while letting tests vary the proposal per
project."""
class _ProjectAwareUsageChatClient(ScriptedChatClient):
"""Selects its reply by scanning the incoming prompt for a known ``project_id`` substring (the
prompt embeds ``project.id`` at run.py:162 and generate.py:48), falling back to a default valid
proposal — so ``run_portfolio``'s single ``client_factory`` stays production-shaped while tests
vary the proposal per project. A THIN subclass: the prompt-scan lives in its selector, the shared
``_inner_get_response`` body in the canonical."""
def __init__(
self, replies: dict[str, str], *, default_reply: str, tokens_per_reply: int = 8
) -> None:
super().__init__(default_reply=default_reply, tokens_per_reply=tokens_per_reply)
self._replies = dict(replies)
table = dict(replies)
def _inner_get_response(
self, *, messages: Sequence[Message], stream: bool, options: Any, **kwargs: Any
) -> Any:
blob = " ".join(getattr(m, "text", "") or "" for m in messages)
reply = next((r for pid, r in self._replies.items() if pid in blob), self._default)
self.call_count += 1
usage = UsageDetails(total_token_count=self._tokens)
if stream:
def _select(blob: str, _role: str) -> str:
return next((r for pid, r in table.items() if pid in blob), default_reply)
async def _agen() -> Any:
yield ChatResponseUpdate(
role="assistant", contents=[{"type": "text", "text": reply}]
)
return self._build_response_stream(_agen())
async def _coro() -> ChatResponse:
return ChatResponse(
messages=[Message(role="assistant", contents=[reply])],
response_id="synthetic",
usage_details=usage,
)
return _coro()
super().__init__(
reply_selector=_select, default_reply=default_reply, tokens_per_reply=tokens_per_reply
)
@pytest.fixture()
@ -163,22 +108,14 @@ def make_portfolio_client_factory() -> Callable[..., Callable[[str], BaseChatCli
return _make
class _RecordingChatClient(SyntheticUsageChatClient):
class _RecordingChatClient(ScriptedChatClient):
"""Records the incoming prompt blob per call into a SHARED sink, then returns a fixed valid
reply. Lets a test assert exactly what text reached the prompt — the probe the Step-1 ExpeL
wiring is made load-bearing against (does a prior verdict reach the hypothesis prompt?)."""
wiring is made load-bearing against (does a prior verdict reach the hypothesis prompt?). A THIN
subclass: the canonical records to the ``sink`` (when given one) and returns the constant reply."""
def __init__(self, sink: list[str], reply: str, *, tokens_per_reply: int = 8) -> None:
super().__init__(default_reply=reply, tokens_per_reply=tokens_per_reply)
self._sink = sink
def _inner_get_response(
self, *, messages: Sequence[Message], stream: bool, options: Any, **kwargs: Any
) -> Any:
self._sink.append(" ".join(getattr(m, "text", "") or "" for m in messages))
return super()._inner_get_response(
messages=messages, stream=stream, options=options, **kwargs
)
super().__init__(reply, sink, tokens_per_reply=tokens_per_reply)
@pytest.fixture()

60
tests/test_scripted_client_consolidation.py Normal file
View file

@ -0,0 +1,60 @@
"""S2.5 (Step 9) consolidation guard (T-2.5e): the four scripted ``_inner_get_response`` bodies
collapsed to ONE canonical client (``simulation.ScriptedChatClient``); conftest's three test doubles
now SUBCLASS it. These grep-guards lock that in — they go RED if a divergent ``_inner_get_response``
is re-added, if a ``src``→``tests`` import creeps in, or if a double stops subclassing the canonical.
The guard is a regression lock over an already-verified consolidation: before the collapse there were
five ``def _inner_get_response`` sites (four scripted + test_step5's own-lineage double); after, two.
"""
from __future__ import annotations
from pathlib import Path
_ROOT = Path(__file__).resolve().parents[1]
def _py_files(base: str) -> list[Path]:
return sorted((_ROOT / base).rglob("*.py"))
def test_inner_get_response_collapsed_to_two_sites() -> None:
"""The four scripted clients collapse to ONE canonical ``_inner_get_response`` (simulation.py);
test_step5's own-lineage double is the only other def. So exactly 2 def-sites remain — NOT 5."""
sites = [
p.relative_to(_ROOT).as_posix()
for base in ("src", "tests")
for p in _py_files(base)
if p.name != Path(__file__).name # this guard file references the pattern in prose
and "def _inner_get_response" in p.read_text(encoding="utf-8")
]
assert sorted(sites) == [
"src/portfolio_optimiser/simulation.py",
"tests/test_step5_refine_loadbearing.py",
], f"expected the four scripted bodies collapsed to one canonical + test_step5's, got: {sites}"
def test_no_src_imports_tests() -> None:
"""No ``src`` module imports from ``tests`` — the canonical lives in ``src/simulation.py`` so
``conftest`` imports ``src``, never the reverse (the forbidden src→tests direction)."""
offenders = [
p.relative_to(_ROOT).as_posix()
for p in _py_files("src")
if ("from tests" in (text := p.read_text(encoding="utf-8")) or "import tests" in text)
]
assert offenders == [], f"src must not import tests: {offenders}"
def test_conftest_doubles_subclass_canonical() -> None:
"""conftest's three test doubles genuinely SUBCLASS the canonical ``ScriptedChatClient``
(delegating the shared body) — so the consolidation is real, not cosmetic."""
from conftest import (
ScriptedChatClient,
SyntheticUsageChatClient,
_ProjectAwareUsageChatClient,
_RecordingChatClient,
)
assert issubclass(SyntheticUsageChatClient, ScriptedChatClient)
assert issubclass(_ProjectAwareUsageChatClient, ScriptedChatClient)
assert issubclass(_RecordingChatClient, ScriptedChatClient)