Save raw metrics exposition in Phase 0 and mark ddev env tools read-only

ddev/src/ddev/ai/flows/openmetrics/phases/inspect_endpoint.py

@@ -18,6 +18,7 @@
 REQUEST_TIMEOUT_SECONDS = 10.0
 RESPONSE_BODY_LIMIT_BYTES = 10 * 1024 * 1024  # 10 MB
 JSONL_FILENAME_SUFFIX = "_metrics.jsonl"
+EXPOSITION_FILENAME_SUFFIX = "_exposition.txt"
 
 
 class EndpointInspectionError(Exception):
@@ -99,13 +100,25 @@ def _write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
         raise EndpointInspectionError(f"Failed to write metrics catalog at {path}: {e}") from e
 
 
+def _write_exposition(path: Path, body: str) -> None:
+    """Atomically write the verbatim endpoint body, for reuse as a test fixture."""
+    tmp_path = path.with_suffix(path.suffix + ".tmp")
+    try:
+        tmp_path.write_text(body, encoding="utf-8")
+        os.replace(tmp_path, path)
+    except OSError as e:
+        _remove_if_exists(tmp_path)
+        raise EndpointInspectionError(f"Failed to write exposition snapshot at {path}: {e}") from e
+
+
 def _build_memory_text(
     url: str,
     status: int,
     content_type: str,
     exposition_format: str,
     metric_count: int,
     jsonl_path: Path,
+    exposition_path: Path,
 ) -> str:
     """Render the markdown memory file describing the inspected endpoint."""
     lines = [
@@ -117,9 +130,11 @@ def _build_memory_text(
         f"- **Exposition format:** {exposition_format}",
         f"- **Metric families detected:** {metric_count}",
         f"- **Metrics catalog:** {jsonl_path}",
+        f"- **Raw exposition snapshot:** {exposition_path}",
         "",
         "Endpoint is reachable and serves a Prometheus/OpenMetrics-compatible body.",
-        "The full list of metrics with metadata is in the catalog file above.",
+        "The full list of metrics with metadata is in the catalog file above. The raw exposition",
+        "snapshot is the verbatim endpoint body, suitable for use as a test fixture.",
     ]
     return "\n".join(lines)
 
@@ -192,6 +207,11 @@ async def execute(self, context: dict[str, Any]) -> PhaseOutcome:
         jsonl_path = (self._checkpoint_manager.memory_dir / f"{self._phase_id}{JSONL_FILENAME_SUFFIX}").resolve()
         _write_jsonl(jsonl_path, rows)
 
+        exposition_path = (
+            self._checkpoint_manager.memory_dir / f"{self._phase_id}{EXPOSITION_FILENAME_SUFFIX}"
+        ).resolve()
+        _write_exposition(exposition_path, body)
+
         metric_count = len(families)
         memory_text = _build_memory_text(
             url=endpoint_url,
@@ -200,6 +220,7 @@ async def execute(self, context: dict[str, Any]) -> PhaseOutcome:
             exposition_format=exposition_format,
             metric_count=metric_count,
             jsonl_path=jsonl_path,
+            exposition_path=exposition_path,
         )
 
         return PhaseOutcome(
@@ -213,5 +234,6 @@ async def execute(self, context: dict[str, Any]) -> PhaseOutcome:
                 "exposition_format": exposition_format,
                 "metric_count": metric_count,
                 "metrics_jsonl_path": str(jsonl_path),
+                "exposition_path": str(exposition_path),
             },
         )

ddev/src/ddev/ai/tools/registry.py

@@ -92,10 +92,10 @@ class ToolSpec:
     "http_get": ToolSpec("http.http_get", "HttpGetTool", read_only=True),
     "ddev_create": ToolSpec("shell.ddev.create", "DdevCreateTool", read_only=False),
     "ddev_test": ToolSpec("shell.ddev.ddev_test", "DdevTestTool", read_only=False),
-    "ddev_env_show": ToolSpec("shell.ddev.env_show", "DdevEnvShowTool", read_only=False),
+    "ddev_env_show": ToolSpec("shell.ddev.env_show", "DdevEnvShowTool", read_only=True),
     "ddev_env_start": ToolSpec("shell.ddev.env_start", "DdevEnvStartTool", read_only=False),
     "ddev_env_stop": ToolSpec("shell.ddev.env_stop", "DdevEnvStopTool", read_only=False),
