feat: implement MLflow Observer (Phase C)

.github/workflows/test.yml

@@ -19,7 +19,7 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          pip install -e ".[dev]"
+          pip install -e ".[dev,observability]"
 
       - name: Run linter (ruff)
         run: |

Docs/mlflow_dashboard_proposal.md

@@ -0,0 +1,99 @@
+# MLflow Dashboard Proposal — ABEvalFlow Observability
+
+Phase C+D of APPENG-5370. Using MLflow's built-in dashboard and experiment tracking to visualize pipeline results, gate scores, certification levels, and LLM token usage — all in one place.
+
+Each pipeline run becomes an MLflow run under an experiment (one experiment per submission). Metrics are logged from our existing PostgreSQL tables.
+
+Please review and share feedback or additional metrics you'd like to see.
+
+---
+
+## Experiment-Level View (per submission)
+
+Each submission (e.g. `hello-world`, `my-skill`) is an MLflow experiment. Every pipeline run is a run within that experiment, enabling run-to-run comparison.
+
+### Logged per run
+
+**Parameters (static per run):**
+| Parameter | Source |
+|-----------|--------|
+| `eval_engine` | `scorecards.eval_engine` |
+| `combination_mode` | `scorecards.combination_mode` |
+| `pipeline_run_id` | `scorecards.pipeline_run_id` |
+| `commit_sha` | `evaluation_runs.commit_sha` |
+| `model_name` | `observability_metrics.model_name` |
+
+**Metrics (tracked over runs):**
+| Metric | Source | What it shows |
+|--------|--------|---------------|
+| `recommendation` | `scorecards.recommendation` | pass/warn/fail verdict |
+| `uplift` | `evaluation_runs.uplift` | Treatment vs control pass rate difference |
+| `treatment_pass_rate` | `evaluation_runs.treatment_pass_rate` | How often treatment variant passed |
+| `control_pass_rate` | `evaluation_runs.control_pass_rate` | How often control variant passed |
+| `mean_reward_gap` | `evaluation_runs.mean_reward_gap` | Reward score difference between variants |
+| `ttest_p_value` | `evaluation_runs.ttest_p_value` | Statistical significance of reward difference |
+| `fisher_p_value` | `evaluation_runs.fisher_p_value` | Statistical significance of pass rate difference |
+| `gates_passed` | `scorecards.gates_passed` | How many gates passed |
+| `gates_failed` | `scorecards.gates_failed` | How many gates failed |
+| `highest_certification` | `scorecards.highest_certification` | Best certification level achieved |
+| `gate_score_evaluation` | `gate_results.score` where gate_name=evaluation | Engine gate score (0-1) |
+| `gate_score_security` | `gate_results.score` where gate_name=security | Security gate score (0-1) |
+| `gate_score_quality` | `gate_results.score` where gate_name=quality | Quality gate score (0-1) |
+| `gate_score_behavioral` | `gate_results.score` where gate_name=behavioral | Behavioral gate score (0-1) |
+| `security_findings_count` | `gate_results.findings_count` where gate_type=security | Number of security issues found |
+| `total_prompt_tokens` | `observability_metrics.total_prompt_tokens` | Prompt tokens sent to LLM |
+| `total_completion_tokens` | `observability_metrics.total_completion_tokens` | Completion tokens from LLM |
+| `total_tokens` | `observability_metrics.total_tokens` | Total LLM tokens used |
+| `llm_calls_count` | `observability_metrics.llm_calls_count` | Number of LLM API calls |
+| `certification_foundational_passed` | `certifications` where level=foundational | Whether foundational checks passed |
+| `certification_trusted_passed` | `certifications` where level=trusted | Whether trusted checks passed |
+| `certification_certified_passed` | `certifications` where level=certified | Whether certified checks passed |
+
+**Artifacts (attached per run):**
+| Artifact | Description |
+|----------|-------------|
+| `scorecard.json` | Full scorecard with all gate results |
+| `report.md` | Human-readable evaluation report |
+| `report.json` | Machine-readable evaluation report |
+
+**Tags:**
+| Tag | Value |
+|-----|-------|
+| `recommendation` | pass / warn / fail |
+| `eval_engine` | harbor / ase / aeh / a2a / mcpchecker |
+| `highest_certification` | none / foundational / trusted / certified |
+
+---
+
+## What MLflow shows out of the box with this data
+
+### Run comparison
+- Compare any two runs side-by-side: metrics, parameters, artifacts
+- See how uplift, pass rate, and gate scores changed between runs
+
+### Metric charts (built-in)
+- `treatment_pass_rate` over runs — is the skill improving?
+- `gate_score_security` over runs — are security scores trending up?
+- `total_tokens` over runs — is token usage growing?
+- `uplift` over runs — is the skill's impact stable?
+
+### Experiment overview
+- Table of all runs with sortable columns (pass rate, uplift, certification, tokens)
+- Filter by tags (e.g. show only failed runs, or only harbor engine)
+
+### Cost tracking
+- `total_tokens` and `total_prompt_tokens` / `total_completion_tokens` per run
+- Compare token usage across submissions to identify expensive evaluations
+
+---
+
+## Summary
+
+| What | Where it comes from | What question it answers |
+|------|-------------------|------------------------|
+| Pass/fail trends | `scorecards.recommendation` | Is the pipeline healthy? |
+| Skill improvement | `evaluation_runs.uplift`, `treatment_pass_rate` | Are skills getting better over time? |
+| Gate health | `gate_results.score`, `gate_results.passed` | Which gates fail most? Are scores improving? |
+| Certification progress | `certifications.passed`, `scorecards.highest_certification` | What levels are submissions reaching? |
+| Token/cost tracking | `observability_metrics.total_tokens`, `llm_calls_count` | How much are evaluations costing? |
+| Security trends | `gate_results.findings_count` where security | Are submissions getting more secure? |

