Building a uv-Based Workflow Evaluation Runner

The workflow now exposes a typed inspection record for each support decision. That gives the evaluation runner something much better than a bare final label: a stable decision key, disposition, service latency, model status, and a place to report the actual number of Jev requests.

This lesson turns that into a uv-managed Python tool that runs a fixed, labelled dataset against two workflow versions and reports:

• Quality: whether each version made the expected business decision.
• Automation coverage: how often it handled eligible tickets automatically, including a safer quality-adjusted version.
• Latency: caller-observed and service-reported percentiles.
• Request volume: actual Jev requests, including calls avoided by deterministic eligibility rules or an open circuit.

The runner calls real versioned workflow endpoints. It is not a replacement for Jev with mocks; its purpose is to compare the behavior of real candidates on the same controlled cases before release.

---

An evaluation is a controlled comparison

A test suite answers, “Did a known invariant break?” An evaluation runner answers, “Did this candidate workflow improve or regress on representative support decisions?”

The distinction matters because your support workflow has both deterministic policy and bounded model judgments. A small change to question wording, thresholds, routing logic, or fallback policy can affect several outcomes at once:

• A candidate may increase automatic routing but also send more unsafe tickets to the wrong queue.
• A candidate may improve decision quality but make two Jev calls where the baseline makes one.
• A candidate may perform well on ordinary billing tickets but regress on ambiguous credential-reset requests.
• A candidate may preserve quality while pushing latency beyond an operational limit.

The evaluation dataset fixes the inputs and expected decision contract. The runner then measures each version under the same conditions.

For this capstone, human-reviewed labels are the source of truth. There is no need for an LLM judge: Jev is making bounded operational decisions, so a deterministic comparison against expected decision keys is more transparent and cheaper to run.

---

Define what “quality” means before writing the runner

A generic metric such as “confidence was high” is not quality. The workflow is useful only if it takes the right business action.

Give every evaluation case an expected decision key. This is more precise than checking the broad outcome alone.

Field	Example	Why it exists
decisionKey	route:billing	The business action expected for this case
outcome	route	Broad decision category
disposition	automated	Whether the system acted automatically, reviewed, deferred, or did nothing
automationEligible	true	Whether this case belongs in the automation-coverage denominator
tags	["billing", "happy_path"]	Enables later slices by scenario type

For example, these two outputs might both have outcome: "route" but are clearly not equivalent:

{ "decisionKey": "route:billing", "outcome": "route" }

{ "decisionKey": "route:technical", "outcome": "route" }

The first is correct for a duplicate-charge ticket; the second is a routing error. Your quality metric should compare decisionKey, not merely the outer shape.

The four metrics

For a fixed dataset of $N$ cases, define:

$$\text{quality}=\frac{\text{cases with expected decision key}}{N}$$

This is deliberately strict: unexpected exceptions and malformed results count as incorrect.

Let $E$ be the number of cases marked automationEligible.

$$\text{raw automation coverage}=\frac{\text{eligible cases with automated disposition}}{E}$$

Raw coverage alone can be misleading: an incorrectly automated decision still raises the number.

$$\text{safe automation coverage}=\frac{\text{eligible cases that are both correct and automated}}{E}$$

For this support workflow, safe automation coverage is the release-relevant coverage metric. It rewards automation only when it is correct.

Finally, request volume is the sum of actual Jev API requests:

$$\text{Jev request volume}=\sum _{i=1}^N\text{Jev requests for case }i$$

Do not infer this from model.attempted. A future workflow might use speculative fan-out or multiple stages: one decision can involve several requests. Instrument the actual Jev adapter and return its counter explicitly.

---

Create the uv project and fixed dataset

Keep this evaluator separate from the production service’s request path, but place it in the same repository or a closely versioned companion repository. It should call real, authenticated internal evaluation endpoints for each candidate workflow version.

uv init support-workflow-evals --python 3.12
cd support-workflow-evals
uv sync

This initial runner uses only the Python standard library, so there is no package dependency to add yet. uv.lock still matters: commit it along with the evaluator so that collaborators and CI recreate the same Python environment.

If you want a quick refresher on the relevant uv flow, Corey Schafer’s video covers project creation and execution without manually activating environments.

Create this layout:

