fix: address R2 adversarial findings — isolate observability from execution

Critical (P0→fixed):
- Observability errors can no longer discard valid execution results
  (all metric/trace calls wrapped in isolated try/except blocks)
- Import-time crash guard: graceful fallback if prometheus_client missing
- _gauge_incremented flag prevents negative gauge values

Security:
- Strict label sanitization: regex [a-zA-Z0-9_-]{1,32} with unknown fallback
- Health probe errors use opaque 'not available' (no type(e).__name__ leak)
- Health probe metadata removed (no internal skill count exposure)

Reliability:
- Health probes now timeout after 5s via ThreadPoolExecutor
- start_trace() auto-finishes orphaned active trace (re-entrancy safe)
- Span.finish() is idempotent (no double-finish corruption)
- max_spans guard returns sentinel span (no parent pollution)

Tests: 1,828 passed (+1 probe timeout test)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: Brian Krafft

openspace/agents/grounding/execution.py

@@ -13,8 +13,12 @@
 import time
 from typing import Any, Dict, List, Optional
 
-from openspace.observability.metrics import metrics as _metrics
-from openspace.observability.tracing import tracer as _tracer
+try:
+    from openspace.observability.metrics import metrics as _metrics
+    from openspace.observability.tracing import tracer as _tracer
+except ImportError:
+    _metrics = None  # type: ignore[assignment]
+    _tracer = None  # type: ignore[assignment]
 from openspace.prompts import GroundingAgentPrompts
 from openspace.utils.logging import Logger
 
@@ -105,16 +109,21 @@ async def process(agent, context: Dict[str, Any]) -> Dict[str, Any]:
         tools=tools,
     )
 
+    # ── Observability setup (isolated — must not kill execution) ──────
+    _gauge_incremented = False
     try:
-        # ── Observability: trace + metrics inside try to prevent leaks ──
         _instr_hash = hashlib.sha256(instruction.encode()).hexdigest()[:12]
