122 lines
5.3 KiB
Python
122 lines
5.3 KiB
Python
"""Spike B — Magentic unbounded-termination footgun (G1/B4) + fan-out state
|
|
isolation (G2/B7).
|
|
|
|
Driving the GA builders with the fake client (de-risked by the Step 2 smoke) *is*
|
|
the experiment here — no live LLM, so this whole spike runs in the quality gate.
|
|
|
|
(a) **Magentic unbounded (G1/B4).** A `StandardMagenticManager` whose progress ledger
|
|
always says "not satisfied / progress being made" never finalizes. With
|
|
`max_round_count=None` it would loop forever; we drive it under the shared harness
|
|
round/iteration guard and confirm the guard is what stops it (it does NOT
|
|
self-terminate). With an explicit `max_round_count` it terminates cleanly.
|
|
|
|
(b) **Fan-out state isolation (G2/B7).** Reusing ONE workflow instance across the three
|
|
reference projects bleeds state (the shared fake clients accumulate calls); a FRESH
|
|
instance per run — via `fresh_workflow()` — shows zero bleed.
|
|
|
|
Token use: 0 — no live LLM (the fake client's "tokens" are word-counts of canned replies).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import warnings
|
|
|
|
from agent_framework import Agent
|
|
from agent_framework.orchestrations import (
|
|
ConcurrentBuilder,
|
|
MagenticBuilder,
|
|
StandardMagenticManager,
|
|
)
|
|
|
|
from spikes._harness import Budget, BudgetExceeded, FakeChatClient, TokenMeter, fake_agent
|
|
|
|
# Cutting an *instrumented* Magentic stream short (the only way to observe an unbounded
|
|
# run) makes OpenTelemetry log a benign "Failed to detach context" on generator close.
|
|
# It is cosmetic — silence it so the spike output stays readable. (Recorded in findings-b.)
|
|
logging.getLogger("opentelemetry.context").setLevel(logging.CRITICAL)
|
|
|
|
# A progress ledger that never reports satisfaction and always claims progress — so the
|
|
# manager has no natural reason to stop. next_speaker points at the single worker.
|
|
_NEVER_SATISFIED_LEDGER = json.dumps(
|
|
{
|
|
"is_request_satisfied": {"reason": "not yet", "answer": False},
|
|
"is_in_loop": {"reason": "no", "answer": False},
|
|
"is_progress_being_made": {"reason": "yes", "answer": True},
|
|
"next_speaker": {"reason": "worker should act", "answer": "worker"},
|
|
"instruction_or_question": {"reason": "continue", "answer": "Keep working."},
|
|
}
|
|
)
|
|
|
|
|
|
def _magentic(max_round_count: int | None) -> object:
|
|
"""Build a Magentic workflow whose manager never self-finalizes."""
|
|
manager = StandardMagenticManager(
|
|
agent=Agent(
|
|
FakeChatClient(default_reply=_NEVER_SATISFIED_LEDGER), "manager", name="manager"
|
|
),
|
|
max_round_count=max_round_count,
|
|
)
|
|
worker = Agent(FakeChatClient(default_reply="worker did some work"), "worker", name="worker")
|
|
return MagenticBuilder(participants=[worker], manager=manager).build()
|
|
|
|
|
|
async def unbounded_magentic_self_terminates(*, guard_rounds: int = 10) -> bool:
|
|
"""Drive an unbounded (`max_round_count=None`) Magentic under the harness guard.
|
|
|
|
Returns ``True`` if the workflow stopped on its own, ``False`` if the external guard
|
|
had to stop it. The footgun (G1/B4) is confirmed when this returns ``False``.
|
|
"""
|
|
workflow = _magentic(None)
|
|
meter = TokenMeter(Budget(max_tokens=10**9, max_rounds=guard_rounds))
|
|
guard_fired = False
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", RuntimeWarning)
|
|
async for _event in workflow.run("Do a never-ending task.", stream=True):
|
|
try:
|
|
meter.tick_round() # one tick per orchestration event (iteration guard)
|
|
except BudgetExceeded:
|
|
guard_fired = True
|
|
break
|
|
return not guard_fired
|
|
|
|
|
|
async def bounded_magentic_terminates(*, max_round_count: int = 2) -> bool:
|
|
"""An explicit `max_round_count` makes the same never-satisfied manager stop cleanly.
|
|
|
|
Returns ``True`` when the workflow runs to completion and yields an output.
|
|
"""
|
|
workflow = _magentic(max_round_count)
|
|
result = await workflow.run("Do a bounded task.")
|
|
return len(result.get_outputs()) >= 1
|
|
|
|
|
|
def fresh_workflow() -> tuple[object, tuple[FakeChatClient, FakeChatClient]]:
|
|
"""B7 mitigation: a factory that builds a FRESH fan-out workflow with FRESH clients
|
|
every call, so no state survives between runs."""
|
|
c1 = FakeChatClient(default_reply="participant one view")
|
|
c2 = FakeChatClient(default_reply="participant two view")
|
|
workflow = ConcurrentBuilder(participants=[fake_agent(c1, "w1"), fake_agent(c2, "w2")]).build()
|
|
return workflow, (c1, c2)
|
|
|
|
|
|
async def shared_instance_max_calls(project_ids: list[str]) -> int:
|
|
"""Reuse ONE fan-out instance across every project — state bleeds: the shared clients
|
|
accumulate calls across runs. Returns the max per-client call count (== len once
|
|
bled)."""
|
|
workflow, clients = fresh_workflow()
|
|
for pid in project_ids:
|
|
await workflow.run(f"Evaluate project {pid}.")
|
|
return max(c.call_count for c in clients)
|
|
|
|
|
|
async def fresh_instance_call_counts(project_ids: list[str]) -> list[int]:
|
|
"""A FRESH instance per project — zero bleed: every run's clients see exactly one
|
|
call. Returns the per-run max call count (each should be 1)."""
|
|
counts: list[int] = []
|
|
for pid in project_ids:
|
|
workflow, clients = fresh_workflow()
|
|
await workflow.run(f"Evaluate project {pid}.")
|
|
counts.append(max(c.call_count for c in clients))
|
|
return counts
|