portfolio-optimiser/tests/spikes/test_e_magentic.py

"""Spike E tests — Magentic exploration loop, MEASURED before anything is built
(order 20260823T162224Z, plan § D.1 spikes S0–S6).

Every test here pins ONE binary outcome the plan's § F assumption table needs. Where an
outcome is version-dependent (the manager's session lifetime changed in orchestrations
1.0.1, upstream regression fix #4371), the test asserts against the STRUCTURAL probe of the
installed manager — never a version string — and carries a non-vacuity control so a zero
can never be the absence of a run.

Pattern: tests/spikes/test_b_footguns.py.
"""

from __future__ import annotations

from pathlib import Path

import pytest

from portfolio_optimiser.budget import BudgetExceeded
from spikes.e_magentic import (
    ExplorationCallRecord,
    checkpoint_until_plan_review,
    expert_liaison_answer_round_trip,
    fresh_manager_contamination,
    manager_budget_enforced,
    manager_keeps_persistent_session,
    plan_review_round_trip,
    reset_signal_resets_participant_session,
    run_resume_subprocess,
    shared_builder_contamination,
    shared_manager_contamination,
    single_use_second_run,
    validator_latency_seconds,
)

# ---------------------------------------------------------------------------
# S1 — B7 state isolation (E1–E4, E7) in the repo's own form
# ---------------------------------------------------------------------------


async def test_e1_second_run_of_one_built_workflow_is_refused() -> None:
    """E1: a built Magentic workflow is SINGLE-USE. The second ``.run()`` raises and makes
    ZERO model calls — stronger than GroupChat 1.9.0's silent empty [2, 0, 0] re-run, because
    a workflow that cannot run cannot fabricate an answer."""
    result = await single_use_second_run()
    assert result["first_ok"] is True
    assert result["second_error"] == "RuntimeError"
    assert "already been completed" in result["second_message"]
    # The refusal is FREE: nothing was spent proving it.
    assert result["manager_calls_added"] == 0
    assert result["worker_calls_added"] == 0


async def test_e3_fresh_manager_per_build_never_contaminates() -> None:
    """E3 — the CONTROL, and the invariant the plan's C.4 rule rests on: a fresh manager
    (fresh builder, fresh agent, fresh client) per exploration leaks nothing across runs, on
    EVERY measured version. This is what ``fresh_exploration_workflow`` will implement."""
    bled, total, kinds = await fresh_manager_contamination()
    assert total >= 4, f"only {total} manager calls — the control would be reporting on nothing"
    assert bled == 0
    # The five-call shape (facts, plan, ledger_UNSAT, ledger_SAT, final) is the discriminator:
    # a contaminated run answers satisfied on its FIRST ledger and never calls the worker.
    assert kinds == ["facts", "plan", "ledger_UNSAT", "ledger_SAT", "final"]


async def test_e2_shared_manager_bleed_tracks_the_persistent_session() -> None:
    """E2: two workflows sharing ONE ``StandardMagenticManager``. Whether run 2's manager still
    sees run 1's task is decided by ONE structural property of the installed manager — whether
    it holds a persistent ``AgentSession`` built in ``__init__`` (1.0.0) or creates a throwaway
    one per call (1.0.1, #4371). The probe reads that property, never a version string."""
    bled, total, kinds = await shared_manager_contamination()
    assert total >= 4, f"only {total} manager calls — a zero here would be vacuous"
    if manager_keeps_persistent_session():
        assert bled == total, "a persistent manager session must bleed EVERY run-2 call"
        # The fabricated-answer signature: satisfied on the first ledger, worker never called.
        assert "ledger_UNSAT" not in kinds
    else:
        assert bled == 0, "a per-call manager session must leak nothing"
        assert kinds == ["facts", "plan", "ledger_UNSAT", "ledger_SAT", "final"]


async def test_e4_shared_builder_bleed_tracks_the_persistent_session() -> None:
    """E4: ONE ``MagenticBuilder``, two ``.build()`` calls. ``manager_agent=`` constructs the
    manager eagerly and hands the SAME instance to every build, so E4 is E2 reached by the
    route a caller is most likely to take by accident. Same discriminator."""
    bled, total, kinds = await shared_builder_contamination()
    assert total >= 4, f"only {total} manager calls — a zero here would be vacuous"
    if manager_keeps_persistent_session():
        assert bled == total
        assert "ledger_UNSAT" not in kinds
    else:
        assert bled == 0
        assert kinds == ["facts", "plan", "ledger_UNSAT", "ledger_SAT", "final"]


async def test_e7_reset_signal_does_not_reset_the_participant_session() -> None:
    """E7: ``MagenticResetSignal`` clears the cache and the conversation but writes the fresh
    session to ``_agent_thread`` — an attribute the executor never reads. Stall-replan therefore
    gives a fresh manager ledger and STALE participants. Measured on 1.0.0 and still true on
    1.0.1: never rely on the reset to empty participant memory."""
    result = await reset_signal_resets_participant_session()
    assert result["cache_cleared"] is True
    assert result["conversation_cleared"] is True
    assert result["session_identity_changed"] is False  # the whole finding
    assert result["orphan_attribute_written"] is True


# ---------------------------------------------------------------------------
# S2 — does BudgetMiddleware fire on the MANAGER's calls? (plan § F / A1, A2)
# ---------------------------------------------------------------------------


async def test_s2_budget_middleware_fires_on_the_manager_path() -> None:
    """A1 + A2: the manager is the most talkative participant, and the plan's hard token cap is
    a lie if agent-level ``ChatMiddleware`` does not reach it. Two halves, both required: the
    typed ``BudgetExceeded`` must LEAVE ``workflow.run`` (A2), and the meter must have been
    CHARGED (A1) — a refusal with a zero meter would prove only that something raised."""
    result = await manager_budget_enforced(max_tokens=1, attach=True)
    assert result["raised"] == "BudgetExceeded"
    assert result["kind"] == "tokens"
    assert result["meter_tokens"] > 0, "the middleware never charged — it did not run"
    assert result["completed"] is False


async def test_s2_control_detaching_the_manager_middleware_lets_the_run_finish() -> None:
    """The detach control the order requires: with no middleware on the manager the SAME
    one-token budget stops nothing and the run completes. Without this the test above could
    pass on an implementation where anything at all raised."""
    result = await manager_budget_enforced(max_tokens=1, attach=False)
    assert result["raised"] is None
    assert result["completed"] is True
    assert result["meter_tokens"] == 0


async def test_s2_budget_exceeded_is_the_repo_type_not_a_look_alike() -> None:
    """The exception that leaves ``workflow.run`` must be the repo's own ``BudgetExceeded``
    carrying ``kind``/``limit``/``observed`` — the triple the 429 channel reads (kø-(y)). An
    orchestration layer that wrapped it in an ``ExceptionGroup`` would make the exploration
    layer's error mapping a fiction, so the object itself is asserted, not just its name."""
    result = await manager_budget_enforced(max_tokens=1, attach=True, return_exception=True)
    exc = result["exception"]
    assert isinstance(exc, BudgetExceeded)
    assert exc.kind == "tokens"
    assert exc.limit == 1
    assert exc.observed > exc.limit  # limit and observed must not be the same number


# ---------------------------------------------------------------------------
# S5 — quick_validate latency (plan C.0 level 1: the tool the hypothesiser calls)
# ---------------------------------------------------------------------------


def test_s5_validator_latency_is_affordable_as_an_in_loop_tool() -> None:
    """S5: ``validate_proposal`` is what ``quick_validate`` wraps. If a single call costs
    seconds, every hypothesis the manager tests costs wall-clock the contract has to budget."""
    median, samples = validator_latency_seconds(runs=20)
    assert samples == 20
    assert median > 0.0, "a zero median means the clock never moved — nothing was measured"
    assert median < 2.0, f"median {median:.3f}s per validate_proposal — budget it in the contract"


def test_call_record_is_a_plain_readable_row() -> None:
    """The record type carries the four facts every contamination verdict is computed from;
    a spike whose evidence cannot be printed is a spike nobody can re-check."""
    record = ExplorationCallRecord(kind="facts", messages=2, sees_alpha=True, sees_beta=False)
    assert record.kind == "facts"
    assert record.sees_alpha is True


@pytest.mark.parametrize("attach", [True, False])
async def test_s2_arms_disagree(attach: bool) -> None:
    """Both arms of S2 run under one parametrisation too, so a future refactor that made the
    two arms identical shows up as a shared outcome rather than as two green tests."""
    result = await manager_budget_enforced(max_tokens=1, attach=attach)
    assert result["completed"] is not attach


# ---------------------------------------------------------------------------
# S3 / S3b — the two HITL doors (plan § C.5 / § C.6, assumptions A3 and A5)
# ---------------------------------------------------------------------------


