Coverage for gco_mcp / mission / engine.py: 100.00%
619 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Five-phase iteration loop driver for the Mission goal-directed loop.
3The :class:`MissionEngine` owns one ``run_iteration`` lifecycle per call: it
4loads the persisted session, walks the iteration through propose → execute
5→ observe → evaluate → decide, persists the resulting record, and writes a
6Final_Report when the verdict is terminal. Every external dependency is
7injected at construction time so unit tests can supply mocks for the tool
8dispatcher, the sampling callable, and the script sandbox runner.
10Why a class rather than a free function? Two reasons.
12* Each phase needs the same handful of dependencies (the backend, the
13 tool dispatcher, the cost-estimator map, the clock). Threading them
14 through every method as positional arguments would be tedious and
15 error-prone; the dataclass shape gives every phase one place to look
16 for them.
17* A test that exercises a single phase in isolation needs to construct
18 a ``MissionEngine`` with stubbed dependencies and call the private
19 method directly. Having the dependencies on the instance — rather
20 than as module-level singletons — keeps the engine pure and free of
21 process-global state.
23Phase contract:
25* Each ``_*_phase`` method is wrapped in a try/finally that emits
26 exactly one ``audit.emit_phase_event`` regardless of whether the body
27 succeeded or raised. The matching :class:`PhaseRecord` is appended to
28 ``record["phases"]`` in the same finally block, so a failed phase
29 still produces a structured record on the iteration.
30* Any phase that raises propagates the exception out of
31 ``run_iteration`` after the engine marks the session as ``failed``,
32 appends the partial iteration, and persists. Subsequent calls to
33 ``run_iteration`` on a ``failed`` session refuse with
34 ``session_failed``.
36Determinism: only the Decide_Phase consults a clock, and it does so by
37calling ``self.now()`` exactly once per call so the value is observable
38and pinnable from tests. The Propose_Phase's deterministic fallback uses
39no clock and no random source. The Execute_Phase reads the clock for
40the per-phase ``started_at`` / ``ended_at`` timestamps but its outputs
41(the tool-call records) do not depend on those values.
43The ``Context`` type is from FastMCP and brings a heavy dependency tree
44(MCP transport, ``contextvars``, etc.) that unit tests do not need. We
45type ``ctx`` as ``Any | None`` so the engine module imports cleanly in
46isolation; the production wiring threads a real :class:`fastmcp.Context`
47through the dispatcher and sampler callables, where its concrete type
48matters. This is the same trade-off the existing ``gco_mcp/tools/*.py``
49modules make for tools that take an injected context.
50"""
52from __future__ import annotations
54import contextlib
55from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence
56from dataclasses import dataclass, field
57from datetime import UTC, datetime
58from typing import Any, cast
60from gco.bedrock import BedrockFTUFormNotAcceptedError
62from . import audit, decide, final_report
63from .checkpoints import mark_checkpoint
64from .predicate import PredicateRejected, evaluate_predicate, parse_predicate
65from .sampling import SamplingFallback, SamplingUsed
66from .types import (
67 TERMINAL_STATES,
68 TERMINAL_VERDICTS,
69 Criterion,
70 CriterionResult,
71 IterationRecord,
72 Observation,
73 PhaseRecord,
74 SessionState,
75 Strategy,
76 ToolCallRecord,
77 VerdictLabel,
78 VerdictReason,
79)
81# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
82# Generated at (UTC): 2026-09-18T02:11:36Z
83# Generated from Git commit: b8faa9689385cea16155a285a7f70cf6d488e512
84# Flowchart(s) generated from this file:
85# * ``MissionEngine.run_iteration`` -> ``diagrams/code_diagrams/gco_mcp/mission/engine.MissionEngine_run_iteration.html``
86# (PNG: ``diagrams/code_diagrams/gco_mcp/mission/engine.MissionEngine_run_iteration.png``)
87# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
88# <pyflowchart-code-diagram> END
91__all__ = [
92 "MissionEngine",
93 "MissionEngineError",
94]
97# ---------------------------------------------------------------------------
98# Error
99# ---------------------------------------------------------------------------
102class MissionEngineError(Exception):
103 """Raised by the engine for stable, code-keyed lifecycle errors.
105 The :attr:`code` attribute carries a short stable string (e.g.
106 ``"session_not_found"``, ``"session_terminal"``, ``"session_paused"``,
107 ``"session_failed"``) that the MCP tool wrappers and the CLI render
108 as a structured tool error. The exception's string form falls back
109 to ``code`` so logs always show something meaningful even when the
110 caller does not pull the attribute out explicitly.
111 """
113 def __init__(self, code: str, *, message: str | None = None) -> None:
114 self.code: str = code
115 super().__init__(message if message is not None else code)
118# ---------------------------------------------------------------------------
119# Phase-name constants (typed)
120# ---------------------------------------------------------------------------
122# Centralised so the audit emitter and PhaseRecord constructor share one
123# spelling for each phase. Matches the ``Literal`` shape declared on
124# :class:`PhaseRecord` and on :func:`audit.emit_phase_event`.
125_PROPOSE = "propose"
126_EXECUTE = "execute"
127_OBSERVE = "observe"
128_EVALUATE = "evaluate"
129_DECIDE = "decide"
132# ---------------------------------------------------------------------------
133# Defaults
134# ---------------------------------------------------------------------------
137def _default_now() -> Callable[[], datetime]:
138 """Return a clock callable that yields the current UTC datetime.
140 Used as the ``default_factory`` for :attr:`MissionEngine.now`. Wrapping
141 the lambda in a function keeps ``mypy --strict`` happy with the
142 ``Callable[[], datetime]`` annotation while preserving the
143 "constructed once per engine, called many times" semantics.
144 """
145 return lambda: datetime.now(UTC)
148# ---------------------------------------------------------------------------
149# Engine
150# ---------------------------------------------------------------------------
153# Type aliases for the injected callables. Loose on purpose: the precise
154# shapes settle in later slices (sampling in slice 6, sandbox in slice 5).
155ToolDispatcher = Callable[[str, dict[str, Any], Any], Awaitable[Any]]
156SamplingCallable = Callable[..., Awaitable[Any]]
157#: Synchronous contribution merged into each iteration's Observation at the
158#: end of the Observe_Phase. Takes the live session, returns a dict whose
159#: optional ``children`` key lands on the Observation verbatim and whose
160#: optional ``metrics`` dict merges via ``metrics.update(...)`` — the same
161#: contract tool results use. Synchronous and pure-by-convention so the
162#: determinism suite can pin its output byte-for-byte. The return is typed
163#: ``object`` because augmenters are caller-supplied plugins:
164#: ``_apply_observation_augmenters`` checks the shape at runtime and ignores
165#: a non-dict contribution, which the type must leave room for.
166ObservationAugmenter = Callable[[SessionState], object]
167SandboxRunner = Callable[
168 [str, Any, ToolDispatcher],
169 Awaitable[tuple[dict[str, Any], list[ToolCallRecord]]],
170]
173@dataclass
174class MissionEngine:
175 """Driver for the Mission five-phase iteration loop.
177 Construction takes every external dependency the engine needs:
179 * ``backend`` — the persistence layer (filesystem, DynamoDB, …).
180 Engine never reaches outside this protocol for state I/O.
181 * ``tool_dispatcher`` — async callable that invokes one MCP tool.
182 Signature ``(tool_name, args, ctx) -> result``. The engine routes
183 every direct ``tool_calls`` invocation through this callable so
184 tests can swap in a stub that returns canned results.
185 * ``sampling_callable`` — optional async callable that produces an
186 LLM-derived next Strategy when the prior Verdict was ``adjust``
187 and the session has ``use_sampling=true``. Loose signature; slice
188 6 finalises it. ``None`` (or any failure inside it) routes the
189 engine to the deterministic fallback strategy.
190 * ``sandbox_runner`` — optional async callable that runs a scripted
191 Strategy in the Mission sandbox. Signature ``(script, ctx,
192 tool_dispatcher) -> (observation_dict, script_call_log)``. ``None``
193 means the engine refuses any scripted strategy with a clear
194 error instead of silently executing operator-supplied code.
195 * ``now`` — injectable clock. Defaults to a UTC clock; tests pin it
196 so deterministic verdicts (the Decide_Phase consults the clock
197 for budget caps and for the cadence resolver) are reproducible.
198 """
200 backend: Any
201 tool_dispatcher: ToolDispatcher
202 sampling_callable: SamplingCallable | None
203 sandbox_runner: SandboxRunner | None
204 now: Callable[[], datetime] = field(default_factory=_default_now)
205 # Optional async callable that drives the Final_Report's
206 # ``lessons`` and ``recommended_followups`` overlay. Loose typing
207 # mirrors :attr:`sampling_callable` so legacy tests that wire a
208 # plain async stub keep working. Production wiring binds it to a
209 # closure over :func:`mcp.mission.sampling.maybe_sample_final_lessons`
210 # — see :meth:`_maybe_sample_final_lessons`. ``None`` (the default)
211 # disables the overlay; the deterministic templates from
212 # :func:`mcp.mission.final_report.build_deterministic_report` stand
213 # on their own in that case.
214 final_lessons_callable: SamplingCallable | None = None
215 # Optional mission-memory store (duck-typed against
216 # :class:`mcp.mission.memory.MissionMemoryStore`) for the
217 # best-effort memory write on terminal verdicts — see
218 # :meth:`_maybe_write_memory`. ``None`` (the default) disables the
219 # write entirely, which is what every directly-constructed test
220 # engine gets; production wiring injects a real store and relies on
221 # the write being swallowed on any failure, because the
222 # Final_Report — not the memory item — is the durable exit
223 # artifact.
224 memory_store: Any | None = None
225 # Optional sequence of :data:`ObservationAugmenter` callables applied at
226 # the end of every Observe_Phase, in order. ``None`` (the default) is
227 # byte-identical to pre-seam behavior — standalone and child sessions
228 # never carry augmenters. The swarm runner injects one on orchestrator
229 # engines to merge the supervised-children snapshot; each augmenter is
230 # best-effort (a raising augmenter records an Observation error instead
231 # of failing the phase, so criteria read inconclusive rather than the
232 # session dying on a snapshot bug).
233 observation_augmenters: Sequence[ObservationAugmenter] | None = None
235 # ------------------------------------------------------------------ #
236 # Public surface
237 # ------------------------------------------------------------------ #
239 async def run_iteration(
240 self,
241 session_id: str,
242 ctx: Any | None = None,
243 ) -> IterationRecord:
244 """Run one full iteration for ``session_id`` and return its record.
246 Lifecycle:
248 1. Load the session; raise ``session_not_found`` when missing.
249 2. Refuse a session in any terminal state (``failed`` →
250 ``session_failed``; ``completed`` / ``terminated`` →
251 ``session_terminal``) or in ``paused`` (``session_paused``).
252 3. Transition ``pending → running`` on the very first iteration
253 and stamp ``session["started_at"]``.
254 4. Allocate a fresh :class:`IterationRecord` and run the five
255 phases in order. Each phase emits exactly one
256 ``audit.emit_phase_event`` regardless of outcome.
257 5. On any phase exception: append the partial iteration to
258 ``session["iterations"]``, mark the session ``failed``, save,
259 and re-raise. The session JSON stays inspectable.
260 6. On success: stamp the verdict on the iteration, append it,
261 update the no-progress counter, save the session.
262 7. On terminal verdict (``complete`` / ``terminate``): transition
263 the session status, write the Final_Report, save again.
264 8. Emit one ``audit.emit_verdict_event`` regardless of outcome.
265 9. Return the iteration record.
266 """
267 session = self.backend.load_session(session_id)
268 if session is None:
269 raise MissionEngineError("session_not_found")
271 # The terminal-state check distinguishes ``failed`` from the
272 # other terminal states because callers (and the tool-error
273 # table) treat them differently — a failed session needs manual
274 # inspection via ``mission_history``, while a completed /
275 # terminated session is simply done.
276 status = session["status"]
277 if status == "failed":
278 raise MissionEngineError("session_failed")
279 if status in TERMINAL_STATES:
280 raise MissionEngineError("session_terminal")
281 if status == "paused":
282 raise MissionEngineError("session_paused")
284 iteration_start = self.now()
286 # First-iteration transition. ``started_at`` is the wall-clock
287 # anchor for the wall-clock-budget computation in Decide_Phase
288 # so we set it exactly once, on the pending → running edge.
289 if session["status"] == "pending":
290 session["status"] = "running"
291 session["started_at"] = iteration_start.isoformat()
293 iteration_index = len(session["iterations"])
294 record = self._make_iteration_record(iteration_index, iteration_start)
296 try:
297 strategy = await self._propose_phase(session, ctx, record)
298 executed_calls = await self._execute_phase(session, strategy, ctx, record)
299 await self._observe_phase(session, strategy, executed_calls, record)
300 await self._evaluate_phase(session, record)
301 verdict, reason = await self._decide_phase(session, record)
302 except Exception:
303 # Persist a failure record so the session JSON remains a
304 # complete history of everything the loop attempted. The
305 # verdict stays at its placeholder value because no
306 # Decide_Phase actually fired; consumers detect the failure
307 # through ``session["status"] == "failed"`` and the failed
308 # phase entry in ``record["phases"]``.
309 record["ended_at"] = self.now().isoformat()
310 session["iterations"].append(record)
311 session["status"] = "failed"
312 session["ended_at"] = record["ended_at"]
313 with contextlib.suppress(Exception):
314 # A save failure during a failure path must not shadow
315 # the original phase exception — the operator's first
316 # need is to see what actually went wrong, not what
317 # went wrong while reporting what went wrong.
318 self.backend.save_session(session)
319 raise
321 # Stamp the verdict on the iteration record before append; the
322 # decide cascade inspects ``len(session["iterations"])`` (i.e.
323 # iterations *before* the current one) so we deliberately
324 # append after Decide_Phase rather than before.
325 record["verdict"] = verdict
326 record["verdict_reason"] = reason
327 record["ended_at"] = self.now().isoformat()
328 session["iterations"].append(record)
330 self._update_session_post_iteration(session, record)
331 self.backend.save_session(session)
333 if verdict in TERMINAL_VERDICTS:
334 await self._finalise_terminal_session(session, record, verdict, reason)
335 self.backend.save_session(session)
337 # One verdict event per iteration regardless of terminal vs
338 # in-progress, so audit consumers see a uniform stream.
339 audit.emit_verdict_event(
340 session_id=session_id,
341 iteration_index=iteration_index,
342 verdict=verdict,
343 verdict_reason=reason,
344 revision_rationale=record.get("revision_rationale"),
345 )
347 return record
349 # ------------------------------------------------------------------ #
350 # Iteration record bootstrap
351 # ------------------------------------------------------------------ #
353 @staticmethod
354 def _make_iteration_record(iteration_index: int, started_at: datetime) -> IterationRecord:
355 """Build the empty :class:`IterationRecord` for a new iteration.
357 Verdict and reason are placeholder values; the Decide_Phase
358 overwrites them before the record is appended. ``ended_at``
359 is set to the empty string and rewritten just before append
360 so the persisted shape always carries an ISO-8601 timestamp.
361 """
362 record: IterationRecord = {
363 "iteration_index": iteration_index,
364 "started_at": started_at.isoformat(),
365 "ended_at": "",
366 "phases": [],
367 "strategy": cast(Strategy, {}),
368 "observation": cast(Observation, {}),
369 "criteria_evaluation": [],
370 "verdict": "continue",
371 "verdict_reason": "in_progress",
372 "checkpoint_evaluated": False,
373 }
374 return record
376 # ------------------------------------------------------------------ #
377 # Phase wrapper
378 # ------------------------------------------------------------------ #
380 async def _run_phase(
381 self,
382 session: SessionState,
383 record: IterationRecord,
384 phase_name: str,
385 body: Callable[[], Awaitable[Any]],
386 ) -> Any:
387 """Execute ``body`` with phase audit + record bookkeeping.
389 Centralises the try/finally that every phase needs:
391 * stamps ``started_at`` from the engine clock,
392 * runs the body,
393 * stamps ``ended_at`` from the engine clock again,
394 * appends a :class:`PhaseRecord` to ``record["phases"]``,
395 * emits exactly one ``audit.emit_phase_event``.
397 On exception, the finally block records ``status="failed"``
398 with the exception's name + message (truncated to 200 chars to
399 match the audit module's existing convention) and re-raises so
400 ``run_iteration`` can drive the failure path.
401 """
402 started_at = self.now().isoformat()
403 status: str = "succeeded"
404 error_message: str | None = None
405 try:
406 return await body()
407 except Exception as exc:
408 status = "failed"
409 error_message = f"{type(exc).__name__}: {exc}"[:200]
410 raise
411 finally:
412 ended_at = self.now().isoformat()
413 phase_record: PhaseRecord = {
414 "phase": cast(Any, phase_name),
415 "status": cast(Any, status),
416 "started_at": started_at,
417 "ended_at": ended_at,
418 }
419 if error_message:
420 phase_record["error_message"] = error_message
421 record["phases"].append(phase_record)
422 audit.emit_phase_event(
423 session_id=session["session_id"],
424 iteration_index=record["iteration_index"],
425 phase=cast(Any, phase_name),
426 status=cast(Any, status),
427 started_at=started_at,
428 ended_at=ended_at,
429 error_message=error_message,
430 )
432 # ------------------------------------------------------------------ #
433 # Phase 1 — propose
434 # ------------------------------------------------------------------ #
436 async def _propose_phase(
437 self,
438 session: SessionState,
439 ctx: Any | None,
440 record: IterationRecord,
441 ) -> Strategy:
442 """Build the Strategy for this iteration.
444 Two paths:
446 * **Sampling path** — when the prior verdict was ``adjust`` AND
447 the session has ``use_sampling=true`` AND a sampling callable
448 is wired, await the callable and adopt its return value as
449 the Strategy. Any exception from the callable, or any return
450 shape that does not look like a Strategy, falls through to
451 the deterministic path. Slice 6 will replace this with a
452 richer prompt-builder + validator.
453 * **Deterministic path** — re-run the most recent successful
454 tool call (using the same args) when one exists, otherwise
455 invoke the first tool in the session's allowlist with empty
456 args. Pure: no clock, no randomness, no external I/O. The
457 resulting Strategy is always a single ``tool_calls`` entry.
459 The chosen Strategy is stored on ``record["strategy"]`` and
460 returned for the Execute_Phase to consume.
461 """
463 async def body() -> Strategy:
464 strategy = await self._build_strategy(session, ctx, record)
465 record["strategy"] = strategy
466 return strategy
468 return cast(Strategy, await self._run_phase(session, record, _PROPOSE, body))
470 async def _build_strategy(
471 self, session: SessionState, ctx: Any | None, record: IterationRecord
472 ) -> Strategy:
473 """Pick the Propose_Phase Strategy via the sampling-or-fallback rule."""
474 if self._should_attempt_sampling(session):
475 sampled = await self._try_sample_strategy(session, ctx, record)
476 if sampled is not None:
477 return sampled
478 return self._deterministic_strategy(session)
480 def _should_attempt_sampling(self, session: SessionState) -> bool:
481 """True iff the prior verdict was ``adjust`` and sampling is wired."""
482 if self.sampling_callable is None:
483 return False
484 if not session.get("use_sampling"):
485 return False
486 iterations = session.get("iterations") or []
487 if not iterations:
488 return False
489 return iterations[-1].get("verdict") == "adjust"
491 async def _try_sample_strategy(
492 self, session: SessionState, ctx: Any | None, record: IterationRecord
493 ) -> Strategy | None:
494 """Call the sampling callable and adopt its return as a Strategy.
496 Returns ``None`` on any failure (exception, non-dict return, dict
497 missing both ``tool_calls`` and ``script``) so the caller can
498 fall back to the deterministic strategy. Shape validation here
499 is intentionally tight: the engine cannot run a ``script``
500 strategy without a sandbox, and we never want a malformed sampler
501 result to cascade into Execute_Phase as an opaque error.
503 Three return shapes are recognised:
505 * :class:`mcp.mission.sampling.SamplingUsed` — the production
506 orchestration helper's accepted-output type. The Strategy is
507 read from ``parsed["next_strategy"]``; the sampler's
508 ``revision_rationale`` is stamped on the iteration ``record``
509 so :func:`mcp.mission.audit.emit_verdict_event` surfaces it
510 in the next iteration's audit trail.
511 * :class:`mcp.mission.sampling.SamplingFallback` — the
512 orchestration helper's rejection / fallback type. The engine
513 treats this exactly like a missing return: ``None`` so the
514 deterministic-fallback path runs. The fallback's own
515 rationale stays on the audit event the helper already emitted;
516 we deliberately do *not* override the engine's deterministic
517 rationale-template here so the verdict path stays fully
518 deterministic when sampling rejects.
519 * Raw ``dict`` (the legacy / test pattern) — kept verbatim so
520 existing engine tests that pass simple async lambdas returning
521 ``{"tool_calls": [...]}`` continue to work without churn.
522 """
523 assert self.sampling_callable is not None # narrowed by caller
524 try:
525 result = await self.sampling_callable(session=session, ctx=ctx)
526 except BedrockFTUFormNotAcceptedError:
527 # Deliberate exception to the swallow-and-fall-back policy above.
528 # Every other sampler failure is potentially transient, so the
529 # deterministic strategy is the better answer. A missing Anthropic
530 # first-time-use form is permanent and account-scoped: it would fail
531 # identically on every remaining iteration, so degrading the whole
532 # run in silence hides a one-line fix. Fail the run instead.
533 raise
534 except Exception:
535 return None
537 # Phase 6.7 result types — the orchestration helper returns
538 # either ``SamplingUsed`` (accept) or ``SamplingFallback``
539 # (reject). The engine maps them onto its existing
540 # "Strategy or fall back" surface.
541 if isinstance(result, SamplingUsed):
542 next_strategy = result.parsed.get("next_strategy")
543 if not isinstance(next_strategy, dict):
544 return None
545 self._capture_sampled_rationale(record, result.parsed)
546 return self._coerce_strategy_dict(next_strategy)
548 if isinstance(result, SamplingFallback):
549 # The fallback's own deterministic rationale is already on
550 # the emitted audit event. The engine routes through its
551 # own deterministic-fallback path so the verdict path stays
552 # fully deterministic — returning ``None`` is the signal.
553 return None
555 # Legacy raw-dict return — preserved verbatim so older tests
556 # and callers that wire a simple ``async def: return {...}``
557 # stub continue to work unchanged.
558 if not isinstance(result, dict):
559 return None
560 return self._coerce_strategy_dict(result)
562 def _coerce_strategy_dict(self, candidate: dict[str, Any]) -> Strategy | None:
563 """Adopt ``candidate`` as a :class:`Strategy` if it is well-shaped.
565 Centralises the structural check (exactly one of ``tool_calls``
566 or ``script`` populated and well-typed) so both the
567 :class:`SamplingUsed` path and the legacy raw-dict path land
568 through one validator. Returns ``None`` for any malformed
569 shape; callers fall back to the deterministic strategy.
570 """
571 if "tool_calls" in candidate:
572 tool_calls = candidate["tool_calls"]
573 if isinstance(tool_calls, list) and tool_calls:
574 return cast(Strategy, dict(candidate))
575 return None
576 if "script" in candidate:
577 script = candidate["script"]
578 if isinstance(script, str) and script:
579 # The sandbox runner is the only thing that can
580 # safely execute a script. If it isn't wired, the
581 # sampled script is unusable — fall back.
582 if self.sandbox_runner is None:
583 return None
584 return cast(Strategy, dict(candidate))
585 return None
586 return None
588 @staticmethod
589 def _capture_sampled_rationale(record: IterationRecord, parsed_payload: dict[str, Any]) -> None:
590 """Stamp the sampler's ``revision_rationale`` on ``record`` if present.
592 The advisory model's rationale lives at
593 ``parsed_payload["revision_rationale"]`` per the
594 Strategy_Revision schema. Recording it on the iteration record
595 means :func:`mcp.mission.audit.emit_verdict_event` (which the
596 engine calls at the end of ``run_iteration`` with
597 ``record.get("revision_rationale")``) emits the model-derived
598 text instead of the deterministic template that the
599 Decide_Phase synthesises for ``adjust`` verdicts. The engine
600 only ever calls this from the sampling-success path, so a
601 rejection / fallback never overrides the deterministic
602 template the Decide_Phase set.
603 """
604 rationale = parsed_payload.get("revision_rationale")
605 if isinstance(rationale, str) and rationale:
606 record["revision_rationale"] = rationale
608 def _deterministic_strategy(self, session: SessionState) -> Strategy:
609 """Build the fallback Strategy when sampling is off or unusable.
611 Re-runs the most recent successful tool call. When the prior
612 call used empty args and there are unmet criteria, the
613 widening rule injects a ``query`` parameter derived from the
614 unmet criterion IDs — this gives catalog-search tools
615 (``find_examples``, ``find_docs``) a chance to return
616 relevant content without needing the sampler. When no
617 successful call exists yet, the first tool in the allowlist
618 runs with widened args if possible, empty args otherwise.
619 """
620 prior_call = self._find_most_recent_successful_call(session)
621 if prior_call is not None:
622 tool_name, args = prior_call
623 widened = self._widen_args(args, session)
624 rationale_suffix = " with widened args" if widened != args else " with prior args"
625 return cast(
626 Strategy,
627 {
628 "tool_calls": [{"tool_name": tool_name, "args": dict(widened)}],
629 "rationale": f"deterministic fallback: re-run {tool_name}{rationale_suffix}",
630 },
631 )
632 allowlist = session.get("tool_allowlist") or []
633 if not allowlist:
634 raise MissionEngineError("propose_no_tool_available")
635 widened = self._widen_args({}, session)
636 return cast(
637 Strategy,
638 {
639 "tool_calls": [{"tool_name": allowlist[0], "args": widened}],
640 "rationale": (
641 "deterministic fallback: invoking first allowlisted "
642 "tool with widened args from unmet criteria"
643 if widened
644 else "deterministic fallback: invoking first allowlisted "
645 "tool with empty args (no prior successful call)"
646 ),
647 },
648 )
650 @staticmethod
651 def _widen_args(args: dict[str, Any], session: SessionState) -> dict[str, Any]:
652 """Inject a ``query`` parameter from unmet criteria when args are empty.
654 The widening rule is intentionally simple: if the args dict
655 has no ``query`` key (or it's empty), extract keywords from
656 the IDs of unmet criteria (splitting on ``_``) and join them
657 as a space-separated query string. This gives catalog-search
658 tools a chance to return relevant content on the deterministic
659 path without needing the sampler.
661 Returns the original args unchanged when:
662 - ``query`` is already populated (don't override operator intent)
663 - No unmet criteria exist (nothing to widen toward)
664 - The session has no iteration history (first call, no eval yet)
665 """
666 # Don't override an existing query.
667 if args.get("query"):
668 return args
670 # Find unmet criteria from the latest iteration's evaluation.
671 iterations = session.get("iterations") or []
672 if not iterations:
673 return args
674 latest = iterations[-1]
675 criteria_eval = latest.get("criteria_evaluation") or []
676 unmet_ids: list[str] = []
677 for result in criteria_eval:
678 if isinstance(result, dict) and result.get("status") == "unmet":
679 cid = result.get("criterion_id", "")
680 if cid:
681 unmet_ids.append(cid)
682 if not unmet_ids:
683 return args
685 # Build a query from the unmet criterion IDs by splitting on
686 # underscores and deduplicating. Skip generic words.
687 skip_words = {"called", "succeeded", "found", "present", "no", "errors", "occurred"}
688 keywords: list[str] = []
689 seen: set[str] = set()
690 for cid in unmet_ids:
691 for word in cid.split("_"):
692 word_lower = word.lower()
693 if word_lower not in skip_words and word_lower not in seen and len(word) > 2:
694 keywords.append(word_lower)
695 seen.add(word_lower)
696 if not keywords:
697 return args
699 widened = dict(args)
700 widened["query"] = " ".join(keywords[:5]) # Cap at 5 keywords
701 return widened
703 @staticmethod
704 def _find_most_recent_successful_call(
705 session: SessionState,
706 ) -> tuple[str, dict[str, Any]] | None:
707 """Walk the iteration history backwards for a successful tool call.
709 Returns ``(tool_name, args)`` for the most recent call whose
710 :class:`ToolCallRecord` has ``status="ok"``. Looks at both the
711 recorded executed-call list (for tool_calls strategies — kept
712 on the iteration's Strategy under ``tool_calls`` after execute)
713 and the script call log (for scripted strategies). Skips
714 iterations whose strategy was a script with no successful
715 embedded tool call.
717 Returns ``None`` when no prior successful call exists across
718 the entire history.
719 """
720 for iteration in reversed(session.get("iterations") or []):
721 # Scripted strategies record their inner calls on
722 # ``script_call_log``; direct tool_calls strategies don't
723 # have that key.
724 for source_key in ("script_call_log",):
725 log = iteration.get(source_key)
726 if not log:
727 continue
728 for call in reversed(log):
729 if (
730 isinstance(call, dict)
731 and call.get("status") == "ok"
732 and isinstance(call.get("tool_name"), str)
733 and isinstance(call.get("args"), dict)
734 ):
735 return call["tool_name"], dict(call["args"])
736 # Direct tool_calls strategies write the executed records
737 # back onto the Strategy under ``tool_calls`` (each entry
738 # carrying the same status / args fields the script log
739 # would carry). This keeps a single lookup path here.
740 strategy = iteration.get("strategy") or {}
741 tool_calls = strategy.get("tool_calls") or []
742 for tc in reversed(tool_calls):
743 if (
744 isinstance(tc, dict)
745 and tc.get("status") == "ok"
746 and isinstance(tc.get("tool_name"), str)
747 and isinstance(tc.get("args"), dict)
748 ):
749 return tc["tool_name"], dict(tc["args"])
750 return None
752 # ------------------------------------------------------------------ #
753 # Phase 2 — execute
754 # ------------------------------------------------------------------ #
756 async def _execute_phase(
757 self,
758 session: SessionState,
759 strategy: Strategy,
760 ctx: Any | None,
761 record: IterationRecord,
762 ) -> list[ToolCallRecord]:
763 """Run the Strategy and return the list of executed tool calls.
765 Two modes:
767 * **tool_calls** — iterate the strategy's ``tool_calls`` in
768 order. For each: gate by the session's allowlist, dispatch
769 via :attr:`tool_dispatcher`, and record the outcome
770 (``ok`` / ``failed`` / ``skipped_not_allowed``). One failed
771 call does not abort the iteration — the next call still
772 runs, the failure lands as one entry, and Observe_Phase
773 surfaces it under ``errors``.
774 * **script** — hand the script to :attr:`sandbox_runner` along
775 with ``ctx`` and the engine's own ``tool_dispatcher`` so the
776 sandbox can safely invoke allowlisted tools as native
777 callables. The runner returns ``(observation_dict,
778 script_call_log)``. The observation is stashed on the record
779 for Observe_Phase to use directly; the script_call_log is
780 stored on ``record["script_call_log"]``.
782 For both modes, every successful call (or each successful
783 embedded call in script mode) is recorded into the iteration
784 record for audit and replay.
785 """
787 async def body() -> list[ToolCallRecord]:
788 if "script" in strategy:
789 return await self._execute_script(session, strategy, ctx, record)
790 return await self._execute_tool_calls(session, strategy, ctx, record)
792 return cast(
793 list[ToolCallRecord],
794 await self._run_phase(session, record, _EXECUTE, body),
795 )
797 async def _execute_tool_calls(
798 self,
799 session: SessionState,
800 strategy: Strategy,
801 ctx: Any | None,
802 record: IterationRecord,
803 ) -> list[ToolCallRecord]:
804 """Run the Strategy's ``tool_calls`` list with allowlist gating."""
805 allowlist = set(session.get("tool_allowlist") or [])
806 executed: list[ToolCallRecord] = []
807 for entry in strategy.get("tool_calls", []) or []:
808 tool_name = entry.get("tool_name") if isinstance(entry, dict) else None
809 args = entry.get("args") if isinstance(entry, dict) else {}
810 if not isinstance(tool_name, str) or not tool_name:
811 # A malformed tool_calls entry is the operator's bug,
812 # but failing the entire iteration over one bad entry
813 # is harsher than the loop semantics demand. Record
814 # it as a failed call and move on.
815 executed.append(
816 {
817 "tool_name": str(tool_name) if tool_name else "<unknown>",
818 "args": args if isinstance(args, dict) else {},
819 "status": "failed",
820 "result_summary": None,
821 "duration_ms": 0,
822 "error_message": "tool_name_missing_or_invalid",
823 }
824 )
825 continue
826 if not isinstance(args, dict):
827 args = {}
828 if tool_name not in allowlist:
829 executed.append(
830 {
831 "tool_name": tool_name,
832 "args": args,
833 "status": "skipped_not_allowed",
834 "result_summary": None,
835 "duration_ms": 0,
836 }
837 )
838 continue
839 executed.append(await self._dispatch_one_call(session, tool_name, args, ctx))
841 # Persist the executed records back onto the Strategy so the
842 # propose-fallback's "most recent successful call" lookup has
843 # a single source of truth on every persisted iteration.
844 record["strategy"]["tool_calls"] = [dict(call) for call in executed]
845 return executed
847 async def _dispatch_one_call(
848 self,
849 session: SessionState,
850 tool_name: str,
851 args: dict[str, Any],
852 ctx: Any | None,
853 ) -> ToolCallRecord:
854 """Invoke one allowlisted tool through the dispatcher."""
855 del session # accepted for symmetry with the execute path; unused
856 started = self.now()
857 try:
858 result = await self.tool_dispatcher(tool_name, args, ctx)
859 except Exception as exc:
860 duration_ms = self._elapsed_ms(started)
861 return {
862 "tool_name": tool_name,
863 "args": args,
864 "status": "failed",
865 "result_summary": None,
866 "duration_ms": duration_ms,
867 "error_message": f"{type(exc).__name__}: {exc}"[:200],
868 }
869 duration_ms = self._elapsed_ms(started)
870 record: ToolCallRecord = {
871 "tool_name": tool_name,
872 "args": args,
873 "status": "ok",
874 "result_summary": result,
875 "duration_ms": duration_ms,
876 }
877 return record
879 async def _execute_script(
880 self,
881 session: SessionState,
882 strategy: Strategy,
883 ctx: Any | None,
884 record: IterationRecord,
885 ) -> list[ToolCallRecord]:
886 """Run a scripted strategy through the wired sandbox runner.
888 Three failure modes get translated into stable engine error
889 codes so the MCP tool wrappers and the CLI render them as
890 structured rejections rather than opaque tracebacks:
892 * ``sandbox_runner is None`` — the engine was constructed
893 without a sandbox. The session-start validator should have
894 rejected any script-bearing Strategy already (scripts go
895 through ``validate_script_ast`` before they reach here),
896 but a sampled-then-injected script could still arrive at
897 this method. Treat it as a validation failure with the
898 equivalent of ``script_rejected``.
899 * :class:`ScriptRejected` from inside the runner — the runner
900 re-validated the script just before execution and the AST
901 gate fired. Re-raise as ``script_rejected``.
902 * :class:`SandboxTerminated` from inside the runner — Monty
903 killed the script for exceeding a duration / memory cap.
904 The cap is a true budget cap, not a code-quality failure,
905 so the engine *swallows* the exception, builds a partial
906 Observation from whatever the script collected before being
907 killed, stashes the partial ``script_call_log`` and a
908 ``sandbox_terminated_reason`` sentinel on the iteration
909 record, and returns a list of partial calls. The Decide_Phase
910 reads the sentinel and emits ``("terminate",
911 "max_wall_clock")`` so the verdict surfaces on the
912 budget-cap path rather than via a phase failure.
914 The sandbox module is imported lazily inside this method so
915 the engine module stays importable on hosts where the
916 underlying ``pydantic_monty`` dependency is absent (CLI-only
917 environments, dry-run validators, etc.). When the lazy
918 import fails, the structured-exception translation is
919 skipped and the original exception bubbles up to the
920 ``run_iteration`` failure path — the engine still records a
921 failed Execute_Phase, which is the right behaviour even
922 without per-class translation.
923 """
924 del session # accepted for symmetry with the execute path; unused
925 if self.sandbox_runner is None:
926 raise MissionEngineError("script_rejected")
927 script = strategy["script"]
928 try:
929 observation_dict, script_call_log = await self.sandbox_runner(
930 script, ctx, self.tool_dispatcher
931 )
932 except Exception as exc:
933 # Late-resolved class lookup: importing the sandbox
934 # module at top of file would pull in
935 # ``pydantic_monty`` on import, which the engine
936 # explicitly does not require (an operator can run
937 # ``mission_validate`` against a stored session JSON
938 # without a working sandbox). Importing here means the
939 # translation is best-effort but the engine module
940 # itself stays loadable everywhere.
941 try:
942 from .sandbox import (
943 SandboxTerminated,
944 ScriptRejected,
945 )
946 except Exception:
947 raise exc from None
948 if isinstance(exc, ScriptRejected):
949 raise MissionEngineError("script_rejected") from exc
950 if isinstance(exc, SandboxTerminated):
951 # The sandbox cap is a budget cap, not a phase
952 # failure: the script ran out of wall clock (or
953 # memory, or hit a runtime / typing / syntax error
954 # mid-run) under operator-supplied limits. Capture
955 # whatever it collected before being killed and
956 # route the verdict through the budget-cap path.
957 #
958 # The sentinel on the iteration record is what the
959 # cascade in ``decide_verdict`` reads to short-
960 # circuit to ``("terminate", "max_wall_clock")``
961 # before any other branch is consulted; without it
962 # the cascade would fall through to the default
963 # ``("continue", "in_progress")`` because no other
964 # cap was breached.
965 record["script_call_log"] = cast(
966 "list[ToolCallRecord]", list(exc.partial_script_call_log)
967 )
968 # Build a minimal Observation from the partial
969 # logs so Evaluate_Phase has the same shape it
970 # would have on a successful sandbox run. Missing
971 # keys (``metrics``, ``events``, etc.) get default
972 # empties; Observe_Phase fills in any timestamps
973 # the partial doesn't carry.
974 partial_observation: dict[str, Any] = {
975 "tool_results": [
976 self._annotate_tool_result(call) for call in exc.partial_script_call_log
977 ],
978 "metrics": {},
979 "events": list(exc.partial_events),
980 }
981 if exc.partial_observations:
982 partial_observation["metrics"]["observations"] = {
983 entry["key"]: entry["value"]
984 for entry in exc.partial_observations
985 if isinstance(entry, dict) and "key" in entry
986 }
987 record["observation"] = cast(Observation, partial_observation)
988 record["sandbox_terminated_reason"] = "max_wall_clock"
989 return cast("list[ToolCallRecord]", list(exc.partial_script_call_log))
990 raise
991 # The sandbox already produced a normalized Observation dict,
992 # so we cache it on the record for Observe_Phase to pick up
993 # directly. This is the only path where Observe_Phase sees a
994 # pre-built Observation.
995 record["script_call_log"] = cast("list[ToolCallRecord]", list(script_call_log))
996 record["observation"] = cast(Observation, dict(observation_dict))
997 return list(script_call_log)
999 def _elapsed_ms(self, started: datetime) -> int:
1000 """Return integer milliseconds elapsed since ``started``."""
1001 delta = self.now() - started
1002 return max(int(delta.total_seconds() * 1000), 0)
1004 # ------------------------------------------------------------------ #
1005 # Phase 3 — observe
1006 # ------------------------------------------------------------------ #
1008 async def _observe_phase(
1009 self,
1010 session: SessionState,
1011 strategy: Strategy,
1012 executed_calls: list[ToolCallRecord],
1013 record: IterationRecord,
1014 ) -> None:
1015 """Normalise tool-call outputs into an :class:`Observation`.
1017 Two paths:
1019 * **Script strategy** — Execute_Phase already stashed the
1020 sandbox's Observation on ``record["observation"]``. Observe
1021 fills in any missing required keys (``tool_results``,
1022 ``metrics``, ``events``, ``phase_started_at`` /
1023 ``phase_ended_at``) so downstream Evaluate_Phase consumers
1024 can rely on the shape.
1025 * **Tool-calls strategy** — build the Observation from the
1026 executed-call records: ``tool_results`` is the list of
1027 ``result_summary`` values (one per call, including failed
1028 ones for stable indexing); ``metrics`` and ``events`` are
1029 merged from any call result that carries those keys at the
1030 top level; ``errors`` is appended for failed or skipped
1031 calls. This is intentionally permissive — a Strategy that
1032 doesn't produce metrics or events leaves those slots empty
1033 rather than raising.
1034 """
1036 async def body() -> None:
1037 phase_started = self.now()
1038 if "script" in strategy:
1039 # The sandbox already produced the Observation. Fill
1040 # in any timestamp slots it didn't populate so the
1041 # shape is uniform for evaluators.
1042 obs = cast(dict[str, Any], record.get("observation") or {})
1043 obs.setdefault("tool_results", [])
1044 obs.setdefault("metrics", {})
1045 obs.setdefault("events", [])
1046 obs.setdefault("phase_started_at", phase_started.isoformat())
1047 obs.setdefault("phase_ended_at", self.now().isoformat())
1048 record["observation"] = cast(Observation, obs)
1049 self._apply_observation_augmenters(session, record)
1050 return
1051 record["observation"] = self._build_observation(executed_calls, phase_started)
1052 self._apply_observation_augmenters(session, record)
1054 await self._run_phase(session, record, _OBSERVE, body)
1056 def _apply_observation_augmenters(self, session: SessionState, record: IterationRecord) -> None:
1057 """Merge each :data:`ObservationAugmenter` contribution into the Observation.
1059 Applied at the end of the Observe_Phase for both strategy shapes.
1060 A contribution's ``children`` list lands on the Observation
1061 verbatim (later augmenters win, matching the ``metrics.update``
1062 last-writer-wins semantics); its ``metrics`` dict merges into
1063 ``observation["metrics"]``. Non-dict contributions are ignored.
1065 Best-effort by design: an augmenter that raises records a
1066 structured entry under ``observation["errors"]`` instead of
1067 failing the phase — downstream criteria read ``unmet`` or
1068 ``inconclusive`` and the deterministic cascade (stagnation,
1069 budget caps) still terminates the session, which is strictly
1070 better than dying inside Observe on a snapshot bug.
1071 """
1072 if not self.observation_augmenters:
1073 return
1074 obs = cast(dict[str, Any], record.get("observation") or {})
1075 for augmenter in self.observation_augmenters:
1076 try:
1077 contribution = augmenter(session)
1078 except Exception as exc: # degrade, never fail the phase
1079 obs.setdefault("errors", []).append(
1080 {
1081 "tool_name": "_observation_augmenter",
1082 "status": "failed",
1083 "error_message": str(exc),
1084 }
1085 )
1086 continue
1087 if not isinstance(contribution, dict):
1088 continue
1089 children = contribution.get("children")
1090 if isinstance(children, list):
1091 obs["children"] = children
1092 extra_metrics = contribution.get("metrics")
1093 if isinstance(extra_metrics, dict):
1094 metrics = obs.setdefault("metrics", {})
1095 if isinstance(metrics, dict):
1096 metrics.update(extra_metrics)
1097 record["observation"] = cast(Observation, obs)
1099 @staticmethod
1100 def _annotate_tool_result(call: ToolCallRecord | dict[str, Any]) -> Any:
1101 """Wrap a call's ``result_summary`` with the per-call call markers.
1103 The Observation's ``tool_results`` list is the canonical input
1104 to predicate criteria and to the dedicated ``tool_call_succeeded``
1105 evaluator. Both consult ``r.get("_status")`` and
1106 ``r.get("tool_name")`` to know which tool produced the entry
1107 and whether the call succeeded — markers the engine adds here
1108 rather than relying on individual tools to inject. The stub
1109 dispatcher used to synthesise these markers in its return,
1110 but the live FastMCP dispatcher returns whatever shape the
1111 underlying tool produces (often a structured ``{"result": [
1112 ... ]}`` dict that doesn't carry call-level metadata).
1114 Strategy:
1116 * **Dict result_summary** — augment in place with ``_status``
1117 and ``tool_name`` only when those keys are absent. This
1118 keeps any caller-supplied marker visible (some tools do
1119 synthesise them) while ensuring evaluators always find
1120 them.
1121 * **Non-dict result_summary** (None, list, primitive) — wrap
1122 in a fresh dict carrying the call's ``_status`` /
1123 ``tool_name`` plus a ``result`` field that holds the
1124 original payload so predicates can still walk into it.
1125 """
1126 result = call.get("result_summary")
1127 status = call.get("status") or "unknown"
1128 tool_name = call.get("tool_name")
1129 if isinstance(result, dict):
1130 annotated = dict(result)
1131 annotated.setdefault("_status", status)
1132 annotated.setdefault("tool_name", tool_name)
1133 return annotated
1134 return {
1135 "_status": status,
1136 "tool_name": tool_name,
1137 "result": result,
1138 }
1140 def _build_observation(
1141 self,
1142 executed_calls: list[ToolCallRecord],
1143 phase_started: datetime,
1144 ) -> Observation:
1145 """Merge a list of :class:`ToolCallRecord` into an :class:`Observation`.
1147 Each ``executed_calls`` entry contributes one annotated dict
1148 to ``observation["tool_results"]`` via
1149 :meth:`_annotate_tool_result` so the entry always carries
1150 the ``_status`` and ``tool_name`` markers the predicate
1151 evaluator and the ``tool_call_succeeded`` evaluator both rely
1152 on, regardless of what shape the underlying tool returned.
1153 """
1154 tool_results: list[Any] = []
1155 metrics: dict[str, Any] = {}
1156 events: list[dict[str, Any]] = []
1157 errors: list[dict[str, Any]] = []
1159 for call in executed_calls:
1160 tool_results.append(self._annotate_tool_result(call))
1161 if call.get("status") == "ok":
1162 result = call.get("result_summary")
1163 # Permissive merge: when a tool's result happens to
1164 # include a top-level ``metrics`` dict or ``events``
1165 # list, lift them into the Observation. Anything else
1166 # stays only in ``tool_results``.
1167 if isinstance(result, dict):
1168 result_metrics = result.get("metrics")
1169 if isinstance(result_metrics, dict):
1170 metrics.update(result_metrics)
1171 result_events = result.get("events")
1172 if isinstance(result_events, list):
1173 for event in result_events:
1174 if isinstance(event, dict):
1175 events.append(event)
1176 else:
1177 # ``failed`` and ``skipped_not_allowed`` both surface as
1178 # errors; the heuristic in decide.py uses "errors that
1179 # didn't appear in the prior Observation" to drive the
1180 # adjust verdict, so a stable shape per error matters.
1181 errors.append(
1182 {
1183 "tool_name": call.get("tool_name"),
1184 "status": call.get("status"),
1185 "error_message": call.get("error_message"),
1186 }
1187 )
1189 observation: Observation = {
1190 "tool_results": tool_results,
1191 "metrics": metrics,
1192 "events": events,
1193 "phase_started_at": phase_started.isoformat(),
1194 "phase_ended_at": self.now().isoformat(),
1195 }
1196 if errors:
1197 observation["errors"] = errors
1198 return observation
1200 # ------------------------------------------------------------------ #
1201 # Phase 4 — evaluate
1202 # ------------------------------------------------------------------ #
1204 async def _evaluate_phase(self, session: SessionState, record: IterationRecord) -> None:
1205 """Walk the session's Criteria and produce :class:`CriterionResult` rows.
1207 The kinds dispatch to per-kind helpers:
1209 * ``metric_threshold`` — dot-path lookup on the Observation,
1210 numeric comparison via the declared operator.
1211 * ``event`` — scan the Observation's ``events`` list for an
1212 entry whose ``event_name`` matches the criterion's target.
1213 * ``predicate`` — evaluate the cached parsed AST against the
1214 Observation. A raised exception lands as ``inconclusive`` so
1215 a malformed predicate cannot crash the loop.
1217 Order in the output list matches the declared order of
1218 ``session["criteria"]`` so iteration audit consumers can pair
1219 results with criteria positionally.
1220 """
1222 async def body() -> None:
1223 observation = cast(dict[str, Any], record.get("observation") or {})
1224 # Build a cumulative observation for predicates: merge all
1225 # prior iterations' tool_results into the current one so
1226 # predicates like ``any('gpu' in str(r) for r in
1227 # obs['tool_results'])`` can see results from prior
1228 # iterations. Metrics and events stay per-iteration (they
1229 # represent the current state, not history).
1230 cumulative_obs = self._build_cumulative_observation(observation, session)
1231 results: list[CriterionResult] = []
1232 for criterion in session.get("criteria") or []:
1233 results.append(
1234 self._evaluate_one_criterion(criterion, observation, cumulative_obs, session)
1235 )
1236 record["criteria_evaluation"] = results
1238 await self._run_phase(session, record, _EVALUATE, body)
1240 @staticmethod
1241 def _build_cumulative_observation(
1242 current_obs: dict[str, Any], session: SessionState
1243 ) -> dict[str, Any]:
1244 """Merge prior iterations' tool_results and metric history into a view.
1246 Two things are cumulative on the returned view:
1248 * ``tool_results`` — every prior iteration's results concatenated with
1249 the current iteration's, so predicate criteria can see results from
1250 all iterations. This enables multi-tool goals where each tool runs in
1251 a different iteration to converge.
1252 * ``metric_history`` — a history-aware map from metric name to the
1253 ordered list of its numeric values across the session
1254 (oldest→newest, current iteration last). This is what lets the
1255 ``metric_trend`` criterion ask "is loss falling across iterations?"
1256 even though the engine keeps the per-iteration ``metrics`` dict
1257 strictly point-in-time. Non-numeric and boolean metric values are
1258 skipped so a stray string reading cannot poison a trend.
1260 ``metrics``, ``events``, and ``errors`` stay per-iteration on the view
1261 because they represent current state: ``metrics`` is the latest
1262 point-in-time reading (a ``metric_threshold`` criterion still compares
1263 the single current value), events are per-iteration signals, and errors
1264 are per-call. The history lives *alongside* them under
1265 ``metric_history`` rather than replacing them, so existing criteria are
1266 unaffected.
1267 """
1268 all_tool_results: list[Any] = []
1269 metric_history: dict[str, list[float]] = {}
1271 def _accumulate_metrics(obs: Mapping[str, Any]) -> None:
1272 metrics = obs.get("metrics")
1273 if not isinstance(metrics, dict):
1274 return
1275 for key, value in metrics.items():
1276 # Mirror the Numeric_Value guard the readers use: int or float,
1277 # never bool. A non-numeric reading contributes no history
1278 # point rather than breaking the series.
1279 if isinstance(value, bool) or not isinstance(value, (int, float)):
1280 continue
1281 metric_history.setdefault(key, []).append(float(value))
1283 for prior in session.get("iterations") or []:
1284 prior_obs = prior.get("observation") or {}
1285 prior_results = prior_obs.get("tool_results")
1286 if isinstance(prior_results, list):
1287 all_tool_results.extend(prior_results)
1288 _accumulate_metrics(prior_obs)
1289 # Append current iteration's results and metrics last so the history is
1290 # ordered oldest→newest with the current reading at the end.
1291 current_results = current_obs.get("tool_results")
1292 if isinstance(current_results, list):
1293 all_tool_results.extend(current_results)
1294 _accumulate_metrics(current_obs)
1295 # Build the cumulative view: tool_results + metric_history are
1296 # cumulative, everything else comes from the current observation.
1297 cumulative: dict[str, Any] = dict(current_obs)
1298 cumulative["tool_results"] = all_tool_results
1299 cumulative["metric_history"] = metric_history
1300 return cumulative
1302 def _evaluate_one_criterion(
1303 self,
1304 criterion: Criterion,
1305 observation: dict[str, Any],
1306 cumulative_obs: dict[str, Any],
1307 session: SessionState,
1308 ) -> CriterionResult:
1309 """Dispatch to the right evaluator and produce a result row."""
1310 criterion_id = criterion["criterion_id"]
1311 # Read ``kind`` as a plain ``str`` rather than the ``CriterionKind``
1312 # literal: the criteria come back from the persisted session, and a
1313 # malformed file can carry a kind the validator never saw. Keeping
1314 # the wider type is what makes the fallback branch below reachable.
1315 kind: str = criterion["kind"]
1316 evaluated_at = self.now().isoformat()
1318 if kind == "metric_threshold":
1319 status, evidence = self._evaluate_metric_threshold(criterion, observation)
1320 elif kind == "metric_trend":
1321 # Trend evaluates against the cumulative observation, where the
1322 # engine accumulates ``metric_history`` across iterations.
1323 status, evidence = self._evaluate_metric_trend(criterion, cumulative_obs)
1324 elif kind == "event":
1325 status, evidence = self._evaluate_event(criterion, observation)
1326 elif kind == "predicate":
1327 # Predicates evaluate against the cumulative observation so
1328 # they can see tool_results from all prior iterations.
1329 status, evidence = self._evaluate_predicate(criterion, cumulative_obs)
1330 elif kind == "tool_call_succeeded":
1331 status, evidence = self._evaluate_tool_call_succeeded(criterion, observation, session)
1332 else:
1333 # Never taken once the validator has run — but if a malformed
1334 # session somehow lands here, surface the bad kind as
1335 # inconclusive rather than raising and tearing down the
1336 # entire iteration.
1337 status = "inconclusive"
1338 evidence = f"unknown_criterion_kind:{kind!r}"
1340 return {
1341 "criterion_id": criterion_id,
1342 "status": cast(Any, status),
1343 "evidence": evidence,
1344 "evaluated_at": evaluated_at,
1345 }
1347 @staticmethod
1348 def _evaluate_metric_threshold(
1349 criterion: Criterion, observation: dict[str, Any]
1350 ) -> tuple[str, Any]:
1351 """Look up the metric by dot-path and compare to ``target``."""
1352 path = criterion.get("metric") or ""
1353 op = criterion.get("op")
1354 target = criterion.get("target")
1355 value: Any = observation
1356 for segment in path.split("."):
1357 if isinstance(value, dict) and segment in value:
1358 value = value[segment]
1359 else:
1360 return "inconclusive", f"metric_path_missing:{path!r}"
1361 if isinstance(value, bool) or not isinstance(value, (int, float)):
1362 return "inconclusive", value
1363 try:
1364 met = _compare_numbers(value, cast(str, op), cast(float, target))
1365 except ValueError:
1366 return "inconclusive", value
1367 return ("met" if met else "unmet"), value
1369 @staticmethod
1370 def _evaluate_metric_trend(
1371 criterion: Criterion, cumulative_obs: dict[str, Any]
1372 ) -> tuple[str, Any]:
1373 """Evaluate a metric's direction across the accumulated history.
1375 Reads the metric's value series from
1376 ``cumulative_obs["metric_history"]`` — the oldest→newest list of
1377 numeric readings the engine accumulates in
1378 :meth:`_build_cumulative_observation`. The ``metric`` dot-path is
1379 resolved against that map: a leading ``metrics.`` segment is stripped
1380 so a criterion can reuse the same ``"metrics.loss"`` path it would use
1381 for ``metric_threshold`` and still address the ``loss`` history series.
1383 The series is trimmed to the most-recent ``window`` points (default:
1384 all available). With fewer than ``min_points`` numeric points (default
1385 2) the criterion is ``inconclusive`` — a trend is undefined on a single
1386 reading, and the loop must never be failed for lack of history. The
1387 verdict compares the last point to the first point of the windowed
1388 series per ``direction``:
1390 * ``decreasing`` → last < first
1391 * ``increasing`` → last > first
1392 * ``non_increasing`` → last <= first
1393 * ``non_decreasing`` → last >= first
1395 Evidence is a structured dict (direction, the windowed points, first /
1396 last, and net delta) so the audit log shows exactly what the verdict
1397 was computed from.
1398 """
1399 path = criterion.get("metric") or ""
1400 direction = criterion.get("direction")
1401 # Resolve the metric key against the history map. Accept both the bare
1402 # key (``"loss"``) and the dot-path form (``"metrics.loss"``) so a
1403 # trend criterion lines up with the metric_threshold convention.
1404 history = cumulative_obs.get("metric_history")
1405 if not isinstance(history, dict):
1406 return "inconclusive", "metric_history_missing"
1407 key = path.split(".", 1)[1] if path.startswith("metrics.") else path
1408 series = history.get(key)
1409 if not isinstance(series, list) or not series:
1410 return "inconclusive", f"metric_history_empty:{key!r}"
1412 # Keep only the numeric points (the accumulator already filters, but be
1413 # defensive against a hand-built cumulative_obs in tests).
1414 points: list[float] = [
1415 float(v) for v in series if not isinstance(v, bool) and isinstance(v, (int, float))
1416 ]
1418 window = criterion.get("window")
1419 if isinstance(window, int) and not isinstance(window, bool) and window > 0:
1420 points = points[-window:]
1422 min_points = criterion.get("min_points")
1423 required_points = (
1424 min_points if isinstance(min_points, int) and not isinstance(min_points, bool) else 2
1425 )
1426 required_points = max(2, required_points)
1427 if len(points) < required_points:
1428 return "inconclusive", {
1429 "reason": "insufficient_history",
1430 "points": points,
1431 "required_points": required_points,
1432 }
1434 first = points[0]
1435 last = points[-1]
1436 delta = last - first
1437 if direction == "decreasing":
1438 met = last < first
1439 elif direction == "increasing":
1440 met = last > first
1441 elif direction == "non_increasing":
1442 met = last <= first
1443 elif direction == "non_decreasing":
1444 met = last >= first
1445 else:
1446 # Unreachable when the validator has run; surface defensively.
1447 return "inconclusive", f"unknown_direction:{direction!r}"
1449 evidence = {
1450 "direction": direction,
1451 "points": points,
1452 "first": first,
1453 "last": last,
1454 "delta": delta,
1455 }
1456 return ("met" if met else "unmet"), evidence
1458 @staticmethod
1459 def _evaluate_event(criterion: Criterion, observation: dict[str, Any]) -> tuple[str, Any]:
1460 """Scan ``observation['events']`` for the named event."""
1461 if "events" not in observation:
1462 return "inconclusive", "events_field_missing"
1463 events = observation.get("events")
1464 if not isinstance(events, list):
1465 return "inconclusive", "events_field_not_a_list"
1466 target = criterion.get("event_name")
1467 for event in events:
1468 if isinstance(event, dict) and event.get("event_name") == target:
1469 return "met", event
1470 return "unmet", None
1472 @staticmethod
1473 def _evaluate_predicate(criterion: Criterion, observation: dict[str, Any]) -> tuple[str, Any]:
1474 """Run the cached parsed AST against the Observation.
1476 The validator caches an :class:`ast.Expression` under
1477 ``_parsed_ast`` when ``validate_criteria`` runs in-process.
1478 Persistence layers strip that key before serialisation (the
1479 AST node is not JSON-safe), so a session reloaded from disk
1480 carries criteria *without* ``_parsed_ast``. We detect the
1481 missing cache and re-parse on demand from ``expression``;
1482 the parser was already accepted at validation time so a
1483 re-parse is a pure no-op short of re-reading the source. A
1484 post-load tampering with ``expression`` would cause
1485 :class:`PredicateRejected`, which we surface as a structured
1486 ``inconclusive`` evidence string rather than letting it
1487 propagate.
1488 """
1489 parsed = criterion.get("_parsed_ast")
1490 if parsed is None:
1491 expression = criterion.get("expression")
1492 if not isinstance(expression, str) or not expression:
1493 return "inconclusive", "predicate_ast_not_cached"
1494 try:
1495 parsed = parse_predicate(expression)
1496 except PredicateRejected as exc:
1497 return "inconclusive", f"predicate_rejected_post_load: {exc.reason}"
1498 try:
1499 value = evaluate_predicate(parsed, observation)
1500 except Exception as exc:
1501 return "inconclusive", f"{type(exc).__name__}: {exc}"
1502 return ("met" if value else "unmet"), value
1504 @staticmethod
1505 def _evaluate_tool_call_succeeded(
1506 criterion: Criterion,
1507 observation: dict[str, Any],
1508 session: SessionState | dict[str, Any] | None = None,
1509 ) -> tuple[str, Any]:
1510 """Count successful tool_results matching the named tool across all iterations.
1512 The criterion is met when at least ``min_count`` (default 1)
1513 entries across the **entire session history** (all prior
1514 iterations' observations plus the current one) have
1515 ``tool_name`` equal to the criterion's ``tool_name`` and
1516 ``_status`` equal to ``"ok"``. This cumulative evaluation
1517 means a multi-tool goal where each tool runs in a different
1518 iteration can still converge — the criterion remembers that
1519 the tool succeeded in a prior iteration even if the current
1520 iteration called a different tool.
1522 Returns a ``(status, evidence)`` tuple where ``evidence`` is
1523 a structured dict so the audit log shows the match shape.
1524 """
1525 target_tool = criterion.get("tool_name")
1526 min_count = criterion.get("min_count", 1)
1528 # Collect tool_results from all prior iterations + the current observation.
1529 all_results: list[dict[str, Any]] = []
1531 # Prior iterations' observations (when session is provided).
1532 if session is not None:
1533 for prior_iteration in session.get("iterations") or []:
1534 prior_obs = prior_iteration.get("observation") or {}
1535 prior_results = prior_obs.get("tool_results")
1536 if isinstance(prior_results, list):
1537 for r in prior_results:
1538 if isinstance(r, dict):
1539 all_results.append(r)
1541 # Current iteration's observation.
1542 current_results = observation.get("tool_results")
1543 if isinstance(current_results, list):
1544 for r in current_results:
1545 if isinstance(r, dict):
1546 all_results.append(r)
1548 if not all_results:
1549 return "inconclusive", "tool_results_field_missing"
1551 successful = [
1552 r for r in all_results if r.get("tool_name") == target_tool and r.get("_status") == "ok"
1553 ]
1554 evidence = {
1555 "tool_name": target_tool,
1556 "min_count": min_count,
1557 "successful_call_count": len(successful),
1558 }
1559 if len(successful) >= min_count:
1560 return "met", evidence
1561 return "unmet", evidence
1563 # ------------------------------------------------------------------ #
1564 # Phase 5 — decide
1565 # ------------------------------------------------------------------ #
1567 async def _decide_phase(
1568 self, session: SessionState, record: IterationRecord
1569 ) -> tuple[VerdictLabel, VerdictReason]:
1570 """Run the deterministic verdict cascade and stamp the record."""
1572 async def body() -> tuple[VerdictLabel, VerdictReason]:
1573 now_value = self.now()
1574 verdict, reason = decide.decide_verdict(session, record, now_value)
1575 checkpoint_evaluated = reason != "cadence_skip"
1576 record["checkpoint_evaluated"] = checkpoint_evaluated
1577 if verdict == "adjust":
1578 record["revision_rationale"] = decide.build_revision_rationale_template(
1579 session, record
1580 )
1581 if checkpoint_evaluated:
1582 # ``last_checkpoint_at`` anchors the every_t_seconds
1583 # cadence; only real (non-skip) verdicts advance it.
1584 mark_checkpoint(session, now_value)
1585 return verdict, reason
1587 return cast(
1588 tuple[VerdictLabel, VerdictReason],
1589 await self._run_phase(session, record, _DECIDE, body),
1590 )
1592 # ------------------------------------------------------------------ #
1593 # Post-iteration housekeeping
1594 # ------------------------------------------------------------------ #
1596 def _update_session_post_iteration(
1597 self, session: SessionState, record: IterationRecord
1598 ) -> None:
1599 """Advance or reset the no-progress counter.
1601 Counter semantics (matching the Decide_Phase's stagnation cap):
1603 * Synthetic ``cadence_skip`` iterations leave the counter
1604 alone — a session whose cadence is ``every_n_iterations``
1605 must not be able to reach ``stagnation_threshold`` purely
1606 because most iterations skip the criteria check.
1607 * On evaluated iterations, compute the per-criterion
1608 improvement against the immediately prior evaluated
1609 iteration. A criterion improved iff its prior status was
1610 ``unmet`` or ``inconclusive`` AND its current status is
1611 ``met``. Any improvement resets the counter to 0; otherwise
1612 the counter increments by 1.
1614 ``record`` has already been appended to
1615 ``session["iterations"]`` by the caller, so the prior
1616 iteration is at index ``-2``.
1617 """
1618 if not record.get("checkpoint_evaluated"):
1619 return
1621 prior_eval = self._previous_evaluated_iteration(session)
1622 if prior_eval is None:
1623 # No prior evaluated iteration: the loop has nothing to
1624 # measure improvement against. Treat as no-improvement
1625 # rather than a forced reset, so the stagnation counter
1626 # tracks "how long since we made measurable progress"
1627 # uniformly across the run.
1628 session["no_progress_counter"] = (session.get("no_progress_counter", 0) or 0) + 1
1629 return
1631 if self._criteria_improved(prior_eval, record["criteria_evaluation"]):
1632 session["no_progress_counter"] = 0
1633 else:
1634 session["no_progress_counter"] = (session.get("no_progress_counter", 0) or 0) + 1
1636 @staticmethod
1637 def _previous_evaluated_iteration(
1638 session: SessionState,
1639 ) -> list[CriterionResult] | None:
1640 """Return the criteria evaluation of the most recent evaluated iteration.
1642 "Evaluated" here means ``checkpoint_evaluated=True``. Skipping
1643 cadence-skip iterations is what makes the no-progress counter
1644 immune to the cadence configuration: it only ever measures
1645 movement between *real* checkpoints. The current iteration is
1646 already appended to ``session["iterations"]`` by the caller,
1647 so we look strictly before it.
1648 """
1649 iterations = session.get("iterations") or []
1650 # Skip the just-appended current iteration (last index).
1651 for prior in reversed(iterations[:-1]):
1652 if prior.get("checkpoint_evaluated"):
1653 return list(prior.get("criteria_evaluation") or [])
1654 return None
1656 @staticmethod
1657 def _criteria_improved(prior: Iterable[object], current: Iterable[object]) -> bool:
1658 """Return True iff any criterion went from not-met to met.
1660 Both lists are ``criteria_evaluation`` rows read back from the
1661 persisted session, so they are typed as ``object`` entries and
1662 every row is shape-checked: a non-dict entry (a corrupt file) is
1663 skipped rather than crashing the stagnation cascade.
1664 """
1665 prior_status = {
1666 result["criterion_id"]: result["status"]
1667 for result in prior
1668 if isinstance(result, dict) and "criterion_id" in result
1669 }
1670 for result in current:
1671 if not isinstance(result, dict):
1672 continue
1673 current_status = result.get("status")
1674 if current_status != "met":
1675 continue
1676 prior_value = prior_status.get(result.get("criterion_id"))
1677 if prior_value in ("unmet", "inconclusive", None):
1678 # ``None`` covers a criterion that did not appear in
1679 # the prior evaluation — treating it as "not met
1680 # before" is consistent with first-time-met being an
1681 # improvement.
1682 return True
1683 return False
1685 # ------------------------------------------------------------------ #
1686 # Terminal-verdict finalisation
1687 # ------------------------------------------------------------------ #
1689 async def _finalise_terminal_session(
1690 self,
1691 session: SessionState,
1692 record: IterationRecord,
1693 verdict: VerdictLabel,
1694 reason: VerdictReason,
1695 ) -> None:
1696 """Transition the session to its terminal status and write the report.
1698 Called by ``run_iteration`` only when the verdict is in
1699 :data:`TERMINAL_VERDICTS`. The status mapping is fixed:
1700 ``complete`` → ``completed``, ``terminate`` → ``terminated``.
1701 ``ended_at`` is anchored on the iteration's own ``ended_at``
1702 so the session's lifecycle window matches the last persisted
1703 iteration's window without an extra clock read.
1705 When :attr:`final_lessons_callable` is wired, the engine awaits
1706 it once to fetch a ``{"lessons": ..., "recommended_followups":
1707 ...}`` overlay and synthesises a tiny synchronous sampler
1708 closure that returns the pre-fetched overlay; the closure is
1709 then handed to :func:`mcp.mission.final_report.write_final_report`
1710 which keeps its existing sync-callable contract. This pre-fetch
1711 bridge is needed because the production helper
1712 (:func:`mcp.mission.sampling.maybe_sample_final_lessons`) is
1713 async while ``write_final_report`` is sync.
1714 """
1715 if verdict == "complete":
1716 session["status"] = "completed"
1717 else: # verdict == "terminate"
1718 session["status"] = "terminated"
1719 session["ended_at"] = record["ended_at"]
1720 session["final_verdict"] = verdict
1721 # Optional sampling overlay for the lessons / followups fields.
1722 # Pre-fetch so the (sync) report writer never has to await.
1723 overlay = await self._maybe_sample_final_lessons(session, verdict, reason)
1724 sampler: Callable[..., dict[str, Any] | None] | None
1725 if overlay is not None:
1727 def _pre_fetched_sampler(
1728 _session: SessionState,
1729 _verdict: VerdictLabel,
1730 _reason: VerdictReason,
1731 ) -> dict[str, Any] | None:
1732 return overlay
1734 sampler = _pre_fetched_sampler
1735 else:
1736 sampler = None
1737 # The Final_Report is the durable exit artifact. We write it
1738 # via the report helper so the persistence path (filesystem
1739 # sibling vs. embedded-on-session for non-filesystem backends)
1740 # is owned by one module.
1741 final_report.write_final_report(self.backend, session, verdict, reason, sampler=sampler)
1743 # Best-effort institutional memory, after the report has landed.
1744 # Reuses the overlay fetched above — never re-samples.
1745 self._maybe_write_memory(session, verdict, reason, overlay)
1747 async def _maybe_sample_final_lessons(
1748 self,
1749 session: SessionState,
1750 verdict: VerdictLabel,
1751 reason: VerdictReason,
1752 ) -> dict[str, Any] | None:
1753 """Fetch the optional Final_Report ``lessons`` / ``followups`` overlay.
1755 Calls :attr:`final_lessons_callable` once (when wired and when
1756 the session opted into sampling) and adapts the return value
1757 into the ``{"lessons": str, "recommended_followups": list[str]}``
1758 shape that
1759 :func:`mcp.mission.final_report.write_final_report` expects.
1761 Three return shapes are recognised:
1763 * :class:`mcp.mission.sampling.SamplingUsed` — production path.
1764 ``parsed["lessons"]`` is a list of strings; the engine joins
1765 them with double newlines so the report's ``lessons`` field
1766 stays a single string. ``parsed["recommended_followups"]`` is
1767 forwarded as a list verbatim.
1768 * Raw ``dict`` (legacy / test pattern) — passed straight
1769 through. The downstream sampler-overlay code in
1770 ``write_final_report`` already validates and silently drops
1771 malformed fields.
1772 * Anything else (including :class:`SamplingFallback`,
1773 ``None``, exceptions) — returns ``None`` so the
1774 deterministic templates from
1775 :func:`mcp.mission.final_report.build_deterministic_report`
1776 stand on their own.
1778 The method swallows any exception raised by the callable
1779 because the Final_Report is the durable exit artifact and a
1780 flaky sampler must not block it from landing.
1781 """
1782 del verdict, reason # forwarded only for symmetry with the legacy Sampler shape
1783 if self.final_lessons_callable is None:
1784 return None
1785 if not session.get("use_sampling"):
1786 return None
1787 try:
1788 result = await self.final_lessons_callable(session=session)
1789 except Exception:
1790 return None
1791 if isinstance(result, SamplingUsed):
1792 lessons = result.parsed.get("lessons")
1793 followups = result.parsed.get("recommended_followups")
1794 overlay: dict[str, Any] = {}
1795 if isinstance(lessons, list) and all(isinstance(item, str) for item in lessons):
1796 # write_final_report expects ``lessons`` as a single
1797 # string; join with blank lines so multi-bullet output
1798 # from the model stays readable on render.
1799 overlay["lessons"] = "\n\n".join(lessons)
1800 if isinstance(followups, list) and all(isinstance(item, str) for item in followups):
1801 overlay["recommended_followups"] = list(followups)
1802 return overlay or None
1803 if isinstance(result, SamplingFallback):
1804 return None
1805 if isinstance(result, dict):
1806 # Legacy raw-dict path — let the downstream overlay
1807 # validator do the structural check.
1808 return result
1809 return None
1811 def _maybe_write_memory(
1812 self,
1813 session: SessionState,
1814 verdict: VerdictLabel,
1815 reason: VerdictReason,
1816 overlay: dict[str, Any] | None,
1817 ) -> None:
1818 """Best-effort mission-memory write on a terminal verdict.
1820 Persists one memory item — the directive, its embedding, and
1821 the report's narrative — through :attr:`memory_store` so future
1822 missions with similar directives can recall this one's lessons.
1824 The narrative fields reuse what the Final_Report just recorded:
1825 the sampled ``overlay`` when one was produced (never re-sampled
1826 — it is the exact dict handed to ``write_final_report`` above),
1827 else the deterministic templates from
1828 :func:`mcp.mission.final_report.build_deterministic_report`,
1829 which is pure and cheap to rebuild.
1831 Every exception is swallowed, mirroring
1832 :meth:`_maybe_sample_final_lessons`: the Final_Report is the
1833 durable exit artifact and memory is strictly additive. An
1834 absent table, a backfilling index, a missing SSM parameter, an
1835 unreachable Bedrock endpoint, or a store bug must all degrade
1836 to "no memory written" — never to a failed mission.
1837 """
1838 if self.memory_store is None:
1839 return
1840 try:
1841 report = final_report.build_deterministic_report(session, verdict, reason)
1842 lessons = report.get("lessons", "")
1843 followups = report.get("recommended_followups", [])
1844 if overlay:
1845 overlay_lessons = overlay.get("lessons")
1846 if isinstance(overlay_lessons, str) and overlay_lessons.strip():
1847 lessons = overlay_lessons
1848 overlay_followups = overlay.get("recommended_followups")
1849 if isinstance(overlay_followups, list) and all(
1850 isinstance(item, str) for item in overlay_followups
1851 ):
1852 followups = overlay_followups
1853 self.memory_store.write_memory(
1854 session,
1855 verdict,
1856 reason,
1857 str(lessons),
1858 [str(item) for item in followups],
1859 )
1860 except Exception:
1861 return
1864# ---------------------------------------------------------------------------
1865# Comparison helper
1866# ---------------------------------------------------------------------------
1869def _compare_numbers(value: float, op: str, target: float) -> bool:
1870 """Apply one of the six allowed numeric comparison operators."""
1871 if op == "<":
1872 return value < target
1873 if op == "<=":
1874 return value <= target
1875 if op == ">":
1876 return value > target
1877 if op == ">=":
1878 return value >= target
1879 if op == "==":
1880 return value == target
1881 if op == "!=":
1882 return value != target
1883 raise ValueError(f"unknown comparison operator: {op!r}")