Revert "fix(ci): use sentinel + robust fallback for preamble stripping"

This reverts commit da479a2b92.

omnigent/cli.py

@@ -267,6 +267,8 @@ _LOCAL_DAEMON_ENV_PREFIXES: tuple[str, ...] = (
     "ANTHROPIC_DEFAULT_",
     "AZURE_OPENAI_",
     "DATABRICKS_",
+    "MLFLOW_",
+    "OTEL_",
     "OMNIGENT_",
     "OPENAI_",
 )

omnigent/host/connect.py

@@ -284,7 +284,7 @@ _RUNNER_ENV_ALLOWLIST: frozenset[str] = frozenset(
     }
 )
 # Locale family (``LC_ALL``, ``LC_CTYPE``, …) — allowed by prefix.
-_RUNNER_ENV_ALLOWLIST_PREFIXES: tuple[str, ...] = ("LC_",)
+_RUNNER_ENV_ALLOWLIST_PREFIXES: tuple[str, ...] = ("LC_", "MLFLOW_", "OTEL_")
 
 # Harness credential / endpoint env vars forwarded host→runner when
 # present. These are the names the harnesses themselves resolve —

omnigent/runner/_entry.py

@@ -824,6 +824,20 @@ async def _run_tunnel_from_env() -> None:
     binding_token = _runner_tunnel_binding_token_from_env()
     parent_pid = _runner_parent_pid_from_env()
     runner_id = get_stable_runner_id()
+
+    # Initialize MLflow tracing in the runner process so the
+    # ExecutorAdapter can emit spans for agent turns, tool calls,
+    # and LLM interactions. No-op when OTEL_EXPORTER_OTLP_ENDPOINT
+    # is unset or mlflow is not installed.
+    try:
+        from omnigent.runtime import telemetry
+
+        telemetry.init()
+    except ImportError:
+        _logger.debug("telemetry init skipped in runner (mlflow not installed)")
+    except Exception:  # noqa: BLE001 — best-effort; tracing failure must not crash the runner
+        _logger.debug("telemetry init failed in runner", exc_info=True)
+
     # Reuse the tunnel's token factory for the app's httpx client so the
     # runner resolves Databricks auth once at boot, not twice.
     app = create_app(auth_token_factory=auth_token_factory)

omnigent/runtime/harnesses/_executor_adapter.py

@@ -36,6 +36,7 @@ import asyncio
 import contextlib
 import json
 import logging
+import os
 import secrets
 import uuid
 from collections import deque
@@ -56,6 +57,7 @@ from omnigent.inner.executor import (
     ToolCallRequest,
     TurnComplete,
 )
