102 lines
4.1 KiB
Python
102 lines
4.1 KiB
Python
"""Step 9 tests — fresh_workflow isolation + maker-checker round cap (FakeChatClient, no LLM).
|
|
|
|
Isolation is asserted on received-message CONTENT (not a call counter — the Fase 1 round-2
|
|
fix): a fresh workflow per run gives each run a clean thread, so a participant sees ONLY its
|
|
own project. The round cap is pinned to the EXACT observed count (retiring the 1-turn=1-round
|
|
off-by-one assumption). Pattern: tests/spikes/test_b_footguns.py + tests/spikes/test_harness.py.
|
|
"""
|
|
|
|
from collections.abc import Callable
|
|
|
|
from agent_framework import BaseChatClient
|
|
from spikes._harness import FakeChatClient
|
|
|
|
from portfolio_optimiser.reference_domain import load_reference_projects
|
|
from portfolio_optimiser.workflow import fresh_workflow
|
|
|
|
|
|
def _recording_factory(created: list[FakeChatClient]) -> Callable[[str], BaseChatClient]:
|
|
def factory(role: str) -> BaseChatClient:
|
|
client = FakeChatClient(default_reply=f"{role} view")
|
|
created.append(client)
|
|
return client
|
|
|
|
return factory
|
|
|
|
|
|
async def test_participant_sees_only_its_own_project() -> None:
|
|
ids = [p.id for p in load_reference_projects()]
|
|
seen_per_run: list[list[str]] = []
|
|
for pid in ids:
|
|
created: list[FakeChatClient] = []
|
|
wf = fresh_workflow(_recording_factory(created), max_rounds=2)
|
|
await wf.run(f"Evaluate project {pid}.")
|
|
blob = " ".join(t for c in created for call in c.received_texts for t in call)
|
|
seen_per_run.append([q for q in ids if q in blob])
|
|
# A fresh instance per run -> each participant sees ONLY its own project (B7 mitigation).
|
|
assert seen_per_run == [[pid] for pid in ids]
|
|
|
|
|
|
async def test_never_converging_debate_halts_at_exactly_max_rounds() -> None:
|
|
created: list[FakeChatClient] = []
|
|
|
|
def factory(role: str) -> BaseChatClient:
|
|
client = FakeChatClient(default_reply="still disagreeing, keep debating")
|
|
created.append(client)
|
|
return client
|
|
|
|
wf = fresh_workflow(factory, max_rounds=3)
|
|
await wf.run("Debate a never-ending question.")
|
|
# with_max_rounds(3) is the hard cap; the debate halts at EXACTLY 3 turns, not <=.
|
|
assert sum(c.call_count for c in created) == 3
|
|
|
|
|
|
def test_layer1_hitl_option_builds() -> None:
|
|
def factory(role: str) -> BaseChatClient:
|
|
return FakeChatClient(default_reply="ok")
|
|
|
|
wf = fresh_workflow(factory, max_rounds=2, enable_layer1_hitl=True)
|
|
assert wf is not None
|
|
|
|
|
|
def test_tools_and_middleware_threaded_to_each_agent(monkeypatch) -> None:
|
|
"""Construction-spy: every agent is built WITH the provided tools + middleware (F7 + F2
|
|
wiring). `.tools` is not a public attr on a built Agent, so assert at the construction
|
|
boundary, not by reading the agent back."""
|
|
from portfolio_optimiser import workflow as wf_mod
|
|
|
|
recorded: list[dict[str, object]] = []
|
|
|
|
class _RecordingAgent:
|
|
def __init__(self, *args: object, **kwargs: object) -> None:
|
|
recorded.append(kwargs)
|
|
|
|
monkeypatch.setattr(wf_mod, "Agent", _RecordingAgent)
|
|
|
|
sentinel_tool = object()
|
|
sentinel_mw = object()
|
|
|
|
def factory(role: str) -> BaseChatClient:
|
|
return FakeChatClient(default_reply="ok")
|
|
|
|
wf_mod.maker_checker_agents(factory, tools=[sentinel_tool], middleware=[sentinel_mw])
|
|
|
|
assert len(recorded) == 2 # proposer + checker
|
|
for kwargs in recorded:
|
|
assert kwargs["tools"] == [sentinel_tool]
|
|
assert kwargs["middleware"] == [sentinel_mw]
|
|
|
|
|
|
async def test_output_from_surfaces_the_proposer_output() -> None:
|
|
"""Behavioral: the debate surfaces the PROPOSER's converged output via output_from=[proposer]
|
|
— without it, get_outputs() yields only the orchestrator's 'reached max rounds' notice
|
|
(verified). Step 4's F1 fix consumes this."""
|
|
marker = "PROPOSER_MARKER_7f3a"
|
|
|
|
def factory(role: str) -> BaseChatClient:
|
|
return FakeChatClient(default_reply=marker if role == "proposer" else "checker view")
|
|
|
|
wf = fresh_workflow(factory, max_rounds=2)
|
|
result = await wf.run("Find a cost-saving measure for FV42.")
|
|
texts = [getattr(o, "text", "") or "" for o in result.get_outputs()]
|
|
assert any(marker in t for t in texts), f"proposer output not surfaced; got {texts!r}"
|