Coverage for gco_mcp / mission / audit.py: 100.00%
116 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Mission-specific audit emitters.
3Three thin wrappers over the existing ``audit_logger`` from ``gco_mcp/audit.py``.
4Each emitter builds a structured dict matching the Mission audit-fields
5schema and writes it as a single ``audit_logger.info(json.dumps(entry))``
6call — the same pattern that ``_build_audit_entry`` and ``emit_startup_log``
7use in ``gco_mcp/audit.py``.
9Three event types, one helper each:
11* ``emit_phase_event`` — one per phase (propose / execute / observe /
12 evaluate / decide), regardless of success or failure. Carries the
13 per-phase ``started_at`` / ``ended_at`` timestamps that the engine
14 measures around the phase body, plus an ``error_message`` field on
15 ``failed`` events.
16* ``emit_verdict_event`` — one per Decide_Phase outcome. Carries the
17 ``verdict`` label and ``verdict_reason``, plus a ``revision_rationale``
18 field when the verdict is ``adjust``.
19* ``emit_sampling_event`` — one per sampling call (whether the call was
20 used, rejected, fell back, was unavailable, or was disabled). The second
21 positional argument is ``iteration_index_or_purpose`` so callers can
22 pass either an integer iteration index (during the loop) or a string /
23 ``None`` for sampling that happens outside any single iteration (e.g.
24 Final_Report fill-in).
26Every entry carries a fresh ``timestamp`` set to ``datetime.now(UTC).isoformat()``
27at emit time — the supplied phase ``started_at`` / ``ended_at`` are recorded
28in their own dedicated fields and are not used as the entry timestamp.
30The Mission package is loaded via ``sys.path``-on-``gco_mcp/`` (the same trick
31``run_mcp.py`` uses for the rest of the MCP modules), so ``audit_logger`` is
32imported as ``from audit import audit_logger`` — matching every
33``gco_mcp/tools/*.py`` module.
35In-process audit ring buffer
36============================
37Mission also installs a bounded in-process collector
38(``MissionAuditCollectorHandler``) on the shared ``gco.mcp.audit``
39logger so the ``mission://sessions/{session_id}/audit-replay``
40resource has a source of phase / verdict entries to feed
41:func:`replay_audit_entries`. The buffer is capped at 5000 entries
42(FIFO eviction) so a long-running process cannot OOM through the
43audit channel; the cap fits comfortably above a 1000-iteration
44session's ~6000 entries since the session would have terminated on
45the iteration cap by then.
46"""
48from __future__ import annotations
50import json
51import logging
52from collections import deque
53from datetime import UTC, datetime
54from typing import Any, Literal
56from audit import audit_logger
58# ---------------------------------------------------------------------------
59# Event-type tags
60# ---------------------------------------------------------------------------
62# Stable ``event_type`` strings so audit consumers can filter Mission events
63# without parsing the rest of the entry. The values are part of the public
64# audit contract — tests, dashboards, and the reconstruction test in
65# ``tests/test_mission_audit.py`` match on these literals.
66EVENT_TYPE_PHASE = "mission_phase_event"
67EVENT_TYPE_VERDICT = "mission_verdict_event"
68EVENT_TYPE_SAMPLING = "mission_sampling_event"
69EVENT_TYPE_SCRIPT_CALL = "mission_script_call_event"
70EVENT_TYPE_CHILD_LIFECYCLE = "mission_child_lifecycle_event"
73# ---------------------------------------------------------------------------
74# Helpers
75# ---------------------------------------------------------------------------
78def _now_iso() -> str:
79 """Return the current UTC time as an ISO 8601 string.
81 Centralised so every Mission emitter records ``timestamp`` in the same
82 format. Matches the pattern used by ``_build_audit_entry`` and
83 ``emit_startup_log`` in ``gco_mcp/audit.py``.
84 """
85 return datetime.now(UTC).isoformat()
88def _emit(entry: dict[str, Any]) -> None:
89 """Serialise ``entry`` and route through the shared audit logger.
91 Centralised so the three public emitters share one log path — and so a
92 future swap to a structured handler only needs to change one site.
93 """
94 audit_logger.info(json.dumps(entry))
97# ---------------------------------------------------------------------------
98# Public emitters
99# ---------------------------------------------------------------------------
102def emit_phase_event(
103 session_id: str,
104 iteration_index: int,
105 phase: Literal["propose", "execute", "observe", "evaluate", "decide"],
106 status: Literal["succeeded", "failed"],
107 started_at: str,
108 ended_at: str,
109 error_message: str | None = None,
110) -> None:
111 """Emit one ``mission_phase_event`` audit entry.
113 Called by ``MissionEngine`` exactly once per phase from a try/finally
114 block, so a failed phase still produces an entry with
115 ``phase_status="failed"`` and the exception's ``error_message``.
117 The ``started_at`` / ``ended_at`` arguments are recorded in their own
118 fields — they describe the phase body, not the audit emit. The entry's
119 own ``timestamp`` field is a fresh ``_now_iso()`` value set at call
120 time so the audit log retains a faithful emission ordering.
121 """
122 entry: dict[str, Any] = {
123 "event_type": EVENT_TYPE_PHASE,
124 "mission_session_id": session_id,
125 "iteration_index": iteration_index,
126 "phase": phase,
127 "phase_status": status,
128 "phase_started_at": started_at,
129 "phase_ended_at": ended_at,
130 "timestamp": _now_iso(),
131 }
132 if error_message:
133 # Match the 200-char truncation that ``_build_audit_entry`` applies
134 # to its own ``error`` field so phase errors don't blow up the log
135 # line on a long traceback summary.
136 entry["error_message"] = error_message[:200]
137 _emit(entry)
140def emit_verdict_event(
141 session_id: str,
142 iteration_index: int,
143 verdict: str,
144 verdict_reason: str,
145 revision_rationale: str | None = None,
146) -> None:
147 """Emit one ``mission_verdict_event`` audit entry.
149 Called by ``MissionEngine`` once per Decide_Phase outcome. The
150 ``revision_rationale`` is meaningful only on the ``adjust`` verdict
151 (the rationale describes why the next iteration is being asked to
152 revise the strategy), but this helper records it whenever the caller
153 supplies a non-empty string — the engine decides when to populate
154 it. This keeps the helper a pure formatter.
155 """
156 entry: dict[str, Any] = {
157 "event_type": EVENT_TYPE_VERDICT,
158 "mission_session_id": session_id,
159 "iteration_index": iteration_index,
160 "verdict": verdict,
161 "verdict_reason": verdict_reason,
162 "timestamp": _now_iso(),
163 }
164 if revision_rationale:
165 entry["revision_rationale"] = revision_rationale
166 _emit(entry)
169def emit_sampling_event(
170 session_id: str,
171 iteration_index_or_purpose: int | str | None,
172 sampling_purpose: str,
173 sampling_status: str,
174 sampling_backend: str,
175 sampling_model_id: str | None = None,
176 model_output_bytes: int | None = None,
177 validation_error: str | None = None,
178 input_tokens: int | None = None,
179 output_tokens: int | None = None,
180) -> None:
181 """Emit one ``mission_sampling_event`` audit entry.
183 The second positional argument carries the iteration index when the
184 sampling call happens inside the loop body, or a string / ``None`` for
185 out-of-loop calls (e.g. final-report ``lessons`` fill-in). The helper
186 routes the value to the right field:
188 * ``int`` → ``iteration_index`` (matches the design's audit-fields
189 table, which lists ``iteration_index`` as present on sampling
190 events that occur during iterations).
191 * non-empty ``str`` → ``sampling_context`` (an out-of-loop label).
192 * ``None`` or empty string → neither field is recorded.
194 ``sampling_model_id`` and ``model_output_bytes`` are recorded only when
195 the sampler actually produced output (typically ``sampling_status="used"``).
196 ``validation_error`` is recorded only when present (typically
197 ``sampling_status="rejected"``). The conditional emission keeps the
198 audit entry from carrying empty / null fields that downstream consumers
199 would otherwise have to filter out.
200 """
201 entry: dict[str, Any] = {
202 "event_type": EVENT_TYPE_SAMPLING,
203 "mission_session_id": session_id,
204 "sampling_purpose": sampling_purpose,
205 "sampling_status": sampling_status,
206 "sampling_backend": sampling_backend,
207 "timestamp": _now_iso(),
208 }
210 # Route the iteration-or-purpose argument. ``int`` → numeric
211 # ``iteration_index``; ``str`` (non-empty) → ``sampling_context``;
212 # ``None`` and empty strings → omitted. ``bool`` is excluded
213 # explicitly because it is a subclass of ``int`` in Python and would
214 # otherwise be silently recorded as ``iteration_index=True``.
215 if isinstance(iteration_index_or_purpose, int) and not isinstance(
216 iteration_index_or_purpose, bool
217 ):
218 entry["iteration_index"] = iteration_index_or_purpose
219 elif isinstance(iteration_index_or_purpose, str) and iteration_index_or_purpose:
220 entry["sampling_context"] = iteration_index_or_purpose
222 if sampling_model_id:
223 entry["sampling_model_id"] = sampling_model_id
224 if model_output_bytes is not None:
225 entry["model_output_bytes"] = model_output_bytes
226 if validation_error:
227 entry["validation_error"] = validation_error[:200]
228 if input_tokens is not None:
229 entry["input_tokens"] = input_tokens
230 if output_tokens is not None:
231 entry["output_tokens"] = output_tokens
233 _emit(entry)
236def emit_script_call_event(
237 session_id: str,
238 iteration_index: int,
239 tool_name: str,
240 status: str,
241 duration_ms: int,
242 error_message: str | None = None,
243) -> None:
244 """Emit one ``mission_script_call_event`` audit entry.
246 Called by the in-script tool wrapper after each invocation of an
247 operator-allowlisted tool from inside a Mission script. The
248 underlying tool call already produced its own ``@audit_logged``
249 entry through the registered tool function; this helper layers a
250 second, distinct audit row tagged ``via_script=True`` so consumers
251 can tell at a glance which calls were driven from a script versus
252 a direct ``tool_calls`` strategy.
254 The ``status`` argument carries the call's terminal state (``ok``
255 / ``failed`` / ``skipped_not_allowed``) and ``duration_ms`` mirrors
256 the per-call timing the wrapper records on its own
257 ``script_call_log`` entries. ``error_message`` is recorded only
258 when supplied and is truncated to 200 characters to match the
259 existing convention in :func:`emit_phase_event` and
260 :func:`emit_sampling_event`.
261 """
262 entry: dict[str, Any] = {
263 "event_type": EVENT_TYPE_SCRIPT_CALL,
264 "via_script": True,
265 "mission_session_id": session_id,
266 "iteration_index": iteration_index,
267 "tool_name": tool_name,
268 "tool_status": status,
269 "duration_ms": duration_ms,
270 "timestamp": _now_iso(),
271 }
272 if error_message:
273 entry["error_message"] = error_message[:200]
274 _emit(entry)
277def emit_child_lifecycle_event(
278 parent_session_id: str,
279 child_session_id: str | None,
280 slot: str,
281 action: str,
282 *,
283 reason: str | None = None,
284 final_status: str | None = None,
285) -> None:
286 """Emit one ``mission_child_lifecycle_event`` audit entry.
288 Called by the swarm runner on every supervised-slot transition:
289 ``spawned``, ``respawned``, ``terminal`` (the child session reached a
290 terminal status on its own), ``aborted`` (the supervisor cascade or
291 ``child_abort`` ended it), and ``respawn_denied`` (the restart policy
292 wanted a replacement but spawn admission refused — ``reason`` carries
293 the admission token).
295 The entry's ``mission_session_id`` is the **parent** (orchestrator)
296 session id so ``entries_for(parent_id)`` reconstructs the fleet's
297 lifecycle history in one filtered read; the child's own engine events
298 keep carrying the child's session id as they always have. The child
299 id rides in its own field.
300 """
301 entry: dict[str, Any] = {
302 "event_type": EVENT_TYPE_CHILD_LIFECYCLE,
303 "mission_session_id": parent_session_id,
304 "child_session_id": child_session_id,
305 "slot": slot,
306 "action": action,
307 "timestamp": _now_iso(),
308 }
309 if reason:
310 entry["reason"] = reason
311 if final_status:
312 entry["final_status"] = final_status
313 _emit(entry)
316__all__ = [
317 "EVENT_TYPE_CHILD_LIFECYCLE",
318 "EVENT_TYPE_PHASE",
319 "EVENT_TYPE_SAMPLING",
320 "EVENT_TYPE_SCRIPT_CALL",
321 "EVENT_TYPE_VERDICT",
322 "MissionAuditCollectorHandler",
323 "emit_child_lifecycle_event",
324 "emit_phase_event",
325 "emit_sampling_event",
326 "emit_script_call_event",
327 "emit_verdict_event",
328 "get_collector",
329 "install_collector",
330 "replay_audit_entries",
331]
334# ---------------------------------------------------------------------------
335# Audit-replay helper
336# ---------------------------------------------------------------------------
339# Default cap on the in-process collector ring buffer. 5000 entries is
340# big enough to cover a session that runs to its iteration budget
341# (six entries per iteration × ~800 iterations) but small enough that
342# a long-running process cannot OOM through the audit channel.
343_DEFAULT_COLLECTOR_CAPACITY = 5000
346class MissionAuditCollectorHandler(logging.Handler):
347 """Bounded ring-buffer logging handler that captures Mission audit JSON.
349 Attached to the shared ``gco.mcp.audit`` logger by
350 :func:`install_collector` so the
351 ``mission://sessions/{session_id}/audit-replay`` resource has a
352 source of phase / verdict entries to feed
353 :func:`replay_audit_entries`. The handler filters by
354 ``event_type`` so non-Mission audit emitters (the standard MCP
355 tool-invocation decorator, the startup-log helper) do not pollute
356 the buffer.
358 Bounded via :class:`collections.deque(maxlen=N)` so a long-running
359 process never grows the buffer without bound. Operators who want a
360 larger or smaller window can construct the handler explicitly with
361 ``capacity=`` or call :func:`install_collector(capacity=...)`.
362 """
364 _MISSION_EVENT_TYPES = frozenset(
365 {
366 EVENT_TYPE_PHASE,
367 EVENT_TYPE_VERDICT,
368 EVENT_TYPE_SAMPLING,
369 EVENT_TYPE_SCRIPT_CALL,
370 EVENT_TYPE_CHILD_LIFECYCLE,
371 }
372 )
374 def __init__(self, capacity: int = _DEFAULT_COLLECTOR_CAPACITY) -> None:
375 super().__init__(level=logging.INFO)
376 self._buffer: deque[dict[str, Any]] = deque(maxlen=capacity)
378 def emit(self, record: logging.LogRecord) -> None:
379 """Capture Mission audit JSON entries into the ring buffer."""
380 try:
381 payload = json.loads(record.getMessage())
382 except TypeError, ValueError:
383 return
384 if not isinstance(payload, dict):
385 return
386 if payload.get("event_type") not in self._MISSION_EVENT_TYPES:
387 return
388 self._buffer.append(payload)
390 def entries_for(self, session_id: str) -> list[dict[str, Any]]:
391 """Return a list copy of every captured entry for ``session_id``."""
392 return [dict(e) for e in list(self._buffer) if e.get("mission_session_id") == session_id]
394 def clear(self) -> None:
395 """Drop every captured entry. Useful for test isolation."""
396 self._buffer.clear()
399# Module-level collector. ``None`` until :func:`install_collector` is
400# called — the resources/__init__.py wiring installs it once at import
401# time so every Mission audit entry the engine emits during the
402# process lifetime is reachable from the audit-replay resource.
403_COLLECTOR: MissionAuditCollectorHandler | None = None
406def install_collector(
407 capacity: int = _DEFAULT_COLLECTOR_CAPACITY,
408) -> MissionAuditCollectorHandler:
409 """Attach a :class:`MissionAuditCollectorHandler` to the audit logger.
411 Idempotent: a second call with the same parameters returns the
412 existing handler. The function exists so test fixtures can clear
413 and re-attach the handler between cases without leaking captured
414 entries across the boundary.
416 Logger level boost. Python's stdlib ``logging`` defaults the root
417 threshold to ``WARNING``, which means an unconfigured caller that
418 never calls ``logging.basicConfig(level=logging.INFO)`` would
419 silently drop every ``audit_logger.info(...)`` call before it
420 reaches a handler — including this collector. The
421 ``mission://sessions/{id}/audit-replay`` resource needs entries
422 to flow regardless of the host's logging setup, so we floor the
423 logger's level at ``INFO`` here. Hosts that have already set a
424 finer threshold (e.g. ``DEBUG``) keep theirs; only the
425 "unconfigured" case is repaired.
426 """
427 global _COLLECTOR
428 if _COLLECTOR is None:
429 _COLLECTOR = MissionAuditCollectorHandler(capacity=capacity)
430 audit_logger.addHandler(_COLLECTOR)
431 # Floor at INFO so audit_logger.info() entries reach the handler
432 # even when the host has not configured logging at all. We never
433 # *raise* the threshold — a host that explicitly set DEBUG keeps
434 # DEBUG.
435 if audit_logger.level == logging.NOTSET or audit_logger.level > logging.INFO:
436 audit_logger.setLevel(logging.INFO)
437 return _COLLECTOR
440def get_collector() -> MissionAuditCollectorHandler | None:
441 """Return the installed collector or ``None`` when nothing is attached."""
442 return _COLLECTOR
445def replay_audit_entries(
446 session_id: str,
447 entries: list[dict[str, Any]],
448) -> list[dict[str, Any]]:
449 """Reconstruct iteration history from a stream of Mission audit entries.
451 Pure function. Each iteration produces five
452 ``mission_phase_event`` entries (one per phase) plus one
453 ``mission_verdict_event`` entry that closes it out, in emission
454 order. This walker filters ``entries`` to the events whose
455 ``mission_session_id`` matches ``session_id``, accumulates phase
456 events into the active iteration's ``phases`` list, and stamps
457 the verdict + reason from the matching verdict event before
458 appending the completed record.
460 Returns a list of dicts shaped like
461 ``{"iteration_index": int, "phases": [{"phase", "status",
462 "started_at", "ended_at", "error_message"}, ...], "verdict": str
463 | None, "verdict_reason": str | None, "revision_rationale": str
464 | None}``. The shape is intentionally narrow — it covers only
465 the fields the audit stream is expected to fully describe, not
466 the strategy / observation / criteria-evaluation fields the
467 engine persists separately to the session backend.
469 A phase event whose ``iteration_index`` jumps ahead of the
470 active iteration before its verdict event has landed flushes
471 the active iteration with ``verdict=None`` / ``verdict_reason
472 =None`` so a malformed audit stream surfaces as a visible
473 sentinel rather than a silent merge. An iteration with no
474 closing verdict event at end-of-stream is appended the same way.
475 """
476 matching = [
477 e for e in entries if isinstance(e, dict) and e.get("mission_session_id") == session_id
478 ]
480 iterations: list[dict[str, Any]] = []
481 current_index: int | None = None
482 current_phases: list[dict[str, Any]] = []
484 def _flush_current(verdict: str | None, reason: str | None, rationale: str | None) -> None:
485 """Append the active iteration to the result list."""
486 if current_index is None:
487 return
488 iterations.append(
489 {
490 "iteration_index": current_index,
491 "phases": list(current_phases),
492 "verdict": verdict,
493 "verdict_reason": reason,
494 "revision_rationale": rationale,
495 }
496 )
498 for entry in matching:
499 event_type = entry.get("event_type")
500 iteration_index = entry.get("iteration_index")
502 if event_type == EVENT_TYPE_PHASE:
503 if current_index is not None and current_index != iteration_index:
504 # New iteration arrived before the prior closed —
505 # flush the prior with sentinel verdict / reason so
506 # the caller can see the orphaned phases.
507 _flush_current(None, None, None)
508 current_phases = []
509 current_index = iteration_index
510 phase_record: dict[str, Any] = {
511 "phase": entry.get("phase"),
512 "status": entry.get("phase_status"),
513 "started_at": entry.get("phase_started_at"),
514 "ended_at": entry.get("phase_ended_at"),
515 }
516 if "error_message" in entry:
517 phase_record["error_message"] = entry["error_message"]
518 current_phases.append(phase_record)
519 elif event_type == EVENT_TYPE_VERDICT:
520 # ``current_index`` may legitimately be ``None`` when a
521 # verdict-only iteration arrives (e.g. a synthetic
522 # ``cadence_skip``); in that case stamp the iteration
523 # index from the verdict event itself.
524 if current_index is None:
525 current_index = iteration_index
526 _flush_current(
527 entry.get("verdict"),
528 entry.get("verdict_reason"),
529 entry.get("revision_rationale"),
530 )
531 current_index = None
532 current_phases = []
534 # Stream ended mid-iteration — flush the unclosed iteration with
535 # null verdict so the caller sees the partial record.
536 if current_index is not None:
537 _flush_current(None, None, None)
539 return iterations