feat(fase1): spike B - magentic unbounded + concurrent state isolation [skip-docs]
This commit is contained in:
parent
9b9a17e2ed
commit
44111113fb
3 changed files with 213 additions and 0 deletions
52
docs/fase1-spikes/findings-b.md
Normal file
52
docs/fase1-spikes/findings-b.md
Normal file
|
|
@ -0,0 +1,52 @@
|
|||
# Spike B findings — Magentic unbounded footgun (G1/B4) + fan-out state isolation (G2/B7)
|
||||
|
||||
**Assumptions:** (G1/B4) Magentic with `limits=None` does NOT self-terminate, so an
|
||||
explicit limit is mandatory; (G2/B7) reusing a shared fan-out builder/workflow instance
|
||||
corrupts state, while a fresh-instance helper prevents it.
|
||||
|
||||
This spike drives the **real** GA builders with the fake client (de-risked by the Step 2
|
||||
builder smoke), so it runs entirely in the quality gate — **no live LLM**.
|
||||
|
||||
## (a) Magentic unbounded termination — CONFIRMED
|
||||
|
||||
- A `StandardMagenticManager` whose progress ledger always reports *not satisfied /
|
||||
progress being made* never finalizes.
|
||||
- With **`max_round_count=None`** the workflow streams orchestration events indefinitely;
|
||||
only the shared harness round/iteration guard (`Budget` + `TokenMeter`) stops it.
|
||||
`unbounded_magentic_self_terminates()` returns **False** → it does **not** self-terminate.
|
||||
- With an explicit **`max_round_count=2`** the same manager stops cleanly
|
||||
("Magentic Orchestrator: Max round count reached") and yields an output.
|
||||
- **Implication (B4):** a stop-criterion/round cap is mandatory at startup for any
|
||||
Magentic use — never rely on natural termination. (Magentic stays experimental and is
|
||||
exercised here ONLY to confirm the footgun, never as the debate default — that remains
|
||||
Group Chat maker-checker, A2/G8.)
|
||||
|
||||
### Minor MAF observation (recorded)
|
||||
|
||||
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, and the
|
||||
in-flight fake response surfaces a `RuntimeWarning: coroutine was never awaited`. Both are
|
||||
cosmetic (no functional effect); the spike silences the OTel context logger and filters the
|
||||
warning. Worth knowing if Fase 2 observes streamed orchestrations.
|
||||
|
||||
## (b) Fan-out state isolation — CONFIRMED
|
||||
|
||||
- **Shared instance:** one `ConcurrentBuilder` workflow reused across the three reference
|
||||
projects → the shared fake clients accumulate calls (max call count == 3) → **state
|
||||
bleed**.
|
||||
- **Fresh instance:** `fresh_workflow()` builds a new workflow + fresh clients per run →
|
||||
each run's clients see exactly one call (`[1, 1, 1]`) → **zero bleed** across all 3
|
||||
projects (B7).
|
||||
- **Implication:** Fase 2 fan-out must build a fresh workflow/executor instance per
|
||||
project run (a factory like `fresh_workflow()`), never reuse a shared instance.
|
||||
|
||||
## Token use
|
||||
|
||||
**0 — no live LLM.** All behavior is driven by the deterministic `FakeChatClient`; its
|
||||
"tokens" are word-counts of canned replies and are not meaningful cost numbers here.
|
||||
|
||||
## Implication for Fase 2
|
||||
|
||||
Both footguns behave exactly as the §15 register predicted. The framework MUST: (1) require
|
||||
explicit round/stop caps for any Magentic-style loop (fail-fast at startup, D6/B4); and
|
||||
(2) use a fresh-instance-per-run factory for fan-out to avoid state corruption (B7).
|
||||
122
spikes/b_footguns.py
Normal file
122
spikes/b_footguns.py
Normal file
|
|
@ -0,0 +1,122 @@
|
|||
"""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
|
||||
39
tests/spikes/test_b_footguns.py
Normal file
39
tests/spikes/test_b_footguns.py
Normal file
|
|
@ -0,0 +1,39 @@
|
|||
"""Spike B tests — Magentic unbounded footgun (G1/B4) + fan-out state isolation (G2/B7).
|
||||
|
||||
These drive the real GA builders with the fake client (no endpoint), de-risked by the
|
||||
Step 2 builder smoke. Pattern: tests/test_backends.py.
|
||||
"""
|
||||
|
||||
from portfolio_optimiser.reference_domain import load_reference_projects
|
||||
from spikes.b_footguns import (
|
||||
bounded_magentic_terminates,
|
||||
fresh_instance_call_counts,
|
||||
shared_instance_max_calls,
|
||||
unbounded_magentic_self_terminates,
|
||||
)
|
||||
|
||||
|
||||
async def test_unbounded_magentic_does_not_self_terminate() -> None:
|
||||
# max_round_count=None never finalizes — only the external guard stops it (B4).
|
||||
self_terminated = await unbounded_magentic_self_terminates(guard_rounds=8)
|
||||
assert self_terminated is False # the guard fired, not a natural stop
|
||||
|
||||
|
||||
async def test_bounded_magentic_terminates_within_limit() -> None:
|
||||
# An explicit max_round_count makes the same never-satisfied manager stop cleanly.
|
||||
assert await bounded_magentic_terminates(max_round_count=2) is True
|
||||
|
||||
|
||||
async def test_shared_instance_bleeds_state_across_projects() -> None:
|
||||
ids = [p.id for p in load_reference_projects()]
|
||||
assert len(ids) == 3
|
||||
bled = await shared_instance_max_calls(ids)
|
||||
# One shared instance reused across all 3 projects -> calls accumulate -> bleed.
|
||||
assert bled == len(ids)
|
||||
|
||||
|
||||
async def test_fresh_instance_zero_bleed_across_projects() -> None:
|
||||
ids = [p.id for p in load_reference_projects()]
|
||||
counts = await fresh_instance_call_counts(ids)
|
||||
# A fresh instance per project -> each run sees exactly one call -> zero bleed.
|
||||
assert counts == [1, 1, 1]
|
||||
Loading…
Add table
Add a link
Reference in a new issue