-        trace = _tracer.start_trace(
+        _tracer.start_trace(
             "grounding.process",
             instruction_hash=_instr_hash,
             instruction_length=len(instruction),
         )
         _metrics.execution_in_flight.labels(agent=_agent_label).inc()
+        _gauge_incremented = True
+    except Exception:
+        logger.debug("Observability setup failed", exc_info=True)
 
+    try:
         while current_iteration < max_iterations:
             current_iteration += 1
             logger.info(
@@ -293,33 +302,42 @@ async def process(agent, context: Dict[str, Any]) -> Dict[str, Any]:
             f"{result.get('status')}"
         )
 
-        # ── Observability: record success metrics ────────────────────
-        _elapsed = time.monotonic() - _exec_start
-        _metrics.execution_latency.labels(agent=_agent_label, status="success").observe(_elapsed)
-        _metrics.execution_iterations.labels(agent=_agent_label).observe(current_iteration)
-        _metrics.execution_total.labels(agent=_agent_label, status="success").inc()
-        _metrics.execution_in_flight.labels(agent=_agent_label).dec()
-        root_span = _tracer.current_span()
-        if root_span:
-            root_span.attributes["iterations"] = current_iteration
-            root_span.attributes["status"] = result.get("status", "unknown")
-        _tracer.finish_trace()
+        # ── Observability: record success (isolated — must not discard result)
+        try:
+            _elapsed = time.monotonic() - _exec_start
+            _metrics.execution_latency.labels(agent=_agent_label, status="success").observe(_elapsed)
+            _metrics.execution_iterations.labels(agent=_agent_label).observe(current_iteration)
+            _metrics.execution_total.labels(agent=_agent_label, status="success").inc()
+            if _gauge_incremented:
+                _metrics.execution_in_flight.labels(agent=_agent_label).dec()
+                _gauge_incremented = False
+            root_span = _tracer.current_span()
+            if root_span:
+                root_span.attributes["iterations"] = current_iteration
+                root_span.attributes["status"] = result.get("status", "unknown")
+            _tracer.finish_trace()
+        except Exception:
+            logger.debug("Observability recording failed", exc_info=True)
 
         return result
 
     except Exception as e:
         logger.error(f"Grounding Agent: Execution failed: {e}")
 
-        # ── Observability: record error metrics ──────────────────────
-        _elapsed = time.monotonic() - _exec_start
-        _metrics.execution_latency.labels(agent=_agent_label, status="error").observe(_elapsed)
-        _metrics.execution_total.labels(agent=_agent_label, status="error").inc()
-        _metrics.execution_in_flight.labels(agent=_agent_label).dec()
-        root_span = _tracer.current_span()
-        if root_span:
-            root_span.add_event("error", error_type=type(e).__name__)
-            root_span.status = "error"
-        _tracer.finish_trace()
+        # ── Observability: record error (isolated — must not mask result)
+        try:
+            _elapsed = time.monotonic() - _exec_start
+            _metrics.execution_latency.labels(agent=_agent_label, status="error").observe(_elapsed)
+            _metrics.execution_total.labels(agent=_agent_label, status="error").inc()
+            if _gauge_incremented:
+                _metrics.execution_in_flight.labels(agent=_agent_label).dec()
+            root_span = _tracer.current_span()
+            if root_span:
+                root_span.add_event("error", error_type=type(e).__name__)
+                root_span.status = "error"
+            _tracer.finish_trace()
+        except Exception:
+            logger.debug("Observability recording failed", exc_info=True)
 
         result = {
             "error": type(e).__name__,
@@ -367,16 +385,19 @@ def _build_retrieved_tools_list(
     return retrieved_tools_list
 
 
+import re
+
 # ── Label sanitization ───────────────────────────────────────────────
 
-# Prometheus label values should be bounded. We keep a short allowlist
-# and map everything else to "unknown" to prevent cardinality bombs.
-_LABEL_MAX_LEN = 64
+# Prometheus label values must be bounded and safe. Strict allowlist
+# prevents cardinality bombs and scraper injection.
+_LABEL_MAX_LEN = 32
+_LABEL_SAFE_RE = re.compile(r"[^a-zA-Z0-9_\-]")
 
 
 def _sanitize_label(value: str) -> str:
-    """Clamp a label value to a safe length and replace spaces."""
+    """Sanitize a string for use as a Prometheus label value."""
     if not value:
         return "unknown"
-    clean = value[:_LABEL_MAX_LEN].replace(" ", "_")
-    return clean
+    clean = _LABEL_SAFE_RE.sub("_", value[:_LABEL_MAX_LEN])
+    return clean or "unknown"

openspace/mcp/server.py

@@ -162,17 +162,17 @@ def _probe_skill_store() -> HealthProbe:
                 _probe_skill_store._cached = SkillStore()
             store = _probe_skill_store._cached
             count = len(store.list_skills()) if hasattr(store, "list_skills") else 0
-            return HealthProbe(ok=True, detail=f"{count} skills", metadata={"count": count})
-        except Exception as exc:
-            return HealthProbe(ok=False, detail=f"{type(exc).__name__}")
+            return HealthProbe(ok=True, detail=f"{count} skills")
+        except Exception:
+            return HealthProbe(ok=False, detail="not available")
 
     def _probe_grounding() -> HealthProbe:
         try:
             from openspace.agents.grounding_agent import GroundingAgent
 
             return HealthProbe(ok=True, detail="module loaded")
-        except Exception as exc:
-            return HealthProbe(ok=False, detail=f"{type(exc).__name__}")
+        except Exception:
+            return HealthProbe(ok=False, detail="not available")
 
     def _probe_mcp_tools() -> HealthProbe:
         return HealthProbe(ok=True, detail="tools registered")

openspace/observability/health.py

@@ -19,11 +19,15 @@
 
 from __future__ import annotations
 
+import concurrent.futures
 import time
 from dataclasses import dataclass, field
 from enum import Enum
 from typing import Any, Callable, Dict, List, Optional
 
+# Per-probe timeout in seconds
+_PROBE_TIMEOUT = 5.0
+
 
 class HealthStatus(str, Enum):
     """Overall system health status."""
@@ -79,25 +83,39 @@ def unregister(self, name: str) -> None:
     def probe_names(self) -> List[str]:
         return list(self._probes.keys())
 
-    def check(self) -> Dict[str, Any]:
-        """Run all probes and return structured health status."""
+    def check(self, timeout: float = _PROBE_TIMEOUT) -> Dict[str, Any]:
+        """Run all probes and return structured health status.
+
+        Each probe is run with a timeout guard to prevent a hung probe
+        from stalling the entire health check.
+        """
         results: Dict[str, Dict[str, Any]] = {}
         failed = 0
         total = len(self._probes)
 
         for name, probe_fn in self._probes.items():
             start = time.monotonic()
             try:
-                probe = probe_fn()
+                with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
+                    future = pool.submit(probe_fn)
+                    probe = future.result(timeout=timeout)
                 probe.latency_ms = (time.monotonic() - start) * 1000
                 results[name] = probe.to_dict()
                 if not probe.ok:
                     failed += 1
-            except Exception as exc:
+            except concurrent.futures.TimeoutError:
+                elapsed = (time.monotonic() - start) * 1000
+                results[name] = HealthProbe(
+                    ok=False,
+                    detail="probe timeout",
+                    latency_ms=elapsed,
+                ).to_dict()
+                failed += 1
+            except Exception:
                 elapsed = (time.monotonic() - start) * 1000
                 results[name] = HealthProbe(
                     ok=False,
-                    detail=f"probe error: {type(exc).__name__}",
+                    detail="probe error",
                     latency_ms=elapsed,
                 ).to_dict()
                 failed += 1

openspace/observability/tracing.py

@@ -56,6 +56,8 @@ def duration_ms(self) -> Optional[float]:
         return (self.end_time - self.start_time) * 1000
 
     def finish(self, status: str = "ok") -> None:
+        if self.end_time is not None:
+            return  # Already finished — idempotent
         self.end_time = time.monotonic()
         self.status = status
 
@@ -125,7 +127,15 @@ def __init__(self, max_traces: int = _MAX_TRACES) -> None:
         self._lock = threading.Lock()
 
     def start_trace(self, name: str = "root", **attrs: Any) -> Trace:
-        """Begin a new trace with a root span."""
+        """Begin a new trace with a root span.
+
+        If a trace is already active, it is finished and stored before
+        the new one starts (prevents orphaned traces).
+        """
+        existing = _current_trace.get()
+        if existing is not None:
+            self.finish_trace()
+
         trace = Trace()
         root = Span(
             name=name,
@@ -196,9 +206,15 @@ def __enter__(self) -> Span:
             return self._span
 
         self._parent = _current_span.get()
-        # Guard against unbounded span growth
+        # Guard against unbounded span growth — return a no-op sentinel
         if len(trace.spans) >= _MAX_SPANS_PER_TRACE:
-            return self._parent or trace.spans[0]  # return existing span, don't add
+            self._span = Span(
+                name=self._name,
+                trace_id=trace.trace_id,
+                parent_id=self._parent.span_id if self._parent else None,
+            )
+            # NOT appended to trace.spans — mutations are silently discarded
+            return self._span
         self._span = Span(
             name=self._name,
             trace_id=trace.trace_id,

tests/test_observability.py

@@ -314,7 +314,7 @@ def bad_probe():
         result = agg.check()
         assert result["status"] == "unhealthy"
         assert result["checks"]["flaky"]["ok"] is False
-        assert "ConnectionError" in result["checks"]["flaky"]["detail"]
+        assert "ConnectionError" in result["checks"]["flaky"]["detail"] or "probe error" in result["checks"]["flaky"]["detail"]
 
     def test_uptime_is_positive(self):
         agg = HealthAggregator()
@@ -587,12 +587,31 @@ def test_probe_exception_counted_as_failure(self):
         result = h.check()
         assert result["failed_probes"] == 1
         assert result["checks"]["explode"]["ok"] is False
-        assert "RuntimeError" in result["checks"]["explode"]["detail"]
+        assert "error" in result["checks"]["explode"]["detail"]
 
     def test_unregister_nonexistent_probe_is_noop(self):
         h = HealthAggregator()
         h.unregister("does_not_exist")  # should not raise
 
+    def test_probe_timeout_returns_failure(self):
+        """A probe that hangs beyond timeout should fail, not block."""
+        import threading
+
+        h = HealthAggregator()
+        hang_event = threading.Event()
+
+        def _hanging_probe():
+            hang_event.wait(10)  # blocks for up to 10s
+            return HealthProbe(ok=True)
+
+        h.register("hangs", _hanging_probe)
+        result = h.check(timeout=0.1)  # 100ms timeout
+        hang_event.set()  # unblock the thread
+
+        assert result["failed_probes"] == 1
+        assert result["checks"]["hangs"]["ok"] is False
+        assert "timeout" in result["checks"]["hangs"]["detail"]
+
     def test_finish_trace_with_no_active_trace_returns_none(self):
         t = ExecutionTracer()
         result = t.finish_trace()