Coverage for gco_mcp / mission / final_report.py: 100.00%
146 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +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
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 returns a dict carrying the
64# two keys — or ``None`` when the call failed and the deterministic
65# templates should be kept.
66Sampler = Callable[
67 [SessionState, VerdictLabel, VerdictReason],
68 "dict[str, Any] | None",
69]
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(
124 cast("list[dict[str, Any]]", list(session.get("criteria") or []))
125 ),
126 "budget": dict(session.get("budget") or {}),
127 "tool_allowlist": list(session.get("tool_allowlist") or []),
128 "checkpoint_cadence": dict(session.get("checkpoint_cadence") or {}),
129 "stagnation_threshold": session.get("stagnation_threshold"),
130 "created_at": session.get("created_at"),
131 "started_at": session.get("started_at"),
132 "ended_at": now_iso,
133 "iterations_run": len(session.get("iterations") or []),
134 "final_verdict": verdict,
135 "final_verdict_reason": reason,
136 "final_criteria_evaluation": _final_criteria_evaluation(session),
137 "lessons": _build_lessons_template(session, verdict, reason),
138 "recommended_followups": _build_followups_template(session, verdict, reason),
139 "iterations": _strip_parsed_ast_from_iterations(session.get("iterations") or []),
140 }
141 swarm_outcomes = build_swarm_children_table(session)
142 if swarm_outcomes is not None:
143 report["swarm_children"] = swarm_outcomes
144 return report
147def build_swarm_children_table(session: SessionState) -> list[dict[str, Any]] | None:
148 """Per-child outcome rows for an orchestrator session's report.
150 Present only on sessions carrying a child registry (``None``
151 otherwise, so standalone and child reports are byte-identical to
152 before). Slot-ordered, built purely from the persisted registry —
153 the settled entries already carry the supervision outcome, so no
154 child session load is needed at report time and an unreadable child
155 cannot fail report assembly.
157 The engine writes the report during terminal finalization, which
158 happens *before* the swarm runner's abort cascade settles the last
159 registry entries; the runner refreshes this table on the written
160 report after the cascade so the durable artifact reflects final
161 states. Public for exactly that caller.
162 """
163 registry = session.get("children")
164 if registry is None:
165 return None
166 rows: list[dict[str, Any]] = []
167 for entry in sorted(registry, key=lambda e: e["slot"]):
168 row: dict[str, Any] = {
169 "slot": entry["slot"],
170 "session_id": entry["session_id"],
171 "spawned_at": entry["spawned_at"],
172 "restart_policy": entry["restart_policy"],
173 "respawn_count": entry["respawn_count"],
174 "iterations_consumed": entry["consumed_iterations"],
175 "settled": bool(entry.get("settled", False)),
176 }
177 lineage = entry.get("prior_session_ids")
178 if lineage:
179 row["prior_session_ids"] = list(lineage)
180 rows.append(row)
181 return rows
184def write_final_report(
185 backend: Any,
186 session: SessionState,
187 verdict: VerdictLabel,
188 reason: VerdictReason,
189 sampler: Sampler | None = None,
190) -> str:
191 """Build, optionally overlay, and persist the Final_Report.
193 The flow is:
195 1. :func:`build_deterministic_report` produces a complete report
196 dict with templated ``lessons`` / ``recommended_followups``.
197 2. When ``sampler`` is supplied, it is called once with
198 ``(session, verdict, reason)``. A returned dict whose ``lessons``
199 and / or ``recommended_followups`` keys are well-typed overlays
200 the corresponding template values; any other return (``None``,
201 a dict missing both keys, or an exception) leaves the templates
202 intact. Sampler failures are logged at WARNING and never
203 propagated — the report must always land.
204 3. The report is persisted alongside (or on) the session, depending
205 on the backend type:
207 * :class:`mcp.mission.state.FilesystemBackend` writes
208 ``<root>/<session_id>.report.json`` using the same temp-file +
209 ``fsync`` + ``os.replace`` atomic pattern as
210 :meth:`FilesystemBackend.save_session`. Returns the absolute
211 path of the report file.
212 * Any other backend (today, the DynamoDB stub) attaches the
213 report dict to the session under ``final_report`` and calls
214 ``backend.save_session(session)``. DynamoDB's single-item
215 ``put_item`` is atomic so no separate dance is needed. Returns
216 ``"dynamodb://{session_id}/report"`` as a stable synthetic
217 identifier.
219 4. ``session["final_report_path"]`` is updated with the returned
220 identifier so callers (and the next ``backend.save_session``)
221 record where the report lives.
222 """
223 report = build_deterministic_report(session, verdict, reason)
225 if sampler is not None:
226 overlay = _safely_invoke_sampler(sampler, session, verdict, reason)
227 if overlay is not None:
228 _apply_sampler_overlay(report, overlay)
230 if isinstance(backend, FilesystemBackend):
231 path = _write_report_to_filesystem(backend, session["session_id"], report)
232 else:
233 path = _attach_report_to_session(backend, session, report)
235 session["final_report_path"] = path
236 return path
239# --------------------------------------------------------------------------
240# Templated narrative
241# --------------------------------------------------------------------------
244def _build_lessons_template(
245 session: SessionState,
246 verdict: VerdictLabel,
247 reason: VerdictReason,
248) -> str:
249 """Deterministic ``lessons`` paragraph for sessions without sampling overlay.
251 A few lines of operator-readable narrative pulling exclusively from
252 the persisted session: the directive, the terminal verdict and
253 reason, the iteration count, and a comma-separated list of unmet or
254 inconclusive criterion ids drawn from the final iteration's
255 evaluation. Stays short and machine-parseable so it is easy to grep
256 or display in a CLI summary.
257 """
258 iterations = session.get("iterations") or []
259 iteration_count = len(iterations)
260 directive = session.get("directive_text", "")
261 # Trim the directive so a verbose multi-line directive does not turn
262 # this paragraph into a wall of text.
263 if len(directive) > 240:
264 directive = directive[:237] + "..."
266 final_eval = _final_criteria_evaluation(session) or []
267 not_met_ids = [
268 result["criterion_id"]
269 for result in final_eval
270 if result.get("status") in ("unmet", "inconclusive")
271 ]
272 not_met_summary = ", ".join(not_met_ids) if not_met_ids else "none"
274 return (
275 f"Mission ended with verdict {verdict!r} (reason {reason!r}) after "
276 f"{iteration_count} iteration(s). Directive: {directive!r}. "
277 f"Outstanding criteria at termination: {not_met_summary}. "
278 "This summary is templated text — re-run with sampling enabled to "
279 "replace it with a model-derived narrative."
280 )
283def _build_followups_template(
284 session: SessionState,
285 verdict: VerdictLabel,
286 reason: VerdictReason,
287) -> list[str]:
288 """Deterministic ``recommended_followups`` for templated reports.
290 Returns 1–3 generic next-step suggestions chosen from the verdict
291 reason. Pure: same inputs → same outputs. Wording stays short so
292 callers can render the list as bullet points in a CLI summary.
294 The ``session`` argument is unused today but kept on the signature
295 so a future enhancement that consults the iteration history (e.g.
296 naming the most-used tool) can be added without changing every
297 call site.
298 """
299 del session # currently unused; kept for signature stability
301 suggestions: list[str] = []
303 if verdict == "complete":
304 suggestions.append(
305 "Persist any artefacts produced by the final iteration so the "
306 "outcome survives beyond the session JSON."
307 )
308 suggestions.append(
309 "Re-run with tighter criteria thresholds to confirm the result "
310 "was not a borderline match."
311 )
312 elif reason == "max_iterations":
313 suggestions.append(
314 "Re-run with a higher max_iterations cap if more iterations "
315 "would plausibly close the remaining gap."
316 )
317 suggestions.append(
318 "Inspect the iteration history for repeated tool sequences and "
319 "consider tightening the strategy revision heuristic."
320 )
321 elif reason == "max_wall_clock":
322 suggestions.append(
323 "Re-run with a higher max_wall_clock_seconds budget, or split "
324 "the directive into smaller sub-goals."
325 )
326 elif reason == "no_progress":
327 suggestions.append(
328 "Re-evaluate criteria thresholds — sustained no-progress may "
329 "indicate the targets are unreachable with the current tool "
330 "allowlist."
331 )
332 suggestions.append(
333 "Widen the tool allowlist or supply a richer directive so the "
334 "loop can explore alternative strategies."
335 )
336 elif reason == "user_abort":
337 suggestions.append(
338 "Resume the session with mission_resume once the manual intervention is complete."
339 )
340 else:
341 suggestions.append(
342 "Inspect the iteration history for the last verdict and adjust "
343 "the directive, criteria, or allowlist accordingly."
344 )
346 suggestions.append(
347 "These suggestions are templated — re-run with sampling enabled to "
348 "replace them with model-derived followups."
349 )
350 return suggestions[:3]
353# --------------------------------------------------------------------------
354# Strip helpers — pure
355# --------------------------------------------------------------------------
358def _strip_parsed_ast_from_criteria(criteria: list[dict[str, Any]]) -> list[dict[str, Any]]:
359 """Return a shallow copy of ``criteria`` with private parser caches removed.
361 The ``validate_criteria`` validator caches the parsed AST under
362 ``_parsed_ast`` on every ``predicate`` Criterion. The Final_Report
363 is meant to be portable JSON, so we strip the cache before
364 serialisation. The strip is also defensive: the report dict is
365 later passed through ``json.dumps``, and an ``ast.Expression``
366 object would raise there with a less obvious error than this.
367 """
368 cleaned: list[dict[str, Any]] = []
369 for criterion in criteria:
370 if not isinstance(criterion, dict):
371 cleaned.append(criterion)
372 continue
373 cleaned.append({k: v for k, v in criterion.items() if k != _PARSED_AST_KEY})
374 return cleaned
377def _strip_parsed_ast_from_iterations(
378 iterations: list[IterationRecord],
379) -> list[dict[str, Any]]:
380 """Return a deep copy of the iteration history with parser caches removed.
382 Walks every nested dict and drops any ``_parsed_ast`` entry it
383 finds. The validators only cache on Criterion entries today, but
384 the strip is intentionally broad so a future code path that
385 accidentally embeds a Criterion (with its cache attached) inside an
386 IterationRecord cannot corrupt the report's JSON serialisation.
387 """
388 cloned = copy.deepcopy(list(iterations))
389 for entry in cloned:
390 _strip_parsed_ast_in_place(entry)
391 return cast(list[dict[str, Any]], cloned)
394def _strip_parsed_ast_in_place(value: Any) -> None:
395 """Recursively delete ``_parsed_ast`` keys from any nested dict."""
396 if isinstance(value, dict):
397 if _PARSED_AST_KEY in value:
398 del value[_PARSED_AST_KEY]
399 for inner in value.values():
400 _strip_parsed_ast_in_place(inner)
401 elif isinstance(value, list):
402 for inner in value:
403 _strip_parsed_ast_in_place(inner)
406def _final_criteria_evaluation(session: SessionState) -> list[dict[str, Any]] | None:
407 """Return the last iteration's ``criteria_evaluation`` list, or ``None``.
409 Used as the ``final_criteria_evaluation`` field on the report so a
410 consumer can answer "which criteria were met at the moment the
411 session ended" without scanning the iteration history.
412 Returns ``None`` when the session ran no iterations — the report is
413 still useful for sessions that terminated at start (e.g. a
414 user_abort before the first iteration).
415 """
416 iterations = session.get("iterations") or []
417 if not iterations:
418 return None
419 last = iterations[-1]
420 evaluation = last.get("criteria_evaluation")
421 if not evaluation:
422 return None
423 return [dict(result) for result in evaluation]
426# --------------------------------------------------------------------------
427# Sampler overlay
428# --------------------------------------------------------------------------
431def _safely_invoke_sampler(
432 sampler: Sampler,
433 session: SessionState,
434 verdict: VerdictLabel,
435 reason: VerdictReason,
436) -> dict[str, Any] | None:
437 """Call ``sampler`` and return its dict, or ``None`` on any failure.
439 A sampler that raises must not block the report from landing — the
440 Final_Report is the durable exit artifact of the loop. Any
441 exception is logged at WARNING and swallowed, leaving the
442 deterministic templates in place. A non-dict return is treated the
443 same way (logged, ignored).
444 """
445 try:
446 result = sampler(session, verdict, reason)
447 except Exception:
448 logger.warning(
449 "Mission sampler raised while building Final_Report for session %s; "
450 "keeping templated lessons / recommended_followups.",
451 session.get("session_id"),
452 exc_info=True,
453 )
454 return None
455 if result is None:
456 return None
457 if not isinstance(result, dict):
458 logger.warning(
459 "Mission sampler returned a non-dict (%s) for session %s; "
460 "keeping templated lessons / recommended_followups.",
461 type(result).__name__,
462 session.get("session_id"),
463 )
464 return None
465 return result
468def _apply_sampler_overlay(report: dict[str, Any], overlay: dict[str, Any]) -> None:
469 """Overwrite ``lessons`` and / or ``recommended_followups`` if well-typed.
471 Each field is overlaid independently: a sampler that produced a
472 valid ``lessons`` string but malformed ``recommended_followups``
473 keeps the lessons replacement and falls back to the template list
474 for the followups. The shape checks are defensive — a sampler is
475 free-form by contract, and silently dropping a malformed field is
476 safer than letting a non-string slip into a downstream consumer.
477 """
478 lessons = overlay.get("lessons")
479 if isinstance(lessons, str) and lessons:
480 report["lessons"] = lessons
482 followups = overlay.get("recommended_followups")
483 if isinstance(followups, list) and all(isinstance(item, str) for item in followups):
484 report["recommended_followups"] = list(followups)
487# --------------------------------------------------------------------------
488# Persistence
489# --------------------------------------------------------------------------
492def _write_report_to_filesystem(
493 backend: FilesystemBackend,
494 session_id: str,
495 report: dict[str, Any],
496) -> str:
497 """Persist ``report`` as ``<root>/<session_id>.report.json`` atomically.
499 Mirrors the temp-file + ``fsync`` + ``os.replace`` pattern from
500 :meth:`FilesystemBackend.save_session`: a partial write leaves the
501 temp file behind but never replaces the existing report file, so a
502 reader concurrent with a writer always sees either the prior
503 version or the new one. Returns the absolute path of the written
504 file.
506 Uses :meth:`FilesystemBackend._ensure_root` to lazily create the
507 backend's root directory on first use; this matches the session
508 writer and avoids duplicating the directory-creation logic here.
509 """
510 backend._ensure_root()
511 final = backend.root / f"{session_id}.report.json"
512 try:
513 tmp = tempfile.NamedTemporaryFile( # noqa: SIM115 - explicit close+replace below
514 mode="w",
515 encoding="utf-8",
516 dir=str(backend.root),
517 prefix=f"{session_id}.report.",
518 suffix=".json.tmp",
519 delete=False,
520 )
521 try:
522 json.dump(report, tmp)
523 tmp.flush()
524 os.fsync(tmp.fileno())
525 finally:
526 tmp.close()
527 if os.name != "nt":
528 with contextlib.suppress(OSError):
529 # Same rationale as the session writer: a successful
530 # fsync is too valuable to abandon over a permission
531 # tightening that the underlying filesystem refused.
532 os.chmod(tmp.name, 0o600)
533 os.replace(tmp.name, final)
534 except OSError as exc:
535 # Re-raise with the underlying message intact so operators see
536 # the real cause (disk full, permission denied) rather than a
537 # wrapped abstraction.
538 raise OSError(str(exc)) from exc
539 return str(final)
542def _attach_report_to_session(
543 backend: Any,
544 session: SessionState,
545 report: dict[str, Any],
546) -> str:
547 """Embed ``report`` on the session and re-save through the backend.
549 Used for backends that do not write sibling files (today, the
550 DynamoDB stub). Returns the synthetic identifier
551 ``"dynamodb://{session_id}/report"`` so the caller has a stable
552 path-like value to record on ``session["final_report_path"]``.
554 The session is mutated in place: the ``final_report`` key carries
555 the report dict so a later ``backend.load_session`` returns the
556 full payload without a second round-trip. The backend's
557 ``save_session`` performs whatever atomicity the storage layer
558 provides (DynamoDB ``put_item`` is single-item-atomic by contract).
559 """
560 # ``final_report`` is not declared on :class:`SessionState`; cast
561 # through ``dict[str, Any]`` so the assignment lands without a
562 # TypedDict-unknown-key complaint while keeping the underlying
563 # session object identity intact.
564 cast(dict[str, Any], session)["final_report"] = report
565 backend.save_session(session)
566 return f"dynamodb://{session['session_id']}/report"