{
  "$type": "site.standard.document",
  "bskyPostRef": {
    "cid": "bafyreigv5fp6doe5jlbetax5zy4mtndn72xld3nw5mztyxidu2aiasqm3i",
    "uri": "at://did:plc:25rdn5elo5izoxrmtis34zuk/app.bsky.feed.post/3mpqbyqfjgxk2"
  },
  "coverImage": {
    "$type": "blob",
    "ref": {
      "$link": "bafkreih7nnowrcdhxgnffq7zolr7jnorotamb2i4njqbo3sgjxvfu45iva"
    },
    "mimeType": "image/webp",
    "size": 486068
  },
  "path": "/wonderlab/workflow-series-05-evaluation-framework-three-layer-testing-and-trace-tracking-2857",
  "publishedAt": "2026-07-03T09:37:42.000Z",
  "site": "https://dev.to",
  "tags": [
    "ai",
    "workflow",
    "evaluation",
    "trace",
    "PrimeSkills",
    "Homepage"
  ],
  "textContent": "##  Why Workflows Need a Dedicated Evaluation Framework\n\nTraditional software testing covers code correctness. Workflows add two layers of uncertainty:\n\n  * **LLM output is non-deterministic** : the same input can produce different results across runs\n  * **Cross-step dependencies** : a Phase 3 problem may only surface at Phase 7, making the debugging chain long\n\n\n\nWithout an evaluation framework, every workflow change requires a full end-to-end run: slow, expensive, incomplete coverage. Three-layer testing decomposes the problem.\n\n##  Three-Layer Evaluation Structure\n\n\n    Layer 3: End-to-end tests (Workflow level)\n      Full pipeline from trigger to completion\n      Test cases: eval/cases.yaml\n      Metrics: completion rate, Phase 4 avg rounds, gate trigger rate\n\n    Layer 2: Integration tests (Phase level)\n      Cross-step data flow is correctly passed\n      Cross-phase routing logic fires correctly\n\n    Layer 1: Unit tests (Step level)\n      Each subagent's output matches its output contract\n      No real LLM calls — validates JSON schema only\n\n\n**Test priority:** Layer 1 should be the most numerous and fastest — catches contract violations in seconds. Layer 3 is the slowest and most expensive — run it only when changes affect the main pipeline.\n\n##  Layer 1: Step-Level Unit Tests\n\nUnit tests verify that subagent output files match the declared schema. No real LLM calls needed.\n\n\n\n    # tests/unit/test_phase3_output.py\n    import json\n    from pathlib import Path\n\n    def test_analysis_output_schema():\n        \"\"\"Phase 3 output must conform to analysis_final.json schema\"\"\"\n        output = json.loads(Path(\"test_fixtures/phase3/analysis_final.json\").read_text())\n\n        assert \"passed\" in output\n        assert isinstance(output[\"passed\"], bool)\n        assert \"confidence\" in output\n        assert 0.0 <= output[\"confidence\"] <= 1.0\n        assert \"root_cause\" in output\n        assert isinstance(output[\"root_cause\"], str | type(None))\n        assert \"evidence\" in output\n        assert isinstance(output[\"evidence\"], list)\n\n        # on failure, error field must be present and non-empty\n        if not output[\"passed\"]:\n            assert \"error\" in output\n            assert output[\"error\"]\n\n    def test_fix_candidate_output_schema():\n        \"\"\"Phase 4 candidate output schema\"\"\"\n        for candidate in [\"candidate_a\", \"candidate_b\", \"candidate_c\"]:\n            output_file = Path(f\"test_fixtures/phase4/{candidate}.json\")\n            if output_file.exists():\n                output = json.loads(output_file.read_text())\n                assert \"passed\" in output\n                assert \"test_coverage\" in output\n                assert isinstance(output[\"test_coverage\"], float)\n\n\n**Test fixtures:** save real run outputs as test data, with one successful path and one failure path per subagent. The fixtures document exactly what the contract looks like in practice.\n\n##  Layer 2: Integration Tests\n\nIntegration tests cover two problem types:\n\n**Data flow tests:** verify that Phase N's output can be consumed by Phase N+1.\n\n\n\n    # tests/integration/test_phase_data_flow.py\n\n    def test_phase1_output_satisfies_phase2_context():\n        \"\"\"Phase 1's bug_info.json must include all fields declared in Phase 2's context_inputs\"\"\"\n        bug_info = json.loads(Path(\"test_fixtures/phase1/bug_info.json\").read_text())\n\n        required_fields = [\"summary\", \"stack_trace\", \"jira_key\", \"attachment_path\"]\n        for field in required_fields:\n            assert field in bug_info, f\"Phase 1 output missing field required by Phase 2: {field}\"\n\n    def test_phase3_routing_logic():\n        \"\"\"Phase 3 completion triggers correct routing based on confidence\"\"\"\n        # high confidence → proceed to Phase 4\n        high_conf = {\"passed\": True, \"confidence\": 0.97, \"root_cause\": \"NPE in parseInput\"}\n        assert route_after_phase3(high_conf) == \"phase_4\"\n\n        # medium confidence → trigger Gate A\n        mid_conf = {\"passed\": True, \"confidence\": 0.75, \"root_cause\": \"...\"}\n        assert route_after_phase3(mid_conf) == \"gate_A\"\n\n        # low confidence + retries remaining → retry Phase 3\n        low_conf = {\"passed\": False, \"confidence\": 0.45}\n        assert route_after_phase3(low_conf, retry_count=1) == \"phase_3_retry\"\n\n        # low confidence + retries exhausted → human escalation\n        assert route_after_phase3(low_conf, retry_count=3) == \"human_escalation\"\n\n\nRouting logic implemented as a pure Python function runs all edge cases in milliseconds with no LLM calls.\n\n##  Layer 3: End-to-End Tests and Metric Baselines\n\n###  Test Case Definitions\n\n\n    # eval/cases.yaml\n    cases:\n      - id: WF-E2E-001\n        name: Happy path (high confidence, first-attempt pass)\n        input:\n          jira_key: AE-MOCK-001\n          bug_description: \"NullPointerException in parseInput() when config=null\"\n        expected_flow:\n          - phase_1: done\n          - phase_2: done\n          - phase_3: done (confidence >= 0.95)\n          - phase_4: done (first candidate passes)\n          - phase_5: done\n          - phase_6: done\n          - phase_7: done\n        expected_metrics:\n          e2e_success: true\n          phase4_rounds: 1\n          gates_triggered: []\n\n      - id: WF-E2E-002\n        name: Low confidence path (Gate A triggered)\n        input:\n          jira_key: AE-MOCK-002\n          bug_description: \"Intermittent crash, no reproducible steps\"\n        expected_flow:\n          - phase_3: done (confidence < 0.95)\n          - gate_A: triggered\n\n      - id: WF-E2E-003\n        name: Fix failure path (all candidates fail, Gate B triggered)\n        input:\n          jira_key: AE-MOCK-003\n        expected_flow:\n          - phase_4: all candidates failed\n          - gate_B: triggered\n        expected_metrics:\n          phase4_rounds: 3\n          gates_triggered: [gate_B]\n\n\n###  Core Metric Definitions\n\n\n    End-to-end completion rate   > 70%\n      = fully automated completions / total triggers\n\n    Phase 4 average rounds       < 1.5\n      = mean phase4_rounds across all runs\n      (close to 1: fix quality is good; close to 3: test pass rate is low)\n\n    Parallel candidate pass rate > 80%\n      = fraction of workflows where at least 1 candidate passed\n      (below 80%: root cause analysis quality or fix strategy needs work)\n\n    Gate trigger rate            < 20%\n      = fraction of workflows that triggered any human gate\n      (above 20%: LLM quality or input data quality has a problem)\n\n\n###  Regression Testing\n\nBefore modifying workflow.md / templates / policy.md, establish a baseline with historical cases:\n\n\n\n    # Step 1: run eval before changes, record baseline\n    python run_eval.py --cases eval/cases.yaml --output baseline_v1.3.json\n\n    # Step 2: make workflow changes\n    # ...\n\n    # Step 3: run the same cases again\n    python run_eval.py --cases eval/cases.yaml --output baseline_v1.4.json\n\n    # Step 4: compare delta\n    python compare_eval.py baseline_v1.3.json baseline_v1.4.json\n\n\n\n    # compare_eval.py output\n    Metric               v1.3    v1.4    Delta\n    ───────────────────────────────────────────\n    e2e_success_rate     78%     82%     +4%  ✓\n    phase4_avg_rounds    1.6     1.4     -0.2 ✓\n    gate_trigger_rate    18%     22%     +4%  ⚠️ (above threshold)\n\n\n`gate_trigger_rate` crossing 20% means this change makes certain paths more likely to trigger human review. Investigate before releasing.\n\n##  Trace Tracking\n\nWithout Trace, every workflow run is a black box. When something goes wrong, the team digs through files, compares timestamps, and guesses execution order. With Langfuse, every run has a queryable chain — open the trace, find the phase, read the span.\n\n###  Three-Layer Trace Structure\n\n\n    from langfuse import Langfuse\n\n    langfuse = Langfuse()\n\n    def run_workflow(jira_key: str) -> None:\n        # Workflow-level trace (top layer)\n        trace = langfuse.trace(\n            name=f\"wf-bug-e2e:{jira_key}\",\n            input={\"jira_key\": jira_key},\n            metadata={\"workflow_version\": \"1.3.0\"}\n        )\n\n        for phase_id in get_pending_phases():\n            # Phase-level span\n            span = trace.span(\n                name=phase_id,\n                input={\"context\": get_phase_context(phase_id)}\n            )\n\n            result = execute_phase(phase_id)\n\n            span.end(\n                output={\"status\": result[\"status\"], \"passed\": result[\"passed\"]},\n                level=\"DEFAULT\" if result[\"passed\"] else \"WARNING\"\n            )\n\n        if gate_triggered:\n            trace.event(\n                name=\"human_gate_A\",\n                metadata={\"triggered_by\": \"low_confidence\", \"value\": confidence}\n            )\n\n\n###  What Trace Answers\n\n\n    How long did each phase take?\n      → span start/end timestamps\n\n    Which phase consumed the most tokens?\n      → span usage field\n\n    What was the raw error when a subagent failed?\n      → span output.error field\n\n    Is Phase 3 confidence within a healthy range across runs?\n      → span output.confidence, aggregated across multiple traces\n\n\nNo more guessing execution order or digging through files.\n\n##  Design Checklist\n\n**Unit tests (Layer 1)**\n\n  * [ ] Every subagent output has a schema validation test\n  * [ ] Fixtures cover both success and failure paths\n  * [ ] No real LLM calls — use saved real outputs as fixtures\n\n\n\n**Integration tests (Layer 2)**\n\n  * [ ] Each phase's output fields align with the next phase's context_inputs\n  * [ ] All routing conditions (high/mid/low confidence, timeout, failure) have test coverage\n  * [ ] Routing logic is implemented as a pure function, runnable in milliseconds\n\n\n\n**End-to-end tests (Layer 3)**\n\n  * [ ] eval/cases.yaml covers happy path, low-confidence path, fix-failure path\n  * [ ] 4 core metrics have defined thresholds\n  * [ ] Baseline delta comparison runs before every release; threshold violations block release\n\n\n\n**Trace tracking**\n\n  * [ ] Every workflow run has a top-level trace\n  * [ ] Every phase has a span recording input, output, and latency\n  * [ ] Human gate triggers are recorded as events with reason metadata\n\n\n\n##  Summary\n\n  1. **Three layers, three speeds** : Layer 1 validates contracts with fixtures in seconds, Layer 2 tests data flow and routing in seconds, Layer 3 runs the full pipeline in minutes — the first two catch most problems before Layer 3 runs\n  2. **Metric baselines are release gates** : if end-to-end completion rate, Phase 4 rounds, candidate pass rate, or gate trigger rate crosses a threshold, the change needs investigation\n  3. **Trace turns black boxes into queryable records** : no more guessing execution order or digging through files — search the Langfuse trace for the run and read the span\n\n\n\n_Check out PrimeSkills — a curated marketplace of AI agents and skills that have been validated in real-world, enterprise-grade workflows. No fluff, just what actually works._\n\n_Find more useful knowledge and interesting products on my Homepage_",
  "title": "Workflow Series (05): Evaluation Framework — Three-Layer Testing and Trace Tracking"
}