diff --git a/src/ucode/agents/codex.py b/src/ucode/agents/codex.py index 2a5c657..94682d2 100644 --- a/src/ucode/agents/codex.py +++ b/src/ucode/agents/codex.py @@ -2,8 +2,12 @@ from __future__ import annotations +import json import os import re +import shutil +import subprocess +from collections.abc import Iterable from pathlib import Path from ucode.agent_updates import available_npm_package_update @@ -23,6 +27,8 @@ from ucode.launcher import exec_or_spawn from ucode.state import mark_tool_managed, save_state from ucode.telemetry import agent_version, ucode_version +from ucode.tracing import tracing_env +from ucode.ui import print_note, print_success, print_warning CODEX_CONFIG_DIR = Path.home() / ".codex" CODEX_PROFILE_NAME = "ucode" @@ -33,6 +39,17 @@ CODEX_MODEL_PROVIDER_NAME = "ucode-databricks" MINIMUM_CODEX_VERSION = (0, 134, 0) MINIMUM_CODEX_VERSION_TEXT = "0.134.0" +MLFLOW_CODEX_PACKAGE = "@mlflow/codex" +# npm integration for Codex, built on the @mlflow/core TypeScript tracing SDK. +MLFLOW_CODEX_PACKAGE_SPEC = f"{MLFLOW_CODEX_PACKAGE}@^0.3.0" +MLFLOW_CODEX_BINARY = "mlflow-codex" +MINIMUM_MLFLOW_CODEX_VERSION = (0, 3, 0) +CODEX_TRACING_NOTIFY = [MLFLOW_CODEX_BINARY, "notify-hook"] +CODEX_TRACING_ENV_KEYS = ( + "MLFLOW_TRACKING_URI", + "MLFLOW_EXPERIMENT_ID", + "MLFLOW_TRACE_LOCATION", +) SPEC: ToolSpec = { @@ -49,6 +66,7 @@ ["model_providers", CODEX_MODEL_PROVIDER_NAME], ["model_providers", CODEX_MODEL_PROVIDER_NAME, "http_headers"], ] +TRACING_MANAGED_KEYS: list[list[str]] = [["notify"]] LEGACY_MANAGED_KEYS: list[list[str]] = [ ["profile"], @@ -79,6 +97,10 @@ def _parse_version(value: str) -> tuple[int, int, int] | None: return int(major), int(minor), int(patch) +def _version_at_least(version: tuple[int, int, int] | None, minimum: tuple[int, int, int]) -> bool: + return version is not None and version >= minimum + + def _installed_version_status() -> tuple[str, bool] | None: version = agent_version(SPEC["binary"]) parsed = _parse_version(version) @@ -87,6 +109,75 @@ def _installed_version_status() -> tuple[str, bool] | None: return version, parsed < MINIMUM_CODEX_VERSION +def _installed_mlflow_codex_version() -> tuple[int, int, int] | None: + if not shutil.which("npm"): + return None + try: + result = subprocess.run( + ["npm", "list", "-g", MLFLOW_CODEX_PACKAGE, "--json", "--depth=0"], + check=False, + capture_output=True, + text=True, + timeout=30, + ) + except (OSError, subprocess.TimeoutExpired): + return None + if result.returncode != 0: + return None + try: + payload = json.loads(result.stdout or "{}") + except json.JSONDecodeError: + return None + deps = payload.get("dependencies") + if not isinstance(deps, dict): + return None + package = deps.get(MLFLOW_CODEX_PACKAGE) + if not isinstance(package, dict): + return None + version = package.get("version") + return _parse_version(version) if isinstance(version, str) else None + + +def ensure_tracing_runtime() -> bool: + """Ensure the UC-capable stock MLflow Codex hook is installed globally.""" + current = _installed_mlflow_codex_version() + if _version_at_least(current, MINIMUM_MLFLOW_CODEX_VERSION) and shutil.which( + MLFLOW_CODEX_BINARY + ): + return True + + if not shutil.which("npm"): + print_warning( + f"Codex tracing needs {MLFLOW_CODEX_PACKAGE_SPEC}, but npm is not available. " + "Install npm, then re-run `ucode configure tracing`." + ) + return False + + print_note(f"Installing {MLFLOW_CODEX_PACKAGE_SPEC} for Codex tracing...") + try: + subprocess.run( + ["npm", "install", "-g", MLFLOW_CODEX_PACKAGE_SPEC], + check=True, + timeout=300, + ) + except (OSError, subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc: + print_warning(f"Could not install {MLFLOW_CODEX_PACKAGE_SPEC}: {exc}") + return False + + installed = _installed_mlflow_codex_version() + if not _version_at_least(installed, MINIMUM_MLFLOW_CODEX_VERSION): + print_warning(f"npm did not install a compatible {MLFLOW_CODEX_PACKAGE} runtime") + return False + if not shutil.which(MLFLOW_CODEX_BINARY): + print_warning( + f"Installed {MLFLOW_CODEX_PACKAGE_SPEC}, but `{MLFLOW_CODEX_BINARY}` is not on PATH" + ) + return False + + print_success("MLflow Codex tracing runtime ready") + return True + + def _use_legacy_layout() -> bool: """Return True when the installed Codex CLI predates per-profile config files. @@ -151,6 +242,32 @@ def render_overlay( return overlay +def _is_tracing_notify(value: object) -> bool: + if not isinstance(value, Iterable): + return False + items = list(value) # tomlkit arrays are iterable but not plain lists. + if not all(isinstance(item, str) for item in items): + return False + return items == CODEX_TRACING_NOTIFY + + +def _apply_tracing_notify(doc: dict, enabled: bool) -> bool: + existing = doc.get("notify") + if enabled: + if existing is not None and not _is_tracing_notify(existing): + print_warning( + "Replacing existing Codex `notify` hook with MLflow tracing; " + "the previous value is preserved in ucode's config backup." + ) + doc["notify"] = list(CODEX_TRACING_NOTIFY) + return True + + if _is_tracing_notify(existing): + doc.pop("notify", None) + return True + return False + + def render_legacy_overlay( workspace: str, model: str | None = None, @@ -316,6 +433,7 @@ def write_tool_config(state: dict, model: str | None = None, provider: str | Non # Databricks endpoint id is pinned. chosen_model = None if provider else _codex_model_id(model or default_model(state)) databricks_profile = state.get("profile") + tracing_enabled = bool(tracing_env(state, "codex")) if _use_legacy_layout(): # Codex < 0.134.0 only reads ~/.codex/config.toml. Write the shared @@ -332,6 +450,7 @@ def write_tool_config(state: dict, model: str | None = None, provider: str | Non ) doc = read_toml_safe(LEGACY_CODEX_CONFIG_PATH) deep_merge_dict(doc, overlay) + _apply_tracing_notify(doc, tracing_enabled) if provider: # deep_merge can't drop keys, so clear a `model` pinned by an # earlier non-provider run that the provider overlay omits. @@ -339,7 +458,10 @@ def write_tool_config(state: dict, model: str | None = None, provider: str | Non if isinstance(profiles, dict) and isinstance(profiles.get(CODEX_PROFILE_NAME), dict): profiles[CODEX_PROFILE_NAME].pop("model", None) write_toml_file(LEGACY_CODEX_CONFIG_PATH, doc) - state = mark_tool_managed(state, "codex", LEGACY_MANAGED_KEYS) + managed_keys = list(LEGACY_MANAGED_KEYS) + if tracing_enabled: + managed_keys += TRACING_MANAGED_KEYS + state = mark_tool_managed(state, "codex", managed_keys) save_state(state) return state @@ -354,12 +476,16 @@ def write_tool_config(state: dict, model: str | None = None, provider: str | Non ) doc = read_toml_safe(CODEX_CONFIG_PATH) deep_merge_dict(doc, overlay) + _apply_tracing_notify(doc, tracing_enabled) if provider: # deep_merge can't drop keys, so clear a `model` pinned by an earlier # non-provider run that the provider overlay omits. doc.pop("model", None) write_toml_file(CODEX_CONFIG_PATH, doc) - state = mark_tool_managed(state, "codex", MANAGED_KEYS) + managed_keys = list(MANAGED_KEYS) + if tracing_enabled: + managed_keys += TRACING_MANAGED_KEYS + state = mark_tool_managed(state, "codex", managed_keys) save_state(state) return state @@ -394,8 +520,23 @@ def _gpt_version_key(entry: tuple[str, tuple[int, int | None, int | None, str]]) def launch(state: dict, tool_args: list[str]) -> None: binary = SPEC["binary"] workspace = state.get("workspace") + for key in CODEX_TRACING_ENV_KEYS: + os.environ.pop(key, None) if workspace: - os.environ["OAUTH_TOKEN"] = get_databricks_token(workspace, state.get("profile")) + token = get_databricks_token(workspace, state.get("profile")) + os.environ["OAUTH_TOKEN"] = token + tracing_env_vars = tracing_env(state, "codex") + if tracing_env_vars: + if not shutil.which(MLFLOW_CODEX_BINARY): + raise RuntimeError( + f"Codex tracing is enabled, but `{MLFLOW_CODEX_BINARY}` is not on PATH. " + f"Run `ucode configure tracing` or `npm install -g {MLFLOW_CODEX_PACKAGE_SPEC}`." + ) + os.environ["DATABRICKS_HOST"] = workspace + os.environ["DATABRICKS_TOKEN"] = token + if state.get("profile"): + os.environ["DATABRICKS_CONFIG_PROFILE"] = str(state["profile"]) + os.environ.update(tracing_env_vars) exec_or_spawn([binary, "--profile", CODEX_PROFILE_NAME, *tool_args]) diff --git a/src/ucode/tracing.py b/src/ucode/tracing.py index fa5fce3..7dd4afd 100644 --- a/src/ucode/tracing.py +++ b/src/ucode/tracing.py @@ -1,16 +1,16 @@ -"""MLflow tracing: route Claude Code sessions to a Databricks experiment. +"""MLflow tracing: route Claude/Codex sessions to a Databricks experiment. -ucode points Claude Code's MLflow integration at a single, pre-provisioned +ucode points supported MLflow integrations at a single, pre-provisioned experiment named ``ucode-traces`` whose traces are stored in a Unity Catalog table. ucode asserts the experiment exists and is UC-backed (it does not create -it), resolves a SQL warehouse for UC writes, and persists the tracking URI, -experiment id, and warehouse id. Auth reuses the workspace profile ucode -already configures in ``~/.databrickscfg``. +it), resolves a SQL warehouse for Claude's Python trace writes, and persists the +tracking URI, experiment id, UC destination, and, for Claude, warehouse id. Auth +reuses the workspace profile ucode already configures in ``~/.databrickscfg``. -Scope: Claude Code only. Its `mlflow autolog claude` Stop hook writes traces to -the UC table. Codex and OpenCode used the `@mlflow/codex`/`@mlflow/opencode` JS -clients, which only reach the classic (non-UC) trace store, so their tracing was -removed. Gemini's exporter is OTLP-only and was never wired here. +Scope: Claude Code and Codex. Claude uses the Python MLflow CLI Stop hook. +Codex uses the `@mlflow/codex` TypeScript package, which needs the UC +destination routed through ``MLFLOW_TRACE_LOCATION`` to use the UC table path. +Gemini's exporter is OTLP-only and is not wired here. This module must not import ``mlflow`` (the heavy optional dependency) or ``ucode.agents`` at import time: agents import the small helpers here, and the @@ -36,12 +36,8 @@ spinner, ) -# Agents whose MLflow integration routes to a Databricks tracking URI. Only -# Claude Code is supported: its `mlflow autolog claude` Stop hook writes traces -# to the experiment's Unity Catalog table. Codex and OpenCode used the -# `@mlflow/codex`/`@mlflow/opencode` JS clients, which only reach the classic -# (non-UC) trace store, so their tracing was removed. -TRACING_AGENTS: tuple[str, ...] = ("claude",) +# Agents whose MLflow integration routes to a Databricks tracking URI. +TRACING_AGENTS: tuple[str, ...] = ("claude", "codex") # Leaf name of the experiment every agent and every user routes to. ucode does # not create it: an admin provisions an experiment with this name whose traces @@ -82,9 +78,11 @@ def _missing_experiment_error(name: str, reason: str | None) -> str: def _missing_warehouse_error(reason: str | None) -> str: - """Actionable error when no SQL warehouse can back UC trace writes. The - warehouse is mandatory: traces to a UC table are silently dropped without - ``MLFLOW_TRACING_SQL_WAREHOUSE_ID``.""" + """Actionable error when Claude's Python trace path has no SQL warehouse. + + Claude requires ``MLFLOW_TRACING_SQL_WAREHOUSE_ID`` for UC trace writes; + Codex's TypeScript trace path does not. + """ detail = f" ({reason})" if reason else "" return ( f"No SQL warehouse is available to write traces to Unity Catalog{detail}.\n" @@ -123,13 +121,24 @@ def tracing_env(state: dict, tool: str) -> dict[str, str]: tracking URI carries the profile, so auth resolves from ``~/.databrickscfg`` without extra vars. - ``MLFLOW_TRACING_SQL_WAREHOUSE_ID`` is required for the experiment's - Unity Catalog trace table: without it the MLflow exporter silently drops - every trace.""" + Claude's Python path uses ``MLFLOW_TRACING_SQL_WAREHOUSE_ID`` for UC writes. + Codex's TypeScript path uses the persisted UC destination through + ``MLFLOW_TRACE_LOCATION`` and does not require a warehouse. + """ cfg = tracing_config(state) entry = agent_tracing(state, tool) if not cfg or not entry: return {} + if tool == "codex": + env = { + # The JS SDK's `databricks://` path shells out without a + # profile when profiles share a host. Give it explicit host/token + # at launch time and use the bare URI here. + "MLFLOW_TRACKING_URI": "databricks", + "MLFLOW_EXPERIMENT_ID": str(entry["experiment_id"]), + "MLFLOW_TRACE_LOCATION": str(cfg["uc_destination"]), + } + return env env = { "MLFLOW_TRACKING_URI": str(cfg["tracking_uri"]), "MLFLOW_EXPERIMENT_ID": str(entry["experiment_id"]), @@ -206,7 +215,8 @@ def _select_tracing_workspace(*, only_enabled: bool = False) -> dict: candidates = _tracing_capable_workspaces(full) if not candidates: raise RuntimeError( - "Claude Code is not configured. Run `ucode configure` for Claude Code first." + "No tracing-capable agent is configured. Run `ucode configure` for Claude Code " + "or Codex first." ) current = full.get("current_workspace") @@ -250,14 +260,18 @@ def _rewrite_agent_configs(state: dict) -> dict: def _install_agent_tracing_deps(state: dict) -> None: - """Install the Claude tracing runtime (pinned mlflow CLI) when Claude is - configured on this workspace and has tracing on. Claude is the only - tracing-capable agent.""" - from ucode.agents import claude + """Install tracing runtimes for configured tracing-capable agents.""" + from ucode.agents import claude, codex + + installers = { + "claude": claude.ensure_tracing_runtime, + "codex": codex.ensure_tracing_runtime, + } configured = _configured_tracing_agents(state) - if "claude" in configured and agent_tracing(state, "claude"): - claude.ensure_tracing_runtime() + for tool in configured: + if agent_tracing(state, tool): + installers[tool]() def configure_tracing_command( @@ -330,13 +344,14 @@ def _enable_tracing_for_state(state: dict) -> dict: if not experiment: raise RuntimeError(_missing_experiment_error(name, reason)) - # A UC-backed experiment needs a SQL warehouse to write traces — without - # `MLFLOW_TRACING_SQL_WAREHOUSE_ID` the exporter silently drops them — so a - # warehouse is mandatory, not optional. - with spinner("Resolving a SQL warehouse for trace storage..."): - warehouse_id, wh_reason = resolve_sql_warehouse_id(workspace, token) - if not warehouse_id: - raise RuntimeError(_missing_warehouse_error(wh_reason)) + warehouse_id = None + if "claude" in configured: + # Claude's Python hook needs a SQL warehouse to write to a UC-backed + # experiment. Codex uses MLFLOW_TRACE_LOCATION and does not need this. + with spinner("Resolving a SQL warehouse for trace storage..."): + warehouse_id, wh_reason = resolve_sql_warehouse_id(workspace, token) + if not warehouse_id: + raise RuntimeError(_missing_warehouse_error(wh_reason)) state["tracing"] = { "enabled": True, @@ -344,8 +359,9 @@ def _enable_tracing_for_state(state: dict) -> dict: "experiment_id": experiment["experiment_id"], "experiment_name": experiment["experiment_name"], "uc_destination": experiment["uc_destination"], - "sql_warehouse_id": warehouse_id, } + if warehouse_id: + state["tracing"]["sql_warehouse_id"] = warehouse_id save_state(state) print_kv("Tracking URI", str(state["tracing"]["tracking_uri"])) @@ -354,7 +370,8 @@ def _enable_tracing_for_state(state: dict) -> dict: f"{experiment['experiment_name']} (id {experiment['experiment_id']})", ) print_kv("Unity Catalog", experiment["uc_destination"]) - print_kv("SQL warehouse", warehouse_id) + if warehouse_id: + print_kv("SQL warehouse", warehouse_id) _install_agent_tracing_deps(state) state = _rewrite_agent_configs(state) diff --git a/tests/test_agent_codex.py b/tests/test_agent_codex.py index 9a68e03..cd6db3c 100644 --- a/tests/test_agent_codex.py +++ b/tests/test_agent_codex.py @@ -3,6 +3,10 @@ from __future__ import annotations import os +import subprocess +from unittest.mock import patch + +import pytest from ucode.agents import codex from ucode.config_io import read_toml_safe @@ -10,6 +14,21 @@ WS = "https://example.databricks.com" +def _enabled_state() -> dict: + return { + "workspace": WS, + "profile": "profile-a", + "codex_models": ["gpt-5"], + "tracing": { + "enabled": True, + "tracking_uri": "databricks://profile-a", + "experiment_id": "111", + "experiment_name": "/Shared/ucode-traces", + "uc_destination": "main.default.ucode_traces", + }, + } + + class TestCodexSpec: def test_binary(self): assert codex.SPEC["binary"] == "codex" @@ -241,6 +260,86 @@ def test_legacy_write_preserves_other_profiles_in_shared_config(self, tmp_path, assert doc["profiles"]["other"]["model_provider"] == "keep" assert doc["profiles"]["ucode"]["model_provider"] == "ucode-databricks" + @pytest.mark.parametrize( + "version,provider,legacy", + [ + ("0.134.0", None, False), + ("0.133.0", None, True), + ("0.134.0", "main.default.openai", False), + ], + ) + def test_writes_tracing_notify_when_enabled( + self, tmp_path, monkeypatch, version, provider, legacy + ): + config_dir = tmp_path / ".codex" + legacy_path = config_dir / "config.toml" + profile_path = config_dir / "ucode.config.toml" + monkeypatch.setattr(codex, "CODEX_CONFIG_PATH", profile_path) + monkeypatch.setattr(codex, "CODEX_BACKUP_PATH", tmp_path / "profile.backup.toml") + monkeypatch.setattr(codex, "LEGACY_CODEX_CONFIG_PATH", legacy_path) + monkeypatch.setattr(codex, "LEGACY_CODEX_BACKUP_PATH", tmp_path / "legacy.backup.toml") + monkeypatch.setattr(codex, "agent_version", lambda binary: version) + monkeypatch.setattr(codex, "save_state", lambda state: None) + + codex.write_tool_config(_enabled_state(), provider=provider) + + doc = read_toml_safe(legacy_path if legacy else profile_path) + assert list(doc["notify"]) == codex.CODEX_TRACING_NOTIFY + if provider: + assert "model" not in doc + + @pytest.mark.parametrize( + "existing_notify,should_warn", + [(["custom", "hook"], True), (codex.CODEX_TRACING_NOTIFY, False)], + ) + def test_updates_existing_notify_when_enabled( + self, tmp_path, monkeypatch, existing_notify, should_warn + ): + config_path = tmp_path / ".codex" / "ucode.config.toml" + config_path.parent.mkdir() + config_path.write_text( + f'notify = ["{existing_notify[0]}", "{existing_notify[1]}"]\n', + encoding="utf-8", + ) + monkeypatch.setattr(codex, "CODEX_CONFIG_PATH", config_path) + monkeypatch.setattr(codex, "CODEX_BACKUP_PATH", tmp_path / "backup.toml") + monkeypatch.setattr(codex, "agent_version", lambda binary: "0.134.0") + monkeypatch.setattr(codex, "save_state", lambda state: None) + + with patch("ucode.agents.codex.print_warning") as warning: + codex.write_tool_config(_enabled_state()) + + doc = read_toml_safe(config_path) + assert list(doc["notify"]) == codex.CODEX_TRACING_NOTIFY + assert warning.call_count == should_warn + + @pytest.mark.parametrize( + "existing_notify,expected_notify", + [ + (codex.CODEX_TRACING_NOTIFY, None), + (["custom", "hook"], ["custom", "hook"]), + ], + ) + def test_updates_notify_when_tracing_disabled( + self, tmp_path, monkeypatch, existing_notify, expected_notify + ): + config_path = tmp_path / ".codex" / "ucode.config.toml" + config_path.parent.mkdir() + config_path.write_text( + f'notify = ["{existing_notify[0]}", "{existing_notify[1]}"]\n', + encoding="utf-8", + ) + monkeypatch.setattr(codex, "CODEX_CONFIG_PATH", config_path) + monkeypatch.setattr(codex, "CODEX_BACKUP_PATH", tmp_path / "backup.toml") + monkeypatch.setattr(codex, "agent_version", lambda binary: "0.134.0") + monkeypatch.setattr(codex, "save_state", lambda state: None) + + codex.write_tool_config({"workspace": WS, "codex_models": ["gpt-5"]}) + + doc = read_toml_safe(config_path) + actual_notify = list(doc["notify"]) if "notify" in doc else None + assert actual_notify == expected_notify + class TestCodexLegacyLayoutDetection: def test_new_codex_uses_modern_layout(self, monkeypatch): @@ -436,3 +535,155 @@ def fake_execvp(binary: str, args: list[str]) -> None: assert os.environ["OAUTH_TOKEN"] == "fresh-token" assert exec_calls == [("codex", ["codex", "--profile", "ucode", "--search"])] + + def test_injects_tracing_env_before_exec(self, monkeypatch): + exec_calls: list[list[str]] = [] + for key in codex.CODEX_TRACING_ENV_KEYS: + monkeypatch.delenv(key, raising=False) + monkeypatch.setattr( + codex, "get_databricks_token", lambda workspace, profile=None: "fresh-token" + ) + monkeypatch.setattr(codex.shutil, "which", lambda binary: f"/bin/{binary}") + monkeypatch.setattr(codex, "exec_or_spawn", lambda args: exec_calls.append(args)) + + codex.launch(_enabled_state(), ["exec", "say hi"]) + + assert os.environ["OAUTH_TOKEN"] == "fresh-token" + assert os.environ["DATABRICKS_HOST"] == WS + assert os.environ["DATABRICKS_TOKEN"] == "fresh-token" + assert os.environ["DATABRICKS_CONFIG_PROFILE"] == "profile-a" + assert os.environ["MLFLOW_TRACKING_URI"] == "databricks" + assert os.environ["MLFLOW_EXPERIMENT_ID"] == "111" + assert os.environ["MLFLOW_TRACE_LOCATION"] == "main.default.ucode_traces" + assert exec_calls == [["codex", "--profile", "ucode", "exec", "say hi"]] + + def test_errors_when_tracing_runtime_missing(self, monkeypatch): + exec_calls: list[list[str]] = [] + monkeypatch.setattr( + codex, "get_databricks_token", lambda workspace, profile=None: "fresh-token" + ) + monkeypatch.setattr(codex.shutil, "which", lambda binary: None) + monkeypatch.setattr(codex, "exec_or_spawn", lambda args: exec_calls.append(args)) + + with pytest.raises(RuntimeError, match="mlflow-codex.*not on PATH"): + codex.launch(_enabled_state(), ["exec", "say hi"]) + + assert exec_calls == [] + + def test_clears_stale_tracing_env_when_disabled(self, monkeypatch): + exec_calls: list[list[str]] = [] + for key in codex.CODEX_TRACING_ENV_KEYS: + monkeypatch.setenv(key, "stale") + monkeypatch.setenv("DATABRICKS_HOST", "https://external.example.com") + monkeypatch.setenv("DATABRICKS_TOKEN", "external-token") + monkeypatch.setenv("DATABRICKS_CONFIG_PROFILE", "external-profile") + monkeypatch.setattr( + codex, "get_databricks_token", lambda workspace, profile=None: "fresh-token" + ) + monkeypatch.setattr(codex, "exec_or_spawn", lambda args: exec_calls.append(args)) + + codex.launch({"workspace": WS}, ["exec", "say hi"]) + + for key in codex.CODEX_TRACING_ENV_KEYS: + assert key not in os.environ + assert os.environ["DATABRICKS_HOST"] == "https://external.example.com" + assert os.environ["DATABRICKS_TOKEN"] == "external-token" + assert os.environ["DATABRICKS_CONFIG_PROFILE"] == "external-profile" + assert os.environ["OAUTH_TOKEN"] == "fresh-token" + assert exec_calls == [["codex", "--profile", "ucode", "exec", "say hi"]] + + +class TestCodexTracingRuntime: + @pytest.mark.parametrize( + "installed_version,binary_ready,should_install", + [ + (None, False, True), + ((0, 2, 0), True, True), + ((0, 3, 0), False, True), + ((0, 3, 0), True, False), + ], + ) + def test_enforces_runtime_version_and_binary( + self, monkeypatch, installed_version, binary_ready, should_install + ): + installed_versions = iter( + [installed_version, (0, 3, 0)] if should_install else [installed_version] + ) + monkeypatch.setattr( + codex, "_installed_mlflow_codex_version", lambda: next(installed_versions) + ) + if installed_version == (0, 3, 0) and not binary_ready: + paths = iter([None, "/bin/mlflow-codex"]) + + def which(binary): + return "/bin/npm" if binary == "npm" else next(paths) + + monkeypatch.setattr(codex.shutil, "which", which) + else: + monkeypatch.setattr(codex.shutil, "which", lambda binary: f"/bin/{binary}") + + with patch.object(codex.subprocess, "run") as run: + assert codex.ensure_tracing_runtime() is True + + if should_install: + assert run.call_args[0][0] == [ + "npm", + "install", + "-g", + "@mlflow/codex@^0.3.0", + ] + else: + run.assert_not_called() + + @pytest.mark.parametrize( + "npm_path,install_error,expected_calls", + [ + (None, None, 0), + ("/bin/npm", subprocess.CalledProcessError(1, "npm"), 1), + ], + ) + def test_warns_when_npm_setup_fails(self, monkeypatch, npm_path, install_error, expected_calls): + monkeypatch.setattr(codex, "_installed_mlflow_codex_version", lambda: None) + monkeypatch.setattr(codex.shutil, "which", lambda binary: npm_path) + + with ( + patch.object(codex.subprocess, "run", side_effect=install_error) as run, + patch("ucode.agents.codex.print_warning") as warning, + ): + assert codex.ensure_tracing_runtime() is False + + assert run.call_count == expected_calls + warning.assert_called_once() + + def test_reads_installed_runtime_version(self, monkeypatch): + payload = '{"dependencies":{"@mlflow/codex":{"version":"0.3.0"}}}' + monkeypatch.setattr(codex.shutil, "which", lambda binary: f"/bin/{binary}") + monkeypatch.setattr( + codex.subprocess, + "run", + lambda *args, **kwargs: subprocess.CompletedProcess(args[0], 0, stdout=payload), + ) + + assert codex._installed_mlflow_codex_version() == (0, 3, 0) + + @pytest.mark.parametrize( + "installed_version,binary_exists", [((0, 2, 0), True), ((0, 3, 0), False)] + ) + def test_runtime_fails_when_install_is_incomplete( + self, monkeypatch, installed_version, binary_exists + ): + versions = iter([None, installed_version]) + monkeypatch.setattr(codex, "_installed_mlflow_codex_version", lambda: next(versions)) + monkeypatch.setattr( + codex.shutil, + "which", + lambda binary: f"/bin/{binary}" if binary_exists else None, + ) + + with ( + patch.object(codex.subprocess, "run"), + patch("ucode.agents.codex.print_warning") as warning, + ): + assert codex.ensure_tracing_runtime() is False + + warning.assert_called_once() diff --git a/tests/test_e2e_tracing.py b/tests/test_e2e_tracing.py index 7e85c63..e464e6f 100644 --- a/tests/test_e2e_tracing.py +++ b/tests/test_e2e_tracing.py @@ -5,24 +5,26 @@ The flow mirrors `ucode configure tracing` + a real agent run: 1. Find the shared, UC-backed `ucode-traces` experiment in the workspace. - 2. Resolve a SQL warehouse (required to write traces to the UC table). - 3. Enable tracing in state and write the agent's config (env + plugin). - 4. Install the agent's tracing runtime (Claude plugin + mlflow CLI). + 2. Resolve a SQL warehouse when the agent runtime needs one. + 3. Enable tracing in state and write the agent's config. + 4. Install the agent's tracing runtime. 5. Launch the agent headless with a trivial prompt so it emits a trace. 6. Poll the experiment via the MLflow SDK until a NEW trace id appears. -Skipped automatically unless UCODE_TEST_WORKSPACE is set, the `claude` binary -is installed, `mlflow` is importable, and the tracing runtime can be set up. +Skipped automatically unless UCODE_TEST_WORKSPACE is set, the agent binary is +installed, `mlflow` is importable, and the tracing runtime can be set up. Verification uses the MLflow Python SDK rather than hand-rolling the SearchTraces V3 REST call (which takes proto `locations`). """ from __future__ import annotations +import json import os import shutil import subprocess import time +import uuid import pytest @@ -31,7 +33,7 @@ # How long to wait for an emitted trace to show up in the experiment. Trace # ingestion is asynchronous, so we poll. -TRACE_POLL_TIMEOUT = int(os.environ.get("UCODE_E2E_TRACE_TIMEOUT", "120")) +TRACE_POLL_TIMEOUT = int(os.environ.get("UCODE_E2E_TRACE_TIMEOUT", "300")) TRACE_POLL_INTERVAL = 5 @@ -57,6 +59,31 @@ def _trace_ids(client, experiment_id: str) -> set[str]: return ids +def _trace_id(trace) -> str: + info = getattr(trace, "info", None) + return str(getattr(info, "trace_id", None) or getattr(info, "request_id", None) or "") + + +def _contains_text(value, text: str) -> bool: + return text in json.dumps(value, default=str) + + +def _span_model(span) -> str | None: + attributes = getattr(span, "attributes", {}) + value = ( + getattr(span, "model_name", None) + or attributes.get("mlflow.llm.model") + or attributes.get("model") + ) + if isinstance(value, str): + try: + decoded = json.loads(value) + except json.JSONDecodeError: + return value + return decoded if isinstance(decoded, str) else value + return value + + @pytest.mark.skip( reason="Tracing e2e is flaky in CI (async trace ingestion races the poll timeout); " "disabled pending a more robust verification path." @@ -171,3 +198,159 @@ def test_claude_session_lands_a_trace(self, tmp_path, monkeypatch, e2e_state, e2 f"No new MLflow trace landed in experiment {experiment_name} (id {experiment_id}) " f"within {TRACE_POLL_TIMEOUT}s. Claude output: {combined[:300]}" ) + + +class TestCodexTracingE2E: + def test_codex_session_lands_a_uc_trace(self, tmp_path, monkeypatch, e2e_state, e2e_workspace): + pytest.importorskip("mlflow", reason="mlflow not installed (pip install mlflow)") + from mlflow import MlflowClient + from mlflow.entities import SpanType + from mlflow.exceptions import MlflowException + + from ucode.agents import codex + from ucode.databricks import get_databricks_token + + _require_binary("codex") + + discovered_model = codex.default_model(e2e_state) + if not discovered_model: + pytest.skip("No Codex-compatible GPT models available on this workspace") + expected_model = codex._codex_model_id(discovered_model) + + token = get_databricks_token(e2e_workspace) + monkeypatch.setenv("DATABRICKS_HOST", e2e_workspace) + monkeypatch.setenv("DATABRICKS_TOKEN", token) + monkeypatch.setenv("DATABRICKS_BEARER", token) + + codex_home = tmp_path / "home" + codex_config_dir = codex_home / ".codex" + monkeypatch.setattr(codex, "CODEX_CONFIG_PATH", codex_config_dir / "ucode.config.toml") + monkeypatch.setattr(codex, "CODEX_BACKUP_PATH", tmp_path / "codex.backup.toml") + monkeypatch.setattr(codex, "LEGACY_CODEX_CONFIG_PATH", codex_config_dir / "config.toml") + monkeypatch.setattr( + codex, "LEGACY_CODEX_BACKUP_PATH", tmp_path / "codex-legacy.backup.toml" + ) + + leaf_name = tracing.experiment_name() + experiment, reason = find_uc_backed_experiment(e2e_workspace, token, leaf_name) + if not experiment: + pytest.skip(f"no UC-backed '{leaf_name}' experiment on this workspace: {reason}") + experiment_id = experiment["experiment_id"] + experiment_name = experiment["experiment_name"] + + # Codex writes through the TypeScript SDK without a SQL warehouse. The + # Python verification client still needs one to read the UC trace tables. + warehouse_id, wh_reason = resolve_sql_warehouse_id(e2e_workspace, token) + if not warehouse_id: + pytest.skip(f"no SQL warehouse available to verify the UC trace: {wh_reason}") + + state = { + **e2e_state, + "workspace": e2e_workspace, + "tracing": { + "enabled": True, + "tracking_uri": tracing.tracking_uri_for_state({"workspace": e2e_workspace}), + "experiment_id": experiment_id, + "experiment_name": experiment_name, + "uc_destination": experiment["uc_destination"], + }, + } + + if not codex.ensure_tracing_runtime(): + pytest.skip("Could not set up the Codex MLflow tracing runtime in this environment") + + with pytest.MonkeyPatch().context() as mp: + mp.setattr("ucode.state.save_state", lambda s: None) + codex.write_tool_config(state, discovered_model) + + marker = f"UCODE_CODEX_E2E_{uuid.uuid4().hex}" + prompt = f"Reply with exactly: {marker}" + env = { + **os.environ, + "HOME": str(codex_home), + "DATABRICKS_HOST": e2e_workspace, + "DATABRICKS_TOKEN": token, + "DATABRICKS_BEARER": token, + "OAUTH_TOKEN": token, + **tracing.tracing_env(state, "codex"), + } + command = codex.validate_cmd("codex") + command[-1] = prompt + result = subprocess.run( + command, + capture_output=True, + text=True, + timeout=120, + env=env, + stdin=subprocess.DEVNULL, + ) + combined = (result.stdout + result.stderr).strip() + assert result.returncode == 0, f"codex run failed: {combined[:500]}" + + monkeypatch.setenv("MLFLOW_TRACING_SQL_WAREHOUSE_ID", warehouse_id) + + client = MlflowClient(tracking_uri="databricks") + uc_destination = str(experiment["uc_destination"]) + expected_trace_prefix = f"trace:/{uc_destination}/" + deadline = time.monotonic() + TRACE_POLL_TIMEOUT + matched_trace = None + candidate_ids: set[str] = set() + while time.monotonic() < deadline: + candidates = client.search_traces( + experiment_ids=[experiment_id], + max_results=100, + include_spans=True, + ) + for candidate in candidates: + trace_id = _trace_id(candidate) + candidate_ids.add(trace_id) + if not trace_id.startswith(expected_trace_prefix): + continue + if not _contains_text(candidate.to_dict(), marker): + continue + try: + fetched = client.get_trace(trace_id, display=False) + except MlflowException: + # Search and span ingestion can become visible at slightly + # different times for UC-backed traces. + continue + if _contains_text(fetched.to_dict(), marker): + matched_trace = fetched + break + if matched_trace is not None: + break + time.sleep(TRACE_POLL_INTERVAL) + + assert matched_trace is not None, ( + f"No Codex trace containing marker {marker!r} landed in experiment " + f"{experiment_name} (id {experiment_id}) within {TRACE_POLL_TIMEOUT}s. " + f"Candidate trace ids: {sorted(candidate_ids)}. Codex output: {combined[:300]}" + ) + + trace_id = _trace_id(matched_trace) + assert trace_id.startswith(expected_trace_prefix) + + spans = matched_trace.data.spans + assert _contains_text(matched_trace.data.request, marker), ( + f"Marker {marker!r} was not present in the trace request" + ) + marker_input_spans = [span for span in spans if _contains_text(span.inputs, marker)] + assert marker_input_spans, f"Marker {marker!r} was not present in any span inputs" + + root_spans = [span for span in spans if span.parent_id is None] + assert len(root_spans) == 1, f"Expected one Codex root span, found {len(root_spans)}" + root_span = root_spans[0] + assert root_span.name == "codex_conversation" + assert root_span.span_type == SpanType.AGENT + assert _span_model(root_span) == expected_model + + llm_spans = [ + span + for span in spans + if span.parent_id == root_span.span_id and span.span_type == SpanType.LLM + ] + assert llm_spans, "Expected at least one LLM span under the Codex root span" + assert any(_span_model(span) == expected_model for span in llm_spans), ( + f"Expected model {expected_model!r} on a Codex LLM span; " + f"recorded models: {[_span_model(span) for span in llm_spans]}" + ) diff --git a/tests/test_tracing.py b/tests/test_tracing.py index 27e5891..4fd99b8 100644 --- a/tests/test_tracing.py +++ b/tests/test_tracing.py @@ -59,9 +59,8 @@ def test_returns_shared_entry(self): assert entry["experiment_id"] == "111" def test_none_for_non_tracing_agents(self): - # Claude is the only tracing-capable agent now. state = _enabled_state() - for tool in ("codex", "opencode", "gemini"): + for tool in ("opencode", "gemini"): assert tracing.agent_tracing(state, tool) is None def test_none_when_disabled(self): @@ -85,10 +84,21 @@ def test_uri_and_experiment(self): "MLFLOW_ENABLE_ASYNC_TRACE_LOGGING": "false", } - def test_empty_for_non_claude_agents(self): - # Only Claude is tracing-capable; codex/opencode get nothing. + def test_codex_uses_bare_databricks_and_trace_location(self): + state = _enabled_state("p") + env = tracing.tracing_env(state, "codex") + assert env == { + "MLFLOW_TRACKING_URI": "databricks", + "MLFLOW_EXPERIMENT_ID": "111", + "MLFLOW_TRACE_LOCATION": "main.default.ucode_traces", + } + + del state["tracing"]["uc_destination"] + with pytest.raises(KeyError, match="uc_destination"): + tracing.tracing_env(state, "codex") + + def test_empty_for_non_tracing_agents(self): state = _enabled_state() - assert tracing.tracing_env(state, "codex") == {} assert tracing.tracing_env(state, "opencode") == {} def test_includes_sql_warehouse_id(self): @@ -351,17 +361,17 @@ def _full(self) -> dict: "workspaces": { "https://a.databricks.com": {"available_tools": ["claude"], "profile": "pa"}, "https://b.databricks.com": {"available_tools": ["claude"], "profile": "pb"}, - # codex/gemini aren't tracing-capable agents → excluded from candidates + # Codex is tracing-capable; Gemini is not. "https://c.databricks.com": {"available_tools": ["codex", "gemini"]}, }, } def test_raises_when_none_configured(self): with patch.object(tracing, "load_full_state", return_value={"workspaces": {}}): - with pytest.raises(RuntimeError, match="Claude Code is not configured"): + with pytest.raises(RuntimeError, match="No tracing-capable agent is configured"): tracing._select_tracing_workspace() - def test_lists_current_first_and_excludes_non_tracing(self): + def test_lists_current_first_and_includes_codex(self): captured: dict = {} def fake_prompt(desc, profiles): @@ -376,7 +386,7 @@ def fake_prompt(desc, profiles): hosts = [host for host, _ in captured["profiles"]] assert hosts[0] == "https://a.databricks.com" - assert "https://c.databricks.com" not in hosts + assert "https://c.databricks.com" in hosts assert state["workspace"] == "https://a.databricks.com" def test_returns_picked_workspace_state(self): @@ -580,22 +590,72 @@ def test_skips_workspace_without_tracing_agent(self): assert rc == 1 -class TestInstallAgentTracingDeps: - """Only Claude has a tracing runtime; it installs when Claude is configured - with tracing on, and is skipped otherwise.""" +class TestEnableTracingForState: + @pytest.mark.parametrize( + ("available_tools", "warehouse_id", "expected_warehouse_calls"), + [ + pytest.param(["codex"], None, 0, id="codex-only"), + pytest.param(["claude", "codex"], "wh123", 1, id="claude-and-codex"), + ], + ) + def test_sql_warehouse_resolution( + self, + available_tools: list[str], + warehouse_id: str | None, + expected_warehouse_calls: int, + ): + experiment = { + "experiment_id": "111", + "experiment_name": "/Shared/ucode-traces", + "uc_destination": "main.default.ucode_traces", + } + state = {"workspace": WS, "available_tools": available_tools} + with ( + patch.object(tracing, "apply_pat_environment"), + patch.object(tracing, "ensure_databricks_auth"), + patch.object(tracing, "get_databricks_token", return_value="tok"), + patch.object(tracing, "find_uc_backed_experiment", return_value=(experiment, None)), + patch.object( + tracing, + "resolve_sql_warehouse_id", + return_value=(warehouse_id, None), + ) as resolve_warehouse, + patch.object(tracing, "save_state"), + patch.object(tracing, "_install_agent_tracing_deps"), + patch.object(tracing, "_rewrite_agent_configs", side_effect=lambda s: s), + ): + out = tracing._enable_tracing_for_state(state) + + assert resolve_warehouse.call_count == expected_warehouse_calls + assert out["tracing"]["uc_destination"] == "main.default.ucode_traces" + assert out["tracing"].get("sql_warehouse_id") == warehouse_id - def test_installs_claude_runtime_when_configured(self): - state = {**_enabled_state(), "available_tools": ["claude"]} - with patch("ucode.agents.claude.ensure_tracing_runtime") as claude_dep: - tracing._install_agent_tracing_deps(state) - claude_dep.assert_called_once() - def test_skips_when_claude_not_configured(self): - # Claude isn't configured on this workspace, so its runtime is skipped. - state = {**_enabled_state(), "available_tools": ["codex"]} - with patch("ucode.agents.claude.ensure_tracing_runtime") as claude_dep: +class TestInstallAgentTracingDeps: + """Tracing runtimes install for configured tracing-capable agents.""" + + @pytest.mark.parametrize( + ("available_tools", "expected_claude_calls", "expected_codex_calls"), + [ + pytest.param(["claude"], 1, 0, id="claude-only"), + pytest.param(["codex"], 0, 1, id="codex-only"), + pytest.param(["claude", "codex"], 1, 1, id="claude-and-codex"), + ], + ) + def test_installs_configured_runtimes( + self, + available_tools: list[str], + expected_claude_calls: int, + expected_codex_calls: int, + ): + state = {**_enabled_state(), "available_tools": available_tools} + with ( + patch("ucode.agents.claude.ensure_tracing_runtime") as claude_dep, + patch("ucode.agents.codex.ensure_tracing_runtime") as codex_dep, + ): tracing._install_agent_tracing_deps(state) - claude_dep.assert_not_called() + assert claude_dep.call_count == expected_claude_calls + assert codex_dep.call_count == expected_codex_calls class TestDisableTracing: