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