open/portfolio-optimiser
1
0
Fork 0
Code Issues Releases 3 Activity

feat(explore): plan-reviewen kan besvares over DAGER (U12 + asynkron U13, rad 3)

Browse source 
F4 gjorde "be om svar, BRUKE svarene" naabar, men bare SYNKRONT: terminal_plan_reviewer
blokkerer loekka paa et menneske ved en terminal, saa svaret maa komme mens prosessen lever.
Maalbilde §3s tidsskala er den andre - eksperten svarer dager senere, i en prosess som aldri
saa kjoeringen.

--checkpoint-dir PARKERER reviewen (FileCheckpointStorage + {run_id}-plan-review.json) og
avslutter; --resume <run_id> leser svaret fra --review-inbox i en fersk interpreter. Det
eneste som krysser prosessgrensen er disk.

MAALT FELLE (Verifiseringsloven ansikt 4): list_checkpoints (_checkpoint.py:386-388) svelger
en blokkert deserialisering til en logger.warning og returnerer TOM liste. Uten BEGGE
MagenticPlanReviewRequest/Response i allowed_checkpoint_types feiler en resume som et FRAVAER,
ikke som en feil. _ALLOWED_CHECKPOINT_TYPES har derfor EN kopi, checkpoint_storage er eneste
konstruksjonssted, og en tom listing ved park raiser CheckpointUnreadable i stedet for aa
skrive et spoersmaal ingen kan besvare.

Budsjettet og revisjons-capen spenner over suspensjonen (meter.charge(parked.tokens_spent) +
trace.ledger.extend), ellers faar hver park et helt budsjett paa nytt. Fail-closed paa
ekspertens egen fil: request_id-mismatch, ord utenfor vokabularet og revise uten innhold
refuseres alle ved navn. hitl.pending_plan_reviews er registeret over hvem som venter.

Load-bearing MAALT: 17 tester, TRETTEN mutasjoner alle roede mot HELE suiten, groenn kontroll
1059 passed / 5 skipped, golden demo-transcript.stdout byte-uendret
(ea8c534773acdbe41ae68f2c55724d69aaf8be4f).

EN MUTASJON FALSIFISERTE SUITEN (vakuoes-gate-klassen, tiende gang): detach av
trace.plan_reviews.extend(parked.plan_reviews) lot HELE suiten staa groenn - capen leser
parked.plan_reviews DIREKTE, saa den binder uansett, og de to foerste legene er identiske
under begge implementasjoner. Gaten maatte bli det TREDJE leget, der artefaktet ellers taper
dag 1s revisjon og to ulike planer deler indeks 1. Ny test skrevet mot mutasjonen foerst.

Aerlighets-grenser: hostet flate NEKTER fortsatt (synkron review ville blokkert baade
requesten og event-loekka som svarer /readiness); en park midt i loepet etter en stall har
ingen naabar sti under det skriptede manuset, saa carry-overen som betjener den drives gjennom
en CRAFTED parkert tilstand.

Ordre 20260825T114645Z-6622513622-from-portfolio-optimiser.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kjell Tore Guttormsen 2026-08-26 12:26:04 +02:00
commit c08ae91809
7 changed files with 1754 additions and 91 deletions
Download patch file Download diff file Expand all files Collapse all files

47
CLAUDE.md
View file

@ -991,6 +991,53 @@ Python ≥3.10. MAF (`agent-framework-core` 1.9.0). Pakkehåndtering: `uv`. To b
review-uten-reviewer-nekten (2, hvorav én i en test som fantes fra før) · detach
reviewer-ingen-spør-nekten (1) · hostet nekt beholder påstanden fra før F4 (1) · strømmene fanget
ved bygge-tid (1).
- **Plan-reviewen kan besvares over DAGER, og det eneste som krysser prosessgrensen er DISK (U12 +
asynkron U13, planens § D.2 rad 3, økt 64):** F4 gjorde «be om svar, BRUKE svarene» nåbar, men
bare SYNKRONT — `terminal_plan_reviewer` blokkerer løkka på et menneske ved en terminal, så
svaret må komme mens prosessen lever. `--checkpoint-dir` PARKERER i stedet reviewen
(`FileCheckpointStorage` + `{run_id}-plan-review.json`), og `--resume <run_id>` leser svaret fra
`--review-inbox` i en prosess som ALDRI så kjøringen. **MÅLT FELLE (ansikt 4):**
`list_checkpoints` (`_checkpoint.py:386-388`) svelger en blokkert deserialisering til en
`logger.warning` og returnerer TOM liste — uten BEGGE
`MagenticPlanReviewRequest`/`…Response` i `allowed_checkpoint_types` feiler en resume som et
FRAVÆR, ikke som en feil, og en test som asserterte «listingen er tom, altså er det ingenting å
gjenoppta» ville vært GRØNN mot nøyaktig den defekten. `_ALLOWED_CHECKPOINT_TYPES` har derfor ÉN
kopi og `checkpoint_storage` er ENESTE konstruksjonssted (BEGGE prosesser må deklarere dem; en
andre kopi er kø-(p)-driften). **Vi er LOUDERE enn rammeverket der det tier:** en tom listing ved
park raiser `CheckpointUnreadable` i stedet for å skrive et spørsmål ingen kan besvare.
**Diskriminatoren er den ANDRE halvdelen:** en dør som skriver en spørsmålsfil og en resume som
leser en svarfil består begge «eksperten ble spurt» — så måltesten krever at et `revise` skrevet
dag 1 får manageren til å REPLANLEGGE og stille et NYTT spørsmål (indeks 1, nytt `request_id`) i
en fersk interpreter, med en approve-kontroll som beviser at døra også kan AVSLUTTE (en gate som
bare kunne parke igjen er en hengning i løkkeklær). **Budsjettet og revisjons-capen spenner over
suspensjonen:** `meter.charge(parked.tokens_spent)` (gjennom `charge`, ikke ved å sette `tokens`
— ladingen re-tester taket) og `trace.ledger.extend(parked.ledger)`, ellers får hver park et helt
budsjett på nytt: S3.4-klassen, ubundet forbruk under vakter som alle ser tilfredse ut. En
`revise` koster to manager-kall, emitterer null ledger og bruker null runde (§ F, A3), så
`max_plan_revisions` er det ENESTE båndet på den. **Fail-closed på ekspertens EGEN fil:**
`request_id`-mismatch, ord utenfor vokabularet og `revise` uten innhold refuseres alle ved navn.
`hitl.pending_plan_reviews` er registeret over hvem som fortsatt venter — tolerant på LESE-siden,
fail-closed på BESLUTTE-siden, og joinen er på `request_id` i BEGGE ender (to distinkte sømmer,
MÅLT: hver har sin egen mutasjon og sin egen røde test). Load-bearing MÅLT
(`tests/test_async_plan_review_loadbearing.py`, 17 tester), tretten mutasjoner alle røde mot HELE
suiten + grønn kontroll 1059/5 og golden `demo-transcript.stdout` BYTE-UENDRET
(`ea8c534773acdbe41ae68f2c55724d69aaf8be4f`): tom `_ALLOWED_CHECKPOINT_TYPES` (11 røde) · park
uten checkpoint (1) · resume alltid-approve (2) · tolerant `request_id` (1) · tolerant vokabular
(1) · `revise` uten feedback (1) · detach `meter.charge` (1) · detach
`trace.plan_reviews.extend` (1) · detach ledger/hypotese-carry-overen (1) ·
`pending_plan_reviews` ignorerer `request_id` (1) · detach to-dører-nekten (1) · detach
outbox/run-id-hoisten (1) · detach `--resume`-armen i `required_scripted_roles` (4 — MAJOR-2s
`KeyError: 'navigator'` på den andre flaten som bygger en utforskning).
**ÉN MUTASJON FALSIFISERTE SUITEN (repoets vakuøs-gate-klasse, TIENDE gang):** `trace.plan_reviews
.extend(parked.plan_reviews)` kunne detaches med HELE suiten grønn (1058/5) — capen leser
`parked.plan_reviews` DIREKTE, så den binder uansett, og de to første legene er identiske under
begge implementasjoner. Gaten måtte derfor bli det TREDJE leget, der artefaktet ellers taper dag
1s revisjon og to ULIKE planer deler indeks 1; den nye testen er rød mot mutasjonen og alene.
**Ærlighets-grenser, uttalt:** den hostede flaten NEKTER fortsatt (en synkron review ville
blokkert både requesten og event-løkka som svarer `/readiness`); en park MIDT i løpet (etter en
stall) har ingen nåbar sti under det skriptede manuset, så carry-overen som betjener den drives
gjennom en CRAFTED parkert tilstand (`budget_stop`-presedensen); og resume-legets
`PlanReviewParked` er et NORMALT utfall, ikke en feil.
- **STATE.md er local-only** (gitignored). Voyage session-state er efemert; STATE.md er kanonisk kontinuitet.
- Prosess: Voyage-plugin (`/trekbrief → /trekplan → /trekexecute → /trekreview`) per større fase.

31
README.md
View file

@ -461,6 +461,37 @@ when the seam is detached, so the loop cannot silently degrade into theater.
— blocking an HTTP request on a human would also block the event loop that answers
`/readiness`.
**Answering it days later (`--checkpoint-dir` / `--resume`).** A domain expert is rarely at the
terminal when the loop reaches the plan, so the same review can be *parked* to disk instead.
`--checkpoint-dir` writes the suspended workflow there and the open question to
`{run_id}-plan-review.json`, and the process exits. Whenever the expert gets to it — another
day, in a process that never saw the run — they drop `{run_id}-plan-review-answer.json` into a
review inbox, and `--resume` picks it up:
```bash
# day 1 — park the review and exit
uv run python -m portfolio_optimiser.run FV42-GSV-E1 --docs-dir <docs> --bundle-dir <bundle> \
--explore "Find the cheapest saving worth testing here" --explore-config exploration.json \
--checkpoint-dir checkpoints --outbox-dir out --run-id r1
# day N — a fresh process, resuming from what is on disk and nothing else
uv run python -m portfolio_optimiser.run FV42-GSV-E1 --docs-dir <docs> --bundle-dir <bundle> \
--outbox-dir out --checkpoint-dir checkpoints --review-inbox inbox --resume r1
```
A `revise` answered this way does the same thing it does at the terminal: the manager replans
and asks again about the **new** plan. The answer names the `request_id` it answers, and a
mismatch is refused rather than applied — two reviews of one run share a file name, so an answer
left over from the previous round would otherwise sign off a plan the expert never saw. The
vocabulary is the same closed one, anything outside it is refused rather than read as approval,
and `revise` with nothing to revise is refused too. `hitl.pending_plan_reviews(outbox, inbox)`
lists every review still waiting on somebody.
The budget and the revision cap span the suspension — the resumed leg starts from what the
parked one already spent, so a park never hands back a fresh budget. `--plan-review` and
`--checkpoint-dir` are refused together (two doors onto one review), as are `--resume` and
`--explore` (two sources of one exploration).
`--explore` is refused together with `--mandate` — they are two sources of one mandate, and
merging would silently overwrite what you wrote. To seed an exploration with a domain expert's
own hypotheses, use `explore(..., seed_approaches=[Approach(...)])`; seeds are always preserved

636
src/portfolio_optimiser/explore.py
View file

@ -30,7 +30,7 @@ from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Final, Literal, TextIO
from agent_framework import Agent, BaseChatClient, FunctionTool, tool
from agent_framework import Agent, BaseChatClient, FileCheckpointStorage, FunctionTool, tool
from agent_framework.orchestrations import (
MagenticBuilder,
MagenticOrchestratorEventType,
@ -122,6 +122,24 @@ def exploration_notice(result: "ExplorationResult") -> str:
)
def parked_notice(parked: "ParkedExploration", *, run_id: str) -> str:
"""The ONE thing a parked exploration says at the terminal it stopped on.
Always a line, for the reason ``exploration_notice`` always is: it is printed only where the
asynchronous door was asked for. It names the run id, because that is the single coordinate
``--resume`` takes, and it is the one an operator will be looking for weeks later.
It does NOT print the plan. The plan is in the question artefact, where the expert who has to
read it will be looking — and this terminal belongs to whoever STARTED the run, who is not
necessarily them.
"""
return (
f"Exploration parked: plan review {parked.index} of run {run_id!r} is waiting for a "
f"human. The question is in {run_id}-plan-review.json; answer it with "
f"{run_id}-plan-review-answer.json in a review inbox, then --resume {run_id}"
)
def load_exploration_contract(path: str | Path) -> ExplorationContract:
"""Fail-fast standalone loader for an exploration's bounds (mirrors ``mandate.load_mandate``).
@ -256,6 +274,11 @@ class ExplorationTrace:
ledger: list[LedgerEntry] = field(default_factory=list)
plan_reviews: list[PlanReview] = field(default_factory=list)
quick_validations: list[QuickValidation] = field(default_factory=list)
#: Tokens spent so far, refreshed as the loop turns rather than written once at the end. The
#: meter is internal to ``explore``, so this is the only way the artefact can report a spend —
#: and updating it per iteration is what makes it readable for a run a cap cut short, which is
#: the same reason the sink exists at all. Across a park it is what the resumed leg adds to.
tokens_spent: int = 0
def trace_payload(trace: ExplorationTrace, *, stop: str | None, completed: bool) -> dict[str, Any]:
@ -271,6 +294,7 @@ def trace_payload(trace: ExplorationTrace, *, stop: str | None, completed: bool)
return {
"completed": completed,
"stop": stop,
"tokens_spent": trace.tokens_spent,
"rounds": [
{
"round_index": entry.round_index,
@ -438,6 +462,202 @@ def terminal_plan_reviewer(
return review
# ---------------------------------------------------------------------------------------------
# U12 — the ASYNCHRONOUS half of the same door: a review answered days later, in another process.
# ---------------------------------------------------------------------------------------------
#: The two types a plan-review checkpoint carries, in ``"module:qualname"`` form.
#:
#: **This tuple is the whole of the measured trap** (plan § F, A4; re-measured against the
#: installed source, ``_workflows/_checkpoint.py:386-388``): ``FileCheckpointStorage`` runs a
#: restricted unpickler, and ``list_checkpoints`` swallows a blocked type into a ``logger.warning``
#: and returns an EMPTY list. Omit either name and the checkpoint is written but comes back
#: unreadable, so a resume fails as an ABSENCE — "nothing to resume" — rather than as an error.
#: ONE copy, used by every process that touches this storage, because both the writing process and
#: the resuming one must declare them and a second copy is the drift kø-(p) exists to prevent.
_ALLOWED_CHECKPOINT_TYPES: Final[tuple[str, ...]] = (
"agent_framework_orchestrations._magentic:MagenticPlanReviewRequest",
"agent_framework_orchestrations._magentic:MagenticPlanReviewResponse",
)
def checkpoint_storage(checkpoint_dir: str | Path) -> FileCheckpointStorage:
"""The ONE construction site for the exploration's checkpoint storage.
A caller that built its own ``FileCheckpointStorage`` would have to remember the allow-list
above, and forgetting it is invisible (see ``_ALLOWED_CHECKPOINT_TYPES``). Routing every
construction through here makes "both processes declare the types" a structural property
rather than a convention two call sites have to keep.
"""
return FileCheckpointStorage(
str(checkpoint_dir), allowed_checkpoint_types=list(_ALLOWED_CHECKPOINT_TYPES)
)
class CheckpointUnreadable(ExplorationError):
"""The exploration stopped at a review but left nothing a later process could resume from.
Raised where the framework is silent: an empty listing means the checkpoint could not be read
back, and parking anyway would hand an expert a question whose answer can never be applied.
Failing here is the fourth face of the verification law written into our own surface.
"""
class ParkedStateError(ExplorationError):
"""A parked-state file that cannot be read as one.
Fail-fast, NOT the tolerant RAW-inbox rule: this file is the run's own suspended state, the
same class as the spend file ``read_spend`` refuses to read loosely. Treating a malformed one
as "no parked run" would silently drop an exploration a human is waiting to answer.
"""
@dataclass(frozen=True)
class ParkedExploration:
"""Everything needed to resume one suspended exploration in a process that never saw it.
Two files cross the boundary and they own different halves. MAF's checkpoint holds the
WORKFLOW's state (the manager's ledgers, the pending request); this holds the EXPLORATION
LAYER's — what has been spent, what the loop has already found, and which question is open.
Neither can reconstruct the other, so both are named here rather than one being inferred.
``tokens_spent`` and ``ledger`` are carried for a reason that is not bookkeeping. Both budget
channels live in the process: a resume with a fresh ``TokenMeter`` and an empty ledger would
get the whole cap AGAIN, once per park — unbounded consumption behind guards that all report
themselves satisfied, which is the class S3.4 split apart. Carrying them makes the cap span
the suspension.
``hypotheses`` are the marked turns the loop produced BEFORE parking. They are verbatim, and
they are carried for the same reason: minting the mandate from only what the resuming process
observed would silently drop everything found before a stalled re-review.
"""
prompt: str
request_id: str
checkpoint_id: str
index: int
plan: str
current_progress: str
is_stalled: bool
bundle_dirs: tuple[str, ...]
contract: ExplorationContract
ledger: tuple[LedgerEntry, ...]
plan_reviews: tuple[PlanReview, ...]
hypotheses: tuple[str, ...]
tokens_spent: int
replans: int
class PlanReviewParked(ExplorationError):
"""The exploration is suspended at a plan review, waiting for a human.
**A raise rather than a return, and the argument is the 429 one** (kø-(y), 14.08). A parked run
produced NO mandate: the loop is stopped mid-plan and nothing has been explored yet. Handing
back an ``ExplorationResult`` would let an automated caller book "explored" for a run that
explored nothing — the same reason an exhausted budget is not a 200 even though it is not a
crash either. The coordinates travel as STRUCTURE on ``parked``, never as ``str(exc)``.
"""
def __init__(self, parked: ParkedExploration) -> None:
super().__init__(
f"exploration parked at plan review {parked.index} "
f"(request {parked.request_id}, checkpoint {parked.checkpoint_id})"
)
self.parked = parked
def parked_payload(parked: ParkedExploration) -> dict[str, Any]:
"""The ONE rendering of a parked exploration into plain data for ``outbox.write_plan_review``.
Plain mappings only, so the RAW output layer stays MAF-free — the ``trace_payload`` precedent,
for the same reason: ``outbox.py`` may not import this module.
"""
return {
"prompt": parked.prompt,
"request_id": parked.request_id,
"checkpoint_id": parked.checkpoint_id,
"index": parked.index,
"plan": parked.plan,
"current_progress": parked.current_progress,
"is_stalled": parked.is_stalled,
"bundle_dirs": list(parked.bundle_dirs),
"contract": parked.contract.model_dump(),
"ledger": [
{
"round_index": entry.round_index,
"is_request_satisfied": entry.is_request_satisfied,
"is_in_loop": entry.is_in_loop,
"is_progress_being_made": entry.is_progress_being_made,
"next_speaker": entry.next_speaker,
"instruction_or_question": entry.instruction_or_question,
"speaker_known": entry.speaker_known,
}
for entry in parked.ledger
],
"plan_reviews": [
{
"index": review.index,
"plan": review.plan,
"is_stalled": review.is_stalled,
"decision": review.decision,
"feedback": review.feedback,
}
for review in parked.plan_reviews
],
"hypotheses": list(parked.hypotheses),
"tokens_spent": parked.tokens_spent,
"replans": parked.replans,
}
def load_parked(payload: Mapping[str, Any]) -> ParkedExploration:
"""Read a parked-state payload back, fail-fast (``ParkedStateError`` on anything missing).
Validation, never repair (the ``write_concept_file`` rule): a payload that has lost, say, its
``checkpoint_id`` describes a suspension nobody can lift, and defaulting it would produce a
resume that looks like one and is not.
"""
try:
return ParkedExploration(
prompt=str(payload["prompt"]),
request_id=str(payload["request_id"]),
checkpoint_id=str(payload["checkpoint_id"]),
index=int(payload["index"]),
plan=str(payload["plan"]),
current_progress=str(payload["current_progress"]),
is_stalled=bool(payload["is_stalled"]),
bundle_dirs=tuple(str(d) for d in payload["bundle_dirs"]),
contract=ExplorationContract.model_validate(payload["contract"]),
ledger=tuple(
LedgerEntry(
round_index=int(row["round_index"]),
is_request_satisfied=bool(row["is_request_satisfied"]),
is_in_loop=bool(row["is_in_loop"]),
is_progress_being_made=bool(row["is_progress_being_made"]),
next_speaker=str(row["next_speaker"]),
instruction_or_question=str(row["instruction_or_question"]),
speaker_known=bool(row["speaker_known"]),
)
for row in payload["ledger"]
),
plan_reviews=tuple(
PlanReview(
index=int(row["index"]),
plan=str(row["plan"]),
is_stalled=bool(row["is_stalled"]),
decision="revise" if row["decision"] == "revise" else "approve",
feedback=str(row["feedback"]),
)
for row in payload["plan_reviews"]
),
hypotheses=tuple(str(h) for h in payload["hypotheses"]),
tokens_spent=int(payload["tokens_spent"]),
replans=int(payload["replans"]),
)
except (KeyError, TypeError, ValueError, ValidationError) as exc:
raise ParkedStateError(f"parked exploration state is unreadable: {exc}") from exc
# ---------------------------------------------------------------------------------------------
# The tools. Level 1 of the three-guarantee table: real computation, ADVISORY verdicts.
# ---------------------------------------------------------------------------------------------
@ -619,6 +839,7 @@ def fresh_exploration_workflow(
bundle_dirs: Sequence[str] = (),
middleware: Sequence[Any] | None = None,
quick_validate_sink: list[QuickValidation] | None = None,
checkpoint_dir: str | None = None,
) -> Any:
"""Build a FRESH Magentic workflow with fresh agents and fresh clients (mirrors
``workflow.fresh_workflow``).
@ -668,14 +889,22 @@ def fresh_exploration_workflow(
middleware=middleware,
)
return MagenticBuilder(
builder = MagenticBuilder(
participants=participants,
manager_agent_factory=_manager_agent,
max_round_count=contract.max_rounds,
max_stall_count=contract.max_stall_count,
max_reset_count=contract.max_reset_count,
enable_plan_review=contract.enable_plan_review,
).build()
)
if checkpoint_dir is not None:
# Measured (spike S4, and re-measured in the plan's own E-table correction): it is the
# BUILDER's ``.with_checkpointing`` that is load-bearing, not ``checkpoint_storage=`` on
# ``run()`` — dropping the latter leaves the whole suite green. Both processes call THIS
# function, so the graph they build is identical, which is what lets a checkpoint written
# by one be restored by the other (``_runner.py:275-279`` matches on the graph signature).
builder = builder.with_checkpointing(checkpoint_storage(checkpoint_dir))
return builder.build()
def _truthy(answer: Any) -> bool:
@ -925,6 +1154,7 @@ async def explore(
meter: TokenMeter | None = None,
success_criteria: str = "",
trace: ExplorationTrace | None = None,
checkpoint_dir: str | None = None,
) -> ExplorationResult:
"""Explore the knowledge bases and return the ``Mandate`` the pipeline should evaluate.
@ -964,10 +1194,23 @@ async def explore(
produced: both budget channels destroy the ``ExplorationResult`` before it exists. When it is
omitted a private one is used, so the returned result is unchanged for every existing caller.
"""
if contract.enable_plan_review and plan_reviewer is None:
if plan_reviewer is not None and checkpoint_dir is not None:
raise ExplorationError(
"enable_plan_review is set but no plan_reviewer was given: the exploration would stop "
"at a review nobody can answer, which is a hang rather than a result"
"a plan_reviewer and a checkpoint_dir are two doors onto one review: the first answers "
"it in this process, the second parks it for another one. Refused rather than ranked, "
"because silently preferring either would block a caller that asked for the other"
)
if contract.enable_plan_review and plan_reviewer is None and checkpoint_dir is None:
raise ExplorationError(
"enable_plan_review is set but no plan_reviewer was given and no checkpoint_dir was "
"offered to park it: the exploration would stop at a review nobody can answer, which "
"is a hang rather than a result"
)
if checkpoint_dir is not None and not contract.enable_plan_review:
raise ExplorationError(
"a checkpoint_dir was given but enable_plan_review is false, so nothing would ever "
"park and the storage would be written and never read (a cap on an event that cannot "
"happen, refused for the reason max_plan_revisions is)"
)
if plan_reviewer is not None and not contract.enable_plan_review:
raise ExplorationError(
@ -1000,16 +1243,10 @@ async def explore(
bundle_dirs=bundle_dirs,
middleware=[BudgetMiddleware(meter)],
quick_validate_sink=trace.quick_validations,
checkpoint_dir=checkpoint_dir,
)
# ONE accumulator per fact, held by the caller (see ``ExplorationTrace``). The local names are
# aliases, never copies — a second list here is the kø-(p) drift this shape exists to prevent.
ledger_log = trace.ledger
plan_reviews = trace.plan_reviews
hypothesis_texts: list[str] = []
seen: set[int] = set()
replans = 0
stop: ExplorationStop | None = None
# ONE span for the whole exploration, opened before the first model call and closed however
# the loop ends — including on a BudgetExceeded, which the span records rather than swallows.
@ -1017,69 +1254,208 @@ async def explore(
# exactly what "tracing is off" has meant since U14.
with exploration_tracer().start_as_current_span(EXPLORATION_SPAN) as span:
result = await workflow.run(prompt)
while True:
replans += _absorb(
result,
ledger_log=ledger_log,
hypotheses=hypothesis_texts,
seen=seen,
span=span,
stop, _ = await _drive(
workflow,
result,
contract=contract,
trace=trace,
hypotheses=hypothesis_texts,
seen=set(),
replans=0,
span=span,
plan_reviewer=plan_reviewer,
checkpoint_dir=checkpoint_dir,
prompt=prompt,
bundle_dirs=bundle_dirs,
meter=meter,
)
return _finish(
prompt=prompt,
stop=stop,
trace=trace,
hypotheses=hypothesis_texts,
bundle_ids=bundle_ids,
seed_approaches=seed_approaches,
success_criteria=success_criteria,
)
async def _drive(
workflow: Any,
result: Any,
*,
contract: ExplorationContract,
trace: ExplorationTrace,
hypotheses: list[str],
seen: set[int],
replans: int,
span: Any,
plan_reviewer: PlanReviewer | None,
checkpoint_dir: str | None,
prompt: str,
bundle_dirs: Sequence[str],
meter: TokenMeter,
) -> tuple[ExplorationStop | None, int]:
"""Drive a built workflow from one ``run()`` result to an ending, answering plan reviews.
**ONE copy of this loop, shared by ``explore`` and ``resume_exploration``.** The asynchronous
door is not a second loop that happens to look like the first: a resumed exploration answers
reviews, absorbs ledgers, mints nothing, and classifies its stop by exactly the same rules, and
two copies of that would drift the moment one of them was corrected (kø-(p)).
Raises ``PlanReviewParked`` when the asynchronous door is armed — the loop stops mid-plan and
the caller decides where to write the question.
"""
# ONE accumulator per fact, held by the caller (see ``ExplorationTrace``). The local names are
# aliases, never copies — a second list here is the kø-(p) drift this shape exists to prevent.
ledger_log = trace.ledger
plan_reviews = trace.plan_reviews
stop: ExplorationStop | None = None
while True:
trace.tokens_spent = meter.tokens
replans += _absorb(
result,
ledger_log=ledger_log,
hypotheses=hypotheses,
seen=seen,
span=span,
)
pending = _pending_plan_reviews(result)
if not pending:
break
request = pending[0]
review = request.data
if plan_reviewer is None:
# The asynchronous door (U12). ``checkpoint_dir`` is what armed it, and the guard in
# ``explore`` refused every other way of arriving here with no reviewer.
assert checkpoint_dir is not None
raise PlanReviewParked(
await _park(
workflow,
request,
checkpoint_dir=checkpoint_dir,
prompt=prompt,
bundle_dirs=bundle_dirs,
contract=contract,
trace=trace,
hypotheses=hypotheses,
meter=meter,
replans=replans,
)
)
pending = _pending_plan_reviews(result)
if not pending:
break
assert plan_reviewer is not None # guarded above; the review implies a reviewer
request = pending[0]
review = request.data
decision = plan_reviewer(
PlanReviewRequest(
decision = plan_reviewer(
PlanReviewRequest(
index=len(plan_reviews),
plan=str(review.plan),
current_progress=str(review.current_progress),
is_stalled=_truthy(review.is_stalled),
)
)
if decision.feedback is None:
plan_reviews.append(
PlanReview(
index=len(plan_reviews),
plan=str(review.plan),
current_progress=str(review.current_progress),
is_stalled=_truthy(review.is_stalled),
decision="approve",
)
)
if decision.feedback is None:
plan_reviews.append(
PlanReview(
index=len(plan_reviews),
plan=str(review.plan),
is_stalled=_truthy(review.is_stalled),
decision="approve",
)
response = MagenticPlanReviewResponse.approve()
else:
# The revision is recorded whether or not it is APPLIED: ``plan_reviews`` is the
# record of what the reviewer DECIDED, and ``stop`` is what says the last one was
# refused.
applied = sum(1 for entry in plan_reviews if entry.decision == "revise")
plan_reviews.append(
PlanReview(
index=len(plan_reviews),
plan=str(review.plan),
is_stalled=_truthy(review.is_stalled),
decision="revise",
feedback=decision.feedback,
)
response = MagenticPlanReviewResponse.approve()
else:
# The revision is recorded whether or not it is APPLIED: ``plan_reviews`` is the
# record of what the reviewer DECIDED, and ``stop`` is what says the last one was
# refused.
applied = sum(1 for entry in plan_reviews if entry.decision == "revise")
plan_reviews.append(
PlanReview(
index=len(plan_reviews),
plan=str(review.plan),
is_stalled=_truthy(review.is_stalled),
decision="revise",
feedback=decision.feedback,
)
)
if applied >= contract.max_plan_revisions:
# Measured (§ F, A3): a revise costs two manager calls, emits no progress
# ledger and consumes no round, then asks AGAIN. Under the round cap alone this
# loop never terminates. Stopping is the honest move — forcing an approve the
# reviewer did not give would be repair, and repair of a human's decision most
# of all.
stop = "plan_revisions_exhausted"
break
response = MagenticPlanReviewResponse.revise(decision.feedback)
result = await workflow.run(responses={request.request_id: response})
)
if applied >= contract.max_plan_revisions:
# Measured (§ F, A3): a revise costs two manager calls, emits no progress
# ledger and consumes no round, then asks AGAIN. Under the round cap alone this
# loop never terminates. Stopping is the honest move — forcing an approve the
# reviewer did not give would be repair, and repair of a human's decision most
# of all.
stop = "plan_revisions_exhausted"
break
response = MagenticPlanReviewResponse.revise(decision.feedback)
result = await workflow.run(responses={request.request_id: response})
if stop is None:
stop = _classify_stop(ledger_log, replans=replans, contract=contract)
trace.tokens_spent = meter.tokens
if stop is None:
stop = _classify_stop(ledger_log, replans=replans, contract=contract)
return stop, replans
discovered = (
_parse_hypotheses(hypothesis_texts, bundle_ids) if stop != "unknown_speaker" else []
async def _park(
workflow: Any,
request: Any,
*,
checkpoint_dir: str,
prompt: str,
bundle_dirs: Sequence[str],
contract: ExplorationContract,
trace: ExplorationTrace,
hypotheses: Sequence[str],
meter: TokenMeter,
replans: int,
) -> ParkedExploration:
"""Freeze the suspended exploration and name the checkpoint a later process resumes from.
``get_latest`` rather than the last entry of a listing: it picks by timestamp
(``_checkpoint.py:424``), while the listing's order is whatever ``Path.glob`` returned.
An EMPTY listing is refused (``CheckpointUnreadable``) rather than parked around. That is the
one place this layer is louder than the framework: a blocked deserialisation is swallowed into
a warning upstream, so the alternative to raising here is a question file whose answer can
never be applied — an exploration that fails as an absence, days later, to somebody who has
already written their answer.
"""
latest = await checkpoint_storage(checkpoint_dir).get_latest(workflow_name=workflow.name)
if latest is None:
raise CheckpointUnreadable(
f"the exploration reached a plan review but no checkpoint could be read back from "
f"{checkpoint_dir!r}: without one the review can never be resumed, so it is refused "
f"here rather than written as a question nobody can answer"
)
review = request.data
return ParkedExploration(
prompt=prompt,
request_id=str(request.request_id),
checkpoint_id=str(latest.checkpoint_id),
index=len(trace.plan_reviews),
plan=str(review.plan),
current_progress=str(review.current_progress),
is_stalled=_truthy(review.is_stalled),
bundle_dirs=tuple(bundle_dirs),
contract=contract,
ledger=tuple(trace.ledger),
plan_reviews=tuple(trace.plan_reviews),
hypotheses=tuple(hypotheses),
tokens_spent=meter.tokens,
replans=replans,
)
def _finish(
*,
prompt: str,
stop: ExplorationStop | None,
trace: ExplorationTrace,
hypotheses: Sequence[str],
bundle_ids: Sequence[str],
seed_approaches: Sequence[Approach],
success_criteria: str,
) -> ExplorationResult:
"""Mint the mandate from a finished drive. Shared, for the reason ``_drive`` is."""
discovered = _parse_hypotheses(hypotheses, bundle_ids) if stop != "unknown_speaker" else []
return ExplorationResult(
mandate=Mandate(
objective=prompt,
@ -1087,9 +1463,133 @@ async def explore(
allow_own_proposals=True,
success_criteria=success_criteria,
),
ledger_log=tuple(ledger_log),
ledger_log=tuple(trace.ledger),
stop=stop,
plan_reviews=tuple(plan_reviews),
plan_reviews=tuple(trace.plan_reviews),
)
async def resume_exploration(
parked: ParkedExploration,
decision: PlanReviewDecision,
*,
checkpoint_dir: str,
profile: Profile | str = Profile.LOCAL,
client_factory: Callable[[str], BaseChatClient] | None = None,
seed_approaches: Sequence[Approach] = (),
success_criteria: str = "",
trace: ExplorationTrace | None = None,
) -> ExplorationResult:
"""Answer a parked plan review and drive the exploration onward, in a process that never ran it.
Everything about the workflow is rebuilt from ``parked`` rather than from argv: the prompt, the
bounds and the bases are what the suspended run used, and the graph must match the checkpoint's
signature (``_runner.py:275-279``) for the restore to be accepted at all. A caller that had to
re-supply them could get one of them wrong and would find out as a restore failure days later.
**The budget spans the suspension.** The meter starts at ``parked.tokens_spent`` and the ledger
at ``parked.ledger``, so the round cap and the token cap measure the whole exploration rather
than this leg of it. Without that a park would hand back a full budget every time it happened.
Parking AGAIN is a normal outcome, not a failure: a revision makes the manager replan and ask
about the NEW plan, which is the second half of "be om svar, BRUKE svarene" on this time-scale.
It leaves by ``PlanReviewParked`` exactly as the first park did.
"""
if trace is None:
trace = ExplorationTrace()
# The carried state is put back BEFORE anything runs: ``_drive`` reads these as its own running
# log, and ``_classify_stop`` counts the ledger to decide whether the ROUND cap bound.
trace.ledger.extend(parked.ledger)
trace.plan_reviews.extend(parked.plan_reviews)
hypothesis_texts = list(parked.hypotheses)
bundle_ids = tuple(_bundle_index(parked.bundle_dirs))
refuse_unroutable_seeds(seed_approaches, bundle_ids)
meter = TokenMeter(
Budget(max_tokens=parked.contract.max_tokens, max_rounds=parked.contract.max_rounds)
)
# Through ``charge``, not by assigning ``tokens``: charging re-tests the cap, so a suspension
# that already spent everything refuses HERE instead of buying one more leg of the loop.
meter.charge(parked.tokens_spent)
if client_factory is None:
from portfolio_optimiser.run import _default_factory
client_factory = _default_factory(profile)
workflow = fresh_exploration_workflow(
client_factory,
contract=parked.contract,
bundle_dirs=parked.bundle_dirs,
middleware=[BudgetMiddleware(meter)],
quick_validate_sink=trace.quick_validations,
checkpoint_dir=checkpoint_dir,
)
# The revision cap counted across the SUSPENSION, over the reviews carried in ``parked``.
# Measured, and it is why the carry-over is load-bearing rather than tidy: a revise costs two
# manager calls, emits no progress ledger and consumes no round (§ F, A3), so a cap that reset
# at every park would leave the asynchronous door with no bound at all — an expert could revise
# forever, one process at a time, under guards that all report themselves satisfied. The same
# arithmetic ``_drive`` does for the synchronous door, and the same refusal to repair: the
# decision is RECORDED and the loop stops, never forced into an approve nobody gave.
applied = sum(1 for entry in parked.plan_reviews if entry.decision == "revise")
trace.plan_reviews.append(
PlanReview(
index=parked.index,
plan=parked.plan,
is_stalled=parked.is_stalled,
decision="approve" if decision.feedback is None else "revise",
feedback=decision.feedback or "",
)
)
if decision.feedback is not None and applied >= parked.contract.max_plan_revisions:
return _finish(
prompt=parked.prompt,
stop="plan_revisions_exhausted",
trace=trace,
hypotheses=hypothesis_texts,
bundle_ids=bundle_ids,
seed_approaches=seed_approaches,
success_criteria=success_criteria,
)
response = (
MagenticPlanReviewResponse.approve()
if decision.feedback is None
else MagenticPlanReviewResponse.revise(decision.feedback)
)
with exploration_tracer().start_as_current_span(EXPLORATION_SPAN) as span:
result = await workflow.run(
responses={parked.request_id: response},
checkpoint_id=parked.checkpoint_id,
checkpoint_storage=checkpoint_storage(checkpoint_dir),
)
stop, _ = await _drive(
workflow,
result,
contract=parked.contract,
trace=trace,
hypotheses=hypothesis_texts,
seen=set(),
replans=parked.replans,
span=span,
plan_reviewer=None,
checkpoint_dir=checkpoint_dir,
prompt=parked.prompt,
bundle_dirs=parked.bundle_dirs,
meter=meter,
)
return _finish(
prompt=parked.prompt,
stop=stop,
trace=trace,
hypotheses=hypothesis_texts,
bundle_ids=bundle_ids,
seed_approaches=seed_approaches,
success_criteria=success_criteria,
)

132
src/portfolio_optimiser/hitl.py
View file

@ -165,6 +165,138 @@ def pending(outbox_dir: str, verdict_dir: str) -> list[PendingProposal]:
return sorted(unjudged, key=lambda p: (p.run_id, p.approach_id, p.verdict_id))
# --- U12: the pending PLAN REVIEWS of parked explorations, and the expert's answer ---------------
# The same registry shape as ``pending`` above, one time-scale earlier: there the outbox holds a
# PROPOSAL awaiting a verdict, here it holds a QUESTION awaiting a decision. Both live in this
# MAF-free module because both are read by an operator tool that must not drag the framework in.
#: The closed answer vocabulary, identical to the terminal door's (``explore.terminal_plan_reviewer``).
#: Two words, matched structurally — a file cannot be re-asked, so anything else is a refusal.
_PLAN_REVIEW_ANSWERS = frozenset({"approve", "revise"})
@dataclass(frozen=True)
class PendingPlanReview:
"""One parked exploration still waiting on a human. ``plan`` is carried because a registry that
only counted questions could not be used to answer one."""
run_id: str
request_id: str
index: int
plan: str
is_stalled: bool
@dataclass(frozen=True)
class PlanReviewAnswer:
"""The expert's decision, as read off a file. ``feedback`` is empty exactly when approving —
the same encoding ``explore.PlanReviewDecision`` uses, kept plain so this module stays
MAF-free and the adapter between them lives at ONE call site."""
run_id: str
request_id: str
decision: str
feedback: str
class PlanReviewAnswerError(ValueError):
"""An answer file that cannot be read as a decision.
Fail-closed, and deliberately NOT the tolerant rule the verdict inbox uses. A dropped verdict
that will not parse is one opinion missing from a fold; an unreadable plan-review answer is the
one thing standing between a suspended run and a plan nobody signed. A ``ValueError`` so the
CLI's existing structured-refusal arm surfaces it as ``rc 1`` rather than a traceback."""
def _answer_path(review_dir: str, run_id: str) -> Path:
return Path(review_dir) / f"{run_id}-plan-review-answer.json"
def read_plan_review_question(outbox_dir: str, run_id: str) -> dict[str, Any] | None:
"""The open question of ``run_id``, or ``None`` when there is none. Tolerant: an outbox with no
such file simply has no parked review."""
return _load_json_dict(Path(outbox_dir) / f"{run_id}-plan-review.json")
def load_plan_review_answer(review_dir: str, run_id: str, *, request_id: str) -> PlanReviewAnswer:
"""Read the expert's answer to ONE named review, fail-closed at every step.
``request_id`` is a required argument rather than something read off the file and trusted: two
reviews of one run share a file name, so an answer left over from the previous round would
otherwise be applied to a plan the expert never saw. A mismatch is REFUSED by name — it is a
stale answer, not an absent one, and the two need different words.
A missing file raises rather than returning ``None``: "not answered yet" is the normal state of
this door, and the caller asking to resume has already said it believes otherwise."""
path = _answer_path(review_dir, run_id)
data = _load_json_dict(path)
if data is None:
raise PlanReviewAnswerError(
f"no answer for plan review {request_id} of run {run_id!r} in {review_dir!r} "
f"(expected {path.name}): the review is still waiting on a human"
)
found = str(data.get("request_id", ""))
if found != request_id:
raise PlanReviewAnswerError(
f"the answer in {path.name} answers plan review {found!r}, but the open review of run "
f"{run_id!r} is {request_id!r}. Refused: an answer to another question is not an "
f"answer to this one"
)
decision = str(data.get("decision", ""))
if decision not in _PLAN_REVIEW_ANSWERS:
raise PlanReviewAnswerError(
f"{path.name} answers {decision!r}, which is outside the vocabulary "
f"{sorted(_PLAN_REVIEW_ANSWERS)}. Refused, never read as a sign-off"
)
feedback = str(data.get("feedback", ""))
if decision == "revise" and not feedback.strip():
raise PlanReviewAnswerError(
f"{path.name} answers 'revise' with nothing to revise: the manager would be asked to "
f"replan against an empty instruction. Say what to change, or answer 'approve'"
)
return PlanReviewAnswer(
run_id=run_id, request_id=request_id, decision=decision, feedback=feedback
)
def pending_plan_reviews(outbox_dir: str, review_dir: str) -> list[PendingPlanReview]:
"""Every parked plan review whose OWN answer has not landed, sorted by ``run_id``.
Mirrors ``pending``: an outbox artefact joined against an inbox, with the join on the key each
side names. The key here is ``request_id`` — an answer to a different review leaves this one
pending rather than quietly clearing it, which is the same fail-closed rule the resume path
applies and for the same reason.
Tolerant on the READ side (an unreadable file in either folder is not a question and not an
answer), fail-closed on the DECIDE side (``load_plan_review_answer``). The registry says who is
waiting; it never decides what they said."""
waiting: list[PendingPlanReview] = []
directory = Path(outbox_dir)
if not directory.is_dir():
return waiting
for file in sorted(directory.glob("*-plan-review.json")):
data = _load_json_dict(file)
if data is None:
continue
run_id = str(data.get("run_id", ""))
request_id = str(data.get("request_id", ""))
if not run_id or not request_id:
continue
answer = _load_json_dict(_answer_path(review_dir, run_id))
if answer is not None and str(answer.get("request_id", "")) == request_id:
continue
waiting.append(
PendingPlanReview(
run_id=run_id,
request_id=request_id,
index=int(data.get("index", 0)),
plan=str(data.get("plan", "")),
is_stalled=bool(data.get("is_stalled", False)),
)
)
return sorted(waiting, key=lambda p: (p.run_id, p.index))
# --- Routing config: self-contained dimension→expert table (fail-fast) ----------------------------
# A minimal MVP stand-in for the S3.5 dimension catalog (kept DISTINCT — see the plan's Non-Goals).
# Field names mirror ``dimension.Dimension`` so the two reconcile cleanly when S3.5 lands. No ``label``

29
src/portfolio_optimiser/outbox.py
View file

@ -189,6 +189,35 @@ def write_exploration(
return path
def write_plan_review(
outbox_dir: str,
run_id: str,
*,
payload: Mapping[str, Any],
) -> Path:
"""Write ``{run_id}-plan-review.json`` — the open question of a PARKED exploration (U12) — and
return its path.
This is the outbox half of the asynchronous HITL door: the run writes the question, the expert
writes the answer into a separate review INBOX, days later. The two folders are never the same
one, for the reason the verdict inbox is never the outbox — a run that read its own output as
input would be answering itself.
Takes an already-rendered plain mapping (``explore.parked_payload``) for the reason
``write_exploration`` does: ``explore`` imports ``agent_framework`` and this layer stays
MAF-free. Byte-deterministic like its neighbours.
**Last write wins**, exactly one open question per run: a revision produces a NEW review of a
REPLANNED plan, and leaving the superseded one on disk would let an expert answer a question
the loop has already moved past. Staleness is caught anyway — the answer names the
``request_id`` it answers — but the file should not invite it."""
directory = Path(outbox_dir)
directory.mkdir(parents=True, exist_ok=True)
path = directory / f"{run_id}-plan-review.json"
path.write_text(_dump({"run_id": run_id, **dict(payload)}), encoding="utf-8")
return path
def write_run_config(
config_dir: str,
run_id: str,

277
src/portfolio_optimiser/run.py
View file

@ -61,9 +61,16 @@ from portfolio_optimiser.explore import (
ExplorationContract,
ExplorationResult,
ExplorationTrace,
ParkedStateError,
PlanReviewDecision,
PlanReviewParked,
explore,
exploration_notice,
load_exploration_contract,
load_parked,
parked_notice,
parked_payload,
resume_exploration,
terminal_plan_reviewer,
trace_payload,
)
@ -92,7 +99,7 @@ from portfolio_optimiser.provenance import ProvenanceStamp
from portfolio_optimiser.reference_domain import Project, load_reference_projects
from portfolio_optimiser.tracing import TracingConfigError, configure_tracing, tracing_notice
from portfolio_optimiser.validator import Rejection, ValidatedProposal, baseline_from_project
from portfolio_optimiser import okf, outbox
from portfolio_optimiser import hitl, okf, outbox
from portfolio_optimiser.semretrieval import (
SEMANTIC_WEIGHT_DEFAULT,
Embedder,
@ -1647,6 +1654,37 @@ def main(argv: list[str] | None = None) -> int:
"round trip is recorded in {run_id}-exploration.json, feedback verbatim. Input that ends "
"without an answer is an error, NEVER a sign-off",
)
parser.add_argument(
"--checkpoint-dir",
default=None,
metavar="DIR",
help="U12 ASYNCHRONOUS HITL door (REQUIRES --explore, --explore-config with "
"enable_plan_review, --run-id and --outbox-dir; refused together with --plan-review): "
"instead of blocking on a human at this terminal, park the exploration's plan review to "
"disk. The workflow's checkpoints go here and the question goes to "
"{run_id}-plan-review.json in the outbox; an expert answers days later by dropping "
"{run_id}-plan-review-answer.json into a review inbox, and --resume picks it up",
)
parser.add_argument(
"--review-inbox",
default=None,
metavar="DIR",
help="where the expert drops their answer to a parked plan review (READ-only, and never "
"the same folder as --outbox-dir: a run that read its own output as input would be "
"answering itself). Required by --resume",
)
parser.add_argument(
"--resume",
default=None,
metavar="RUN_ID",
help="resume the exploration parked under RUN_ID (REQUIRES --checkpoint-dir and "
"--review-inbox): read the open question from the outbox, the answer from the review "
"inbox, and drive the exploration onward in THIS process. The prompt, the bounds and the "
"knowledge bases are read from the parked state, not from argv — the workflow has to be "
"rebuilt exactly as it was for the checkpoint to be accepted at all. A revision makes the "
"manager replan and park a NEW question; an approval lets the run continue into the "
"pipeline as usual",
)
parser.add_argument(
"--mcp-config",
default=None,
@ -1793,6 +1831,11 @@ def main(argv: list[str] | None = None) -> int:
# this list is silently dropped rather than refused — which is the whole reason the
# list enumerates every distinguishable flag instead of the ones that would misbehave.
"--plan-review": args.plan_review,
# The three U12 flags, listed for exactly that reason: report mode returns before the
# resume dispatch, so an omission here is a silent drop, not a refusal.
"--checkpoint-dir": args.checkpoint_dir is not None,
"--review-inbox": args.review_inbox is not None,
"--resume": args.resume is not None,
}
if any(report_forbidden.values()):
print(
@ -1843,6 +1886,11 @@ def main(argv: list[str] | None = None) -> int:
# --explore" would tell an operator who wrote --portfolio --plan-review to add the one
# flag this mode also refuses.
"--plan-review": args.plan_review,
# And the asynchronous half of the same door, on the same side of the partition and by
# NAME for the same reason.
"--checkpoint-dir": args.checkpoint_dir,
"--review-inbox": args.review_inbox,
"--resume": args.resume,
}
offending = [name for name, value in single_only.items() if value]
if offending:
@ -1952,6 +2000,108 @@ def main(argv: list[str] | None = None) -> int:
file=sys.stderr,
)
return 1
# --- U12, the asynchronous half. Every refusal names its flags, and every one of them fires
# BEFORE the first model call: a resume that is going to be refused must be refused while it
# is still free (the økt-57 hoist), and a park that cannot write its question must not run at
# all — the whole point of the door is that somebody can answer it afterwards.
if args.checkpoint_dir is not None and args.plan_review:
print(
"run refused: --plan-review and --checkpoint-dir are two doors onto one review — the "
"first answers it at this terminal, the second parks it for another process. Refused "
"rather than ranked: silently preferring either would block an operator who asked for "
"the other",
file=sys.stderr,
)
return 1
if args.resume is not None:
if args.explore is not None:
print(
"run refused: --resume and --explore are two sources of one exploration. --resume "
"continues the one recorded in the parked state (its own prompt, bounds and "
"bases); --explore starts a new one. Merging would silently drop a prompt",
file=sys.stderr,
)
return 1
if args.mandate is not None:
print(
"run refused: --resume and --mandate are two sources of one mandate — the resumed "
"exploration SHAPES one (the --explore + --mandate refusal, one time-scale later)",
file=sys.stderr,
)
return 1
if args.run_id is not None:
print(
"run refused: --resume and --run-id are two sources of one run id. --resume names "
"the parked run, and the resumed leg keeps writing under that same id",
file=sys.stderr,
)
return 1
if args.live_dry_run:
print(
"run refused: --resume and --live-dry-run contradict each other (the drill stops "
"before the first model call; resuming an exploration is model calls) — pick one",
file=sys.stderr,
)
return 1
if args.checkpoint_dir is None:
print(
"run refused: --resume requires --checkpoint-dir (the workflow state a resume "
"restores from lives there; without it there is nothing to resume)",
file=sys.stderr,
)
return 1
if args.review_inbox is None:
print(
"run refused: --resume requires --review-inbox (the expert's answer lives there, "
"and a resume with no answer would have to invent one)",
file=sys.stderr,
)
return 1
if not args.outbox_dir:
print(
"run refused: --resume requires --outbox-dir (the open question was written "
"there as {run_id}-plan-review.json, and it is what names the review to answer)",
file=sys.stderr,
)
return 1
if not args.bundle_dir:
print(
"run refused: --resume requires --bundle-dir (the resumed exploration navigates "
"knowledge bases, exactly as the parked one did)",
file=sys.stderr,
)
return 1
# ONE run id across the suspension. --run-id was refused above precisely so this
# assignment is the only source, and the resumed leg keeps writing under the id the parked
# leg used — an artefact set split across two ids would describe two runs that never were.
args.run_id = args.resume
elif args.checkpoint_dir is not None and args.explore is None:
print(
"run refused: --checkpoint-dir requires --explore (to park a plan review) or --resume "
"(to lift one); on its own it names a folder nothing would ever be written to",
file=sys.stderr,
)
return 1
if args.review_inbox is not None and args.resume is None:
print(
"run refused: --review-inbox requires --resume (the answers there are read by a "
"resume and by nothing else, so the folder would be named and never opened)",
file=sys.stderr,
)
return 1
if args.checkpoint_dir is not None and args.explore is not None:
# The HOIST again, and it is the one that matters most here: the question artefact IS the
# asynchronous door. Without somewhere to write it the exploration would spend its whole
# budget and then have no way to say what it stopped to ask — a park indistinguishable
# from a crash, days before anybody noticed.
if not args.outbox_dir or not args.run_id:
print(
"run refused: --checkpoint-dir requires --outbox-dir and --run-id, settled BEFORE "
"the exploration runs: the parked question is written as "
"{run_id}-plan-review.json, and without it the review could never be answered",
file=sys.stderr,
)
return 1
if args.explore_config is not None and args.explore is None:
print(
"run refused: --explore-config requires --explore (the bounds describe an exploration "
@ -2022,7 +2172,19 @@ def main(argv: list[str] | None = None) -> int:
# The two messages share the token ``enable_plan_review`` and must NOT share their
# distinguishing wording: a test asserting on the shared substring passes against a
# surface missing one of the branches entirely (measured in økt 57 on --explore).
if exploration_contract.enable_plan_review and not args.plan_review:
if args.checkpoint_dir is not None and not exploration_contract.enable_plan_review:
print(
"run refused: --checkpoint-dir was given but --explore-config sets "
"enable_plan_review false, so nothing would ever park and the checkpoints would "
"be written and never read (refused, never silently ignored)",
file=sys.stderr,
)
return 1
if (
exploration_contract.enable_plan_review
and not args.plan_review
and args.checkpoint_dir is None
):
# The refusal SURVIVES F4 — a run must never stop at a review nobody can answer — but
# its old wording ("the synchronous door is the library API") stopped being true the
# moment this CLI grew one, so it names the flag instead. A claim a surface makes about
@ -2030,7 +2192,8 @@ def main(argv: list[str] | None = None) -> int:
print(
"run refused: --explore-config sets enable_plan_review but no reviewer was "
"offered, so the run would stop at a review nobody can answer. Add --plan-review "
"to answer it at this terminal, or set enable_plan_review to false",
"to answer it at this terminal, or --checkpoint-dir to park it for an expert to "
"answer later, or set enable_plan_review to false",
file=sys.stderr,
)
return 1
@ -2093,7 +2256,13 @@ def main(argv: list[str] | None = None) -> int:
# door must know about them BEFORE loading the file, or a missing one crashes deep inside
# ``explore()`` instead of being refused here, at the door, by name.
required_scripted_roles: Sequence[str] = _SCRIPTED_ROLES
if args.explore is not None:
if args.explore is not None or args.resume is not None:
# ``--resume`` rebuilds the SAME workflow with the SAME three participants, so it needs
# the same three replies. Measured, not reasoned: without ``--resume`` here the child
# process died on ``KeyError: 'navigator'`` deep inside ``fresh_exploration_workflow``
# — the identical defect MAJOR-2 closed for ``--explore`` in økt 62, reappearing on the
# second surface that builds an exploration. A gate that names one door and not the
# other is the drift this comment exists to stop happening a third time.
required_scripted_roles = _SCRIPTED_ROLES + _EXPLORATION_SCRIPTED_ROLES
try:
replies = _load_scripted_replies(args.scripted_replies, required_scripted_roles)
@ -2120,27 +2289,86 @@ def main(argv: list[str] | None = None) -> int:
# marked hypothesis, an exhausted budget, and (since F4) a plan review the operator left
# unanswered — is the RUN failing, not the caller erring, and leaves as it does for the debate
# today.
if args.explore is not None:
assert (
exploration_contract is not None
) # guarded above: --explore requires --explore-config
# The resume's two loads happen HERE, before the trace block below: they are refusals, and a
# refusal must not first overwrite {run_id}-exploration.json with an empty trace — the record
# of what the PARKED leg did is the only evidence of the run so far. This is also the økt-57
# hoist in its purest form: not answered yet is the NORMAL state of this door, so it has to be
# free. Both errors are ``ValueError``s (``PlanReviewAnswerError``) or ``ExplorationError``
# (``ParkedStateError``), and both are caught by NAME rather than left to escape as tracebacks.
resumed: tuple[Any, PlanReviewDecision] | None = None
if args.resume is not None:
question = hitl.read_plan_review_question(args.outbox_dir, args.resume)
if question is None:
print(
f"run refused: no parked plan review for run {args.resume!r} in "
f"{args.outbox_dir!r} (expected {args.resume}-plan-review.json) — there is "
f"nothing to resume",
file=sys.stderr,
)
return 1
try:
parked_state = load_parked(question)
answer = hitl.load_plan_review_answer(
args.review_inbox, args.resume, request_id=parked_state.request_id
)
except (hitl.PlanReviewAnswerError, ParkedStateError) as exc:
print(f"run refused: {exc}", file=sys.stderr)
return 1
resumed = (
parked_state,
PlanReviewDecision.approve()
if answer.decision == "approve"
else PlanReviewDecision.revise(answer.feedback),
)
if args.explore is not None or resumed is not None:
exploration_trace = ExplorationTrace()
exploration: ExplorationResult | None = None
parked_now: PlanReviewParked | None = None
try:
exploration = asyncio.run(
explore(
args.explore,
contract=exploration_contract,
bundle_dirs=(args.bundle_dir,),
profile=args.profile,
client_factory=scripted_client_factory,
trace=exploration_trace,
# The F4 door. Built here and never inside ``explore()``: the loop owns the
# seam, the CLI owns which reviewer fills it, and a library that reached for
# stdin on its own would answer for a caller that never offered to.
plan_reviewer=terminal_plan_reviewer() if args.plan_review else None,
if resumed is not None:
# The parked state, not argv, is what rebuilds the workflow: the graph has to match
# the checkpoint's signature for the restore to be accepted at all, so an operator
# who had to re-supply the prompt and the bounds could get one wrong and find out
# as a restore failure days later.
parked_state, decision = resumed
exploration = asyncio.run(
resume_exploration(
parked_state,
decision,
checkpoint_dir=args.checkpoint_dir,
profile=args.profile,
client_factory=scripted_client_factory,
trace=exploration_trace,
)
)
)
else:
assert (
exploration_contract is not None
) # guarded above: --explore requires --explore-config
exploration = asyncio.run(
explore(
args.explore,
contract=exploration_contract,
bundle_dirs=(args.bundle_dir,),
profile=args.profile,
client_factory=scripted_client_factory,
trace=exploration_trace,
# The F4 door. Built here and never inside ``explore()``: the loop owns the
# seam, the CLI owns which reviewer fills it, and a library that reached
# for stdin on its own would answer for a caller that never offered to.
plan_reviewer=terminal_plan_reviewer() if args.plan_review else None,
# The U12 door. Mutually exclusive with the one above, refused at the top.
checkpoint_dir=args.checkpoint_dir,
)
)
except PlanReviewParked as parked_exc:
# NOT an error, and not a completed run either — the third channel, for the reason
# ``BudgetExceeded`` has its own: the exploration produced no mandate, so returning one
# would let a caller book "explored" for a loop suspended mid-plan. Caught here rather
# than left to escape, because parking is what the operator ASKED for by giving
# --checkpoint-dir; the artefact is where a machine reads that it happened.
parked_now = parked_exc
finally:
# From a ``finally``, exactly as ``write_parse_failures`` is (Fase 1b, funn 1): the run
# that most needs this evidence is the one a cap cut short, and that run returns
@ -2156,6 +2384,13 @@ def main(argv: list[str] | None = None) -> int:
completed=exploration is not None,
),
)
if parked_now is not None:
outbox.write_plan_review(
args.outbox_dir, args.run_id, payload=parked_payload(parked_now.parked)
)
print(parked_notice(parked_now.parked, run_id=args.run_id))
return 0
assert exploration is not None # the only other way out of the block above is an exception
print(exploration_notice(exploration))
mandate = exploration.mandate

689
tests/test_async_plan_review_loadbearing.py Normal file
View file

@ -0,0 +1,689 @@
"""U12 + asynchronous U13 (plan § D.2 row 3) — a plan review a human answers over DAYS.
F4 (økt 63) made "still spørsmål, be om svar, bruke svarene" reachable from the CLI, but only
SYNCHRONOUSLY: ``terminal_plan_reviewer`` blocks the loop on a human at a terminal, so the answer
has to arrive while the process is alive. Målbilde §3's time-scale is the other one — the expert
answers days later, in a process that never saw the run — and that is impossible without carrying
the workflow's state to disk.
**The measured trap this row is built around** (§ F row A4, and confirmed here against the
INSTALLED source rather than the plan's prose): ``FileCheckpointStorage.list_checkpoints``
(``_workflows/_checkpoint.py:386-388``) swallows a deserialisation failure into a
``logger.warning`` and returns an EMPTY list. Without ``MagenticPlanReviewRequest`` and
``MagenticPlanReviewResponse`` in ``allowed_checkpoint_types``, a resume therefore fails as an
ABSENCE — "nothing to resume" — not as an error. A test asserting "the listing is empty, so there
is nothing to resume" would be GREEN against exactly that defect, which is why every test here
asserts that the resume DID something instead.
**The discriminator, in both halves.** A door that writes a question file and a resume that reads
an answer file both pass "the expert was asked" while failing the målbilde. So the goal test drives
the F4 T1 shape across TWO process boundaries: ``revise`` written into an inbox on day 1 must reach
the manager, make it replan, and produce a SECOND question about the NEW plan — which an
always-approve resume, or one that discards the answer, cannot produce.
The witness is the artefacts, never scraped stdout: ``{run_id}-plan-review.json`` is the question
and ``{run_id}-exploration.json`` is the record of what was decided.
"""
from __future__ import annotations
import json
import subprocess
import sys
from pathlib import Path
from typing import Any
import pytest
from portfolio_optimiser import explore as ex
from portfolio_optimiser import hitl, run
_REPO = Path(__file__).resolve().parents[1]
_BUNDLE_DIR = _REPO / "shared" / "examples" / "bygg-energi-mikro"
_PID = "BYGG-KONTOR-NORD"
_RUN_ID = "async-review"
_PROPOSER_REPLY = json.dumps(
{
"measure": "LED-retrofit",
"affected_items": [{"code": "ENERGI-TOTAL-EL", "quantity": 300000, "unit_cost": 1.0}],
"claimed_saving_nok": 30000,
}
)
_MANAGER_REPLY = json.dumps(
{
"is_request_satisfied": {"reason": "r", "answer": True},
"is_in_loop": {"reason": "r", "answer": False},
"is_progress_being_made": {"reason": "r", "answer": True},
"next_speaker": {"reason": "r", "answer": "hypothesiser"},
"instruction_or_question": {"reason": "r", "answer": "go"},
}
)
_REPLIES = {
"proposer": _PROPOSER_REPLY,
"checker": "VERDICT: APPROVE",
"manager": _MANAGER_REPLY,
"navigator": "NAVIGATOR: read the index.",
"hypothesiser": "HYPOTHESIS: " + json.dumps({"label": "Night setback", "rationale": "y"}),
}
_FEEDBACK = "Also test night setback on the ventilation."
# ---------------------------------------------------------------------------------------------
# Fixture plumbing: the two operator surfaces, driven exactly as an operator would drive them
# ---------------------------------------------------------------------------------------------
def _config_file(tmp_path: Path, **overrides: Any) -> str:
path = tmp_path / "exploration.json"
path.write_text(
json.dumps(
{
"max_rounds": 4,
"max_tokens": 200_000,
"max_stall_count": 2,
"max_reset_count": 1,
"max_plan_revisions": 2,
"enable_plan_review": True,
**overrides,
}
),
encoding="utf-8",
)
return str(path)
def _replies_file(tmp_path: Path) -> str:
path = tmp_path / "replies.json"
path.write_text(json.dumps(_REPLIES), encoding="utf-8")
return str(path)
def _park_argv(tmp_path: Path, **config: Any) -> list[str]:
"""Day 1: explore with the ASYNCHRONOUS door armed. No reviewer at this terminal."""
return [
_PID,
"--docs-dir",
str(_BUNDLE_DIR),
"--bundle-dir",
str(_BUNDLE_DIR),
"--explore",
"Find the cheapest saving.",
"--explore-config",
_config_file(tmp_path, **config),
"--scripted-replies",
_replies_file(tmp_path),
"--outbox-dir",
str(tmp_path / "outbox"),
"--checkpoint-dir",
str(tmp_path / "checkpoints"),
"--run-id",
_RUN_ID,
]
def _resume_argv(tmp_path: Path) -> list[str]:
"""Day N: a process that never saw the run, resuming from the checkpoint and the answer."""
return [
_PID,
"--docs-dir",
str(_BUNDLE_DIR),
"--bundle-dir",
str(_BUNDLE_DIR),
"--scripted-replies",
_replies_file(tmp_path),
"--outbox-dir",
str(tmp_path / "outbox"),
"--checkpoint-dir",
str(tmp_path / "checkpoints"),
"--review-inbox",
str(tmp_path / "review-inbox"),
"--resume",
_RUN_ID,
]
def _question(tmp_path: Path) -> dict[str, Any]:
path = tmp_path / "outbox" / f"{_RUN_ID}-plan-review.json"
assert path.exists(), "a parked exploration must leave the question where an expert can read it"
return json.loads(path.read_text(encoding="utf-8"))
def _artefact(tmp_path: Path) -> dict[str, Any]:
path = tmp_path / "outbox" / f"{_RUN_ID}-exploration.json"
assert path.exists(), "the exploration artefact must be written even when the run parked"
return json.loads(path.read_text(encoding="utf-8"))
def _answer(tmp_path: Path, **payload: Any) -> None:
"""The expert's side of the loop: a file dropped into the review inbox, days later."""
inbox = tmp_path / "review-inbox"
inbox.mkdir(parents=True, exist_ok=True)
body = {"run_id": _RUN_ID, "request_id": _question(tmp_path)["request_id"], **payload}
(inbox / f"{_RUN_ID}-plan-review-answer.json").write_text(json.dumps(body), encoding="utf-8")
def _resume_in_a_fresh_process(tmp_path: Path) -> subprocess.CompletedProcess[str]:
"""The resume runs in its OWN interpreter, because an in-process resume would prove nothing.
The whole claim of U12 is that the only thing crossing the boundary is what is on disk. The
subprocess precedent is ``spikes/e_magentic_resume.py`` and ``test_hosting_loadbearing.py``.
"""
return subprocess.run(
[sys.executable, "-m", "portfolio_optimiser.run", *_resume_argv(tmp_path)],
capture_output=True,
text=True,
cwd=str(_REPO),
)
# ---------------------------------------------------------------------------------------------
# 1. THE GOAL — asked on day 1, answered on day N, and the answer USED
# ---------------------------------------------------------------------------------------------
def test_an_answer_written_days_later_reaches_the_manager_and_produces_a_new_question(
tmp_path, capsys
) -> None:
"""T1: the whole row, across two process boundaries.
Day 1 parks at the plan review. The expert writes ``revise`` into the inbox. A FRESH
interpreter resumes from the checkpoint alone — and because the feedback reached the manager,
the manager replans and asks AGAIN about the NEW plan.
RED against a resume that discards the answer, and RED against one that always approves: both
yield a single review and no second question. RED against an in-process-only door: there is no
checkpoint for the child to resume from.
"""
assert run.main(_park_argv(tmp_path)) == 0, capsys.readouterr().err
first = _question(tmp_path)
assert first["index"] == 0
assert first["request_id"], "without the request id the answer can never be routed back"
_answer(tmp_path, decision="revise", feedback=_FEEDBACK)
completed = _resume_in_a_fresh_process(tmp_path)
assert completed.returncode == 0, completed.stderr
second = _question(tmp_path)
assert second["index"] == 1, (
"a revision must produce a SECOND question about the replanned plan — a resume that "
"discarded the answer would finish, or park again on the SAME review"
)
assert second["request_id"] != first["request_id"], (
"the replanned review is a new request; reusing the old id would route the next answer "
"into a request the orchestrator has already retired"
)
reviews = _artefact(tmp_path)["plan_reviews"]
assert [r["decision"] for r in reviews] == ["revise"], (
"the record must carry the decision the expert actually made"
)
assert reviews[0]["feedback"] == _FEEDBACK, (
"what a human told the loop is worth nothing paraphrased"
)
def test_the_answer_is_carried_all_the_way_to_a_completed_exploration(tmp_path, capsys) -> None:
"""T2: the control for T1 — the door does not only re-ask, it can also FINISH.
A gate that could only ever park again would be a hang wearing a loop's clothes. Answering
``approve`` must let the exploration conclude and the pipeline run, leaving NO open question.
"""
assert run.main(_park_argv(tmp_path)) == 0, capsys.readouterr().err
_answer(tmp_path, decision="approve")
completed = _resume_in_a_fresh_process(tmp_path)
assert completed.returncode in (0, 1), completed.stderr
assert "Traceback" not in completed.stderr, completed.stderr
artefact = _artefact(tmp_path)
assert artefact["completed"] is True, (
"an approved review must let the exploration conclude, not park again"
)
assert [r["decision"] for r in artefact["plan_reviews"]] == ["approve"]
assert (
hitl.pending_plan_reviews(str(tmp_path / "outbox"), str(tmp_path / "review-inbox")) == []
), "a concluded exploration leaves no question waiting for anybody"
# ---------------------------------------------------------------------------------------------
# 2. THE MEASURED TRAP — a checkpoint that cannot be read back
# ---------------------------------------------------------------------------------------------
def test_the_parked_checkpoint_can_actually_be_read_back(tmp_path, capsys) -> None:
"""T3: the trap, asserted POSITIVELY.
``list_checkpoints`` turns a blocked deserialisation into an empty list, so the failure mode
is silence. This asserts the opposite of silence: the id the question file names resolves to a
checkpoint that loads. RED the moment ``_ALLOWED_CHECKPOINT_TYPES`` stops naming both types.
"""
import asyncio
assert run.main(_park_argv(tmp_path)) == 0, capsys.readouterr().err
checkpoint_id = _question(tmp_path)["checkpoint_id"]
assert checkpoint_id, "parking on an unreadable checkpoint is an unanswerable question"
storage = ex.checkpoint_storage(str(tmp_path / "checkpoints"))
loaded = asyncio.run(storage.load(checkpoint_id))
assert loaded.checkpoint_id == checkpoint_id
def test_a_park_with_no_readable_checkpoint_refuses_instead_of_writing_a_dead_question(
tmp_path, capsys, monkeypatch
) -> None:
"""T4: fail LOUDLY where the framework fails silently.
If the listing comes back empty there is nothing to resume from, and writing the question
anyway would hand an expert a review whose answer can never be applied — the fourth face of
the verification law, built into our own surface. Simulated by emptying the allow-list, which
is exactly what produces an empty listing in the installed source.
It leaves as a RAISE, not an rc-1 refusal, and that is the consistent call rather than a
softer one: argv was fine and the loop had already spent, so this is the run failing — the
same channel ``BudgetExceeded`` and an unreadable marked hypothesis use. What the door owes is
that it fails LOUDLY where the framework fails silently, and that no dead question is left
behind for somebody to answer into the void.
"""
monkeypatch.setattr(ex, "_ALLOWED_CHECKPOINT_TYPES", ())
with pytest.raises(ex.CheckpointUnreadable):
run.main(_park_argv(tmp_path))
capsys.readouterr()
assert not (tmp_path / "outbox" / f"{_RUN_ID}-plan-review.json").exists(), (
"a question nobody can answer must not be written at all"
)
# ---------------------------------------------------------------------------------------------
# 3. FAIL-CLOSED ON THE EXPERT'S OWN INPUT (the F4 rule, on a file instead of a terminal)
# ---------------------------------------------------------------------------------------------
def test_an_answer_outside_the_vocabulary_is_refused_never_read_as_a_sign_off(
tmp_path, capsys
) -> None:
"""T5: the closed vocabulary survives the move from stdin to a file.
``terminal_plan_reviewer`` re-asks anything it does not recognise; a file cannot be re-asked,
so the only honest answer is a refusal. Reading it as approval would sign a plan nobody signed.
"""
assert run.main(_park_argv(tmp_path)) == 0, capsys.readouterr().err
_answer(tmp_path, decision="looks fine to me")
completed = _resume_in_a_fresh_process(tmp_path)
assert completed.returncode == 1, completed.stdout
assert "looks fine to me" in completed.stderr, completed.stderr
assert _artefact(tmp_path)["plan_reviews"] == [], (
"an unreadable answer must not be recorded as a decision"
)
def test_a_revision_with_nothing_to_revise_is_refused(tmp_path, capsys) -> None:
"""T6: ``revise`` without feedback is the same defect wearing a valid token.
``PlanReviewDecision.revise`` refuses an empty revision at the library door too, and that is
exactly why the assertion here is on the SHAPE of the failure rather than on the exit code:
measured, removing the inbox guard still gives rc 1 and still puts the word "revise" on
stderr — as a TRACEBACK out of the library. A test that stopped at those two facts could not
tell a fail-closed door from an unhandled exception, so it asserts the structured refusal.
"""
assert run.main(_park_argv(tmp_path)) == 0, capsys.readouterr().err
_answer(tmp_path, decision="revise", feedback=" ")
completed = _resume_in_a_fresh_process(tmp_path)
assert completed.returncode == 1, completed.stdout
assert "Traceback" not in completed.stderr, completed.stderr
assert "run refused" in completed.stderr, completed.stderr
assert "revise" in completed.stderr.lower(), completed.stderr
def test_an_answer_to_a_different_review_is_refused_never_applied_to_this_one(
tmp_path, capsys
) -> None:
"""T7: staleness is a refusal, not a silent misapplication.
Two reviews of one run share a file name, so the answer names the ``request_id`` it answers.
An answer left over from the previous round must not be applied to the current question — that
would sign off a plan the expert never saw.
"""
assert run.main(_park_argv(tmp_path)) == 0, capsys.readouterr().err
_answer(tmp_path, decision="approve")
inbox = tmp_path / "review-inbox" / f"{_RUN_ID}-plan-review-answer.json"
stale = json.loads(inbox.read_text(encoding="utf-8"))
stale["request_id"] = "a-request-from-last-week"
inbox.write_text(json.dumps(stale), encoding="utf-8")
completed = _resume_in_a_fresh_process(tmp_path)
assert completed.returncode == 1, completed.stdout
assert "a-request-from-last-week" in completed.stderr, completed.stderr
def test_a_resume_with_no_answer_yet_refuses_before_spending_anything(tmp_path, capsys) -> None:
"""T8: "not answered yet" is the normal state of this door, and it must be cheap.
The hoist rule from økt 57: a refusal that fires AFTER the model calls is indistinguishable
from one that fires before, by exit code alone — so this asserts that NOTHING was spent.
"""
assert run.main(_park_argv(tmp_path)) == 0, capsys.readouterr().err
before = json.loads(
(tmp_path / "outbox" / f"{_RUN_ID}-plan-review.json").read_text(encoding="utf-8")
)
completed = _resume_in_a_fresh_process(tmp_path)
assert completed.returncode == 1, completed.stdout
after = json.loads(
(tmp_path / "outbox" / f"{_RUN_ID}-plan-review.json").read_text(encoding="utf-8")
)
assert after == before, (
"a resume with no answer must not touch the run at all — an unchanged question file is "
"what proves the exploration was never restarted"
)
# ---------------------------------------------------------------------------------------------
# 4. THE PENDING REGISTRY (hitl.py) — who is still waiting on whom
# ---------------------------------------------------------------------------------------------
def test_a_parked_review_is_pending_until_its_own_answer_lands(tmp_path, capsys) -> None:
"""T9: ``hitl.pending_plan_reviews`` is the machine-readable "still waiting", mirroring
``hitl.pending`` for proposals: an outbox question whose answer is not yet in the inbox.
The join is on ``request_id``, so an answer to a DIFFERENT review leaves the question pending
rather than quietly clearing it — the same fail-closed rule the resume applies.
"""
outbox, inbox = str(tmp_path / "outbox"), str(tmp_path / "review-inbox")
assert run.main(_park_argv(tmp_path)) == 0, capsys.readouterr().err
waiting = hitl.pending_plan_reviews(outbox, inbox)
assert [p.run_id for p in waiting] == [_RUN_ID]
assert waiting[0].plan, "an expert cannot answer a review that does not show them the plan"
_answer(tmp_path, decision="approve")
inbox_file = tmp_path / "review-inbox" / f"{_RUN_ID}-plan-review-answer.json"
wrong = json.loads(inbox_file.read_text(encoding="utf-8"))
wrong["request_id"] = "someone-elses-review"
inbox_file.write_text(json.dumps(wrong), encoding="utf-8")
assert hitl.pending_plan_reviews(outbox, inbox) == waiting, (
"an answer to another review must not clear this one"
)
_answer(tmp_path, decision="approve")
assert hitl.pending_plan_reviews(outbox, inbox) == []
# ---------------------------------------------------------------------------------------------
# 5. THE BUDGET MUST SPAN THE SUSPENSION, NOT RESTART WITH IT
# ---------------------------------------------------------------------------------------------
def test_a_resumed_exploration_does_not_get_a_fresh_budget(tmp_path, capsys) -> None:
"""T10: the hole a park would otherwise open.
Both budget channels live in the process: a fresh ``TokenMeter`` and an empty ledger mean a
resumed exploration could spend its whole cap AGAIN, once per park — unbounded consumption
behind guards that all look satisfied (the S3.4 class). The suspended state therefore carries
what was already spent, and the resume starts from it.
RED when the carry-over is detached: the resumed run then reports a spend of its own calls
only, and the ledger restarts at round 1.
"""
assert run.main(_park_argv(tmp_path)) == 0, capsys.readouterr().err
parked = ex.load_parked(_question(tmp_path))
assert parked.tokens_spent > 0, (
"the parked run made model calls; a zero here would make the assertion below vacuous"
)
_answer(tmp_path, decision="approve")
assert _resume_in_a_fresh_process(tmp_path).returncode in (0, 1)
artefact = _artefact(tmp_path)
assert artefact["tokens_spent"] > parked.tokens_spent, (
"the resumed exploration must add to the day-1 spend, never start over from zero"
)
def test_the_revision_cap_is_counted_across_the_suspension_not_restarted_by_it(
tmp_path, capsys
) -> None:
"""T11: the hole the carry-over closes, and the reason it exists at all.
A revise costs two manager calls, emits no ledger and consumes no round (§ F, A3), so
``max_plan_revisions`` is the ONLY bound on it. With ``max_plan_revisions=1`` a second revision
must be refused — and refused on the strength of what the FIRST process did, which only the
carried ``plan_reviews`` can say.
RED when the carry-over is detached: every leg then counts zero prior revisions, the cap never
binds, and the run parks a third time. That mutation left the entire suite green before this
test existed, which is exactly the vacuous-gate class it was written against.
"""
assert run.main(_park_argv(tmp_path, max_plan_revisions=1)) == 0, capsys.readouterr().err
_answer(tmp_path, decision="revise", feedback=_FEEDBACK)
assert _resume_in_a_fresh_process(tmp_path).returncode == 0
assert _question(tmp_path)["index"] == 1, "the first revision must be applied"
_answer(tmp_path, decision="revise", feedback="And once more.")
second = _resume_in_a_fresh_process(tmp_path)
assert second.returncode in (0, 1), second.stderr
assert "Traceback" not in second.stderr, second.stderr
artefact = _artefact(tmp_path)
assert artefact["stop"] == "plan_revisions_exhausted", (
"the second revision is over the cap and must STOP the exploration, never be sent"
)
assert (
hitl.pending_plan_reviews(str(tmp_path / "outbox"), str(tmp_path / "review-inbox")) == []
), "a stopped exploration must not leave a third question waiting for anybody"
def test_the_review_history_survives_every_leg_not_just_the_last(tmp_path, capsys) -> None:
"""T16: the carried ``plan_reviews`` are the RECORD, and the record is the only witness.
Written because the mutation that detaches ``trace.plan_reviews.extend(parked.plan_reviews)``
left the ENTIRE suite green (measured, økt 64): the revision cap counts
``parked.plan_reviews`` DIRECTLY, so it binds either way, and the first two legs cannot tell
the difference — a park with an empty carried history and one with none look identical until
there are two reviews to carry. An unmeasured seam is this repo's recurring defect class, so
the gate is the THIRD leg, where the two implementations finally diverge.
RED when the carry-over is detached: the artefact then records only the review the LAST
process saw — the day-1 revision vanishes from the run's own history — and the next question
is numbered 1 again, so two distinct reviews of one run share an index.
"""
second_feedback = "And check the pumps while you are at it."
assert run.main(_park_argv(tmp_path, max_plan_revisions=2)) == 0, capsys.readouterr().err
assert _question(tmp_path)["index"] == 0
_answer(tmp_path, decision="revise", feedback=_FEEDBACK)
assert _resume_in_a_fresh_process(tmp_path).returncode == 0
assert _question(tmp_path)["index"] == 1, "the first revision must be applied"
_answer(tmp_path, decision="revise", feedback=second_feedback)
third = _resume_in_a_fresh_process(tmp_path)
assert third.returncode == 0, third.stderr
assert _question(tmp_path)["index"] == 2, (
"the third question is the third review of this run — a resume that dropped the carried "
"history would number it 1 again, and two different plans would share one index"
)
reviews = _artefact(tmp_path)["plan_reviews"]
assert [r["feedback"] for r in reviews] == [_FEEDBACK, second_feedback], (
"the record must carry every decision the expert made, not only the most recent one: "
"what a human told the loop on day 1 is not superseded by what they said on day 2"
)
def test_what_the_first_process_found_survives_into_the_resumed_mandate(tmp_path) -> None:
"""T12: the other half of the carry-over — the loop's own findings.
A plan review can fire mid-run after a stall, and everything the loop found before it would be
lost if the mandate were minted from only what the resuming process observed. Driven through a
CRAFTED parked state (the ``budget_stop`` precedent) because the scripted manager never stalls,
so the mid-run park has no reachable path today — but the carry-over that serves it does.
RED when the ledger and hypotheses are dropped on resume: the mandate then names only what the
second leg saw, and the returned ledger restarts at the resumed round.
"""
import asyncio
import dataclasses
from portfolio_optimiser.simulation import scripted_factory
factory = scripted_factory(_REPLIES, [])
contract = ex.load_exploration_contract(_config_file(tmp_path))
checkpoints = str(tmp_path / "checkpoints")
with pytest.raises(ex.PlanReviewParked) as caught:
asyncio.run(
ex.explore(
"Find the cheapest saving.",
contract=contract,
bundle_dirs=(str(_BUNDLE_DIR),),
client_factory=factory,
checkpoint_dir=checkpoints,
)
)
carried = dataclasses.replace(
caught.value.parked,
hypotheses=("HYPOTHESIS: " + json.dumps({"label": "Carried", "rationale": "found first"}),),
ledger=(
ex.LedgerEntry(
round_index=1,
is_request_satisfied=False,
is_in_loop=False,
is_progress_being_made=True,
next_speaker="hypothesiser",
instruction_or_question="keep going",
speaker_known=True,
),
),
)
result = asyncio.run(
ex.resume_exploration(
carried,
ex.PlanReviewDecision.approve(),
checkpoint_dir=checkpoints,
client_factory=factory,
)
)
assert "Carried" in [a.label for a in result.mandate.approaches], (
"a hypothesis the FIRST process found must reach the mandate the second one mints"
)
assert result.ledger_log[0].instruction_or_question == "keep going", (
"the resumed ledger must continue the suspended run's, not restart it"
)
# ---------------------------------------------------------------------------------------------
# 6. THE CLI PARTITION — two doors onto one review, and the preconditions
# ---------------------------------------------------------------------------------------------
def test_the_two_review_doors_are_refused_together(tmp_path, capsys) -> None:
"""T11: ``--plan-review`` and ``--checkpoint-dir`` are two sources of one answer.
Refused rather than ranked, for the reason ``--explore`` + ``--mandate`` is: silently
preferring one would mean an operator who asked for the asynchronous door got the synchronous
one, and found out by being blocked at a terminal.
"""
rc = run.main([*_park_argv(tmp_path), "--plan-review"])
err = capsys.readouterr().err
assert rc == 1
assert "--plan-review" in err and "--checkpoint-dir" in err, err
def test_an_asynchronous_door_without_somewhere_to_put_the_question_is_refused_early(
tmp_path, capsys
) -> None:
"""T12: the økt-57 hoist. The question artefact IS the door — without ``--outbox-dir`` and
``--run-id`` the exploration would spend a full budget and then have nowhere to say what it
was waiting for. Refused before the first model call; the empty checkpoint dir is what proves
nothing ran.
"""
argv = [a for a in _park_argv(tmp_path) if a not in ("--outbox-dir", str(tmp_path / "outbox"))]
rc = run.main(argv)
err = capsys.readouterr().err
assert rc == 1
assert "--outbox-dir" in err and "--checkpoint-dir" in err, err
assert not (tmp_path / "checkpoints").exists() or not list(
(tmp_path / "checkpoints").iterdir()
), "the refusal must fire before the exploration starts, not after it has paid"
def test_a_review_with_neither_door_is_still_refused_and_says_which_two_exist(
tmp_path, capsys
) -> None:
"""T13: the F4 refusal is WIDENED, never weakened.
``enable_plan_review`` with no reviewer at all is still a hang. The message must now name BOTH
doors, because an operator told only about ``--plan-review`` cannot find the asynchronous one.
"""
argv = [
a
for a in _park_argv(tmp_path)
if a not in ("--checkpoint-dir", str(tmp_path / "checkpoints"))
]
rc = run.main(argv)
err = capsys.readouterr().err
assert rc == 1
assert "--plan-review" in err and "--checkpoint-dir" in err, err
@pytest.mark.parametrize(
"drop,expected",
[
(("--checkpoint-dir",), "--checkpoint-dir"),
(("--review-inbox",), "--review-inbox"),
],
)
def test_a_resume_names_the_coordinate_it_is_missing(
tmp_path, capsys, drop: tuple[str, ...], expected: str
) -> None:
"""T14: a resume needs the checkpoints AND the inbox. Refused by NAME rather than falling
through to a generic message, so an operator is told which of the two to add.
"""
argv = _resume_argv(tmp_path)
for flag in drop:
i = argv.index(flag)
del argv[i : i + 2]
rc = run.main(argv)
err = capsys.readouterr().err
assert rc == 1
assert expected in err, err
def test_resume_and_explore_are_refused_together(tmp_path, capsys) -> None:
"""T15: two sources of one exploration. ``--resume`` continues the exploration recorded in the
parked file; ``--explore`` starts a new one. Merging them would silently drop one prompt.
"""
argv = [
*_resume_argv(tmp_path),
"--explore",
"A different question entirely.",
"--explore-config",
_config_file(tmp_path),
]
rc = run.main(argv)
err = capsys.readouterr().err
assert rc == 1
assert "--resume" in err and "--explore" in err, err