async def test_s3_plan_review_round_trip_revises_then_approves() -> None:
    """A3: the request/response round-trip the plan's synchronous HITL rests on.

    Four facts in one run: the review STOPS the workflow before any output; a ``revise`` replans
    and asks AGAIN (so an always-revising expert is an unbounded loop unless the contract caps
    it); the revise costs manager calls but NO ledger call (it is not a round); and an
    ``approve`` lets the loop finish."""
    result = await plan_review_round_trip()
    assert result["pending_before_review"] == 1
    assert result["request_type"] == "MagenticPlanReviewRequest"
    assert result["is_stalled"] is False
    assert result["stopped_without_output"] is True
    # Before the review the manager has only surveyed and planned -- it never reached a ledger.
    assert result["kinds_before_review"] == ["facts", "plan"]
    # The measured cost of one revise: two manager calls, zero ledger calls, zero rounds.
    assert result["revise_manager_calls"] == ["facts_update", "plan_update"]
    assert not any(k.startswith("ledger") for k in result["revise_manager_calls"])
    # ... and it asks again -- this is why max_plan_revisions must exist in the contract.
    assert result["pending_after_revise"] == 1
    assert result["outputs_after_approve"] == ["FINAL: the worker did it."]


async def test_s3b_expert_liaison_answer_reaches_the_manager() -> None:
    """A5: ``AgentApprovalExecutor`` as the ``expert_liaison`` PARTICIPANT — door 3 of § C.6.

    Reachable, and the expert's words do become context: the liaison sees them, and the manager
    sees them in a later prompt. The cost is two round-trips per human turn -- ``from_strings``
    feeds the answer back into the liaison and does NOT resume the manager (measured: zero
    manager calls between the two requests); only ``approve`` forwards the output."""
    answer = "EXPERT-SAYS-TEST-THE-LED-RETROFIT"
    result = await expert_liaison_answer_round_trip(answer=answer)
    assert result["reachable"] is True
    assert result["manager_calls_between_requests"] == []  # from_strings alone resumes nothing
    assert result["second_request"] == 1  # ... it asks again, with the answer folded in
    assert result["liaison_saw_answer"] is True
    assert result["manager_saw_answer"] is True  # the point: it becomes context for the next round
    assert result["outputs"] == ["FINAL: the worker did it."]


async def test_s3b_control_an_unsent_sentinel_never_reaches_the_manager() -> None:
    """The control that makes the assertion above non-vacuous. Same run, same expert answer --
    but the manager's prompts are scanned for a sentinel the expert NEVER sent. It must come
    back absent. Without this arm, ``manager_saw_answer is True`` would be equally consistent
    with a scanner that matches anything, and door 3 would look proven when it was not."""
    result = await expert_liaison_answer_round_trip(
        answer="EXPERT-SAYS-TEST-THE-LED-RETROFIT", probe="NEVER-SENT-SENTINEL-XYZ"
    )
    assert result["reachable"] is True
    assert result["liaison_saw_answer"] is True  # the run really happened
    assert result["manager_saw_answer"] is False  # ... and the unsent sentinel is nowhere in it


# ---------------------------------------------------------------------------
# S4 — resume a pending plan review in a NEW PROCESS (plan U12, assumption A4)
# ---------------------------------------------------------------------------


async def test_s4_pending_plan_review_resumes_in_a_fresh_process(tmp_path: Path) -> None:
    """A4: the asynchronous HITL time-scale. The parent runs until the plan review stops it and
    leaves checkpoints on disk; a SEPARATE interpreter -- which never saw the run -- answers the
    pending request from the checkpoint alone and drives the workflow to its final answer.

    In-process resume would prove nothing here: U12's whole claim is that the expert can answer
    days later from a file inbox."""
    storage = tmp_path / "checkpoints"
    storage.mkdir()
    first = await checkpoint_until_plan_review(str(storage))
    assert first["request_id"], "no plan review was raised -- nothing to resume"
    assert first["outputs"] == []
    assert first["checkpoint_ids"], "no checkpoint was written -- the resume would be vacuous"

    resumed = run_resume_subprocess(
        str(storage), request_id=first["request_id"], checkpoint_id=first["checkpoint_ids"][-1]
    )
    assert resumed["outputs"] == ["FINAL: the worker did it."]
    assert resumed["pending_after_resume"] == 0
    # The fresh process really did the remaining work -- not a replay of a cached answer.
    assert any(k.startswith("ledger") for k in resumed["manager_kinds"])