feat(fase1): spike A - group chat maker-checker vs single-agent [skip-docs]

spikes/a_groupchat.py

@@ -0,0 +1,191 @@
+"""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:
+        _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"
+    )