support-workflow-evals/
├── data/
│   └── support-triage-v1.jsonl
├── reports/
├── eval-config.json
├── runner.py
├── pyproject.toml
└── uv.lock

Use JSON Lines for the dataset: one independently reviewable case per line. Here is data/support-triage-v1.jsonl:

{"id":"billing-duplicate-charge-001","state":{"status":"open","subject":"Charged twice","message":"My card was charged twice for my subscription. Please help.","accountTier":"standard"},"expected":{"decisionKey":"route:billing","outcome":"route","disposition":"automated"},"automationEligible":true,"tags":["billing","happy_path"]}
{"id":"credential-risk-001","state":{"status":"open","subject":"Reset access","message":"I cannot log in. Here is my password so you can verify the account: example-only-secret.","accountTier":"standard"},"expected":{"decisionKey":"security_review","outcome":"security_review","disposition":"review"},"automationEligible":false,"tags":["security","sensitive_content"]}
{"id":"ambiguous-returns-001","state":{"status":"open","subject":"Returns","message":"returns","accountTier":"standard"},"expected":{"decisionKey":"manual_triage","outcome":"manual_triage","disposition":"review"},"automationEligible":false,"tags":["ambiguous","minimal_context"]}
{"id":"closed-ticket-001","state":{"status":"closed","subject":"Old issue","message":"This was already resolved.","accountTier":"standard"},"expected":{"decisionKey":"no_action:closed","outcome":"no_action","disposition":"no_action"},"automationEligible":false,"tags":["deterministic","no_model_needed"]}

The examples above are intentionally small. Your real dataset should grow from reviewed support decisions and production failure analysis.

Two data-handling rules are non-negotiable:

1. Keep the dataset fixed during a comparison. Do not edit labels, cases, or the expected contract between baseline and candidate runs.
2. Treat it as sensitive engineering data. Redact customer identifiers and secrets, avoid committing raw support text to a public repository, and restrict access as appropriate.

A useful practical split is:

• support-triage-v1.jsonl: stable release-gate dataset.
• support-triage-candidates.jsonl: newly discovered cases under review.
• Later, after human labelling, promote candidate cases into a versioned stable set.

---

Define the endpoint contract once

Your Python runner should not replicate TypeScript policy logic. It should invoke each real workflow version and consume a small evaluation response contract.

An internal endpoint can return a shape like this:

{
  "decision": {
    "decisionKey": "route:billing",
    "outcome": "route",
    "disposition": "automated"
  },
  "inspection": {
    "latency": {
      "totalMs": 487,
      "modelMs": 461
    },
    "model": {
      "attempted": true,
      "status": "completed"
    }
  },
  "metrics": {
    "jevRequestCount": 1
  }
}

This contract deliberately builds on the decision inspector from the previous lesson:

• decision supplies the domain action used for quality scoring.
• inspection.latency.totalMs is the service’s own measured end-to-end time.
• metrics.jevRequestCount is an adapter-level counter, not a guess based on inspection fields.

For a circuit-open fallback, a valid response might instead include:

{
  "decision": {
    "decisionKey": "manual_triage:jev_circuit_open",
    "outcome": "manual_triage",
    "disposition": "review"
  },
  "inspection": {
    "latency": {
      "totalMs": 3,
      "modelMs": null
    },
    "model": {
      "attempted": false,
      "status": "circuit_open"
    }
  },
  "metrics": {
    "jevRequestCount": 0
  }
}

That is still a completed evaluation case. It may be incorrect relative to its expected decision, but it is operationally distinct from a server crash or malformed response.

Keep version identities immutable

Use Git SHA-like names rather than vague labels such as new:

{
  "dataset": "data/support-triage-v1.jsonl",
  "timeoutMs": 8000,
  "versions": [
    {
      "name": "baseline-4a72c1e",
      "url": "http://localhost:3000/internal/eval/triage?version=baseline-4a72c1e"
    },
    {
      "name": "candidate-9d03f6b",
      "url": "http://localhost:3000/internal/eval/triage?version=candidate-9d03f6b"
    }
  ]
}

Save this as eval-config.json.

The endpoint must be access-controlled. Put any internal endpoint token in an environment variable such as EVAL_AUTH_TOKEN, never in the config file, dataset, report, or source code.

---

Implement the Python runner

Create runner.py. It has four responsibilities:

