portfolio-optimiser/spikes/b_footguns.py

"""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 built `ConcurrentBuilder` workflow
    across the three reference projects bleeds *conversation state*: MAF accumulates the
    shared thread across `.run()` calls, so each run's participants receive the PRIOR
    projects' prompts and replies (project N contaminates N+1's context). A FRESH instance
    per run — via `fresh_workflow()` — gives each run a clean thread (zero contamination).
    The observable is the message history each participant *receives*, captured via
    `FakeChatClient.received_texts` — NOT a call counter (a counter would rise for any
    reused mutable object and prove nothing about MAF state).

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)


def _projects_seen(received_texts: list[str], project_ids: list[str]) -> list[str]:
    """Which project ids appear anywhere in the messages an agent received on one call.

    The observable for genuine state bleed: if a later run's agent sees an EARLIER
    project's id, the workflow carried that project's conversation forward."""
    blob = " ".join(received_texts)
    return [pid for pid in project_ids if pid in blob]


async def shared_instance_conversation_bleed(project_ids: list[str]) -> list[list[str]]:
    """Reuse ONE built fan-out workflow across every project. MAF accumulates the shared
    thread across `.run()` calls, so run N's participants also receive runs 0..N-1's
    prompts/replies — genuine cross-run state corruption (G2/B7).

    Returns, per run (in order), which project ids were visible to a participant on that
    run. With a reused instance this grows monotonically: ``[[p0], [p0, p1], [p0, p1, p2]]``."""
    workflow, clients = fresh_workflow()
    for pid in project_ids:
        await workflow.run(f"Evaluate project {pid}.")
    # One participant is representative: concurrent fan-out feeds every participant the
    # same accumulated thread. clients[0] was called once per run, in order.
    rep = clients[0]
    return [_projects_seen(call_view, project_ids) for call_view in rep.received_texts]


async def fresh_instance_conversation_isolation(project_ids: list[str]) -> list[list[str]]:
    """A FRESH instance per project (the B7 mitigation): each run gets a clean thread, so
    a participant sees ONLY its own project — zero cross-run contamination.

    Returns, per run, the project ids visible to a participant; each should be exactly
    its own: ``[[p0], [p1], [p2]]``."""
    seen_per_run: list[list[str]] = []
    for pid in project_ids:
        workflow, clients = fresh_workflow()
        await workflow.run(f"Evaluate project {pid}.")
        # fresh client -> exactly one call this run; read its single received view.
        seen_per_run.append(_projects_seen(clients[0].received_texts[0], project_ids))
    return seen_per_run