abevalflow/db/observer.py

@@ -11,6 +11,7 @@
 
 import logging
 import uuid
+from pathlib import Path
 from typing import Protocol
 
 from abevalflow.report import AnalysisResult
@@ -46,7 +47,19 @@ def discover_observers() -> list[ResultsObserver]:
     observers: list[ResultsObserver] = []
 
     if os.environ.get("MLFLOW_TRACKING_URI"):
-        logger.info("MLflow observer requested but not yet implemented — skipping")
+        try:
+            from abevalflow.observability.mlflow_observer import MLflowObserver
+
+            prefix = os.environ.get("MLFLOW_EXPERIMENT_PREFIX", "abevalflow")
+            observers.append(
+                MLflowObserver(
+                    tracking_uri=os.environ["MLFLOW_TRACKING_URI"],
+                    experiment_prefix=prefix,
+                )
+            )
+            logger.info("MLflow observer enabled (uri=%s)", os.environ["MLFLOW_TRACKING_URI"])
+        except ImportError:
+            logger.warning("MLflow observer requested but mlflow package not installed — skipping")
 
     if os.environ.get("LANGFUSE_PUBLIC_KEY"):
         logger.info("Langfuse observer requested but not yet implemented — skipping")
@@ -58,10 +71,23 @@ def notify_observers(
     observers: list[ResultsObserver],
     result: AnalysisResult,
     run_id: uuid.UUID,
+    report_dir: Path | None = None,
+    pipeline_run_id: str | None = None,
 ) -> None:
     """Invoke all observers, catching and logging any errors."""
+    kwargs: dict = {}
+    if report_dir is not None:
+        kwargs["report_dir"] = report_dir
+    if pipeline_run_id is not None:
+        kwargs["pipeline_run_id"] = pipeline_run_id
+
     for obs in observers:
         try:
