Coverage for gco_mcp / mission / final_report.py: 100.00%
146 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Mission Final_Report writer.
3Builds and persists the durable JSON artifact that ends a Mission_Session.
4The report captures the directive, criteria, budget, allowlist, cadence,
5the full iteration history (with private parser caches stripped), and the
6terminal verdict. Two surfaces:
8* :func:`build_deterministic_report` — pure: takes a session and the
9 terminal ``(verdict, reason)`` tuple, returns a dict containing only
10 fields that can be derived from the session payload without consulting
11 any LLM. The ``lessons`` and ``recommended_followups`` slots are
12 pre-populated with templated text so a Mission running with sampling
13 disabled — or with a sampling backend that fails — still produces a
14 complete, useful report.
15* :func:`write_final_report` — calls :func:`build_deterministic_report`,
16 optionally overlays the sampler-supplied ``lessons`` /
17 ``recommended_followups``, persists the report, and updates
18 ``session["final_report_path"]``. Returns the persisted-path identifier.
20The writer is deliberately backend-aware. :class:`FilesystemBackend` writes
21the report as a sibling file at ``<root>/<session_id>.report.json`` using
22the same temp-file + ``fsync`` + ``os.replace`` atomic pattern that
23:meth:`FilesystemBackend.save_session` uses, so a reader concurrent with a
24writer never sees a partial JSON document. Other backends (today, the
25:class:`DynamoDBBackend` stub) embed the report on the session under a
26``final_report`` key and re-save the session — DynamoDB's single-item
27``put_item`` is atomic, so no separate dance is needed. The synthetic
28identifier returned in that case is ``"dynamodb://{session_id}/report"`` so
29callers always have a stable string to record on
30``session["final_report_path"]``.
31"""
33from __future__ import annotations
35import contextlib
36import copy
37import json
38import logging
39import os
40import tempfile
41from collections.abc import Callable, Iterable
42from datetime import UTC, datetime
43from typing import Any, cast
45from .state import FilesystemBackend
46from .types import IterationRecord, SessionState, VerdictLabel, VerdictReason
48__all__ = [
49 "build_deterministic_report",
50 "build_swarm_children_table",
51 "write_final_report",
52]
54logger = logging.getLogger(__name__)
57# --------------------------------------------------------------------------
58# Type aliases
59# --------------------------------------------------------------------------
61# A sampler callable supplies LLM-derived ``lessons`` /
62# ``recommended_followups`` overlays for the report. It receives the
63# session and the terminal verdict tuple, and is expected to return a dict
64# carrying the two keys — or ``None`` when the call failed and the
65# deterministic templates should be kept. The return is typed ``object``
66# because the sampler is a caller-supplied plugin whose output
67# ``_safely_invoke_sampler`` validates at runtime (a non-dict is logged and
68# ignored); promising a dict here would make that check dead code.
69Sampler = Callable[[SessionState, VerdictLabel, VerdictReason], object]
72# Private cache key written by ``validate_criteria`` onto every
73# ``predicate`` Criterion. We strip it from anything that lands in the
74# report so the artifact stays portable JSON.
75_PARSED_AST_KEY = "_parsed_ast"
78# --------------------------------------------------------------------------
79# Public surface
80# --------------------------------------------------------------------------
83def build_deterministic_report(
84 session: SessionState,
85 verdict: VerdictLabel,
86 reason: VerdictReason,
87) -> dict[str, Any]:
88 """Return the Final_Report dict using only deterministic session fields.
90 The returned dict carries:
92 * Identification — ``session_id`` and the verbatim ``directive_text``.
93 * Configuration snapshot — ``criteria`` (with the cached parser AST
94 stripped), ``budget``, ``tool_allowlist``, ``checkpoint_cadence``,
95 and ``stagnation_threshold``.
96 * Lifecycle timestamps — ``created_at``, ``started_at`` (``None``
97 when the session never ran a real iteration), and a fresh
98 ``ended_at`` set to the current UTC time.
99 * Outcome — ``iterations_run``, ``final_verdict``,
100 ``final_verdict_reason``, ``final_criteria_evaluation`` (the last
101 iteration's per-Criterion results, or ``None`` when no iteration
102 ran).
103 * Iteration history — ``iterations``, deep-copied with private
104 ``_parsed_ast`` keys stripped throughout.
105 * Templated narrative — ``lessons`` and ``recommended_followups``
106 pre-populated with deterministic template text so a session that
107 ran with sampling disabled, or whose sampler failed, still
108 produces a useful report. :func:`write_final_report` overlays
109 these two fields when a working sampler is supplied.
111 Pure: depends only on the session payload and the verdict tuple, and
112 produces nothing that a caller could not regenerate from the same
113 inputs. The single ``datetime.now`` call records the moment the
114 report was assembled — that is itself the deterministic function of
115 "now I am writing the report" rather than business logic that
116 consults the clock.
117 """
118 now_iso = datetime.now(UTC).isoformat()
120 report: dict[str, Any] = {
121 "session_id": session["session_id"],
122 "directive_text": session["directive_text"],
123 "criteria": _strip_parsed_ast_from_criteria(session.get("criteria") or []),
124 "budget": dict(session["budget"]),
125 "tool_allowlist": list(session.get("tool_allowlist") or []),
126 "checkpoint_cadence": dict(session["checkpoint_cadence"]),
127 "stagnation_threshold": session.get("stagnation_threshold"),
128 "created_at": session.get("created_at"),
129 "started_at": session.get("started_at"),
130 "ended_at": now_iso,
131 "iterations_run": len(session.get("iterations") or []),
132 "final_verdict": verdict,
133 "final_verdict_reason": reason,
134 "final_criteria_evaluation": _final_criteria_evaluation(session),
135 "lessons": _build_lessons_template(session, verdict, reason),
136 "recommended_followups": _build_followups_template(session, verdict, reason),
137 "iterations": _strip_parsed_ast_from_iterations(session.get("iterations") or []),
138 }
139 swarm_outcomes = build_swarm_children_table(session)
140 if swarm_outcomes is not None:
141 report["swarm_children"] = swarm_outcomes
142 return report
145def build_swarm_children_table(session: SessionState) -> list[dict[str, Any]] | None:
146 """Per-child outcome rows for an orchestrator session's report.
148 Present only on sessions carrying a child registry (``None``
149 otherwise, so standalone and child reports are byte-identical to
150 before). Slot-ordered, built purely from the persisted registry —
151 the settled entries already carry the supervision outcome, so no
152 child session load is needed at report time and an unreadable child
153 cannot fail report assembly.
155 The engine writes the report during terminal finalization, which
156 happens *before* the swarm runner's abort cascade settles the last
157 registry entries; the runner refreshes this table on the written
158 report after the cascade so the durable artifact reflects final
159 states. Public for exactly that caller.
160 """
161 registry = session.get("children")
162 if registry is None:
163 return None
164 rows: list[dict[str, Any]] = []
165 for entry in sorted(registry, key=lambda e: e["slot"]):
166 row: dict[str, Any] = {
167 "slot": entry["slot"],
168 "session_id": entry["session_id"],
169 "spawned_at": entry["spawned_at"],
170 "restart_policy": entry["restart_policy"],
171 "respawn_count": entry["respawn_count"],
172 "iterations_consumed": entry["consumed_iterations"],
173 "settled": bool(entry.get("settled", False)),
174 }
175 lineage = entry.get("prior_session_ids")
176 if lineage:
177 row["prior_session_ids"] = list(lineage)
178 rows.append(row)
179 return rows
182def write_final_report(
183 backend: Any,
184 session: SessionState,
185 verdict: VerdictLabel,
186 reason: VerdictReason,
187 sampler: Sampler | None = None,
188) -> str:
189 """Build, optionally overlay, and persist the Final_Report.
191 The flow is:
193 1. :func:`build_deterministic_report` produces a complete report
194 dict with templated ``lessons`` / ``recommended_followups``.
195 2. When ``sampler`` is supplied, it is called once with
196 ``(session, verdict, reason)``. A returned dict whose ``lessons``
197 and / or ``recommended_followups`` keys are well-typed overlays
198 the corresponding template values; any other return (``None``,
199 a dict missing both keys, or an exception) leaves the templates
200 intact. Sampler failures are logged at WARNING and never
201 propagated — the report must always land.
202 3. The report is persisted alongside (or on) the session, depending
203 on the backend type:
205 * :class:`mcp.mission.state.FilesystemBackend` writes
206 ``<root>/<session_id>.report.json`` using the same temp-file +
207 ``fsync`` + ``os.replace`` atomic pattern as
208 :meth:`FilesystemBackend.save_session`. Returns the absolute
209 path of the report file.
210 * Any other backend (today, the DynamoDB stub) attaches the
211 report dict to the session under ``final_report`` and calls
212 ``backend.save_session(session)``. DynamoDB's single-item
213 ``put_item`` is atomic so no separate dance is needed. Returns
214 ``"dynamodb://{session_id}/report"`` as a stable synthetic
215 identifier.
217 4. ``session["final_report_path"]`` is updated with the returned
218 identifier so callers (and the next ``backend.save_session``)
219 record where the report lives.
220 """
221 report = build_deterministic_report(session, verdict, reason)
223 if sampler is not None:
224 overlay = _safely_invoke_sampler(sampler, session, verdict, reason)
225 if overlay is not None:
226 _apply_sampler_overlay(report, overlay)
228 if isinstance(backend, FilesystemBackend):
229 path = _write_report_to_filesystem(backend, session["session_id"], report)
230 else:
231 path = _attach_report_to_session(backend, session, report)
233 session["final_report_path"] = path
234 return path
237# --------------------------------------------------------------------------
238# Templated narrative
239# --------------------------------------------------------------------------
242def _build_lessons_template(
243 session: SessionState,
244 verdict: VerdictLabel,
245 reason: VerdictReason,
246) -> str:
247 """Deterministic ``lessons`` paragraph for sessions without sampling overlay.
249 A few lines of operator-readable narrative pulling exclusively from
250 the persisted session: the directive, the terminal verdict and
251 reason, the iteration count, and a comma-separated list of unmet or
252 inconclusive criterion ids drawn from the final iteration's
253 evaluation. Stays short and machine-parseable so it is easy to grep
254 or display in a CLI summary.
255 """
256 iterations = session.get("iterations") or []
257 iteration_count = len(iterations)
258 directive = session.get("directive_text", "")
259 # Trim the directive so a verbose multi-line directive does not turn
260 # this paragraph into a wall of text.
261 if len(directive) > 240:
262 directive = directive[:237] + "..."
264 final_eval = _final_criteria_evaluation(session) or []
265 not_met_ids = [
266 result["criterion_id"]
267 for result in final_eval
268 if result.get("status") in ("unmet", "inconclusive")
269 ]
270 not_met_summary = ", ".join(not_met_ids) if not_met_ids else "none"
272 return (
273 f"Mission ended with verdict {verdict!r} (reason {reason!r}) after "
274 f"{iteration_count} iteration(s). Directive: {directive!r}. "
275 f"Outstanding criteria at termination: {not_met_summary}. "
276 "This summary is templated text — re-run with sampling enabled to "
277 "replace it with a model-derived narrative."
278 )
281def _build_followups_template(
282 session: SessionState,
283 verdict: VerdictLabel,
284 reason: VerdictReason,
285) -> list[str]:
286 """Deterministic ``recommended_followups`` for templated reports.
288 Returns 1–3 generic next-step suggestions chosen from the verdict
289 reason. Pure: same inputs → same outputs. Wording stays short so
290 callers can render the list as bullet points in a CLI summary.
292 The ``session`` argument is unused today but kept on the signature
293 so a future enhancement that consults the iteration history (e.g.
294 naming the most-used tool) can be added without changing every
295 call site.
296 """
297 del session # currently unused; kept for signature stability
299 suggestions: list[str] = []
301 if verdict == "complete":
302 suggestions.append(
303 "Persist any artefacts produced by the final iteration so the "
304 "outcome survives beyond the session JSON."
305 )
306 suggestions.append(
307 "Re-run with tighter criteria thresholds to confirm the result "
308 "was not a borderline match."
309 )
310 elif reason == "max_iterations":
311 suggestions.append(
312 "Re-run with a higher max_iterations cap if more iterations "
313 "would plausibly close the remaining gap."
314 )
315 suggestions.append(
316 "Inspect the iteration history for repeated tool sequences and "
317 "consider tightening the strategy revision heuristic."
318 )
319 elif reason == "max_wall_clock":
320 suggestions.append(
321 "Re-run with a higher max_wall_clock_seconds budget, or split "
322 "the directive into smaller sub-goals."
323 )
324 elif reason == "no_progress":
325 suggestions.append(
326 "Re-evaluate criteria thresholds — sustained no-progress may "
327 "indicate the targets are unreachable with the current tool "
328 "allowlist."
329 )
330 suggestions.append(
331 "Widen the tool allowlist or supply a richer directive so the "
332 "loop can explore alternative strategies."
333 )
334 elif reason == "user_abort":
335 suggestions.append(
336 "Resume the session with mission_resume once the manual intervention is complete."
337 )
338 else:
339 suggestions.append(
340 "Inspect the iteration history for the last verdict and adjust "
341 "the directive, criteria, or allowlist accordingly."
342 )
344 suggestions.append(
345 "These suggestions are templated — re-run with sampling enabled to "
346 "replace them with model-derived followups."
347 )
348 return suggestions[:3]
351# --------------------------------------------------------------------------
352# Strip helpers — pure
353# --------------------------------------------------------------------------
356def _strip_parsed_ast_from_criteria(criteria: Iterable[object]) -> list[Any]:
357 """Return a shallow copy of ``criteria`` with private parser caches removed.
359 The ``validate_criteria`` validator caches the parsed AST under
360 ``_parsed_ast`` on every ``predicate`` Criterion. The Final_Report
361 is meant to be portable JSON, so we strip the cache before
362 serialisation. The strip is also defensive: the report dict is
363 later passed through ``json.dumps``, and an ``ast.Expression``
364 object would raise there with a less obvious error than this.
366 ``criteria`` is typed as ``object`` entries because the list is read
367 back from the persisted session: a corrupt file can put a non-dict
368 there, and such an entry passes through verbatim so the corruption
369 stays visible in the report rather than crashing its assembly.
370 """
371 cleaned: list[Any] = []
372 for criterion in criteria:
373 if not isinstance(criterion, dict):
374 cleaned.append(criterion)
375 continue
376 cleaned.append({k: v for k, v in criterion.items() if k != _PARSED_AST_KEY})
377 return cleaned
380def _strip_parsed_ast_from_iterations(
381 iterations: list[IterationRecord],
382) -> list[dict[str, Any]]:
383 """Return a deep copy of the iteration history with parser caches removed.
385 Walks every nested dict and drops any ``_parsed_ast`` entry it
386 finds. The validators only cache on Criterion entries today, but
387 the strip is intentionally broad so a future code path that
388 accidentally embeds a Criterion (with its cache attached) inside an
389 IterationRecord cannot corrupt the report's JSON serialisation.
390 """
391 cloned = copy.deepcopy(list(iterations))
392 for entry in cloned:
393 _strip_parsed_ast_in_place(entry)
394 return cast(list[dict[str, Any]], cloned)
397def _strip_parsed_ast_in_place(value: Any) -> None:
398 """Recursively delete ``_parsed_ast`` keys from any nested dict."""
399 if isinstance(value, dict):
400 if _PARSED_AST_KEY in value:
401 del value[_PARSED_AST_KEY]
402 for inner in value.values():
403 _strip_parsed_ast_in_place(inner)
404 elif isinstance(value, list):
405 for inner in value:
406 _strip_parsed_ast_in_place(inner)
409def _final_criteria_evaluation(session: SessionState) -> list[dict[str, Any]] | None:
410 """Return the last iteration's ``criteria_evaluation`` list, or ``None``.
412 Used as the ``final_criteria_evaluation`` field on the report so a
413 consumer can answer "which criteria were met at the moment the
414 session ended" without scanning the iteration history.
415 Returns ``None`` when the session ran no iterations — the report is
416 still useful for sessions that terminated at start (e.g. a
417 user_abort before the first iteration).
418 """
419 iterations = session.get("iterations") or []
420 if not iterations:
421 return None
422 last = iterations[-1]
423 evaluation = last.get("criteria_evaluation")
424 if not evaluation:
425 return None
426 return [dict(result) for result in evaluation]
429# --------------------------------------------------------------------------
430# Sampler overlay
431# --------------------------------------------------------------------------
434def _safely_invoke_sampler(
435 sampler: Sampler,
436 session: SessionState,
437 verdict: VerdictLabel,
438 reason: VerdictReason,
439) -> dict[str, Any] | None:
440 """Call ``sampler`` and return its dict, or ``None`` on any failure.
442 A sampler that raises must not block the report from landing — the
443 Final_Report is the durable exit artifact of the loop. Any
444 exception is logged at WARNING and swallowed, leaving the
445 deterministic templates in place. A non-dict return is treated the
446 same way (logged, ignored).
447 """
448 try:
449 result = sampler(session, verdict, reason)
450 except Exception:
451 logger.warning(
452 "Mission sampler raised while building Final_Report for session %s; "
453 "keeping templated lessons / recommended_followups.",
454 session.get("session_id"),
455 exc_info=True,
456 )
457 return None
458 if result is None:
459 return None
460 if not isinstance(result, dict):
461 logger.warning(
462 "Mission sampler returned a non-dict (%s) for session %s; "
463 "keeping templated lessons / recommended_followups.",
464 type(result).__name__,
465 session.get("session_id"),
466 )
467 return None
468 return result
471def _apply_sampler_overlay(report: dict[str, Any], overlay: dict[str, Any]) -> None:
472 """Overwrite ``lessons`` and / or ``recommended_followups`` if well-typed.
474 Each field is overlaid independently: a sampler that produced a
475 valid ``lessons`` string but malformed ``recommended_followups``
476 keeps the lessons replacement and falls back to the template list
477 for the followups. The shape checks are defensive — a sampler is
478 free-form by contract, and silently dropping a malformed field is
479 safer than letting a non-string slip into a downstream consumer.
480 """
481 lessons = overlay.get("lessons")
482 if isinstance(lessons, str) and lessons:
483 report["lessons"] = lessons
485 followups = overlay.get("recommended_followups")
486 if isinstance(followups, list) and all(isinstance(item, str) for item in followups):
487 report["recommended_followups"] = list(followups)
490# --------------------------------------------------------------------------
491# Persistence
492# --------------------------------------------------------------------------
495def _write_report_to_filesystem(
496 backend: FilesystemBackend,
497 session_id: str,
498 report: dict[str, Any],
499) -> str:
500 """Persist ``report`` as ``<root>/<session_id>.report.json`` atomically.
502 Mirrors the temp-file + ``fsync`` + ``os.replace`` pattern from
503 :meth:`FilesystemBackend.save_session`: a partial write leaves the
504 temp file behind but never replaces the existing report file, so a
505 reader concurrent with a writer always sees either the prior
506 version or the new one. Returns the absolute path of the written
507 file.
509 Uses :meth:`FilesystemBackend._ensure_root` to lazily create the
510 backend's root directory on first use; this matches the session
511 writer and avoids duplicating the directory-creation logic here.
512 """
513 backend._ensure_root()
514 final = backend.root / f"{session_id}.report.json"
515 try:
516 tmp = tempfile.NamedTemporaryFile( # noqa: SIM115 - explicit close+replace below
517 mode="w",
518 encoding="utf-8",
519 dir=str(backend.root),
520 prefix=f"{session_id}.report.",
521 suffix=".json.tmp",
522 delete=False,
523 )
524 try:
525 json.dump(report, tmp)
526 tmp.flush()
527 os.fsync(tmp.fileno())
528 finally:
529 tmp.close()
530 if os.name != "nt":
531 with contextlib.suppress(OSError):
532 # Same rationale as the session writer: a successful
533 # fsync is too valuable to abandon over a permission
534 # tightening that the underlying filesystem refused.
535 os.chmod(tmp.name, 0o600)
536 os.replace(tmp.name, final)
537 except OSError as exc:
538 # Re-raise with the underlying message intact so operators see
539 # the real cause (disk full, permission denied) rather than a
540 # wrapped abstraction.
541 raise OSError(str(exc)) from exc
542 return str(final)
545def _attach_report_to_session(
546 backend: Any,
547 session: SessionState,
548 report: dict[str, Any],
549) -> str:
550 """Embed ``report`` on the session and re-save through the backend.
552 Used for backends that do not write sibling files (today, the
553 DynamoDB stub). Returns the synthetic identifier
554 ``"dynamodb://{session_id}/report"`` so the caller has a stable
555 path-like value to record on ``session["final_report_path"]``.
557 The session is mutated in place: the ``final_report`` key carries
558 the report dict so a later ``backend.load_session`` returns the
559 full payload without a second round-trip. The backend's
560 ``save_session`` performs whatever atomicity the storage layer
561 provides (DynamoDB ``put_item`` is single-item-atomic by contract).
562 """
563 # ``final_report`` is not declared on :class:`SessionState`; cast
564 # through ``dict[str, Any]`` so the assignment lands without a
565 # TypedDict-unknown-key complaint while keeping the underlying
566 # session object identity intact.
567 cast(dict[str, Any], session)["final_report"] = report
568 backend.save_session(session)
569 return f"dynamodb://{session['session_id']}/report"