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