-    "ddev_env_test": ToolSpec("shell.ddev.env_test", "DdevEnvTestTool", read_only=False),
+    "ddev_env_test": ToolSpec("shell.ddev.env_test", "DdevEnvTestTool", read_only=True),
     "ddev_release_changelog": ToolSpec("shell.ddev.release_changelog", "DdevReleaseChangelogTool", read_only=False),
     "ddev_validate": ToolSpec("shell.ddev.validate", "DdevValidateTool", read_only=False),
     "spawn_subagent": ToolSpec(

ddev/tests/ai/flows/openmetrics/phases/test_inspect_endpoint.py

@@ -360,20 +360,23 @@ def test_parse_exposition_raises_on_zero_families():
 
 def test_build_memory_text_renders_all_fields(tmp_path):
     jsonl_path = tmp_path / "inspect_endpoint_metrics.jsonl"
+    exposition_path = tmp_path / "inspect_endpoint_exposition.txt"
     text = _build_memory_text(
         url="http://example.test:9100/metrics",
         status=200,
         content_type="text/plain; version=0.0.4",
         exposition_format="prometheus",
         metric_count=2,
         jsonl_path=jsonl_path,
+        exposition_path=exposition_path,
     )
     assert "http://example.test:9100/metrics" in text
     assert "200" in text
     assert "text/plain; version=0.0.4" in text
     assert "prometheus" in text
     assert "2" in text
     assert str(jsonl_path) in text
+    assert str(exposition_path) in text
 
 
 # ---------------------------------------------------------------------------
@@ -589,3 +592,51 @@ async def test_jsonl_failure_propagates_as_phase_failure(flow_dir, message_queue
     blocker.mkdir()
 
     await _assert_phase_fails(phase, mgr, message_queue, error_contains="Failed to write metrics catalog")
+
+
+# ---------------------------------------------------------------------------
+# Raw exposition sidecar — new tests
+# ---------------------------------------------------------------------------
+
+
+async def test_exposition_path_in_checkpoint(flow_dir, message_queue, monkeypatch):
+    _install_mock_transport(monkeypatch, _ok_handler(200, PROMETHEUS_BODY, "text/plain"))
+    phase, mgr = _make_phase(flow_dir, message_queue)
+
+    await phase.process_message(PhaseTrigger(id="start", phase_id=None))
+
+    path_str = mgr.read()[PHASE_ID]["exposition_path"]
+    assert isinstance(path_str, str)
+    assert os.path.isabs(path_str)
+    assert os.path.exists(path_str)
+
+
+async def test_exposition_file_holds_verbatim_body(flow_dir, message_queue, monkeypatch):
+    _install_mock_transport(monkeypatch, _ok_handler(200, PROMETHEUS_BODY, "text/plain"))
+    phase, _mgr = _make_phase(flow_dir, message_queue)
+
+    await phase.process_message(PhaseTrigger(id="start", phase_id=None))
+
+    exposition = (flow_dir / f"{PHASE_ID}_exposition.txt").read_text(encoding="utf-8")
+    assert exposition == PROMETHEUS_BODY
+
+
+async def test_memory_text_includes_exposition_path(flow_dir, message_queue, monkeypatch):
+    _install_mock_transport(monkeypatch, _ok_handler(200, PROMETHEUS_BODY, "text/plain"))
+    phase, mgr = _make_phase(flow_dir, message_queue)
+
+    await phase.process_message(PhaseTrigger(id="start", phase_id=None))
+
+    assert mgr.read()[PHASE_ID]["exposition_path"] in mgr.memory_content(PHASE_ID)
+
+
+async def test_exposition_failure_propagates_as_phase_failure(flow_dir, message_queue, monkeypatch):
+    _install_mock_transport(monkeypatch, _ok_handler(200, PROMETHEUS_BODY, "text/plain"))
+    phase, mgr = _make_phase(flow_dir, message_queue)
+
+    # Block only the exposition temp path (the JSONL is written first and succeeds).
+    blocker = flow_dir / f"{PHASE_ID}_exposition.txt.tmp"
+    blocker.mkdir()
+
+    await _assert_phase_fails(phase, mgr, message_queue, error_contains="Failed to write exposition snapshot")
+    assert not (flow_dir / f"{PHASE_ID}_exposition.txt").exists()