portfolio-optimiser/spikes/a_groupchat.py

"""Spike A — Group Chat maker-checker vs single-agent baseline (U3 / G7).

De-risks the assumption that a maker-checker debate (proposer · critic · validator)
beats a single agent by enough to justify its multiplicative token cost.

Two layers (gate-green contract):

* **Logic layer (always green, no endpoint):** ``Arm``, ``make_termination``,
  ``verdict`` and ``render_comparison`` — the falsifiable cheaper/better decision,
  tested with *varied* inputs so the verdict function is non-tautological.
* **Integration/live layer (gated):** ``run_live`` builds the real maker-checker via
  ``GroupChatBuilder`` and a single-agent baseline against a reference project carrying
  a *planted flaw*, measures real numbers, and writes them to ``findings-a.md``. It
  ``skip``s without a LOCAL endpoint — the empirical better/cheaper verdict is honestly
  endpoint-dependent.
"""

from __future__ import annotations

from collections.abc import Callable, Sequence
from dataclasses import dataclass
from pathlib import Path

from agent_framework import Agent, Message
from agent_framework.orchestrations import GroupChatBuilder

from spikes._harness import live_local_client_or_skip

DEFAULT_ROUNDS = 3
"""N fixed at 3 (reviewer refinement #1) — a fair maker-checker bound under G7
token-discipline. If a live run cannot converge within 3, bump to <=5 and re-record."""

_FINDINGS = Path(__file__).resolve().parent.parent / "docs" / "fase1-spikes" / "findings-a.md"
_FLAW_MARKERS = ("negative", "infeasible", "exceeds", "violat", "invalid")


@dataclass(frozen=True)
class Arm:
    """Measured outcome of one debate arm."""

    label: str
    rounds: int
    stalls: int
    tokens: int
    caught_flaw: bool


def make_termination(n_rounds: int = DEFAULT_ROUNDS) -> Callable[[Sequence[Message]], bool]:
    """Return a GroupChat ``TerminationCondition`` that stops once ``n_rounds`` turns
    have been taken.

    The condition receives the running conversation (a flat ``list[Message]``) and
    returns ``True`` once it has at least ``n_rounds`` turns — a deterministic,
    endpoint-free stop the logic layer can test directly. The live arm also sets the
    builder's own ``with_max_rounds`` as a hard safety cap.
    """
    if n_rounds <= 0:
        raise ValueError(f"n_rounds must be positive, got {n_rounds}")

    def terminate(conversation: Sequence[Message]) -> bool:
        return len(conversation) >= n_rounds

    return terminate


def verdict(mc: Arm, single: Arm) -> dict[str, object]:
    """Decide whether the maker-checker arm is both *better* and *affordable*.

    * ``better``     — maker-checker caught the planted flaw and the single agent did not.
    * ``affordable`` — maker-checker spent no more than 3x the single agent's tokens (G7).
    * ``passed``     — both of the above.
    """
    better = mc.caught_flaw and not single.caught_flaw
    affordable = mc.tokens <= 3 * single.tokens
    return {
        "better": better,
        "affordable": affordable,
        "passed": better and affordable,
        "mc_tokens": mc.tokens,
        "single_tokens": single.tokens,
        "token_ratio": (mc.tokens / single.tokens) if single.tokens else float("inf"),
    }


def render_comparison(mc: Arm, single: Arm) -> str:
    """Render a markdown comparison table + verdict line for the findings note."""
    v = verdict(mc, single)
    ratio = v["token_ratio"]
    ratio_str = "inf" if ratio == float("inf") else f"{ratio:.2f}x"
    lines = [
        "| Arm | rounds | stalls | tokens | caught flaw |",
        "|-----|-------:|-------:|-------:|:-----------:|",
        f"| {mc.label} | {mc.rounds} | {mc.stalls} | {mc.tokens} | {mc.caught_flaw} |",
        f"| {single.label} | {single.rounds} | {single.stalls} | {single.tokens} | {single.caught_flaw} |",
        "",
        f"- token ratio (mc/single): **{ratio_str}** (affordable if <= 3x: {v['affordable']})",
        f"- better (mc caught & single missed): **{v['better']}**",
        f"- **verdict passed: {v['passed']}**",
    ]
    return "\n".join(lines)