1. Load and validate a fixed JSONL dataset.
2. Send every case to every workflow version.
3. Alternate version order for adjacent cases to reduce simple “first version always gets warmer network conditions” bias.
4. Write a report containing records, summaries, and baseline-relative deltas.

from __future__ import annotations

import argparse
import hashlib
import json
import math
import os
import time
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen


@dataclass(frozen=True)
class EvalCase:
    case_id: str
    state: dict[str, Any]
    expected_key: str
    expected_outcome: str
    expected_disposition: str
    automation_eligible: bool
    tags: list[str]


@dataclass(frozen=True)
class WorkflowVersion:
    name: str
    url: str


@dataclass
class RunRecord:
    case_id: str
    version: str
    expected_key: str
    observed_key: str | None
    observed_outcome: str | None
    observed_disposition: str | None
    correct: bool
    caller_latency_ms: int
    service_latency_ms: int | None
    jev_request_count: int | None
    error: str | None
    tags: list[str]


def load_cases(path: Path) -> list[EvalCase]:
    cases: list[EvalCase] = []
    seen_ids: set[str] = set()

    for line_number, line in enumerate(path.read_text().splitlines(), start=1):
        if not line.strip():
            continue

        raw = json.loads(line)
        case_id = raw["id"]

        if case_id in seen_ids:
            raise ValueError(f"Duplicate case id at line {line_number}: {case_id}")

        seen_ids.add(case_id)
        expected = raw["expected"]

        cases.append(
            EvalCase(
                case_id=case_id,
                state=raw["state"],
                expected_key=expected["decisionKey"],
                expected_outcome=expected["outcome"],
                expected_disposition=expected["disposition"],
                automation_eligible=raw["automationEligible"],
                tags=raw.get("tags", []),
            )
        )

    if not cases:
        raise ValueError("Evaluation dataset contains no cases.")

    return cases


def percentile(values: list[int], percent: float) -> int | None:
    if not values:
        return None

    ordered = sorted(values)
    index = math.ceil(percent * len(ordered)) - 1
    return ordered[index]


def post_case(
    version: WorkflowVersion,
    case: EvalCase,
    timeout_seconds: float,
) -> RunRecord:
    payload = json.dumps(
        {
            "caseId": case.case_id,
            "ticket": case.state,
        }
    ).encode("utf-8")

    headers = {"Content-Type": "application/json"}
    token = os.environ.get("EVAL_AUTH_TOKEN")

    if token:
        headers["Authorization"] = f"Bearer {token}"

    request = Request(
        version.url,
        data=payload,
        headers=headers,
        method="POST",
    )

    started = time.perf_counter()

    try:
        with urlopen(request, timeout=timeout_seconds) as response:
            body = json.loads(response.read().decode("utf-8"))

        caller_latency_ms = round((time.perf_counter() - started) * 1000)

        decision = body["decision"]
        inspection = body["inspection"]
        metrics = body["metrics"]

        observed_key = decision["decisionKey"]
        observed_outcome = decision["outcome"]
        observed_disposition = decision["disposition"]

        return RunRecord(
            case_id=case.case_id,
            version=version.name,
            expected_key=case.expected_key,
            observed_key=observed_key,
            observed_outcome=observed_outcome,
            observed_disposition=observed_disposition,
            correct=observed_key == case.expected_key,
            caller_latency_ms=caller_latency_ms,
            service_latency_ms=inspection["latency"]["totalMs"],
            jev_request_count=metrics["jevRequestCount"],
            error=None,
            tags=case.tags,
        )

    except (HTTPError, URLError, TimeoutError, KeyError, TypeError, ValueError) as error:
        caller_latency_ms = round((time.perf_counter() - started) * 1000)

        return RunRecord(
            case_id=case.case_id,
            version=version.name,
            expected_key=case.expected_key,
            observed_key=None,
            observed_outcome=None,
            observed_disposition=None,
            correct=False,
            caller_latency_ms=caller_latency_ms,
            service_latency_ms=None,
            jev_request_count=None,
            error=f"{type(error).__name__}: {error}",
            tags=case.tags,
        )


def rate(numerator: int, denominator: int) -> float | None:
    if denominator == 0:
        return None

    return numerator / denominator


