feat(fase2): fresh_workflow factory + maker-checker GroupChat

This commit is contained in:
Kjell Tore Guttormsen 2026-06-24 13:41:30 +02:00
commit a0a8edd58c
2 changed files with 142 additions and 0 deletions

View file

@ -0,0 +1,83 @@
"""fresh_workflow isolation factory + maker-checker GroupChat (B7 / B4 / Layer-1 HITL).
Two Fase 1 findings are load-bearing here:
* **B7 (state isolation):** a *reused* built workflow accumulates the MAF conversation thread
across ``.run()`` calls, so project N contaminates N+1. The mitigation is a FACTORY that
builds a FRESH ``GroupChatBuilder`` with FRESH chat clients **per project run** zero
cross-run bleed. ``fresh_workflow`` is that factory.
* **B4 (bounded debate):** a never-converging debate does NOT self-terminate. ``with_max_rounds``
is the hard cap; the budget middleware (Step 4, wired by the orchestrator) is the external
token guard. ``make_termination`` is a higher turn-count safety net, not the sole stop.
**Layer-1 HITL** optionally enables the GA in-run synchronous review gate
(``with_request_info(agents=[checker])``) no checkpoint (research 01: durable checkpoint
resume is fragile; the durable verdict is captured out-of-band in Step 12).
"""
from __future__ import annotations
from collections.abc import Callable, Sequence
from agent_framework import Agent, BaseChatClient, Message
from agent_framework.orchestrations import GroupChatBuilder
_MAKER_CHECKER_ROLES = ("proposer", "checker")
_INSTRUCTIONS = {
"proposer": "You propose one concrete cost-saving measure for the project.",
"checker": "You critique the proposal and flag any constraint violation.",
}
def make_termination(n_turns: int) -> Callable[[Sequence[Message]], bool]:
"""A GroupChat ``TerminationCondition`` that stops once ``n_turns`` turns have been taken.
Deterministic and endpoint-free; the hard cap is ``with_max_rounds`` (B4)."""
if n_turns <= 0:
raise ValueError(f"n_turns must be positive, got {n_turns}")
def terminate(conversation: Sequence[Message]) -> bool:
return len(conversation) >= n_turns
return terminate
def maker_checker_agents(client_factory: Callable[[str], BaseChatClient]) -> list[Agent]:
"""FRESH proposer + checker agents, each backed by a FRESH client (zero cross-run state)."""
return [
Agent(client_factory(role), _INSTRUCTIONS[role], name=role)
for role in _MAKER_CHECKER_ROLES
]
def fresh_workflow(
client_factory: Callable[[str], BaseChatClient],
*,
max_rounds: int = 3,
enable_layer1_hitl: bool = False,
) -> object:
"""Build a FRESH maker-checker GroupChat with FRESH clients per call (B7). Bounded by
``with_max_rounds`` (B4) plus a higher turn-count termination safety net. ``client_factory``
is called once per role, so each run owns its own clients no state survives between runs.
"""
agents = maker_checker_agents(client_factory)
names = [a.name for a in agents]
counter = {"n": 0}
def select(_state: object) -> str:
choice = names[counter["n"] % len(names)]
counter["n"] += 1
return choice
builder = (
GroupChatBuilder(
participants=agents,
selection_func=select,
# Safety net well above the hard cap; with_max_rounds is the binding bound (B4).
termination_condition=make_termination(max_rounds * len(names) + 1),
)
.with_max_rounds(max_rounds)
)
if enable_layer1_hitl:
# Layer-1: in-run synchronous review on the checker (no checkpoint — research 01).
builder = builder.with_request_info(agents=[agents[-1]])
return builder.build()

59
tests/test_workflow.py Normal file
View file

@ -0,0 +1,59 @@
"""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