def _word_tokens(text: str) -> int:
    return len(text.split())


def _caught_flaw(texts: Sequence[str]) -> bool:
    blob = " ".join(texts).lower()
    return any(marker in blob for marker in _FLAW_MARKERS)


async def run_live(
    *, n_rounds: int = DEFAULT_ROUNDS, write_findings: bool = True
) -> dict[str, object]:
    """Gated empirical arm: run maker-checker vs single-agent against a reference
    project carrying a planted flaw, measure, and write ``findings-a.md``.

    ``skip``s when no LOCAL endpoint is configured (no silent egress, D6). Token counts
    are a word-count proxy over assistant outputs — honest and endpoint-dependent.
    """
    client = live_local_client_or_skip()  # pytest.skip if PORTFOLIO_LOCAL_* unset

    # Planted flaw: a savings candidate that violates an obvious constraint.
    task = (
        "A project proposes cutting cost code 'B20' by removing 1200 m2 of formwork, "
        "but the project only has 800 m2 — a NEGATIVE residual quantity. "
        "Decide whether this savings proposal is valid."
    )

    # --- maker-checker arm (3 roles) ---
    proposer = Agent(client, "You propose one concrete cost-saving measure.", name="proposer")
    critic = Agent(
        client, "You critique the proposal and flag any constraint violation.", name="critic"
    )
    validator = Agent(
        client, "You give the final valid/invalid decision with a reason.", name="validator"
    )
    names = ["proposer", "critic", "validator"]
    counter = {"n": 0}

    def select(_state: object) -> str:
        choice = names[counter["n"] % len(names)]
        counter["n"] += 1
        return choice

    mc_wf = (
        GroupChatBuilder(
            participants=[proposer, critic, validator],
            selection_func=select,
            termination_condition=make_termination(n_rounds * len(names)),
        )
        .with_max_rounds(n_rounds)
        .build()
    )
    mc_res = await mc_wf.run(task)
    mc_texts = [o.text for o in mc_res.get_outputs() if getattr(o, "text", None)]
    mc = Arm(
        label="maker-checker",
        rounds=min(counter["n"], n_rounds),
        stalls=0,
        tokens=sum(_word_tokens(t) for t in mc_texts),
        caught_flaw=_caught_flaw(mc_texts),
    )

    # --- single-agent baseline ---
    solo = Agent(client, "You decide whether the savings proposal is valid.", name="solo")
    solo_res = await solo.run(task)
    solo_text = getattr(solo_res, "text", "") or ""
    single = Arm(
        label="single-agent",
        rounds=1,
        stalls=0,
        tokens=_word_tokens(solo_text),
        caught_flaw=_caught_flaw([solo_text]),
    )

    table = render_comparison(mc, single)
    if write_findings:
        # Guard the parent dir: the spike may run after the documented disposal
        # (`rm -rf docs/fase1-spikes`), and a missing dir must not discard the
        # just-measured live result with a bare FileNotFoundError.
        _FINDINGS.parent.mkdir(parents=True, exist_ok=True)
        _FINDINGS.write_text(_findings_doc(table), encoding="utf-8")
    return {"mc": mc, "single": single, "verdict": verdict(mc, single), "table": table}


def _findings_doc(table: str) -> str:
    return (
        "# Spike A findings — Group Chat maker-checker vs single-agent (U3 / G7)\n\n"
        "**Assumption:** a maker-checker debate beats a single agent by enough to justify "
        "its multiplicative token cost.\n\n"
        "## Measured (live arm)\n\n"
        f"{table}\n\n"
        "## Token use\n\n"
        "Token counts above are a word-count proxy over assistant outputs (live arm). "
        "The empirical better/cheaper verdict is **endpoint-dependent** — it is only "
        "produced when a LOCAL OpenAI-compatible endpoint is configured "
        "(`PORTFOLIO_LOCAL_BASE_URL` + `PORTFOLIO_LOCAL_MODEL`). Without an endpoint the "
        "live arm is skipped and the logic layer (verdict function) is what the quality "
        "gate proves.\n"
    )