portfolio-optimiser/spikes/e_magentic.py
Kjell Tore Guttormsen 9f0843ab6d test(spikes): S0-S6 maalt for Magentic-utforskningssloeyfa (ORDRE 20260823T162224Z) [skip-docs]
Maaling foer bygging. Alle aatte antakelsene i planens § F flyttet fra «umaalt» til
et maalt utfall; ingenting bygget, ingenting i src/.

S0 (versjon): 1.0.1 loeser paa core 1.9.0. Diffen mot 1.0.0 er upstream #4371 -
`StandardMagenticManager.__init__` mistet sin persistente `AgentSession`, og hvert
manager-kall mynter naa en engangs-sesjon. **E2 OG E4 er dermed BORTE (4/4 -> 0/5).**
E1 (single-use) og E7 (orphan `_agent_thread`, :1369) staar. Ordrens «felles hvis»
(E7 staar -> revert) hviler paa at 1.0.1 ikke kjoeper noe; den kjoepte noe stoerre
enn det som ble haapet, saa laasen staar paa 1.0.1 I PAAVENTE AV OPERATOEREN.

S1 (B7): E1-E4 + E7 i repoets form. Versjons-sensitiviteten testes mot en STRUKTURELL
sonde (holder manageren en persistent sesjon?), aldri en versjonsstreng - den sier
AARSAKEN og overlever en versjon planen ikke har sett.
S2 (budsjett): A1+A2 GROENNE. `BudgetMiddleware` fyrer paa manager-stien
(`meter.tokens == 8`), og `BudgetExceeded` forlater `workflow.run` som repoets EGEN
type med `kind`/`limit`/`observed` intakt - ikke pakket i en ExceptionGroup.
S3 (plan review): rundturen virker; en revise koster 2 manager-kall, 0 ledger-kall,
0 runder, og SPOER PAA NYTT -> `max_plan_revisions` maa inn i kontrakten.
S3b: doer 3 staar. To rundturer per menneskesvar; `from_strings` gjenopptar IKKE
manageren, kun `approve` gjoer det. Pris: `AgentApprovalExecutor` er ikke re-eksportert.
S4 (resume i NY prosess): GROENN. Pris: `FileCheckpointStorage` nekter aa deserialisere
plan-review-typene uten `allowed_checkpoint_types` - uten det feiler resume som et
FRAVAER (tom listing), ikke som en feil.
S5: median `validate_proposal` 13,6 ms - fritt kallbart i loekka.
S6 (scratch-venv, ingenting lagt til pyproject): 2 `workflow.run`-spans, men
`enable_console_exporters` skriver til STDOUT og ville oedelagt golden-transkriptet;
`ConsoleSpanExporter(out=sys.stderr)` gir spanene paa stderr OG byte-identisk stdout.

Klienten er repoets `ScriptedChatClient` og budsjett-typene er PRODUKSJONENS - en bar
`BaseChatClient` no-op-er middleware, og `spikes/_harness.py`s egen kopi er nettopp
grunnen til at koe-(y) fantes.

Load-bearing MAALT mot HELE suiten, groenn kontroll 920/5: konstant persistent-sesjon
(2 roede) · aldri fest middleware paa manageren (3 roede, detach-armen groenn) · detach
markoer-registreringen (1 roed) · flipp `_route`-rekkefoelgen (4 roede) · resume uten
`checkpoint_id` (1 roed) · builder uten `with_checkpointing` (1 roed) · tom
`_ALLOWED_CHECKPOINT_TYPES` (1 roed). Og EN falsifisert: resume uten
`checkpoint_storage=` gir 0 roede - planens E-tabell navngir feil detach-punkt, og
det er skrevet inn i § F i stedet for aa staa som en gate som ikke kan bli roed.

Planens V1-sti ble portabel i 2eb4622 (pakke-gaten var roed paa HEAD siden 2e33905).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UTEa7uw2JMxgijx8k8XgxG
2026-08-23 19:36:00 +02:00

646 lines
27 KiB
Python

"""Spike E — the Magentic exploration loop, measured BEFORE it is built
(order 20260823T162224Z; plan ``docs/plan/2026-08-23-magentic-utforskningssloeyfe.md`` § D.1).
Nothing here is production code and nothing here is wired into ``src/``. Each function is one
measurement whose outcome moves a row of the plan's § F assumption table from "umålt" to a fact.
**The client is the repo's own ``ScriptedChatClient``, not an ad-hoc fake.** The scratch scripts
this spike ports used a bare ``BaseChatClient``, on which ``BudgetMiddleware`` is silently a
no-op (measured, ``simulation.py:373-375``) — so a budget claim proved with one would have proved
nothing. ``ScriptedChatClient`` subclasses the LAYERED ``OpenAIChatCompletionClient``, which is
what makes S2 a real measurement.
**The budget types are the PRODUCTION ones** (``portfolio_optimiser.budget``), deliberately NOT
``spikes/_harness.py``'s private copy. The harness copy is exactly why ``tick_round``'s
``observed`` went four raise-sites without coverage (kø-(y)); S2's whole question is whether the
SHIPPED middleware reaches the manager, and only the shipped object can answer it.
**Routing is on the joined prompt blob, in a fixed priority order, and that order is
load-bearing.** ``ScriptedChatClient``'s selector receives the concatenation of every message in
the call, so one manager call carries two markers (the plan prompt is built on a history that
still holds the pre-survey text — measured: 1 ambiguous call in 5). Checking the later-stage
marker first resolves it; the five-kind call shape each experiment asserts is what proves the
routing stayed correct.
"""
from __future__ import annotations
import json
import statistics
import subprocess
import sys
import time
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from agent_framework import Agent, FileCheckpointStorage
from agent_framework.orchestrations import (
AgentRequestInfoResponse,
MagenticAgentExecutor,
MagenticBuilder,
MagenticPlanReviewResponse,
MagenticResetSignal,
StandardMagenticManager,
)
# ``AgentApprovalExecutor`` is the ONE name S3b needs that the package does not re-export
# (measured: ``hasattr(agent_framework.orchestrations, "AgentApprovalExecutor")`` is False while
# its response type IS public). Reaching into the private module is therefore part of the S3b
# FINDING, not an oversight: door 3 of § C.6 currently costs a private-API dependency, and that
# is a fact the operator's decision needs to carry.
from agent_framework_orchestrations._orchestration_request_info import AgentApprovalExecutor
import portfolio_optimiser
from portfolio_optimiser.budget import Budget, BudgetExceeded, BudgetMiddleware, TokenMeter
from portfolio_optimiser.ir import SavingsProposal
from portfolio_optimiser.okf import load_ir_projection, load_optional_cost_baseline
from portfolio_optimiser.simulation import ScriptedChatClient
from portfolio_optimiser.validator import validate_proposal
TASK_ALPHA = "TASK-ALPHA: find the saving in the alpha project."
TASK_BETA = "TASK-BETA: find the saving in the beta project."
WORKER_SENTINEL = "WORKER-SAW"
def _ledger(*, satisfied: bool, speaker: str) -> str:
"""A progress ledger naming ``speaker`` as the next talker.
The name is a PARAMETER because an unknown ``next_speaker`` is a silent footgun: the
orchestrator does not error on it, it quietly produces a final answer with zero participant
work (``_magentic.py:1128-1131``). Hard-coding "worker" here made the S3b run — whose
participant is the ``expert_liaison`` — finish without ever asking anyone (measured).
"""
return json.dumps(
{
"is_request_satisfied": {
"reason": "the participant replied" if satisfied else "no one has spoken yet",
"answer": satisfied,
},
"is_in_loop": {"reason": "no", "answer": False},
"is_progress_being_made": {"reason": "yes", "answer": True},
"next_speaker": {"reason": "it does the work", "answer": speaker},
"instruction_or_question": {
"reason": "done" if satisfied else "kick off",
"answer": "none" if satisfied else "Do the work now.",
},
}
)
@dataclass(frozen=True)
class ExplorationCallRecord:
"""One manager call, reduced to the facts every verdict here reads.
``sees_marker`` is the HITL half (S3b): whether a caller-supplied sentinel — an expert's
answer injected mid-run — had reached this manager prompt. It defaults to ``False`` so the
contamination experiments, which supply no marker, are unchanged.
"""
kind: str
messages: int
sees_alpha: bool
sees_beta: bool
sees_marker: bool = False
def _route(blob: str, speaker: str) -> tuple[str, str]:
"""Map a manager prompt blob to ``(kind, reply)``.
Order matters: the later-stage marker is tested FIRST because an earlier stage's text is
still present in the joined blob. Reversing two of these silently changes which prompt a
kind is attributed to, which is why every experiment asserts the resulting call shape.
"""
if "provide the final answer" in blob:
return "final", "FINAL: the worker did it."
if "pure JSON format" in blob:
if WORKER_SENTINEL in blob:
return "ledger_SAT", _ledger(satisfied=True, speaker=speaker)
return "ledger_UNSAT", _ledger(satisfied=False, speaker=speaker)
if "went wrong on this last run" in blob:
return "plan_update", "PLAN-UPDATE: ask the worker again."
if "rewrite the following fact sheet" in blob:
return "facts_update", "FACTS-UPDATE: still nothing."
if "bullet-point plan" in blob:
return "plan", "PLAN: - ask the worker"
if "pre-survey" in blob:
return "facts", "FACTS: nothing given."
return "unknown", "{}"
def _manager_client(
records: list[ExplorationCallRecord],
*,
marker: str | None = None,
speaker: str = "worker",
) -> ScriptedChatClient:
"""A manager-shaped scripted client appending one record per call to ``records``.
``records`` is CALLER-owned (the parse-failure-capture precedent): the evidence must survive
however the run ended, including a run cut short by ``BudgetExceeded`` mid-way.
"""
def _select(blob: str, _role: str) -> str:
kind, reply = _route(blob, speaker)
records.append(
ExplorationCallRecord(
kind=kind,
messages=blob.count("\n") + 1,
sees_alpha="TASK-ALPHA" in blob,
sees_beta="TASK-BETA" in blob,
sees_marker=marker is not None and marker in blob,
)
)
return reply
return ScriptedChatClient(reply_selector=_select, role="manager")
def _worker_client(seen: list[str]) -> ScriptedChatClient:
def _select(blob: str, _role: str) -> str:
which = "ALPHA" if "TASK-ALPHA" in blob else ("BETA" if "TASK-BETA" in blob else "NOTHING")
seen.append(which)
return f"{WORKER_SENTINEL}-{which}"
return ScriptedChatClient(reply_selector=_select, role="worker")
def _manager_agent(client: ScriptedChatClient, *, middleware: Sequence[Any] | None = None) -> Agent:
return Agent(
client,
"You are the Magentic manager.",
name="manager",
description="plans the work",
middleware=middleware,
)
def _worker_agent(client: ScriptedChatClient) -> Agent:
return Agent(client, "You are the worker.", name="worker", description="does the work")
async def _run(workflow: Any, task: str) -> dict[str, Any]:
try:
result = await workflow.run(task)
outputs = [str(o) for o in (result.get_outputs() or [])]
return {"ok": True, "outputs": outputs}
except Exception as exc: # noqa: BLE001 - the failure mode IS the measurement
return {"ok": False, "error": type(exc).__name__, "message": str(exc)}
def manager_keeps_persistent_session() -> bool:
"""Does the INSTALLED ``StandardMagenticManager`` hold one ``AgentSession`` for its whole
life, or mint a throwaway one per call?
This is the single structural property that decides E2 and E4. Orchestrations 1.0.0 assigns
``self._session = self._agent.create_session()`` in ``__init__``; 1.0.1 removed that line and
creates the session inside the call instead (upstream regression fix #4371). Probing the
attribute rather than the version string states the CAUSE, and keeps the spike honest across
a version the plan has not seen.
"""
manager = StandardMagenticManager(agent=_manager_agent(_manager_client([])))
return hasattr(manager, "_session")
async def single_use_second_run() -> dict[str, Any]:
"""E1: build one Magentic workflow, run it twice. Measures the second run's outcome and
what it COST — a refusal that still made model calls would be a different finding."""
records: list[ExplorationCallRecord] = []
worker_seen: list[str] = []
workflow = MagenticBuilder(
participants=[_worker_agent(_worker_client(worker_seen))],
manager_agent=_manager_agent(_manager_client(records)),
max_round_count=6,
).build()
first = await _run(workflow, TASK_ALPHA)
manager_before, worker_before = len(records), len(worker_seen)
second = await _run(workflow, TASK_BETA)
return {
"first_ok": first["ok"],
"second_error": second.get("error"),
"second_message": second.get("message", ""),
"manager_calls_added": len(records) - manager_before,
"worker_calls_added": len(worker_seen) - worker_before,
}
async def _two_runs(
build_second: Callable[[list[ExplorationCallRecord]], Any],
*,
shared_records: bool,
) -> tuple[int, int, list[str]]:
"""Run ALPHA, then BETA, and report how many of run 2's manager calls still see ALPHA.
``shared_records`` says whether run 2's manager is the same object as run 1's — when it is,
run 2's records are the tail of one list; when it is not, they are their own list. Both
shapes reduce to the same verdict triple so the callers stay comparable.
"""
records: list[ExplorationCallRecord] = []
first_worker: list[str] = []
first = MagenticBuilder(
participants=[_worker_agent(_worker_client(first_worker))],
manager_agent=_manager_agent(_manager_client(records)),
max_round_count=6,
).build()
await _run(first, TASK_ALPHA)
split = len(records)
second_records = records if shared_records else []
second = build_second(second_records)
await _run(second, TASK_BETA)
tail = records[split:] if shared_records else second_records
bled = sum(1 for record in tail if record.sees_alpha)
return bled, len(tail), [record.kind for record in tail]
async def shared_manager_contamination() -> tuple[int, int, list[str]]:
"""E2: two builders sharing ONE ``StandardMagenticManager`` instance.
Built directly (not via ``_two_runs``) because run 1 must go through the SAME manager object,
which ``manager=`` accepts and ``manager_agent=`` does not.
"""
records: list[ExplorationCallRecord] = []
shared = StandardMagenticManager(
agent=_manager_agent(_manager_client(records)), max_round_count=6
)
first = MagenticBuilder(
participants=[_worker_agent(_worker_client([]))], manager=shared
).build()
await _run(first, TASK_ALPHA)
split = len(records)
second = MagenticBuilder(
participants=[_worker_agent(_worker_client([]))], manager=shared
).build()
await _run(second, TASK_BETA)
tail = records[split:]
return sum(1 for r in tail if r.sees_alpha), len(tail), [r.kind for r in tail]
async def shared_builder_contamination() -> tuple[int, int, list[str]]:
"""E4: ONE ``MagenticBuilder``, ``.build()`` twice. ``manager_agent=`` constructs the manager
eagerly and hands the same instance to every build — the accidental route into E2."""
records: list[ExplorationCallRecord] = []
builder = MagenticBuilder(
participants=[_worker_agent(_worker_client([]))],
manager_agent=_manager_agent(_manager_client(records)),
max_round_count=6,
)
await _run(builder.build(), TASK_ALPHA)
split = len(records)
await _run(builder.build(), TASK_BETA)
tail = records[split:]
return sum(1 for r in tail if r.sees_alpha), len(tail), [r.kind for r in tail]
async def fresh_manager_contamination() -> tuple[int, int, list[str]]:
"""E3, the control and the mitigation: a fresh builder, agent and client per exploration."""
def _second(records: list[ExplorationCallRecord]) -> Any:
return MagenticBuilder(
participants=[_worker_agent(_worker_client([]))],
manager_agent=_manager_agent(_manager_client(records)),
max_round_count=6,
).build()
return await _two_runs(_second, shared_records=False)
async def reset_signal_resets_participant_session() -> dict[str, Any]:
"""E7: does ``MagenticResetSignal`` actually give the participant a clean session?
Measured no: the fresh session is written to ``_agent_thread``, which nothing reads, while
the live ``_session`` keeps its identity. The stall-replan path therefore hands the manager a
clean ledger and the participants their old memory.
"""
executor = MagenticAgentExecutor(_worker_agent(_worker_client([])))
session_before = id(executor._session)
executor._cache.append("sentinel") # type: ignore[arg-type]
executor._full_conversation.append("sentinel") # type: ignore[arg-type]
await executor.handle_magentic_reset(MagenticResetSignal(), None) # type: ignore[arg-type]
return {
"cache_cleared": len(executor._cache) == 0,
"conversation_cleared": len(executor._full_conversation) == 0,
"session_identity_changed": id(executor._session) != session_before,
"orphan_attribute_written": hasattr(executor, "_agent_thread"),
}
async def manager_budget_enforced(
*, max_tokens: int, attach: bool, return_exception: bool = False
) -> dict[str, Any]:
"""S2: put the SHIPPED ``BudgetMiddleware`` on the manager agent and see whether the typed
refusal leaves ``workflow.run``.
``attach=False`` is the detach control: the same one-token budget with no middleware must
stop nothing. Without that arm the positive test would pass on any implementation in which
something, anything, raised.
"""
meter = TokenMeter(Budget(max_tokens=max_tokens, max_rounds=8))
middleware = [BudgetMiddleware(meter)] if attach else None
workflow = MagenticBuilder(
participants=[_worker_agent(_worker_client([]))],
manager_agent=_manager_agent(_manager_client([]), middleware=middleware),
max_round_count=6,
).build()
raised: str | None = None
kind: str | None = None
exception: BaseException | None = None
completed = False
try:
result = await workflow.run(TASK_ALPHA)
completed = bool(result.get_outputs())
except BudgetExceeded as exc:
raised, kind, exception = type(exc).__name__, exc.kind, exc
except Exception as exc: # noqa: BLE001 - a DIFFERENT exception type is itself the finding
raised, exception = type(exc).__name__, exc
payload: dict[str, Any] = {
"raised": raised,
"kind": kind,
"completed": completed,
"meter_tokens": meter.tokens,
}
if return_exception:
payload["exception"] = exception
return payload
# ---------------------------------------------------------------------------
# S3 / S3b — the two HITL doors (plan § C.5, § C.6)
# ---------------------------------------------------------------------------
def _pending_requests(result: Any) -> list[Any]:
return [event for event in result if event.type == "request_info"]
async def plan_review_round_trip() -> dict[str, Any]:
"""S3: ``enable_plan_review=True`` → the run stops with a ``MagenticPlanReviewRequest`` and no
output → ``revise(...)`` replans and asks AGAIN → ``approve()`` lets the loop run.
The manager-call count PER revise is the number the contract needs: a revise costs model
calls but is not counted as a round by the orchestration (measured: no ledger call), so an
uncapped reviser is an unbounded spend the plan's ``max_plan_revisions`` has to bound.
"""
records: list[ExplorationCallRecord] = []
workflow = MagenticBuilder(
participants=[_worker_agent(_worker_client([]))],
manager_agent=_manager_agent(_manager_client(records)),
max_round_count=6,
enable_plan_review=True,
).build()
first = await workflow.run(TASK_ALPHA)
pending = _pending_requests(first)
stopped_without_output = not first.get_outputs()
review_kinds = [r.kind for r in records]
before_revise = len(records)
revised = await workflow.run(
responses={pending[0].request_id: MagenticPlanReviewResponse.revise("Test the LED case.")}
)
revise_calls = [r.kind for r in records[before_revise:]]
pending_after_revise = _pending_requests(revised)
approve_id = (
pending_after_revise[0].request_id if pending_after_revise else pending[0].request_id
)
approved = await workflow.run(responses={approve_id: MagenticPlanReviewResponse.approve()})
return {
"pending_before_review": len(pending),
"request_type": type(pending[0].data).__name__ if pending else None,
"is_stalled": bool(pending[0].data.is_stalled) if pending else None,
"stopped_without_output": stopped_without_output,
"kinds_before_review": review_kinds,
"revise_manager_calls": revise_calls,
"pending_after_revise": len(pending_after_revise),
"outputs_after_approve": [str(o) for o in (approved.get_outputs() or [])],
}
async def expert_liaison_answer_round_trip(
*, answer: str, probe: str | None = None
) -> dict[str, Any]:
"""S3b: an ``AgentApprovalExecutor`` standing in as the ``expert_liaison`` PARTICIPANT — the
third door of § C.6, the one that lets the manager ask a question MID-run.
Measured shape, two round-trips per human turn: the manager picks the liaison → its output
becomes a ``request_info`` → ``from_strings([answer])`` feeds the human's words back INTO the
liaison, which runs again → a second ``request_info`` → ``approve()`` forwards the liaison's
(now informed) output to the manager, which resumes. ``from_strings`` alone does NOT resume
the manager: measured zero manager calls between the two requests.
``probe`` is the CONTROL knob: the manager's prompts are scanned for IT instead of for
``answer``, so a sentinel the expert never sent must come back absent. Without that arm,
"the manager saw the answer" could equally mean "the scanner matches anything".
"""
records: list[ExplorationCallRecord] = []
liaison_prompts: list[str] = []
def _liaison_reply(blob: str, _role: str) -> str:
liaison_prompts.append(blob)
if answer in blob:
return f"{WORKER_SENTINEL}-LIAISON heard: {answer}"
return f"{WORKER_SENTINEL}-LIAISON has no expert input yet."
liaison = AgentApprovalExecutor(
Agent(
ScriptedChatClient(reply_selector=_liaison_reply, role="expert_liaison"),
"You relay the expert's answers.",
name="expert_liaison",
description="asks the human expert",
)
)
workflow = MagenticBuilder(
participants=[liaison],
manager_agent=_manager_agent(
_manager_client(records, marker=probe or answer, speaker="expert_liaison")
),
max_round_count=6,
).build()
first = await workflow.run(TASK_ALPHA)
asked = _pending_requests(first)
before_answer = len(records)
answered = await workflow.run(
responses={asked[0].request_id: AgentRequestInfoResponse.from_strings([answer])}
)
manager_calls_on_answer = [r.kind for r in records[before_answer:]]
second = _pending_requests(answered)
approved = (
await workflow.run(responses={second[0].request_id: AgentRequestInfoResponse.approve()})
if second
else answered
)
return {
"reachable": bool(asked),
"request_type": type(asked[0].data).__name__ if asked else None,
"manager_calls_between_requests": manager_calls_on_answer,
"second_request": len(second),
"liaison_saw_answer": any(answer in prompt for prompt in liaison_prompts),
"manager_saw_answer": any(r.sees_marker for r in records),
"outputs": [str(o) for o in (approved.get_outputs() or [])],
}
# ---------------------------------------------------------------------------
# S4 — resume a pending plan review in a NEW PROCESS (plan U12)
# ---------------------------------------------------------------------------
async def checkpoint_until_plan_review(storage_dir: str) -> dict[str, Any]:
"""Run until the plan review stops the workflow, leaving checkpoints on disk.
This is the FIRST half of S4 and runs in the parent process; the resume half must run in a
separate interpreter (``spikes.e_magentic_resume``), because a resume that quietly rode on
live in-process objects would prove nothing about the asynchronous file inbox U12 needs.
"""
records: list[ExplorationCallRecord] = []
workflow = _plan_review_workflow(records, storage_dir)
result = await workflow.run(TASK_ALPHA)
pending = _pending_requests(result)
checkpoints = await _checkpoint_storage(storage_dir).list_checkpoints(
workflow_name=workflow.name
)
return {
"request_id": pending[0].request_id if pending else None,
"checkpoint_ids": [c.checkpoint_id for c in checkpoints],
"outputs": [str(o) for o in (result.get_outputs() or [])],
}
# Measured, and a real cost of the asynchronous HITL door: ``FileCheckpointStorage`` refuses to
# unpickle a plan-review request unless its type is declared. Without this the checkpoint file is
# written but comes back UNREADABLE ("Checkpoint deserialization blocked for type ..."), and the
# listing is empty — a resume that fails as an absence rather than as an error, which is exactly
# the shape the fourth face of the verification law warns about. Both processes must declare it.
_ALLOWED_CHECKPOINT_TYPES = [
"agent_framework_orchestrations._magentic:MagenticPlanReviewRequest",
"agent_framework_orchestrations._magentic:MagenticPlanReviewResponse",
]
def _checkpoint_storage(storage_dir: str) -> FileCheckpointStorage:
return FileCheckpointStorage(storage_dir, allowed_checkpoint_types=_ALLOWED_CHECKPOINT_TYPES)
def _plan_review_workflow(records: list[ExplorationCallRecord], storage_dir: str) -> Any:
"""The workflow BOTH processes build — identical construction, so the only thing carried
across the process boundary is the checkpoint on disk."""
return (
MagenticBuilder(
participants=[_worker_agent(_worker_client([]))],
manager_agent=_manager_agent(_manager_client(records)),
max_round_count=6,
enable_plan_review=True,
)
.with_checkpointing(_checkpoint_storage(storage_dir))
.build()
)
async def resume_from_checkpoint(
storage_dir: str, *, request_id: str, checkpoint_id: str
) -> dict[str, Any]:
"""The SECOND half of S4, called by ``spikes.e_magentic_resume`` in a fresh interpreter.
**Measured, and it contradicts the plan's E-table:** ``checkpoint_storage=`` on ``run()`` is
NOT the load-bearing seam here — removing it leaves the whole suite green (920 passed),
because ``.with_checkpointing(...)`` on the builder already gave this workflow its storage.
The two arguments that ARE load-bearing are ``checkpoint_id=`` (drop it → red) and the
builder's ``.with_checkpointing(...)`` (drop it → red). It is passed anyway, explicitly,
because an exploration layer that builds its workflow WITHOUT checkpointing and resumes by
handing storage in at call time is a legitimate second shape — but a criterion that names it
as the detach point would be a gate that cannot go red.
"""
records: list[ExplorationCallRecord] = []
workflow = _plan_review_workflow(records, storage_dir)
result = await workflow.run(
responses={request_id: MagenticPlanReviewResponse.approve()},
checkpoint_id=checkpoint_id,
checkpoint_storage=_checkpoint_storage(storage_dir),
)
return {
"manager_kinds": [r.kind for r in records],
"pending_after_resume": len(_pending_requests(result)),
"outputs": [str(o) for o in (result.get_outputs() or [])],
}
def run_resume_subprocess(
storage_dir: str, *, request_id: str, checkpoint_id: str
) -> dict[str, Any]:
"""Launch ``spikes.e_magentic_resume`` in a FRESH interpreter and parse its one JSON line.
``sys.executable`` is this venv's Python, and the repo root is the working directory, so the
child imports the same tree the parent did without any path juggling. A non-zero exit is
surfaced with the child's stderr attached: a resume that failed must read as a failed resume,
never as an empty result.
"""
completed = subprocess.run(
[sys.executable, "-m", "spikes.e_magentic_resume", storage_dir, request_id, checkpoint_id],
capture_output=True,
text=True,
cwd=str(Path(__file__).resolve().parents[1]),
)
if completed.returncode != 0:
raise RuntimeError(
f"resume subprocess exited {completed.returncode}: {completed.stderr.strip()}"
)
return dict(json.loads(completed.stdout.strip().splitlines()[-1]))
def micro_bundle_dir() -> str:
"""The repo's own anchored micro bundle — the one bundle that ships BOTH a cost baseline and
an IR projection, so a latency number measured here is measured through the WHOLE gate
(stage 0 reconciliation + CBC solve + 512-sample Monte Carlo), not a subset of it."""
return str(
Path(portfolio_optimiser.__file__).parent
/ "data"
/ "bundles"
/ "bygg-energi-baseline-mikro"
)
def micro_proposal() -> SavingsProposal:
"""The bundle's IR projection as a validated ``SavingsProposal``, carrying an assumption band.
The band is not decoration: without one, ``validator._monte_carlo`` falls back to each item's
own ``unit_cost`` and every draw is identical — a cheaper computation than any real hypothesis
would trigger, so a latency measured without it would understate the in-loop cost.
"""
projection = dict(load_ir_projection(micro_bundle_dir()))
projection.pop("_note", None)
item = projection["affected_items"][0]
unit_cost = float(item["unit_cost"])
projection["assumptions"] = {item["code"]: (unit_cost * 0.9, unit_cost * 1.1)}
return SavingsProposal.model_validate(projection)
def validator_latency_seconds(*, runs: int) -> tuple[float, int]:
"""S5: median wall-clock of one ``validate_proposal`` against the micro reference bundle —
the cost the hypothesiser's ``quick_validate`` tool would pay per call, per hypothesis."""
baseline = load_optional_cost_baseline(micro_bundle_dir())
proposal = micro_proposal()
timings: list[float] = []
for _ in range(runs):
start = time.perf_counter()
validate_proposal(proposal, baseline=baseline)
timings.append(time.perf_counter() - start)
return statistics.median(timings), len(timings)