+from omnigent.inner.tracing import TracingContext, is_tracing_enabled
 from omnigent.runtime.harnesses._scaffold import HarnessApp, PolicyVerdictPayload, TurnContext
 from omnigent.runtime.tool_output import cap_tool_output
 from omnigent.server.schemas import (
@@ -105,6 +107,34 @@ _OBSERVED_TOOL_CALL_STATUS = "in_progress"
 _MCP_TOOL_NAME_PREFIX = "mcp__"
 
 
+def _finalize_trace_status(response_id: str) -> None:
+    """PATCH the trace status to OK on the MLflow server.
+
+    OTLP-ingested traces stay "In progress" because the server has
+    no signal that all spans have arrived. This call explicitly
+    marks the trace as complete after the OTel provider is flushed.
+    """
+    try:
+        from omnigent.runtime.telemetry import trace_id_from_response_id
+
+        trace_id = trace_id_from_response_id(response_id)
+        request_id = f"tr-{trace_id}"
+
+        tracking_uri = os.environ.get("MLFLOW_TRACKING_URI") or os.environ.get(
+            "OTEL_EXPORTER_OTLP_ENDPOINT", ""
+        )
+        if not tracking_uri:
+            return
+        import httpx
+
+        httpx.Client(timeout=5).patch(
+            f"{tracking_uri.rstrip('/')}/api/2.0/mlflow/traces/{request_id}",
+            json={"status": "OK"},
+        ).close()
+    except Exception:
+        _logger.debug("failed to finalize trace status", exc_info=True)
+
+
 def _strip_mcp_tool_prefix(name: str) -> str:
     """
     Strip the Claude SDK MCP tool prefix from a tool name.
@@ -212,6 +242,10 @@ class ExecutorAdapter(HarnessApp):
         # suppress-observed mitigation that introduced the
         # end-of-turn ordering regression this queue resolves.
         self._pending_mcp_call_ids: deque[str] = deque()
+        # Per-session tracing context. Created lazily on the first
+        # turn when tracing is enabled; reused across turns so the
+        # span parent chain stays rooted on the session's executor.
+        self._tracing_ctx: TracingContext | None = None
 
     async def run_turn(self, request: CreateResponseRequest, ctx: TurnContext) -> None:
         """
@@ -301,6 +335,24 @@ class ExecutorAdapter(HarnessApp):
         # previous turn. Clearing makes each turn's correlation
         # window self-contained.
         self._pending_mcp_call_ids.clear()
+
+        # --- Tracing setup ------------------------------------------------
+        # Create a TracingContext per turn when tracing is enabled.
+        # The trace_context_for_response wrapper derives the W3C
+        # trace ID from the response_id so operators can look up
+        # traces by response ID without a mapping table.
+        tracing = is_tracing_enabled()
+        if tracing and self._tracing_ctx is None:
+            self._tracing_ctx = TracingContext()
+        tctx = self._tracing_ctx if tracing else None
+        agent_span = None
+        # Active tool span for correlating ToolCallRequest → ToolCallComplete.
+        _active_tool_span = None
+        _active_tool_parent = None
+
+        user_message = _extract_last_user_message(request.input)
+        # --- End tracing setup --------------------------------------------
+
         # Watcher for mid-turn steering injections. The scaffold
         # routes incoming steering events with
         # ``previous_response_id == ctx.response_id`` onto
@@ -316,29 +368,97 @@ class ExecutorAdapter(HarnessApp):
             name=f"executor-adapter-injection-watch:{ctx.response_id}",
         )
         try:
-            async for event in executor.run_turn(
-                messages=messages,
-                tools=tools,
-                system_prompt=system_prompt,
-                config=config,
-            ):
-                if ctx.cancelled.is_set():
-                    # Cancellation arrived mid-stream — stop emitting
-                    # further events and ask the inner executor to
-                    # interrupt. The scaffold's terminal event handler
-                    # will emit response.cancelled on return.
-                    await executor.interrupt_session(self._session_key)
-                    return
-                self._translate_event(event, ctx)
-                if isinstance(event, TurnComplete):
-                    # Scaffold emits response.completed automatically
-                    # when run_turn returns; nothing more to do.
-                    return
-                if isinstance(event, ExecutorError):
-                    # Re-raise so the scaffold's terminal-event path
-                    # surfaces response.failed with the underlying
-                    # error message.
-                    raise RuntimeError(f"inner executor error: {event.message}")
+            # Wrap the executor loop in the trace context so all
+            # MLflow spans share the response-derived trace ID.
+            # The context manager is built outside the `with` so we
+            # can fall back to nullcontext if the response_id format
+            # doesn't match (e.g. 24-char hex vs expected 32).
+            trace_cm: contextlib.AbstractContextManager[None] = contextlib.nullcontext()
+            if tctx:
+                try:
+                    from omnigent.runtime.telemetry import trace_context_for_response
+
+                    trace_cm = trace_context_for_response(response_id=ctx.response_id)
+                except Exception:
+                    _logger.debug("trace_context_for_response unavailable", exc_info=True)
+            with trace_cm:
+                if tctx is not None:
+                    agent_span = tctx.start_agent_span(
+                        agent_name=request.model or "unknown",
+                        user_message=user_message,
+                        model=request.model_override or request.model,
+                    )
+
+                response_text: str | None = None
+                async for event in executor.run_turn(
+                    messages=messages,
+                    tools=tools,
+                    system_prompt=system_prompt,
+                    config=config,
+                ):
+                    if ctx.cancelled.is_set():
+                        if tctx is not None and agent_span is not None:
+                            from omnigent.runtime.telemetry import record_cancellation
+
+                            record_cancellation(agent_span)
+                            tctx.end_agent_span(agent_span, response=None, status="ERROR")
+                            agent_span = None
+                        await executor.interrupt_session(self._session_key)
+                        return
+                    # --- Tracing: emit spans per event ---
+                    if tctx is not None:
+                        if isinstance(event, ToolCallRequest):
+                            _active_tool_parent = tctx._current_span
+                            _active_tool_span = tctx.start_tool_span(
+                                _strip_mcp_tool_prefix(event.name),
+                                event.args or {},
+                            )
+                        elif isinstance(event, ToolCallComplete):
+                            if _active_tool_span is not None:
+                                tctx.end_tool_span(
+                                    _active_tool_span,
+                                    result=event.result,
+                                    status="ERROR" if event.error else "OK",
+                                    error=event.error,
+                                    duration_ms=event.duration_ms,
+                                    parent_span=_active_tool_parent,
+                                )
+                                _active_tool_span = None
+                                _active_tool_parent = None
+                        elif isinstance(event, TurnComplete):
+                            response_text = event.response
+                            if event.usage is not None:
+                                from omnigent.runtime.telemetry import record_llm_usage
+
+                                # Record usage on the agent span for
+                                # aggregate visibility.
+                                record_llm_usage(agent_span, event.usage)
+                    # --- End tracing ---
+                    self._translate_event(event, ctx)
+                    if isinstance(event, TurnComplete):
+                        if tctx is not None and agent_span is not None:
+                            tctx.end_agent_span(agent_span, response=response_text)
+                            agent_span = None
+                        return
+                    if isinstance(event, ExecutorError):
+                        if tctx is not None and agent_span is not None:
+                            tctx.end_agent_span(
+                                agent_span,
+                                response=None,
+                                status="ERROR",
+                                error=event.message,
+                            )
+                            agent_span = None
+                        raise RuntimeError(f"inner executor error: {event.message}")
+        except BaseException:
+            # End agent span on unhandled exceptions so it's not
+            # left open (which would leak on the OTel provider).
+            if tctx is not None and agent_span is not None:
+                tctx.end_agent_span(
+                    agent_span, response=None, status="ERROR", error="unhandled exception"
+                )
+                agent_span = None
+            raise
         finally:
             # Stop the injection watcher and let it drain so a
             # late ``next_injection`` doesn't fire after we've
