Coverage for gco_mcp / resources / mission.py: 100.00%

1"""Mission session resources (``mission://sessions/{session_id}``, ``.../report``, ``.../audit-replay``). 

2 

3[gated by GCO_ENABLE_MISSION] 

4 

5Three resource templates that surface live Mission_Session state, the 

6durable Final_Report artifact, and the in-process audit trail through the 

7FastMCP resource layer: 

8 

9* ``mission://sessions/{session_id}`` — returns the JSON-serialised live 

10 session payload. Returns a JSON error envelope when the session is 

11 unknown so that tool-only clients (which call the synthetic 

12 ``read_resource`` tool produced by the Resources As Tools transform) 

13 receive a stable string body rather than a transport-level error. 

14* ``mission://sessions/{session_id}/report`` — returns the Final_Report 

15 JSON for a terminal session. Raises a not-found error when the 

16 session is missing, not yet terminal, or its report has not been 

17 written; FastMCP maps that exception to the MCP ``-32002 Resource 

18 not found`` code on the wire. 

19* ``mission://sessions/{session_id}/audit-replay`` — reconstructs the 

20 session's iteration history from the in-process audit collector 

21 (installed at registration time so the buffer always exists). 

22 

23All three templates are gated by :data:`feature_flags.FLAG_MISSION` at 

24registration time. When the flag is unset, :func:`register` is a no-op 

25so importing this module from ``resources/__init__.py`` is always safe. 

26""" 

27 

28from __future__ import annotations 

29 

30import json 

31import sys 

32from collections.abc import Mapping 

33from pathlib import Path 

34from typing import Any 

35 

36from feature_flags import FLAG_MISSION, is_enabled 

37 

38# Mission package lives under ``gco_mcp/mission/``; the path-injection 

39# pattern matches the rest of the MCP module surface so ``import 

40# mission.*`` resolves without making the ``mcp`` directory a package. 

41sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) 

42 

43 

44def _strip_private_fields(session: Mapping[str, Any]) -> dict[str, Any]: 

45 """Return a JSON-safe copy of ``session`` with private criterion keys dropped. 

46 

47 Thin alias over :func:`mission.validation.strip_private_fields` — 

48 the canonical implementation lives next to ``validate_criteria`` 

49 (which creates the ``_parsed_ast`` keys) so a single function 

50 governs the JSON-safety contract across CLI, MCP tools, and MCP 

51 resources. 

52 """ 

53 from mission.validation import strip_private_fields 

54 

55 return strip_private_fields(session) 

56 

57 

58def _make_not_found(message: str) -> Exception: 

59 """Construct the best available "not found" exception for a missing resource. 

60 

61 Prefers :class:`fastmcp.exceptions.NotFoundError` because the 

62 FastMCP error-handling middleware maps it (along with 

63 :class:`KeyError` and :class:`FileNotFoundError`) to MCP error code 

64 ``-32002`` when the method namespace is ``resources/``. Falls back 

65 to :class:`fastmcp.exceptions.ResourceError` and then to 

66 :class:`KeyError` so the resource layer still surfaces a structured 

67 not-found regardless of the exact FastMCP build in use. 

68 """ 

69 try: 

70 from fastmcp.exceptions import NotFoundError 

71 

72 return NotFoundError(message) 

73 except ImportError: 

74 pass 

75 try: 

76 from fastmcp.exceptions import ResourceError 

77 

78 return ResourceError(message) 

79 except ImportError: 

80 pass 

81 return KeyError(message) 

82 

83 

84def _session_resource(session_id: str) -> str: 

85 """Return the live JSON for the Mission session ``session_id``. 

86 

87 Returns a JSON error envelope (``{"error": "session_not_found", 

88 "session_id": ...}``) when no record exists rather than raising, 

89 because the synthetic ``read_resource`` tool from the Resources As 

90 Tools transform expects a string body. 

91 """ 

92 from mission.state import get_backend 

93 

94 backend = get_backend() 

95 session = backend.load_session(session_id) 

96 if session is None: 

97 return json.dumps({"error": "session_not_found", "session_id": session_id}) 

98 return json.dumps(_strip_private_fields(session), default=str) 

99 

100 

101def _session_report_resource(session_id: str) -> str: 

102 """Return the persisted Final_Report JSON for a terminal session. 

103 

104 Raises a not-found error (mapped by the FastMCP error-handling 

105 middleware to MCP code ``-32002``) when the session is missing, 

106 when it is not yet in a terminal state, or when the matching 

107 report payload has not been written. The :class:`FilesystemBackend` 

108 stores reports as sibling ``<session_id>.report.json`` files; other 

109 backends embed the report under ``session["final_report"]``. 

110 """ 

111 from mission.state import FilesystemBackend, get_backend 

112 from mission.types import TERMINAL_STATES 

113 

114 backend = get_backend() 

115 session = backend.load_session(session_id) 

116 if session is None: 

117 raise _make_not_found(f"Mission session {session_id!r} not found") 

118 

119 status = session.get("status") 

120 if status not in TERMINAL_STATES: 

121 raise _make_not_found(f"Mission session {session_id!r} is not terminal (status={status!r})") 

122 

123 # ``FilesystemBackend`` persists the report next to the session 

124 # JSON. Read it back verbatim so the on-disk artifact is the 

125 # authoritative payload. 

126 if isinstance(backend, FilesystemBackend): 

127 report_path = backend.root / f"{session_id}.report.json" 

128 try: 

129 return report_path.read_text(encoding="utf-8") 

130 except FileNotFoundError as err: 

131 raise _make_not_found( 

132 f"Mission session {session_id!r} terminal but report not found" 

133 ) from err 

134 

135 # Other backends (today, the DynamoDB stub) embed the report on 

136 # the session itself under ``final_report``. 

137 report = session.get("final_report") 

138 if report is None: 

139 raise _make_not_found(f"Mission session {session_id!r} terminal but report not found") 

140 return json.dumps(report, default=str) 

141 

142 

143def _session_audit_replay_resource(session_id: str) -> str: 

144 """Return the iteration history reconstructed from in-process audit entries. 

145 

146 Reads from the :class:`mission.audit.MissionAuditCollectorHandler` 

147 ring buffer attached at server start, projects through 

148 :func:`mission.audit.replay_audit_entries`, and returns 

149 ``{"session_id": session_id, "iterations": [...], "note": 

150 "Reconstructed from in-process audit handler"}`` as JSON. 

151 

152 Returns ``iterations: []`` when the session is unknown to the 

153 handler — does NOT 404. The motivation: a Mission session that 

154 finished before the resource was first read still has a valid 

155 entry in the persistent session backend, but its audit entries 

156 may have aged out of the bounded ring buffer. An empty 

157 reconstruction is the honest answer; a 404 would imply the 

158 session never existed. 

159 """ 

160 from mission.audit import get_collector, replay_audit_entries 

161 

162 collector = get_collector() 

163 note = "Reconstructed from in-process audit handler" 

164 if collector is None: 

165 # No handler attached (e.g. a CLI process that imported the 

166 # resources package without the install hook firing). Empty 

167 # reconstruction. 

168 return json.dumps( 

169 {"session_id": session_id, "iterations": [], "note": note}, 

170 default=str, 

171 ) 

172 

173 entries = collector.entries_for(session_id) 

174 iterations = replay_audit_entries(session_id, entries) 

175 return json.dumps( 

176 {"session_id": session_id, "iterations": iterations, "note": note}, 

177 default=str, 

178 ) 

179 

180 

181def register(mcp_instance: Any) -> None: 

182 """Register Mission resource templates against the shared MCP server. 

183 

184 Gated by :data:`feature_flags.FLAG_MISSION`: when the flag is unset 

185 this function is a no-op so importing this module from 

186 :mod:`resources` stays side-effect-free. With the flag set, three 

187 templates are registered and become reachable via the synthetic 

188 ``read_resource`` tool from the Resources As Tools transform. 

189 """ 

190 if not is_enabled(FLAG_MISSION): 

191 return 

192 # Install the in-process collector when the resource module is 

193 # registered so the audit-replay path always has a buffer to 

194 # read from. Idempotent — call-safe across reloads. 

195 from mission.audit import install_collector 

196 

197 install_collector() 

198 

199 mcp_instance.resource("mission://sessions/{session_id}")(_session_resource) 

200 mcp_instance.resource("mission://sessions/{session_id}/report")(_session_report_resource) 

201 mcp_instance.resource("mission://sessions/{session_id}/audit-replay")( 

202 _session_audit_replay_resource 

203 )