+            if kwargs:
+                obs.on_evaluation_stored(result, run_id, **kwargs)
+            else:
+                obs.on_evaluation_stored(result, run_id)
+        except TypeError:
             obs.on_evaluation_stored(result, run_id)
         except Exception:
             logger.warning(

abevalflow/observability/mlflow_observer.py

@@ -0,0 +1,181 @@
+"""MLflow observer for logging evaluation results as MLflow experiment runs.
+
+One experiment per submission, one run per pipeline execution. Logs metrics,
+parameters, tags, and artifacts. Auto-activates when MLFLOW_TRACKING_URI is set.
+
+Observer failures are isolated and never fail the pipeline.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import uuid
+from pathlib import Path
+
+import mlflow
+
+from abevalflow.report import AnalysisResult
+
+logger = logging.getLogger(__name__)
+
+
+class MLflowObserver:
+    """Logs evaluation results to MLflow as experiment runs."""
+
+    def __init__(self, tracking_uri: str, experiment_prefix: str = "abevalflow") -> None:
+        self.tracking_uri = tracking_uri
+        self.experiment_prefix = experiment_prefix
+
+    def on_evaluation_stored(
+        self,
+        result: AnalysisResult,
+        run_id: uuid.UUID,
+        report_dir: Path | None = None,
+        pipeline_run_id: str | None = None,
+    ) -> None:
+
+        mlflow.set_tracking_uri(self.tracking_uri)
+
+        experiment_name = f"{self.experiment_prefix}/{result.submission_name}"
+        experiment = mlflow.get_experiment_by_name(experiment_name)
+        if experiment is None:
+            experiment_id = mlflow.create_experiment(experiment_name)
+        else:
+            experiment_id = experiment.experiment_id
+
+        s = result.summary
+        p = result.provenance
+        effective_run_id = pipeline_run_id or p.pipeline_run_id or str(run_id)
+        effective_commit = p.commit_sha or ""
+
+        with mlflow.start_run(experiment_id=experiment_id, run_name=effective_run_id):
+            mlflow.log_params(
+                {
+                    "eval_engine": p.eval_engine,
+                    "commit_sha": effective_commit,
+                    "pipeline_run_id": effective_run_id,
+                }
+            )
+
+            mlflow.log_metrics(
+                {
+                    "uplift": s.uplift,
+                    "treatment_pass_rate": s.treatment.pass_rate,
+                    "control_pass_rate": s.control.pass_rate,
+                    "treatment_n_trials": s.treatment.n_trials,
+                    "control_n_trials": s.control.n_trials,
+                }
+            )
+
+            if s.mean_reward_gap is not None:
+                mlflow.log_metric("mean_reward_gap", s.mean_reward_gap)
+            if s.ttest_p_value is not None:
+                mlflow.log_metric("ttest_p_value", s.ttest_p_value)
+            if s.fisher_p_value is not None:
+                mlflow.log_metric("fisher_p_value", s.fisher_p_value)
+            if s.treatment.mean_reward is not None:
+                mlflow.log_metric("treatment_mean_reward", s.treatment.mean_reward)
+            if s.control.mean_reward is not None:
+                mlflow.log_metric("control_mean_reward", s.control.mean_reward)
+
+            mlflow.set_tags(
+                {
+                    "recommendation": s.recommendation.value,
+                    "eval_engine": p.eval_engine,
+                    "submission_name": result.submission_name,
+                    "db_run_id": str(run_id),
+                }
+            )
+
+            if report_dir and report_dir.is_dir():
+                for artifact_name in ("scorecard.json", "report.md", "report.json"):
+                    artifact_path = report_dir / artifact_name
+                    if artifact_path.is_file():
+                        mlflow.log_artifact(str(artifact_path))
+
+                _log_scorecard_metrics(report_dir)
+                _log_observability_metrics(report_dir)
+
+        logger.info(
+            "MLflow: logged run for %s (experiment=%s)",
+            result.submission_name,
+            experiment_name,
+        )
+
+
+def _log_scorecard_metrics(report_dir: Path) -> None:
+    """Log gate scores and certification from scorecard.json if present."""
+
+    scorecard_path = report_dir / "scorecard.json"
+    if not scorecard_path.is_file():
+        return
+
+    try:
+        scorecard = json.loads(scorecard_path.read_text())
+
+        if "highest_certification" in scorecard:
+            mlflow.set_tag("highest_certification", scorecard["highest_certification"])
+
+        if "gates_passed" in scorecard:
+            mlflow.log_metric("gates_passed", scorecard["gates_passed"])
+        if "gates_failed" in scorecard:
+            mlflow.log_metric("gates_failed", scorecard["gates_failed"])
+
+        for gate in scorecard.get("gates", []):
+            key = gate.get("policy_key") or gate.get("gate_name", "unknown")
+            if "score" in gate:
+                mlflow.log_metric(f"gate_score_{key}", gate["score"])
+            if "findings_count" in gate or "findings" in gate:
+                count = gate.get("findings_count", len(gate.get("findings", [])))
+                mlflow.log_metric(f"gate_findings_{key}", count)
+
+        cert = scorecard.get("certification")
+        if cert:
+            for level_name in ("foundational", "trusted", "certified"):
+                level_data = cert.get(level_name)
+                if level_data:
+                    mlflow.log_metric(
+                        f"certification_{level_name}_passed",
+                        1.0 if level_data.get("passed") else 0.0,
+                    )
+    except Exception as e:
+        logger.warning("Failed to log scorecard metrics: %s", e)
+
+
+def _log_observability_metrics(report_dir: Path) -> None:
+    """Log token usage from metrics checkpoint if present."""
+
+    checkpoint_path = report_dir / "_metrics_checkpoint.json"
+    if not checkpoint_path.is_file():
+        return
+
+    try:
+        checkpoint = json.loads(checkpoint_path.read_text())
+
+        token_usage = checkpoint.get("token_usage", {})
+        total_prompt = 0
+        total_completion = 0
+        total_calls = 0
+
+        for phase_name, usage in token_usage.items():
+            total_prompt += usage.get("prompt_tokens", 0)
+            total_completion += usage.get("completion_tokens", 0)
+            total_calls += usage.get("call_count", 0)
+
+        if total_prompt > 0 or total_completion > 0:
+            mlflow.log_metrics(
+                {
+                    "total_prompt_tokens": total_prompt,
+                    "total_completion_tokens": total_completion,
+                    "total_tokens": total_prompt + total_completion,
+                    "llm_calls_count": total_calls,
+                }
+            )
+
+        model_name = checkpoint.get("model_name")
+        if model_name:
+            mlflow.set_tag("model_name", model_name)
+
+    except Exception as e:
+        logger.warning("Failed to log observability metrics: %s", e)

pipeline/tasks/post/store.yaml

@@ -81,6 +81,10 @@ spec:
       type: string
       default: "main"
       description: Branch or SHA of the pipeline repo to use for scripts
+    - name: mlflow-tracking-uri
+      type: string
+      default: ""
+      description: MLflow tracking URI. Observer activates only when set.
   workspaces:
     - name: source
       description: Shared workspace containing evaluation artifacts
@@ -94,6 +98,8 @@ spec:
               name: ab-eval-db-credentials
               key: database-url
               optional: true
+        - name: MLFLOW_TRACKING_URI
+          value: "$(params.mlflow-tracking-uri)"
       script: |
         #!/usr/bin/env bash
         set -euo pipefail
@@ -124,7 +130,7 @@ spec:
         export PYTHONPATH="$PIPELINE_DIR"
 
         # --- 2. Install dependencies ---
-        pip install --quiet --no-cache-dir pydantic sqlalchemy "psycopg[binary]" "tenacity>=8.2" pyyaml
+        pip install --quiet --no-cache-dir pydantic sqlalchemy "psycopg[binary]" "tenacity>=8.2" pyyaml mlflow
 
         # --- 3. Run store script ---
         REPORT_DIR="$(workspaces.source.path)/reports/$(params.submission-name)"

pyproject.toml

@@ -22,6 +22,9 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
+observability = [
+    "mlflow>=2.10",
+]
 dev = [
     "pytest>=8.0",
     "pytest-cov>=4.1",

scripts/store_results.py

@@ -432,7 +432,13 @@ def store(
 
         observers = discover_observers()
         if observers:
-            notify_observers(observers, result, ev_run.id)
+            notify_observers(
+                observers,
+                result,
+                ev_run.id,
+                report_dir=report_dir,
+                pipeline_run_id=effective_run_id,
+            )
 
     return True