@@ -348,6 +468,22 @@ class ExecutorAdapter(HarnessApp):
             injection_watcher.cancel()
             with contextlib.suppress(asyncio.CancelledError):
                 await injection_watcher
+            # Flush the OTel provider and finalize the trace status
+            # on the MLflow server. Without the flush, the
+            # BatchSpanProcessor may not have exported the final
+            # spans. Without the PATCH, the OTLP-ingested trace
+            # stays "In progress" because the server has no
+            # signal that all spans have arrived.
+            if tctx is not None:
+                try:
+                    from opentelemetry import trace as otel_trace
+
+                    provider = otel_trace.get_tracer_provider()
+                    if hasattr(provider, "force_flush"):
+                        provider.force_flush(timeout_millis=5000)
+                except Exception:
+                    pass
+                _finalize_trace_status(ctx.response_id)
             # Clear the per-turn pointers so a stray late callback
             # (e.g. one fired after the SDK's stream closed) sees
             # ``None`` and returns an explicit error rather than
@@ -1241,6 +1377,45 @@ async def _bridge_one_dispatch(
     return {"result": parsed}
 
 
+def _extract_last_user_message(
+    input_value: str | list[dict[str, Any]],
+) -> str:
+    """Extract the last user message text from a request input.
+
+    Handles both conversation-history shape (list of message items
+    with ``role``/``content``) and single-turn shape (plain string
+    or content-block list). Used by tracing to populate the agent
+    span's ``user_message`` input.
+
+    :param input_value: The request's ``input`` field.
+    :returns: The text of the last user message, or empty string.
+    """
+    if isinstance(input_value, str):
+        return input_value
+    # Conversation-history shape: find last user message
+    last_user_text = ""
+    for item in input_value:
+        role = item.get("role")
+        if role == "user":
+            content = item.get("content")
+            if isinstance(content, str):
+                last_user_text = content
+            elif isinstance(content, list):
+                parts = []
+                for block in content:
+                    text = block.get("text")
+                    if isinstance(text, str):
+                        parts.append(text)
+                if parts:
+                    last_user_text = "\n".join(parts)
+        # Single-turn content-block shape (no role key)
+        elif role is None:
+            text = item.get("text")
+            if isinstance(text, str):
+                last_user_text = text
+    return last_user_text
+
+
 def _extract_user_text(
     input_value: str | list[dict[str, Any]],
 ) -> str:

omnigent/runtime/harnesses/_runner.py

@@ -317,6 +317,18 @@ def main(argv: list[str] | None = None) -> None:
         the live process arguments. Tests pass an explicit list.
     """
     args = _parse_args(argv if argv is not None else sys.argv[1:])
+
+    # Initialize MLflow tracing in the harness subprocess so
+    # ExecutorAdapter can emit spans for agent turns, tool calls,
+    # and LLM interactions. No-op when OTEL_EXPORTER_OTLP_ENDPOINT
+    # is unset or mlflow is not installed.
+    try:
+        from omnigent.runtime import telemetry
+
+        telemetry.init()
+    except Exception:
+        pass  # mlflow not installed or init failed; tracing disabled
+
     app = _load_harness_app(args.harness, args.module, args.conversation_id)
     if args.parent_pid is not None:
         _set_pdeathsig()

omnigent/runtime/telemetry.py

@@ -285,10 +285,15 @@ def trace_id_from_response_id(response_id: str) -> str:
     if not response_id.startswith(_RESP_PREFIX):
         raise ValueError(f"Expected {_RESP_PREFIX!r} prefix, got {response_id!r}")
     hex_part = response_id[len(_RESP_PREFIX) :]
-    if len(hex_part) != _HEX_LEN:
+    if len(hex_part) > _HEX_LEN:
         raise ValueError(
-            f"Expected {_HEX_LEN} hex chars after prefix, got {len(hex_part)} in {response_id!r}"
+            f"Expected at most {_HEX_LEN} hex chars after prefix, "
+            f"got {len(hex_part)} in {response_id!r}"
         )
+    # Zero-pad short hex suffixes (e.g. 24-char harness-allocated
+    # IDs) to a valid 128-bit W3C trace ID. The padding preserves
+    # uniqueness — the original hex is a prefix of the trace ID.
+    hex_part = hex_part.ljust(_HEX_LEN, "0")
     try:
         int(hex_part, 16)
     except ValueError as exc:
@@ -603,6 +608,14 @@ def init() -> None:
         import mlflow.tracing
 
         mlflow.tracing.enable()
+
+        # Enable the inner tracing module so TracingContext spans are
+        # created for every agent turn. Without this, telemetry.init()
+        # sets up the OTel provider but no spans are emitted because the
+        # per-session tracing flag stays False.
+        from omnigent.inner.tracing import enable_tracing
+
+        enable_tracing()
     except ImportError:
         # mlflow is an optional dependency (`omnigent[tracing]`). When it
         # is absent, tracing is simply disabled — degrade quietly rather

tests/runtime/test_telemetry.py

@@ -270,14 +270,23 @@ def test_trace_id_from_response_id_wrong_prefix() -> None:
         telemetry.trace_id_from_response_id("conv_" + _RESP_HEX)
 
 
-def test_trace_id_from_response_id_wrong_length() -> None:
+def test_trace_id_from_response_id_short_hex_zero_padded() -> None:
     """
-    An ID with a hex suffix shorter than 32 chars raises ValueError.
-    Guards against accidentally passing a truncated ID or some
-    non-UUID4 value.
+    A short hex suffix (< 32 chars) is zero-padded to 32 chars.
+    Harness-allocated response IDs use 24-char hex; the padding
+    produces a valid 128-bit W3C trace ID.
     """
-    with pytest.raises(ValueError, match="32 hex chars"):
-        telemetry.trace_id_from_response_id("resp_abcdef")
+    result = telemetry.trace_id_from_response_id("resp_abcdef")
+    assert result == "abcdef" + "0" * 26
+    assert len(result) == 32
+
+
+def test_trace_id_from_response_id_too_long() -> None:
+    """
+    A hex suffix longer than 32 chars raises ValueError.
+    """
+    with pytest.raises(ValueError, match="at most"):
+        telemetry.trace_id_from_response_id("resp_" + "a" * 33)
 
 
 def test_trace_id_from_response_id_invalid_hex() -> None: