open/portfolio-optimiser
1
0
Fork 0
Code Issues Releases 4 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

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,
)