def summarize(records: list[RunRecord], cases: list[EvalCase]) -> dict[str, Any]:
    eligible_ids = {
        case.case_id
        for case in cases
        if case.automation_eligible
    }

    correct_count = sum(record.correct for record in records)
    error_count = sum(record.error is not None for record in records)

    automated_eligible = [
        record
        for record in records
        if record.case_id in eligible_ids
        and record.observed_disposition == "automated"
    ]

    safe_automated_eligible = [
        record
        for record in automated_eligible
        if record.correct
    ]

    caller_latencies = [record.caller_latency_ms for record in records]

    service_latencies = [
        record.service_latency_ms
        for record in records
        if record.service_latency_ms is not None
    ]

    known_request_counts = [
        record.jev_request_count
        for record in records
        if record.jev_request_count is not None
    ]

    return {
        "cases": len(records),
        "quality": rate(correct_count, len(records)),
        "correctCases": correct_count,
        "errors": error_count,
        "automationEligibleCases": len(eligible_ids),
        "rawAutomationCoverage": rate(
            len(automated_eligible),
            len(eligible_ids),
        ),
        "safeAutomationCoverage": rate(
            len(safe_automated_eligible),
            len(eligible_ids),
        ),
        "callerLatencyMs": {
            "p50": percentile(caller_latencies, 0.50),
            "p95": percentile(caller_latencies, 0.95),
        },
        "serviceLatencyMs": {
            "p50": percentile(service_latencies, 0.50),
            "p95": percentile(service_latencies, 0.95),
        },
        "jevRequests": {
            "knownTotal": sum(known_request_counts),
            "meanPerCase": rate(sum(known_request_counts), len(records)),
            "unknownCount": len(records) - len(known_request_counts),
        },
    }


def compare_to_baseline(
    baseline: dict[str, Any],
    candidate: dict[str, Any],
) -> dict[str, Any]:
    return {
        "qualityPercentagePoints": round(
            100 * (candidate["quality"] - baseline["quality"]),
            2,
        ),
        "safeAutomationCoveragePercentagePoints": round(
            100 * (
                candidate["safeAutomationCoverage"]
                - baseline["safeAutomationCoverage"]
            ),
            2,
        ),
        "callerP95LatencyMs": (
            candidate["callerLatencyMs"]["p95"]
            - baseline["callerLatencyMs"]["p95"]
        ),
        "knownJevRequestDelta": (
            candidate["jevRequests"]["knownTotal"]
            - baseline["jevRequests"]["knownTotal"]
        ),
    }


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--config", required=True)
    parser.add_argument("--out", required=True)
    args = parser.parse_args()

    config_path = Path(args.config)
    config = json.loads(config_path.read_text())

    dataset_path = Path(config["dataset"])
    cases = load_cases(dataset_path)

    versions = [
        WorkflowVersion(name=item["name"], url=item["url"])
        for item in config["versions"]
    ]

    if len(versions) < 2:
        raise ValueError("Provide a baseline and at least one candidate version.")

    timeout_seconds = config.get("timeoutMs", 8000) / 1000
    records: list[RunRecord] = []

    for index, case in enumerate(cases):
        ordered_versions = versions if index % 2 == 0 else list(reversed(versions))

        for version in ordered_versions:
            record = post_case(version, case, timeout_seconds)
            records.append(record)

            status = "correct" if record.correct else "incorrect"
            print(f"{version.name} {case.case_id}: {status}")

    summaries = {
        version.name: summarize(
            [record for record in records if record.version == version.name],
            cases,
        )
        for version in versions
    }

    baseline_name = versions[0].name
    baseline_summary = summaries[baseline_name]

    comparisons = {
        version.name: compare_to_baseline(
            baseline_summary,
            summaries[version.name],
        )
        for version in versions[1:]
    }

    report = {
        "dataset": {
            "path": str(dataset_path),
            "sha256": hashlib.sha256(dataset_path.read_bytes()).hexdigest(),
            "caseCount": len(cases),
        },
        "baseline": baseline_name,
        "summaries": summaries,
        "comparisons": comparisons,
        "records": [asdict(record) for record in records],
    }

    output_path = Path(args.out)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(json.dumps(report, indent=2))
    print(f"Wrote report: {output_path}")


if __name__ == "__main__":
    main()

Run it with:

