Read snapshots from where the gateway puts them

run_tests.py

@@ -8,8 +8,9 @@
 import sys
 
 # Remove ROS 2 paths before importing pytest
-sys.path = [p for p in sys.path if '/opt/ros' not in p]
+sys.path = [p for p in sys.path if "/opt/ros" not in p]
 
 if __name__ == "__main__":
     import pytest
+
     sys.exit(pytest.main(sys.argv[1:] or ["-v", "tests/"]))

src/ros2_medkit_mcp/client.py

@@ -814,18 +814,6 @@ async def list_all_faults(
             return await self._raw_get_items("/faults", params)
         return _extract_items(await self._call(faults.list_all_faults.asyncio))
 
-    async def get_fault_snapshots(
-        self, entity_id: str, fault_code: str, entity_type: str = "components"
-    ) -> dict[str, Any]:
-        return await self._raw_request(
-            "GET",
-            f"/{quote(entity_type, safe='')}/{quote(entity_id, safe='')}"
-            f"/faults/{quote(fault_code, safe='')}/snapshots",
-        )
-
-    async def get_system_fault_snapshots(self, fault_code: str) -> dict[str, Any]:
-        return await self._raw_request("GET", f"/faults/{quote(fault_code, safe='')}/snapshots")
-
     # ==================== Data ====================
 
     async def get_component_data(

src/ros2_medkit_mcp/mcp_app.py

@@ -49,14 +49,14 @@
     ExecuteScriptArgs,
     ExecuteUpdateArgs,
     ExecutionArgs,
-    ExtendedDataRecords,
     ExtendLockArgs,
     FaultGetArgs,
     FaultItem,
     FaultsListArgs,
-    FaultSnapshotsArgs,
+    FaultStatusDetail,
     FreezeFrameSnapshot,
     FunctionIdArgs,
+    GatewaySnapshot,
     GetConfigurationArgs,
     GetCyclicSubArgs,
     GetLockArgs,
@@ -84,7 +84,6 @@
     SetLogConfigurationArgs,
     SubareasArgs,
     SubcomponentsArgs,
-    SystemFaultSnapshotsArgs,
     ToolResult,
     UpdateCyclicSubArgs,
     UpdateExecutionArgs,
@@ -170,12 +169,18 @@ def format_fault_item(item: FaultItem) -> str:
     lines = [f"Fault: {item.code}"]
     if item.fault_name:
         lines[0] += f" - {item.fault_name}"
-    if item.severity:
-        lines.append(f"  Severity: {item.severity}")
+    if item.severity is not None:
+        # The gateway sends a number; name it so the model does not have to
+        # guess the scale. Unknown values print as-is.
+        labels = {0: "INFO", 1: "WARNING", 2: "ERROR", 3: "CRITICAL"}
+        label = labels.get(item.severity) if isinstance(item.severity, int) else None
+        lines.append(f"  Severity: {item.severity}" + (f" ({label})" if label else ""))
     if item.status:
-        lines.append(
-            f"  Status: {item.status.value if hasattr(item.status, 'value') else item.status}"
-        )
+        if isinstance(item.status, FaultStatusDetail):
+            status_text = item.status.aggregated_status or "unknown"
+        else:
+            status_text = item.status.value if hasattr(item.status, "value") else str(item.status)
+        lines.append(f"  Status: {status_text}")
     if item.is_confirmed is not None:
         lines.append(f"  Confirmed: {item.is_confirmed}")
     if item.is_current is not None:
@@ -250,6 +255,44 @@ def format_snapshot(snapshot: FreezeFrameSnapshot | RosbagSnapshot) -> str:
     return "\n".join(lines)
 
 
+def format_gateway_snapshot(snapshot: GatewaySnapshot) -> str:
+    """Format one environment_data.snapshots entry for display.
+
+    Args:
+        snapshot: A freeze frame or rosbag entry as the gateway sends it.
+
+    Returns:
+        Formatted string describing the snapshot.
+    """
+    lines = [f"  Snapshot: {snapshot.name or snapshot.type}"]
+    if snapshot.captured_at:
+        lines.append(f"    Captured At: {snapshot.captured_at}")
+
+    if snapshot.type == "rosbag":
+        # The URI carries the recording id, which is what distinguishes one
+        # occurrence's black box from another's. A rosbag entry without one is not
+        # something a client can fetch, so say that rather than print "None".
+        if snapshot.bulk_data_uri:
+            lines.append(f"    Download URI: {snapshot.bulk_data_uri}")
+        else:
+            lines.append("    Download URI: unavailable")
+        # `is not None`, not truthiness: a zero-byte recording is a real answer and
+        # hiding its size reads as "the gateway did not report one".
+        if snapshot.size_bytes is not None:
+            lines.append(f"    File Size: {snapshot.size_bytes / (1024 * 1024):.2f} MB")
+        if snapshot.duration_sec is not None:
+            lines.append(f"    Duration: {snapshot.duration_sec:.2f}s")
+        if snapshot.format:
+            lines.append(f"    Format: {snapshot.format}")
+    elif snapshot.data is not None:
+        # JSON, not Python repr: the payload is often a whole message body
+        # (Twist, Odometry), and True/None/'single quotes' would teach the
+        # model a format nothing else speaks.
+        lines.append(f"    Data: {json.dumps(snapshot.data, default=str)}")
+
+    return "\n".join(lines)
+
+
 def format_environment_data(env_data: EnvironmentData) -> str:
     """Format environment data for LLM readability.
 
@@ -261,18 +304,31 @@ def format_environment_data(env_data: EnvironmentData) -> str:
     """
     lines = ["\nEnvironment Data:"]
 
-    if env_data.extended_data_records:
-        records = env_data.extended_data_records
-
-        if records.freeze_frame_snapshots:
-            lines.append(f"  Freeze Frame Snapshots ({len(records.freeze_frame_snapshots)}):")
-            for snap in records.freeze_frame_snapshots:
-                lines.append(format_snapshot(snap))
-
-        if records.rosbag_snapshots:
-            lines.append(f"  Rosbag Snapshots ({len(records.rosbag_snapshots)}):")
-            for snap in records.rosbag_snapshots:
-                lines.append(format_snapshot(snap))
+    records = env_data.extended_data_records
+    if records and (records.first_occurrence or records.last_occurrence):
+        if records.first_occurrence:
+            lines.append(f"  First Occurrence: {records.first_occurrence}")
+        if records.last_occurrence:
+            lines.append(f"  Last Occurrence: {records.last_occurrence}")
+
+    # Read from environment_data.snapshots, which is where the gateway actually
+    # puts them, discriminated by `type`. The freezeFrameSnapshots /
+    # rosbagSnapshots containers under extended_data_records are never populated,
+    # so everything below used to render nothing at all.
+    freeze_frames = [s for s in env_data.snapshots if s.type == "freeze_frame"]
+    recordings = [s for s in env_data.snapshots if s.type == "rosbag"]
+
+    if freeze_frames:
+        lines.append(f"  Freeze Frame Snapshots ({len(freeze_frames)}):")
+        for snap in freeze_frames:
+            lines.append(format_gateway_snapshot(snap))
+
+    if recordings:
+        # Plural on purpose: since ros2_medkit#620 a fault keeps one black box
+        # per occurrence, and each is downloaded by its own URI.
+        lines.append(f"  Rosbag Recordings ({len(recordings)}):")
+        for snap in recordings:
+            lines.append(format_gateway_snapshot(snap))
 
     return "\n".join(lines)
 
@@ -316,41 +372,6 @@ def format_fault_response(fault_data: dict[str, Any]) -> list[TextContent]:
     return [TextContent(type="text", text="\n".join(lines))]
 
 
-def format_snapshots_response(snapshots_data: dict[str, Any]) -> list[TextContent]:
-    """Format a snapshots response for LLM readability.
-
-    Args:
-        snapshots_data: Snapshots response dictionary from the API.
-
-    Returns:
-        Formatted TextContent list.
-    """
-    lines = ["Diagnostic Snapshots:"]
-
-    # Try to validate as ExtendedDataRecords
-    try:
-        records = ExtendedDataRecords.model_validate(snapshots_data)
-
-        if records.freeze_frame_snapshots:
-            lines.append(f"\nFreeze Frame Snapshots ({len(records.freeze_frame_snapshots)}):")
-            for snap in records.freeze_frame_snapshots:
-                lines.append(format_snapshot(snap))
-
-        if records.rosbag_snapshots:
-            lines.append(f"\nRosbag Snapshots ({len(records.rosbag_snapshots)}):")
-            for snap in records.rosbag_snapshots:
-                lines.append(format_snapshot(snap))
-
-        if not records.freeze_frame_snapshots and not records.rosbag_snapshots:
-            lines.append("  No snapshots available.")
-
-    except Exception:
-        # Fallback to raw JSON
-        lines.append(json.dumps(snapshots_data, indent=2, default=str))
-
-    return [TextContent(type="text", text="\n".join(lines))]
-
-
 # ==================== Bulk Data Formatting ====================
 
 
@@ -521,22 +542,14 @@ async def download_rosbags_for_fault(
             )
         ]
 
-    # Get extended data records
-    records = env_data.get("extendedDataRecords") or env_data.get("extended_data_records")
-    if not records:
-        return [
-            TextContent(
-                type="text",
-                text=f"No snapshot data found for fault {fault_code}",
-            )
-        ]
-
-    # Get rosbag snapshots
-    rosbag_snapshots = records.get("rosbagSnapshots") or records.get("rosbag_snapshots", [])
+    # Snapshots live in environment_data.snapshots, discriminated by `type`.
+    # The rosbagSnapshots / freezeFrameSnapshots containers under
+    # extended_data_records are never populated by the gateway, so reading them
+    # made this tool answer "no rosbag snapshots" for every fault that had them.
+    snapshots = env_data.get("snapshots") or []
+    rosbag_snapshots = [s for s in snapshots if s.get("type") == "rosbag"]
     if not rosbag_snapshots:
-        freeze_frames = records.get("freezeFrameSnapshots") or records.get(
-            "freeze_frame_snapshots", []
-        )
+        freeze_frames = [s for s in snapshots if s.get("type") == "freeze_frame"]
         if freeze_frames:
             return [
                 TextContent(
@@ -559,7 +572,10 @@ async def download_rosbags_for_fault(
     errors: list[str] = []
 
     for snap in rosbag_snapshots:
-        snap_id = snap.get("snapshotId") or snap.get("snapshot_id", "unknown")
+        # The recording's own name. There is no snapshotId on the wire, and a
+        # fault can hold several recordings, so falling back to the fault code
+        # would give every one of them the same identity in the report below.
+        snap_id = snap.get("name") or "unknown"
         bulk_uri = snap.get("bulkDataUri") or snap.get("bulk_data_uri")
 
         if not bulk_uri:
@@ -646,8 +662,6 @@ async def download_rosbags_for_fault(
     "ros2_medkit_delete_all_configurations": "ros2_medkit_delete_all_configurations",
     "ros2_medkit_all_faults_list": "ros2_medkit_all_faults_list",
     "ros2_medkit_clear_all_faults": "ros2_medkit_clear_all_faults",
-    "ros2_medkit_fault_snapshots": "ros2_medkit_fault_snapshots",
-    "ros2_medkit_system_fault_snapshots": "ros2_medkit_system_fault_snapshots",
     "ros2_medkit_data_categories": "ros2_medkit_data_categories",
     "ros2_medkit_data_groups": "ros2_medkit_data_groups",
     "ros2_medkit_bulkdata_categories": "ros2_medkit_bulkdata_categories",
@@ -731,8 +745,6 @@ async def download_rosbags_for_fault(
     "sovd_delete_all_configurations": "ros2_medkit_delete_all_configurations",
     "sovd_all_faults_list": "ros2_medkit_all_faults_list",
     "sovd_clear_all_faults": "ros2_medkit_clear_all_faults",
-    "sovd_fault_snapshots": "ros2_medkit_fault_snapshots",
-    "sovd_system_fault_snapshots": "ros2_medkit_system_fault_snapshots",
     "sovd_data_categories": "ros2_medkit_data_categories",
     "sovd_data_groups": "ros2_medkit_data_groups",
     "sovd_bulkdata_categories": "ros2_medkit_bulkdata_categories",
@@ -1012,44 +1024,6 @@ async def list_tools() -> list[Tool]:
                     "required": ["entity_id"],
                 },
             ),
-            Tool(
-                name="ros2_medkit_fault_snapshots",
-                description="Get diagnostic snapshots for a specific fault. Contains data captured at fault occurrence time.",
-                inputSchema={
-                    "type": "object",
-                    "properties": {
-                        "entity_id": {
-                            "type": "string",
-                            "description": "The entity identifier",
-                        },
-                        "fault_code": {
-                            "type": "string",
-                            "description": "The fault code",
-                        },
-                        "entity_type": {
-                            "type": "string",
-                            "enum": ["components", "apps", "areas", "functions"],
-                            "description": "Entity type",
-                            "default": "components",
-                        },
-                    },
-                    "required": ["entity_id", "fault_code"],
-                },
-            ),
-            Tool(
-                name="ros2_medkit_system_fault_snapshots",
-                description="Get system-wide diagnostic snapshots for a fault code.",
-                inputSchema={
-                    "type": "object",
-                    "properties": {
-                        "fault_code": {
-                            "type": "string",
-                            "description": "The fault code",
-                        },
-                    },
-                    "required": ["fault_code"],
-                },
-            ),
             Tool(
                 name="ros2_medkit_area_components",
                 description="List all components within a specific area. Use ros2_medkit_areas_list first to discover valid area IDs (e.g., 'perception', 'control', 'diagnostics').",
@@ -2778,18 +2752,6 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
                 result = await client.clear_all_faults(args.entity_id, args.entity_type)
                 return format_json_response(result)
 
-            elif normalized_name == "ros2_medkit_fault_snapshots":
-                args = FaultSnapshotsArgs(**arguments)
-                snapshots = await client.get_fault_snapshots(
-                    args.entity_id, args.fault_code, args.entity_type
-                )
-                return format_snapshots_response(snapshots)
-
-            elif normalized_name == "ros2_medkit_system_fault_snapshots":
-                args = SystemFaultSnapshotsArgs(**arguments)
-                snapshots = await client.get_system_fault_snapshots(args.fault_code)
-                return format_snapshots_response(snapshots)
-
             # ==================== Entity Data ====================
 
             elif normalized_name == "ros2_medkit_entity_data":