feat(fase1): shared spike harness + builder smoke (budget caps, fake client, live gate) [skip-docs]
This commit is contained in:
parent
ffbfe00317
commit
432476346c
3 changed files with 314 additions and 3 deletions
188
spikes/_harness.py
Normal file
188
spikes/_harness.py
Normal file
|
|
@ -0,0 +1,188 @@
|
|||
"""Shared test seams for the Fase 1 de-risk spikes (throwaway, dev-only).
|
||||
|
||||
Three reusable pieces every spike leans on:
|
||||
|
||||
1. **Cost/stop invariant (B4 / D6).** ``Budget`` refuses to start without positive
|
||||
token and round caps (``ValueError`` on a bad construction argument), and
|
||||
``TokenMeter`` raises ``BudgetExceeded`` the moment a cap is crossed at runtime.
|
||||
Two exception types is intentional: ``ValueError`` = bad ctor argument (you never
|
||||
even started), ``BudgetExceeded`` = a cap was breached while running.
|
||||
|
||||
2. **Deterministic fake model.** ``FakeChatClient`` subclasses the GA
|
||||
``BaseChatClient`` with scripted, deterministic replies and counts "tokens" by
|
||||
word-count — no network, no endpoint. ``fake_agent`` wraps it in a real
|
||||
``agent_framework.Agent`` so the orchestration builders get genuine participants.
|
||||
The Step 2 builder smoke (in ``tests/spikes/test_harness.py``) proves this client
|
||||
can actually drive the GA ``GroupChatBuilder`` / ``ConcurrentBuilder``.
|
||||
|
||||
3. **Gated live arm.** ``live_local_client_or_skip`` builds an
|
||||
``agent_framework.openai.OpenAIChatClient`` against an OpenAI-compatible LOCAL
|
||||
endpoint **directly** (the D2 ``LocalBackend`` seam is deliberately left un-wired
|
||||
until Fase 2, so ``src/`` stays untouched), or ``pytest.skip``s when the
|
||||
``PORTFOLIO_LOCAL_*`` env is unset. No silent egress (D6).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
from agent_framework import (
|
||||
Agent,
|
||||
BaseChatClient,
|
||||
ChatResponse,
|
||||
ChatResponseUpdate,
|
||||
Message,
|
||||
)
|
||||
|
||||
|
||||
class BudgetExceeded(RuntimeError):
|
||||
"""Raised the moment a runtime cap (tokens or rounds) is crossed (B4).
|
||||
|
||||
Carries the breached ``kind`` ("tokens" | "rounds"), the ``limit`` that was
|
||||
set, and the ``observed`` value that crossed it — a structured stop event,
|
||||
never a silent hang.
|
||||
"""
|
||||
|
||||
def __init__(self, kind: str, limit: int, observed: int) -> None:
|
||||
self.kind = kind
|
||||
self.limit = limit
|
||||
self.observed = observed
|
||||
super().__init__(f"budget exceeded: {kind} limit={limit} observed={observed}")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Budget:
|
||||
"""Hard token + round/iteration caps, required at startup (A4 / D6).
|
||||
|
||||
Refuses to construct without positive caps — fail-fast, never an unbounded loop.
|
||||
"""
|
||||
|
||||
max_tokens: int
|
||||
max_rounds: int
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.max_tokens <= 0:
|
||||
raise ValueError(f"max_tokens must be positive, got {self.max_tokens}")
|
||||
if self.max_rounds <= 0:
|
||||
raise ValueError(f"max_rounds must be positive, got {self.max_rounds}")
|
||||
|
||||
|
||||
class TokenMeter:
|
||||
"""Accumulates token and round usage against a ``Budget``; raises the moment
|
||||
a cap is crossed."""
|
||||
|
||||
def __init__(self, budget: Budget) -> None:
|
||||
self.budget = budget
|
||||
self.tokens = 0
|
||||
self.rounds = 0
|
||||
|
||||
def charge(self, tokens: int) -> int:
|
||||
"""Add ``tokens`` to the running total; raise ``BudgetExceeded`` if over cap."""
|
||||
self.tokens += tokens
|
||||
if self.tokens > self.budget.max_tokens:
|
||||
raise BudgetExceeded("tokens", self.budget.max_tokens, self.tokens)
|
||||
return self.tokens
|
||||
|
||||
def tick_round(self) -> int:
|
||||
"""Increment the round counter; raise ``BudgetExceeded`` if over cap."""
|
||||
self.rounds += 1
|
||||
if self.rounds > self.budget.max_rounds:
|
||||
raise BudgetExceeded("rounds", self.budget.max_rounds, self.rounds)
|
||||
return self.rounds
|
||||
|
||||
|
||||
def _word_tokens(text: str) -> int:
|
||||
"""Token proxy: word count. Deterministic, endpoint-free."""
|
||||
return len(text.split())
|
||||
|
||||
|
||||
class FakeChatClient(BaseChatClient):
|
||||
"""A deterministic, network-free ``BaseChatClient`` for driving MAF agents in tests.
|
||||
|
||||
Returns scripted replies in order; once the script is exhausted it falls back to
|
||||
``default_reply``. Counts "tokens" by word-count of each reply it emits, exposing
|
||||
``total_tokens`` and ``call_count`` for the spike measurements.
|
||||
"""
|
||||
|
||||
OTEL_PROVIDER_NAME = "fake"
|
||||
|
||||
def __init__(self, scripted: Sequence[str] | None = None, *, default_reply: str = "ok") -> None:
|
||||
super().__init__()
|
||||
self._scripted: list[str] = list(scripted or [])
|
||||
self._idx = 0
|
||||
self._default = default_reply
|
||||
self.total_tokens = 0
|
||||
self.call_count = 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
|
||||
self.total_tokens += _word_tokens(reply)
|
||||
return reply
|
||||
|
||||
def _inner_get_response(
|
||||
self,
|
||||
*,
|
||||
messages: Sequence[Message],
|
||||
stream: bool,
|
||||
options: Any,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
# Matches the GA BaseChatClient contract: return a ResponseStream when
|
||||
# streaming, otherwise an awaitable resolving to a ChatResponse.
|
||||
reply = self._next_reply()
|
||||
|
||||
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="fake",
|
||||
)
|
||||
|
||||
return _coro()
|
||||
|
||||
|
||||
def fake_agent(
|
||||
client: BaseChatClient,
|
||||
name: str,
|
||||
instructions: str = "You are a terse participant. Answer in one short line.",
|
||||
) -> Agent:
|
||||
"""Build a minimal real ``agent_framework.Agent`` backed by ``client`` so the
|
||||
orchestration builders get a genuine participant."""
|
||||
return Agent(client, instructions, name=name)
|
||||
|
||||
|
||||
def live_local_client_or_skip() -> Any:
|
||||
"""Build an ``OpenAIChatClient`` against the OpenAI-compatible LOCAL endpoint
|
||||
(``PORTFOLIO_LOCAL_BASE_URL`` + ``PORTFOLIO_LOCAL_MODEL``), or ``pytest.skip``
|
||||
when unset.
|
||||
|
||||
The D2 ``LocalBackend`` seam is intentionally NOT used here — its live wiring is a
|
||||
Fase 2 concern; the throwaway spike builds the client directly so ``src/`` stays
|
||||
untouched. No silent egress: without the env vars the live arm simply skips (D6).
|
||||
"""
|
||||
base_url = os.environ.get("PORTFOLIO_LOCAL_BASE_URL")
|
||||
model = os.environ.get("PORTFOLIO_LOCAL_MODEL")
|
||||
if not base_url or not model:
|
||||
import pytest
|
||||
|
||||
pytest.skip(
|
||||
"LOCAL endpoint not configured "
|
||||
"(set PORTFOLIO_LOCAL_BASE_URL and PORTFOLIO_LOCAL_MODEL to run the live arm)"
|
||||
)
|
||||
|
||||
from agent_framework.openai import OpenAIChatClient
|
||||
|
||||
# Most local OpenAI-compatible servers (Ollama / LM Studio) accept any non-empty
|
||||
# key; allow an override but default to a dummy so construction never blocks.
|
||||
api_key = os.environ.get("PORTFOLIO_LOCAL_API_KEY", "local")
|
||||
return OpenAIChatClient(model=model, api_key=api_key, base_url=base_url)
|
||||
|
|
@ -1,3 +0,0 @@
|
|||
"""Throwaway Fase 1 de-risk spike tests — dev-only, never shipped in the wheel;
|
||||
safe to delete after findings are recorded.
|
||||
"""
|
||||
126
tests/spikes/test_harness.py
Normal file
126
tests/spikes/test_harness.py
Normal file
|
|
@ -0,0 +1,126 @@
|
|||
"""Step 2: shared harness invariants + the builder smoke that de-risks Spikes A/B.
|
||||
|
||||
The builder smoke is the one place that proves ``FakeChatClient`` can actually drive
|
||||
the GA ``GroupChatBuilder`` / ``ConcurrentBuilder``. If it could not, that would itself
|
||||
be a primary de-risk finding (escalate) — but it does (see findings note).
|
||||
|
||||
Pattern: tests/test_backends.py (parametrize + pytest.raises).
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from agent_framework import Message
|
||||
from agent_framework.orchestrations import ConcurrentBuilder, GroupChatBuilder
|
||||
|
||||
from spikes._harness import (
|
||||
Budget,
|
||||
BudgetExceeded,
|
||||
FakeChatClient,
|
||||
TokenMeter,
|
||||
fake_agent,
|
||||
live_local_client_or_skip,
|
||||
)
|
||||
|
||||
|
||||
# --- Budget: refuse to start without positive caps (A4 / fail-fast) ---
|
||||
|
||||
|
||||
@pytest.mark.parametrize("caps", [(0, 5), (5, 0), (-1, 5), (5, -3)])
|
||||
def test_budget_rejects_non_positive_caps(caps: tuple[int, int]) -> None:
|
||||
with pytest.raises(ValueError):
|
||||
Budget(*caps)
|
||||
|
||||
|
||||
def test_budget_accepts_positive_caps() -> None:
|
||||
b = Budget(max_tokens=100, max_rounds=3)
|
||||
assert b.max_tokens == 100
|
||||
assert b.max_rounds == 3
|
||||
|
||||
|
||||
# --- TokenMeter: structured stop the moment a cap is crossed (B4) ---
|
||||
|
||||
|
||||
def test_token_meter_charges_until_cap() -> None:
|
||||
meter = TokenMeter(Budget(max_tokens=10, max_rounds=3))
|
||||
assert meter.charge(4) == 4
|
||||
assert meter.charge(6) == 10 # exactly at cap is fine
|
||||
with pytest.raises(BudgetExceeded) as exc:
|
||||
meter.charge(1)
|
||||
assert exc.value.kind == "tokens"
|
||||
assert exc.value.limit == 10
|
||||
assert exc.value.observed == 11
|
||||
|
||||
|
||||
def test_token_meter_ticks_rounds_until_cap() -> None:
|
||||
meter = TokenMeter(Budget(max_tokens=100, max_rounds=2))
|
||||
assert meter.tick_round() == 1
|
||||
assert meter.tick_round() == 2
|
||||
with pytest.raises(BudgetExceeded) as exc:
|
||||
meter.tick_round()
|
||||
assert exc.value.kind == "rounds"
|
||||
assert exc.value.limit == 2
|
||||
|
||||
|
||||
# --- FakeChatClient: deterministic scripted replies + word-count tokens ---
|
||||
|
||||
|
||||
async def test_fake_client_returns_scripted_replies_in_order() -> None:
|
||||
client = FakeChatClient(["one two", "three"], default_reply="fallback word")
|
||||
r1 = await client.get_response([Message(role="user", contents=["hi"])])
|
||||
r2 = await client.get_response([Message(role="user", contents=["hi"])])
|
||||
r3 = await client.get_response([Message(role="user", contents=["hi"])]) # script exhausted
|
||||
assert r1.text == "one two"
|
||||
assert r2.text == "three"
|
||||
assert r3.text == "fallback word"
|
||||
assert client.call_count == 3
|
||||
assert client.total_tokens == 2 + 1 + 2 # word counts
|
||||
|
||||
|
||||
# --- Live gate: skip (no silent egress) when the LOCAL endpoint is unconfigured ---
|
||||
|
||||
|
||||
def test_live_client_skips_without_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("PORTFOLIO_LOCAL_BASE_URL", raising=False)
|
||||
monkeypatch.delenv("PORTFOLIO_LOCAL_MODEL", raising=False)
|
||||
with pytest.raises(pytest.skip.Exception):
|
||||
live_local_client_or_skip()
|
||||
|
||||
|
||||
# --- Builder smoke: the front-loaded de-risk for Spikes A/B ---
|
||||
|
||||
|
||||
async def test_builder_smoke_concurrent_runs_with_fake_agents() -> None:
|
||||
c1 = FakeChatClient(default_reply="alpha view")
|
||||
c2 = FakeChatClient(default_reply="beta view")
|
||||
workflow = ConcurrentBuilder(
|
||||
participants=[fake_agent(c1, "alpha"), fake_agent(c2, "beta")]
|
||||
).build()
|
||||
result = await workflow.run("Evaluate this trivial task.")
|
||||
assert result is not None
|
||||
assert c1.call_count == 1
|
||||
assert c2.call_count == 1
|
||||
|
||||
|
||||
async def test_builder_smoke_groupchat_runs_and_respects_round_cap() -> None:
|
||||
g1 = FakeChatClient(default_reply="proposer line")
|
||||
g2 = FakeChatClient(default_reply="critic line")
|
||||
names = ["proposer", "critic"]
|
||||
counter = {"n": 0}
|
||||
|
||||
def select(state: object) -> str:
|
||||
choice = names[counter["n"] % len(names)]
|
||||
counter["n"] += 1
|
||||
return choice
|
||||
|
||||
workflow = (
|
||||
GroupChatBuilder(
|
||||
participants=[fake_agent(g1, "proposer"), fake_agent(g2, "critic")],
|
||||
selection_func=select,
|
||||
)
|
||||
.with_max_rounds(3)
|
||||
.build()
|
||||
)
|
||||
result = await workflow.run("Debate this trivial task.")
|
||||
assert result is not None
|
||||
# The fake agents actually spoke; the round cap forced completion (no hang).
|
||||
assert g1.call_count + g2.call_count >= 1
|
||||
assert counter["n"] <= 3
|
||||
Loading…
Add table
Add a link
Reference in a new issue