export EVAL_AUTH_TOKEN="your-internal-evaluation-token"

uv run python runner.py \
  --config eval-config.json \
  --out reports/triage-comparison.json

The runner alternates version order by case. Case 1 runs baseline then candidate; case 2 runs candidate then baseline. This is not a substitute for a proper performance experiment, but it avoids the obvious bias of letting one version always run after connections, caches, or rate limits have changed.

---

Read the report as an engineering decision, not a single score

A report summary might look like this:

{
  "baseline": "baseline-4a72c1e",
  "summaries": {
    "baseline-4a72c1e": {
      "quality": 0.88,
      "safeAutomationCoverage": 0.61,
      "callerLatencyMs": { "p50": 506, "p95": 940 },
      "jevRequests": {
        "knownTotal": 46,
        "meanPerCase": 0.92,
        "unknownCount": 0
      }
    },
    "candidate-9d03f6b": {
      "quality": 0.92,
      "safeAutomationCoverage": 0.68,
      "callerLatencyMs": { "p50": 521, "p95": 1064 },
      "jevRequests": {
        "knownTotal": 53,
        "meanPerCase": 1.06,
        "unknownCount": 0
      }
    }
  },
  "comparisons": {
    "candidate-9d03f6b": {
      "qualityPercentagePoints": 4.0,
      "safeAutomationCoveragePercentagePoints": 7.0,
      "callerP95LatencyMs": 124,
      "knownJevRequestDelta": 7
    }
  }
}

This candidate improved quality and safe automation coverage, but it also increased p95 latency and Jev request volume. That is not automatically a rejection or acceptance. It is the evidence needed for the next lesson’s release gate.

Inspect individual records before trusting aggregates

Aggregate metrics tell you that something changed. The records array tells you where.

For any incorrect result, inspect:

• expected_key and observed_key
• case tags such as security, ambiguous, or deterministic
• observed_disposition
• service latency and caller latency
• Jev request count
• any error string

A few interpretation patterns are especially valuable:

Finding	Likely meaning	Next action
Higher raw coverage, flat safe coverage	More automated decisions, but not more correct automation	Inspect new automatically handled errors
Quality improves, requests rise	Additional Jev stage may be worthwhile	Check whether the quality gain justifies cost and latency
P95 rises, median stable	Tail failures, retries, or rate limits may dominate	Inspect records near the latency tail
unknownCount is nonzero	The evaluator could not obtain a structured workflow result	Treat this as an operational regression, not missing data
Deterministic closed-ticket cases call Jev	Eligibility logic regressed	Fix locally; model calls should remain at zero
Security tags regress	Candidate may be unsafe despite better overall quality	Add a category-specific release constraint

Do not “fix” an inconvenient result by editing expected labels after seeing the candidate. If a label was genuinely wrong, correct it in a separate dataset revision, record why, and rerun both versions against the new immutable dataset.

---

Make the runner trustworthy

Before relying on it in CI, verify these implementation properties:

• The dataset hash appears in every report.
• Every configured version receives every dataset case exactly once.
• A malformed endpoint response produces an explicit failed record rather than silently disappearing.
• A known timeout or Jev fallback returned by the workflow remains a structured decision with jevRequestCount: 0 or the actual count.
• A transport failure has unknownCount incremented, so reported request volume is not falsely precise.
• callerLatencyMs measures the complete evaluator-side request, while serviceLatencyMs comes from the decision inspector and excludes network overhead.
• safeAutomationCoverage, rather than raw automation rate, is used when discussing increased automation.
• The evaluator runs with a real internal endpoint and real Jev integration credentials, while secrets remain outside version control.

For an initial local run, a dataset of 10 to 20 carefully reviewed cases is enough to validate the mechanism. For a meaningful release comparison, include ordinary cases, ambiguous requests, sensitive-content cases, deterministic short-circuits, fallback scenarios, and every prior production failure that should never recur.

---

You now have a uv-based Python evaluation runner that compares real workflow versions on one fixed labelled dataset. It records strict decision quality, distinguishes raw from safe automation coverage, measures latency from two useful vantage points, and counts actual Jev usage rather than guessing from model status.

Next, you will turn this report into a release gate: predefined targets for quality, coverage, p95 latency, request volume, and fallback behavior will decide whether a candidate is allowed to ship.