phase2-runtime-ab.yml

name: Phase 2 Runtime A-B

on:
  pull_request:
    types: [opened, synchronize, reopened, labeled]
  workflow_dispatch:
    inputs:
      ab_factor:
        description: "Independent A/B factor (text cache requires block-active)"
        required: true
        default: "scenario-config"
        type: choice
        options: ["scenario-config", "legacy-text-component-cache"]
      scenario:
        description: "Runtime workload (choose block-active for text cache)"
        required: true
        default: "static-steady"
        type: choice
        options: ["static-steady", "block-idle", "block-active", "block-direct-write"]
      runs:
        description: "Restart-isolated runs (4, 8, or formal 12)"
        required: true
        default: "12"
        type: choice
        options: ["4", "8", "12"]
      items:
        description: "Logical items (max 8192) or fully tracked workload blocks (max 1024)"
        required: true
        default: "1024"
        type: string
      warmup_seconds:
        description: "Warmup after the real TCP client is ready"
        required: true
        default: "120"
        type: string
      settle_seconds:
        description: "Scene settling time before sampling"
        required: true
        default: "20"
        type: string
      measure_seconds:
        description: "MSPT/TPS sampling window (diagnostic when Spark is enabled)"
        required: true
        default: "180"
        type: string
      spark_profile_mode:
        description: "Optional 4-run Spark diagnostic (cpu=slow ticks, cpu-all=steady-state)"
        required: true
        default: "none"
        type: choice
        options: ["none", "cpu", "cpu-all", "alloc"]

permissions:
  contents: read

jobs:
  clean-runtime-ab:
    name: Paper 26.1.2 runtime ABBA
    if: >-
      github.event_name == 'workflow_dispatch' ||
      github.event.action != 'labeled' ||
      github.event.label.name == 'phase2-runtime-formal'
    runs-on: ubuntu-latest
    timeout-minutes: 130
    env:
      CAMPAIGN_AB_FACTOR: ${{ github.event_name == 'workflow_dispatch' && inputs.ab_factor || 'scenario-config' }}
      CAMPAIGN_SCENARIO: ${{ github.event_name == 'workflow_dispatch' && inputs.scenario || 'static-steady' }}
      CAMPAIGN_RUNS: ${{ github.event_name == 'workflow_dispatch' && inputs.runs || github.event.action == 'labeled' && '12' || '4' }}
      CAMPAIGN_ITEMS: ${{ github.event_name == 'workflow_dispatch' && inputs.items || github.event.action == 'labeled' && '4096' || '1024' }}
      CAMPAIGN_WARMUP_SECONDS: ${{ github.event_name == 'workflow_dispatch' && inputs.warmup_seconds || github.event.action == 'labeled' && '120' || '10' }}
      CAMPAIGN_SETTLE_SECONDS: ${{ github.event_name == 'workflow_dispatch' && inputs.settle_seconds || github.event.action == 'labeled' && '20' || '5' }}
      CAMPAIGN_MEASURE_SECONDS: ${{ github.event_name == 'workflow_dispatch' && inputs.measure_seconds || github.event.action == 'labeled' && '180' || '10' }}
      CAMPAIGN_SPARK_PROFILE_MODE: ${{ github.event_name == 'workflow_dispatch' && inputs.spark_profile_mode || 'none' }}
      CAMPAIGN_EVIDENCE_KIND: ${{ github.event_name == 'workflow_dispatch' && inputs.spark_profile_mode != 'none' && 'diagnostic' || 'clean' }}

    steps:
      - name: Check out immutable test source
        uses: actions/checkout@v4
        with:
          persist-credentials: false

      - name: Set up Java 25
        uses: actions/setup-java@v4
        with:
          distribution: temurin
          java-version: "25"

      - name: Set up Node 24 for the isolated protocol peer
        uses: actions/setup-node@v4
        with:
          node-version: "24"

      - name: Set up Gradle
        uses: gradle/actions/setup-gradle@v4

      - name: Validate harness source
        run: |
          bash -n tools/perf/prepare-phase2-protocol-client.sh
          bash -n tools/perf/run-phase2-runtime-once.sh
          node --check tools/perf/phase2-protocol-client.js
          pwsh -NoProfile -File tools/perf/analyze-phase2-abba.ps1 -SelfTest
          pwsh -NoProfile -File tools/perf/analyze-phase2-pcap.ps1 -SelfTest

      - name: Build and test the production plugin
        run: ./gradlew clean check shadowJar --no-daemon --no-build-cache --rerun-tasks

      - name: Download stable Paper 26.1.2
        env:
          PAPER_USER_AGENT: InteractionVisualizer-Phase2/1.0 (https://github.com/EllanServer/InteractionVisualizer)
        run: |
          mkdir -p phase2-dependencies
          BUILDS=$(curl --fail --silent --show-error \
            -H "User-Agent: $PAPER_USER_AGENT" \
            https://fill.papermc.io/v3/projects/paper/versions/26.1.2/builds)
          PAPER_URL=$(echo "$BUILDS" | jq -r 'first(.[] | select(.channel == "STABLE") | .downloads."server:default".url) // empty')
          test -n "$PAPER_URL"
          curl --fail --location --show-error \
            -H "User-Agent: $PAPER_USER_AGENT" \
            --output phase2-dependencies/paper.jar "$PAPER_URL"

      - name: Prepare immutable protocol client artifact
        run: bash tools/perf/prepare-phase2-protocol-client.sh phase2-dependencies/protocol-client

      - name: Run restart-isolated runtime campaign
        env:
          AB_FACTOR: ${{ env.CAMPAIGN_AB_FACTOR }}
          SCENARIO: ${{ env.CAMPAIGN_SCENARIO }}
          RUNS: ${{ env.CAMPAIGN_RUNS }}
          ITEMS: ${{ env.CAMPAIGN_ITEMS }}
          WARMUP_SECONDS: ${{ env.CAMPAIGN_WARMUP_SECONDS }}
          SETTLE_SECONDS: ${{ env.CAMPAIGN_SETTLE_SECONDS }}
          MEASURE_SECONDS: ${{ env.CAMPAIGN_MEASURE_SECONDS }}
          SPARK_PROFILE_MODE: ${{ env.CAMPAIGN_SPARK_PROFILE_MODE }}
          EVIDENCE_KIND: ${{ env.CAMPAIGN_EVIDENCE_KIND }}
        run: |
          set -euo pipefail
          [[ "$AB_FACTOR" == scenario-config || "$AB_FACTOR" == legacy-text-component-cache ]]
          [[ "$SCENARIO" == static-steady || "$SCENARIO" == block-idle || \
             "$SCENARIO" == block-active || "$SCENARIO" == block-direct-write ]]
          [[ "$RUNS" =~ ^(4|8|12)$ ]]
          [[ "$ITEMS" =~ ^[0-9]+$ ]]
          if [[ "$SCENARIO" == block-* ]]; then
            # TileEntityUpdate.CheckingRange=1 covers the centered 32x32
            # block footprint completely; larger command scenes are valid but
            # are not full-workload clean ABBA evidence.
            (( ITEMS >= 1 && ITEMS <= 1024 ))
          else
            (( ITEMS >= 1 && ITEMS <= 8192 ))
          fi
          [[ "$WARMUP_SECONDS" =~ ^[0-9]+$ ]] && (( WARMUP_SECONDS >= 10 ))
          [[ "$SETTLE_SECONDS" =~ ^[0-9]+$ ]] && (( SETTLE_SECONDS >= 5 ))
          [[ "$MEASURE_SECONDS" =~ ^[0-9]+$ ]] && (( MEASURE_SECONDS >= 10 ))
          [[ "$SPARK_PROFILE_MODE" =~ ^(none|cpu|cpu-all|alloc)$ ]]
          if [[ "$AB_FACTOR" == legacy-text-component-cache ]]; then
            if [[ "$SCENARIO" != block-active ]]; then
              echo "legacy-text-component-cache A/B is isolated to block-active" >&2
              exit 64
            fi
            if [[ "$SPARK_PROFILE_MODE" == none && "$RUNS" != 12 ]]; then
              echo "clean legacy-text-component-cache evidence is formal-only and requires runs=12" >&2
              exit 64
            fi
            if [[ "$SPARK_PROFILE_MODE" != none && "$RUNS" != 4 ]]; then
              echo "profiled legacy-text-component-cache diagnostics require runs=4" >&2
              exit 64
            fi
            if (( ITEMS < 100 )); then
              echo "legacy-text-component-cache A/B requires at least 100 workload blocks" >&2
              exit 64
            fi
          fi
          if [[ "$SPARK_PROFILE_MODE" == cpu-all && \
                "$SCENARIO" != block-active && "$SCENARIO" != block-direct-write ]]; then
            echo "Spark cpu-all profiling is isolated to block-active or block-direct-write" >&2
            exit 64
          fi
          if [[ "$SPARK_PROFILE_MODE" != none && "$SPARK_PROFILE_MODE" != cpu-all && \
                "$SCENARIO" != block-direct-write ]]; then
            echo "Spark cpu/alloc profiling is isolated to block-direct-write" >&2
            exit 64
          fi
          if [[ "$SPARK_PROFILE_MODE" != none && "$RUNS" != 4 ]]; then
            echo "Spark profiling is diagnostic-only and requires runs=4" >&2
            exit 64
          fi
          if [[ "$SCENARIO" == block-direct-write ]]; then
            # Direct BlockState writes intentionally emit no Bukkit event.
            # Allow the 600-tick audit to begin plus enough ticks for every
            # per-display iterator to cover the <=1024-block scene.
            (( MEASURE_SECONDS >= 45 ))
          fi

          PLUGIN_JAR=$(find build/libs -maxdepth 1 -type f -name 'InteractionVisualizer-*.jar' ! -name '*-sources.jar' ! -name '*-benchmark.jar' -print -quit)
          test -n "$PLUGIN_JAR"
          if [[ "$SPARK_PROFILE_MODE" == none ]]; then
            [[ "$EVIDENCE_KIND" == clean ]]
            CAPTURE_METHOD=none
          else
            [[ "$EVIDENCE_KIND" == diagnostic ]]
            CAPTURE_METHOD="spark-$SPARK_PROFILE_MODE"
          fi
          EVIDENCE_ROOT="phase2-results/$EVIDENCE_KIND"
          mkdir -p "$EVIDENCE_ROOT"
          MANIFEST="$EVIDENCE_ROOT/abba-manifest.csv"
          printf 'Scenario,Block,Position,Variant,RunId,AbFactor,LegacyTextComponentCacheDisableProperty,LegacyTextComponentCacheEnabled,LegacyTextCacheRequests,LegacyTextCacheMisses,LegacyTextCacheHits,LegacyTextCacheHitRate,LegacyTextSameRawFastPaths,ConfigSha256,JvmArgumentsSha256,JvmArgumentsNormalizedSha256,StackSha256,ArtifactSha256,CaptureMethod,SourcePath\n' > "$MANIFEST"
          ARTIFACT_SHA=$(sha256sum "$PLUGIN_JAR" | awk '{print $1}')
          STACK_SHA=$(
            {
              sha256sum phase2-dependencies/paper.jar
              sha256sum phase2-dependencies/protocol-client/client-build-manifest.json
              sha256sum tools/perf/run-phase2-runtime-once.sh
              java -version 2>&1
              printf '%s\n' \
                "abFactor=$AB_FACTOR" "scenario=$SCENARIO" "sparkProfile=$SPARK_PROFILE_MODE"
            } | sha256sum | awk '{print $1}'
          )

          for run_number in $(seq 1 "$RUNS"); do
            block=$(( (run_number - 1) / 4 + 1 ))
            position=$(( (run_number - 1) % 4 + 1 ))
            if (( block % 2 == 1 )); then pattern=ABBA; else pattern=BAAB; fi
            variant=${pattern:$((position - 1)):1}
            if [[ "$AB_FACTOR" == scenario-config ]]; then
              run_id=$(printf '%s_%s_%02d' "${SCENARIO//-/_}" "$variant" "$run_number")
            else
              run_id=$(printf '%s_text_cache_%s_%02d' "${SCENARIO//-/_}" "$variant" "$run_number")
            fi

            PHASE2_PLUGIN_JAR="$PLUGIN_JAR" \
            PHASE2_PAPER_JAR=phase2-dependencies/paper.jar \
            PHASE2_CLIENT_ROOT=phase2-dependencies/protocol-client \
            PHASE2_OUTPUT_ROOT="$EVIDENCE_ROOT" \
            PHASE2_RUN_ID="$run_id" \
            PHASE2_SCENARIO="$SCENARIO" \
            PHASE2_VARIANT="$variant" \
            PHASE2_AB_FACTOR="$AB_FACTOR" \
            PHASE2_ITEM_COUNT="$ITEMS" \
            PHASE2_WARMUP_SECONDS="$WARMUP_SECONDS" \
            PHASE2_SETTLE_SECONDS="$SETTLE_SECONDS" \
            PHASE2_MEASURE_SECONDS="$MEASURE_SECONDS" \
            PHASE2_CAPTURE_ENABLED=0 \
            PHASE2_PROTOCOL_TRACE_ENABLED=0 \
            PHASE2_SPARK_PROFILE_MODE="$SPARK_PROFILE_MODE" \
            bash tools/perf/run-phase2-runtime-once.sh

            python3 - "$MANIFEST" "$EVIDENCE_ROOT" "$SCENARIO" "$block" "$position" \
              "$variant" "$run_id" "$AB_FACTOR" "$STACK_SHA" "$ARTIFACT_SHA" \
              "$CAPTURE_METHOD" <<'PY'
          import csv
          import json
          from pathlib import Path
          import sys

          (
              manifest_text,
              evidence_root_text,
              scenario,
              block,
              position,
              variant,
              run_id,
              ab_factor,
              stack_sha,
              artifact_sha,
              capture_method,
          ) = sys.argv[1:]
          manifest_path = Path(manifest_text)
          evidence_root = Path(evidence_root_text)
          metrics_path = evidence_root / run_id / "iv-perf.json"
          run_manifest_path = evidence_root / run_id / "run-manifest.json"
          metrics = json.loads(metrics_path.read_text(encoding="utf-8"))
          provenance = json.loads(run_manifest_path.read_text(encoding="utf-8"))
          cache = provenance.get("legacyTextComponentCache", {})
          if provenance.get("abFactor") != ab_factor or metrics.get("abFactor") != ab_factor:
              raise SystemExit(f"A/B factor provenance mismatch for {run_id}")
          cache_metric_mapping = {
              "enabled": "legacyTextComponentCache",
              "requests": "legacyTextCacheRequests",
              "misses": "legacyTextCacheMisses",
              "hits": "legacyTextCacheHits",
              "hitRate": "legacyTextCacheHitRate",
              "sameRawFastPaths": "legacyTextSameRawFastPaths",
          }
          for provenance_field, metrics_field in cache_metric_mapping.items():
              if cache.get(provenance_field) != metrics.get(metrics_field):
                  raise SystemExit(
                      f"cache provenance mismatch for {run_id}: {provenance_field}/{metrics_field}"
                  )
          if cache.get("disableProperty") != metrics.get("legacyTextComponentCacheDisableProperty"):
              raise SystemExit(f"cache property provenance mismatch for {run_id}")
          if provenance.get("jvmArgumentsSha256") != metrics.get("jvmArgumentsSha256"):
              raise SystemExit(f"JVM argument provenance mismatch for {run_id}")
          if (provenance.get("jvmArgumentsNormalizedSha256")
                  != metrics.get("jvmArgumentsNormalizedSha256")):
              raise SystemExit(f"normalized JVM argument provenance mismatch for {run_id}")
          process_command_line = provenance.get("jvmDiagnostics", {}).get("processCommandLine", {})
          if (process_command_line.get("formalEvidenceReady") is not True
                  or process_command_line.get("capturedFromProcCmdline") is not True
                  or process_command_line.get("jvmArgumentsSha256")
                  != provenance.get("jvmArgumentsSha256")
                  or process_command_line.get("jvmArgumentsNormalizedSha256")
                  != provenance.get("jvmArgumentsNormalizedSha256")):
              raise SystemExit(f"live JVM process command-line provenance mismatch for {run_id}")
          row = [
              scenario,
              block,
              position,
              variant,
              run_id,
              ab_factor,
              str(cache.get("disableProperty")).lower(),
              str(cache.get("enabled")).lower(),
              cache.get("requests"),
              cache.get("misses"),
              cache.get("hits"),
              cache.get("hitRate"),
              cache.get("sameRawFastPaths"),
              provenance.get("configSha256"),
              provenance.get("jvmArgumentsSha256"),
              provenance.get("jvmArgumentsNormalizedSha256"),
              stack_sha,
              artifact_sha,
              capture_method,
              f"{run_id}/iv-perf.json",
          ]
          with manifest_path.open("a", encoding="utf-8", newline="") as stream:
              csv.writer(stream, lineterminator="\n").writerow(row)
          PY
          done

          if [[ "$SPARK_PROFILE_MODE" == none ]]; then
            minimum_seconds=$(( MEASURE_SECONDS - 2 ))
            incomplete=()
            if [[ "$RUNS" != 12 ]]; then incomplete=(-AllowIncomplete); fi
            for metric in msptP95 msptP99 msptP999 msptMean; do
              pwsh -NoProfile -File tools/perf/analyze-phase2-abba.ps1 "$MANIFEST" \
                -Scenario "$SCENARIO" -Metric "$metric" -Direction LowerIsBetter \
                -MinimumSeconds "$minimum_seconds" "${incomplete[@]}" \
                -OutputJson "$EVIDENCE_ROOT/$metric.analysis.json" -Overwrite
            done
            # At a healthy 20 TPS cap this is only an overload/non-regression
            # diagnostic. MSPT remains the optimization effect-size metric.
            pwsh -NoProfile -File tools/perf/analyze-phase2-abba.ps1 "$MANIFEST" \
              -Scenario "$SCENARIO" -Metric observedTps -Direction HigherIsBetter \
              -MinimumSeconds "$minimum_seconds" "${incomplete[@]}" \
              -OutputJson "$EVIDENCE_ROOT/observedTps.analysis.json" -Overwrite
          fi

          if [[ "$SCENARIO" == block-* ]]; then
            # Active/direct-write CPU time has a meaningful LowerIsBetter
            # direction after lifecycle/config assertions pass. An idle
            # candidate can legitimately perform zero checks in a short
            # window, so it stays diagnostic instead of entering log ratios.
            # Raw check counts never carry a universal performance direction.
            if [[ "$SPARK_PROFILE_MODE" == none && "$SCENARIO" != block-idle ]]; then
              pwsh -NoProfile -File tools/perf/analyze-phase2-abba.ps1 "$MANIFEST" \
                -Scenario "$SCENARIO" -Metric blockUpdateMs -Direction LowerIsBetter \
                -MinimumSeconds "$minimum_seconds" "${incomplete[@]}" \
                -OutputJson "$EVIDENCE_ROOT/blockUpdateMs.analysis.json" -Overwrite
            fi

            python3 - "$MANIFEST" "$SCENARIO" "$RUNS" "$ITEMS" "$CAPTURE_METHOD" "$AB_FACTOR" \
              "$EVIDENCE_ROOT/blockUpdateChecks.evidence.json" <<'PY'
          from pathlib import Path
          import csv
          import json
          import math
          import os
          import re
          import statistics
          import sys

          manifest_path = Path(sys.argv[1])
          scenario = sys.argv[2]
          expected_runs = int(sys.argv[3])
          expected_items = int(sys.argv[4])
          capture_method = sys.argv[5]
          ab_factor = sys.argv[6]
          output_path = Path(sys.argv[7])
          records = []
          with manifest_path.open(encoding="utf-8", newline="") as stream:
              rows = list(csv.DictReader(stream))
          if len(rows) != expected_runs:
              raise SystemExit(f"block evidence expected {expected_runs} runs, found {len(rows)}")

          def parse_bool(value, field, run_id):
              if value == "true":
                  return True
              if value == "false":
                  return False
              raise SystemExit(f"invalid {field} for {run_id}: {value!r}")

          def parse_nonnegative_int(value, field, run_id):
              try:
                  parsed = int(value)
              except (TypeError, ValueError) as error:
                  raise SystemExit(f"invalid {field} for {run_id}: {value!r}") from error
              if parsed < 0:
                  raise SystemExit(f"invalid {field} for {run_id}: {parsed!r}")
              return parsed

          for row in rows:
              if row["Scenario"] != scenario:
                  raise SystemExit(f"manifest scenario drift: {row['Scenario']} != {scenario}")
              if row["CaptureMethod"] != capture_method:
                  raise SystemExit(
                      f"manifest capture-method drift: {row['CaptureMethod']} != {capture_method}"
                  )
              if row["AbFactor"] != ab_factor:
                  raise SystemExit(f"manifest A/B factor drift: {row['AbFactor']} != {ab_factor}")
              source_path = manifest_path.parent / row["SourcePath"]
              metrics = json.loads(source_path.read_text(encoding="utf-8"))
              run_id = row["RunId"]
              if metrics.get("label") != run_id:
                  raise SystemExit(f"metrics label mismatch for {run_id}")
              checks = metrics.get("blockUpdateChecks")
              elapsed_ms = metrics.get("blockUpdateMs")
              if isinstance(checks, bool) or not isinstance(checks, int) or checks < 0:
                  raise SystemExit(f"invalid blockUpdateChecks for {run_id}: {checks!r}")
              if (isinstance(elapsed_ms, bool) or not isinstance(elapsed_ms, (int, float))
                      or not math.isfinite(elapsed_ms) or elapsed_ms < 0):
                  raise SystemExit(f"invalid blockUpdateMs for {run_id}: {elapsed_ms!r}")
              if checks > 0 and elapsed_ms <= 0:
                  raise SystemExit(f"{run_id} recorded {checks} checks but no elapsed time")
              tick_samples = metrics.get("tickSamples")
              if isinstance(tick_samples, bool) or not isinstance(tick_samples, int) or tick_samples <= 0:
                  raise SystemExit(f"invalid tickSamples for {run_id}: {tick_samples!r}")
              variant = row["Variant"]
              expected_event_driven = (
                  ab_factor == "legacy-text-component-cache" or variant == "B"
              )
              if metrics.get("eventDrivenBlockUpdates") is not expected_event_driven:
                  raise SystemExit(f"event-driven config mismatch for {run_id}")
              disable_property = parse_bool(
                  row["LegacyTextComponentCacheDisableProperty"],
                  "LegacyTextComponentCacheDisableProperty",
                  run_id,
              )
              cache_enabled = parse_bool(
                  row["LegacyTextComponentCacheEnabled"],
                  "LegacyTextComponentCacheEnabled",
                  run_id,
              )
              expected_disable_property = (
                  ab_factor == "legacy-text-component-cache" and variant == "A"
              )
              if disable_property is not expected_disable_property:
                  raise SystemExit(f"legacy text cache property mismatch for {run_id}")
              if cache_enabled is not (not disable_property):
                  raise SystemExit(f"legacy text cache enabled state mismatch for {run_id}")
              requests = parse_nonnegative_int(row["LegacyTextCacheRequests"], "requests", run_id)
              misses = parse_nonnegative_int(row["LegacyTextCacheMisses"], "misses", run_id)
              hits = parse_nonnegative_int(row["LegacyTextCacheHits"], "hits", run_id)
              same_raw_fast_paths = parse_nonnegative_int(
                  row["LegacyTextSameRawFastPaths"], "sameRawFastPaths", run_id
              )
              try:
                  hit_rate = float(row["LegacyTextCacheHitRate"])
              except ValueError as error:
                  raise SystemExit(f"invalid cache hit rate for {run_id}") from error
              if (not math.isfinite(hit_rate) or not 0.0 <= hit_rate <= 1.0
                      or misses > requests or hits != requests - misses):
                  raise SystemExit(f"inconsistent legacy text cache metrics for {run_id}")
              derived_hit_rate = hits / requests if requests else 0.0
              if not math.isclose(hit_rate, derived_hit_rate, rel_tol=0.0, abs_tol=0.000001):
                  raise SystemExit(f"legacy text cache hit rate mismatch for {run_id}")
              metric_expectations = {
                  "abFactor": ab_factor,
                  "jvmArgumentsSha256": row["JvmArgumentsSha256"],
                  "jvmArgumentsNormalizedSha256": row["JvmArgumentsNormalizedSha256"],
                  "legacyTextComponentCacheDisableProperty": disable_property,
                  "legacyTextComponentCache": cache_enabled,
                  "legacyTextCacheRequests": requests,
                  "legacyTextCacheMisses": misses,
                  "legacyTextCacheHits": hits,
                  "legacyTextSameRawFastPaths": same_raw_fast_paths,
              }
              for field, expected in metric_expectations.items():
                  if metrics.get(field) != expected:
                      raise SystemExit(f"{field} evidence mismatch for {run_id}")
              if not math.isclose(
                  metrics.get("legacyTextCacheHitRate", -1), hit_rate,
                  rel_tol=0.0, abs_tol=0.000001,
              ):
                  raise SystemExit(f"legacyTextCacheHitRate evidence mismatch for {run_id}")
              config_sha = row["ConfigSha256"]
              jvm_arguments_sha = row["JvmArgumentsSha256"]
              jvm_arguments_normalized_sha = row["JvmArgumentsNormalizedSha256"]
              for field, value in (
                  ("ConfigSha256", config_sha),
                  ("JvmArgumentsSha256", jvm_arguments_sha),
                  ("JvmArgumentsNormalizedSha256", jvm_arguments_normalized_sha),
              ):
                  if re.fullmatch(r"[0-9a-f]{64}", value) is None:
                      raise SystemExit(f"invalid {field} for {run_id}: {value!r}")
              provenance_path = source_path.parent / "run-manifest.json"
              provenance = json.loads(provenance_path.read_text(encoding="utf-8"))
              if (provenance.get("abFactor") != ab_factor
                      or provenance.get("configSha256") != config_sha
                      or provenance.get("jvmArgumentsSha256") != jvm_arguments_sha
                      or provenance.get("jvmArgumentsNormalizedSha256")
                      != jvm_arguments_normalized_sha):
                  raise SystemExit(f"run provenance mismatch for {run_id}")
              cache_provenance = provenance.get("legacyTextComponentCache", {})
              expected_cache_provenance = {
                  "propertyName": "interactionvisualizer.disableLegacyTextComponentCache",
                  "disableProperty": disable_property,
                  "enabled": cache_enabled,
                  "requests": requests,
                  "misses": misses,
                  "hits": hits,
                  "hitRate": hit_rate,
                  "sameRawFastPaths": same_raw_fast_paths,
              }
              if cache_provenance != expected_cache_provenance:
                  raise SystemExit(f"legacy text cache provenance mismatch for {run_id}")
              records.append({
                  "block": int(row["Block"]),
                  "position": int(row["Position"]),
                  "variant": variant,
                  "runId": run_id,
                  "abFactor": ab_factor,
                  "configSha256": config_sha,
                  "jvmArgumentsSha256": jvm_arguments_sha,
                  "jvmArgumentsNormalizedSha256": jvm_arguments_normalized_sha,
                  "legacyTextComponentCacheDisableProperty": disable_property,
                  "legacyTextComponentCacheEnabled": cache_enabled,
                  "legacyTextCacheRequests": requests,
                  "legacyTextCacheMisses": misses,
                  "legacyTextCacheHits": hits,
                  "legacyTextCacheHitRate": hit_rate,
                  "legacyTextSameRawFastPaths": same_raw_fast_paths,
                  "blockUpdateChecks": checks,
                  "tickSamples": tick_samples,
                  "checksPerTick": checks / tick_samples,
                  "blockUpdateMs": elapsed_ms,
                  "msPerCheck": elapsed_ms / checks if checks else None,
                  "sourcePath": source_path.as_posix(),
              })

          def summarize(variant):
              selected = [record for record in records if record["variant"] == variant]
              checks = [record["blockUpdateChecks"] for record in selected]
              checks_per_tick = [record["checksPerTick"] for record in selected]
              times = [record["blockUpdateMs"] for record in selected]
              cache_requests = [record["legacyTextCacheRequests"] for record in selected]
              cache_misses = [record["legacyTextCacheMisses"] for record in selected]
              cache_hits = [record["legacyTextCacheHits"] for record in selected]
              cache_hit_rates = [record["legacyTextCacheHitRate"] for record in selected]
              same_raw_fast_paths = [record["legacyTextSameRawFastPaths"] for record in selected]
              return {
                  "runCount": len(selected),
                  "blockUpdateChecks": {
                      "values": checks,
                      "median": statistics.median(checks),
                      "mean": statistics.fmean(checks),
                  },
                  "checksPerTick": {
                      "values": checks_per_tick,
                      "median": statistics.median(checks_per_tick),
                      "mean": statistics.fmean(checks_per_tick),
                  },
                  "blockUpdateMs": {
                      "values": times,
                      "median": statistics.median(times),
                      "mean": statistics.fmean(times),
                  },
                  "legacyTextComponentCache": {
                      "enabled": sorted({record["legacyTextComponentCacheEnabled"] for record in selected}),
                      "disableProperty": sorted({
                          record["legacyTextComponentCacheDisableProperty"] for record in selected
                      }),
                      "requests": cache_requests,
                      "misses": cache_misses,
                      "hits": cache_hits,
                      "hitRates": cache_hit_rates,
                      "sameRawFastPaths": same_raw_fast_paths,
                      "meanHitRate": statistics.fmean(cache_hit_rates),
                  },
              }

          by_variant = {"A": summarize("A"), "B": summarize("B")}
          config_hashes = sorted({record["configSha256"] for record in records})
          config_hashes_by_variant = {
              variant: sorted({
                  record["configSha256"]
                  for record in records if record["variant"] == variant
              })
              for variant in ("A", "B")
          }
          jvm_hashes_by_variant = {
              variant: sorted({
                  record["jvmArgumentsSha256"]
                  for record in records if record["variant"] == variant
              })
              for variant in ("A", "B")
          }
          normalized_jvm_hashes = sorted({
              record["jvmArgumentsNormalizedSha256"] for record in records
          })
          if any(len(hashes) != 1 for hashes in jvm_hashes_by_variant.values()):
              raise SystemExit("JVM argument SHA drifted within a variant")
          if any(len(hashes) != 1 for hashes in config_hashes_by_variant.values()):
              raise SystemExit("config SHA drifted within a variant")
          if len(normalized_jvm_hashes) != 1:
              raise SystemExit("normalized JVM argument SHA drifted across the campaign")
          if ab_factor == "legacy-text-component-cache":
              if len(config_hashes) != 1:
                  raise SystemExit("cache A/B variants do not share the same config SHA")
              if jvm_hashes_by_variant["A"] == jvm_hashes_by_variant["B"]:
                  raise SystemExit("cache A/B variants unexpectedly share one JVM argument SHA")
              for record in records:
                  if record["legacyTextCacheRequests"] <= 0:
                      raise SystemExit(f"cache A/B workload was not exercised in {record['runId']}")
                  if (record["variant"] == "A"
                          and record["legacyTextCacheMisses"] != record["legacyTextCacheRequests"]):
                      raise SystemExit(f"disabled cache reported hits in {record['runId']}")
                  if (record["variant"] == "B"
                          and record["legacyTextCacheHitRate"] < 0.90):
                      raise SystemExit(
                          "enabled cache missed the 90% steady-state hit-rate guard in "
                          f"{record['runId']}: {record['legacyTextCacheHitRate']:.6f}"
                      )
          else:
              if jvm_hashes_by_variant["A"] != jvm_hashes_by_variant["B"]:
                  raise SystemExit("scenario-config variants unexpectedly changed JVM arguments")
              if config_hashes_by_variant["A"] == config_hashes_by_variant["B"]:
                  raise SystemExit("scenario-config variants unexpectedly share one config SHA")
          cadence_guard = {
              "evaluated": False,
              "reason": "only block-active workloads with at least 100 blocks have a stable mixed-type cadence",
          }
          if scenario == "block-active" and expected_items >= 100:
              baseline = by_variant["A"]["checksPerTick"]["median"]
              candidate = by_variant["B"]["checksPerTick"]["median"]
              if baseline <= 0:
                  raise SystemExit("block-active baseline recorded no checks")
              ratio = candidate / baseline
              if ab_factor == "legacy-text-component-cache":
                  minimum_ratio = 0.95
                  maximum_ratio = 1.05
                  cadence_interpretation = (
                      "Both cache variants use the same event-driven configuration, so their executed "
                      "block-check cadence must remain equivalent."
                  )
              else:
                  minimum_ratio = 0.60
                  maximum_ratio = 0.85
                  cadence_interpretation = (
                      "The active candidate aggregate rate must remain consistent with all three furnace "
                      "cadences while eliminating idle bee polling and noisy level-event invalidations."
                  )
              cadence_guard = {
                  "evaluated": True,
                  "minimumCandidateToBaselineRatio": minimum_ratio,
                  "maximumCandidateToBaselineRatio": maximum_ratio,
                  "candidateToBaselineRatio": ratio,
                  "passed": minimum_ratio <= ratio <= maximum_ratio,
                  "interpretation": cadence_interpretation,
              }

          if capture_method == "none":
              interpretation = (
                  "blockUpdateChecks is workload evidence, not a standalone optimization objective; "
                  "fewer checks can mean either effective event-driven scheduling or missing work. "
                  "For block-idle, zero candidate checks/time in a short window is valid and blockUpdateMs "
                  "has no ratio analysis. Interpret all scenarios with machine-validated scene actions, "
                  "blockUpdateMs, MSPT and TPS."
              )
          else:
              interpretation = (
                  "Instrumented Spark workload evidence only. blockUpdateChecks can establish executed work, "
                  "but MSPT, TPS and blockUpdateMs were collected under profiler overhead and must not be "
                  "used as clean performance effect sizes. Use the saved profile to attribute CPU cost, then "
                  "confirm any optimization in a separate captureMethod=none campaign."
              )

          evidence = {
              "schemaVersion": 2,
              "analysisType": "diagnostic-work-evidence",
              "scenario": scenario,
              "abFactor": ab_factor,
              "requestedRuns": expected_runs,
              "expectedItems": expected_items,
              "captureMethod": capture_method,
              "performanceEvidenceReady": capture_method == "none",
              "formalComplete": expected_runs == 12 and capture_method == "none",
              "blockUpdateChecksDirection": None,
              "blockUpdateMsDirection": (
                  "LowerIsBetter" if capture_method == "none" and scenario != "block-idle" else None
              ),
              "interpretation": interpretation,
              "activeCadenceGuard": cadence_guard,
              "provenance": {
                  "configSha256": config_hashes,
                  "configSha256ByVariant": config_hashes_by_variant,
                  "jvmArgumentsSha256ByVariant": jvm_hashes_by_variant,
                  "jvmArgumentsNormalizedSha256": normalized_jvm_hashes[0],
                  "legacyTextComponentCacheProperty": (
                      "interactionvisualizer.disableLegacyTextComponentCache"
                  ),
                  "legacyTextProcessingTreatmentScope": [
                      "sharedComponentCache",
                      "perEntitySameRawFastPath",
                  ],
              },
              "byVariant": by_variant,
              "runs": records,
          }
          output_path.write_text(
              json.dumps(evidence, ensure_ascii=False, indent=2) + "\n",
              encoding="utf-8",
          )

          step_summary = os.environ.get("GITHUB_STEP_SUMMARY")
          if step_summary:
              with open(step_summary, "a", encoding="utf-8", newline="\n") as stream:
                  stream.write(f"### {scenario} / {ab_factor} block update evidence\n\n")
                  stream.write(
                      f"Capture method: `{capture_method}`; clean performance evidence ready: "
                      f"`{str(capture_method == 'none').lower()}`.\n\n"
                  )
                  stream.write("`blockUpdateChecks` is diagnostic work evidence (no universal direction).\n\n")
                  stream.write(
                      "| Run | Variant | Checks/tick | Cache | Requests | Misses | Hits | Hit rate | Same raw |\n"
                  )
                  stream.write("|---|---:|---:|---:|---:|---:|---:|---:|---:|\n")
                  for record in records:
                      stream.write(
                          f"| {record['runId']} | {record['variant']} | "
                          f"{record['checksPerTick']:.6f} | "
                          f"{'on' if record['legacyTextComponentCacheEnabled'] else 'off'} | "
                          f"{record['legacyTextCacheRequests']} | {record['legacyTextCacheMisses']} | "
                          f"{record['legacyTextCacheHits']} | "
                          f"{record['legacyTextCacheHitRate']:.6f} | "
                          f"{record['legacyTextSameRawFastPaths']} |\n"
                      )
                  if cadence_guard["evaluated"]:
                      stream.write(
                          "\nActive cadence guard: "
                          f"`{'pass' if cadence_guard['passed'] else 'fail'}`; "
                          f"candidate/baseline=`{cadence_guard['candidateToBaselineRatio']:.6f}` "
                          f"(required `{cadence_guard['minimumCandidateToBaselineRatio']:.2f}.."
                          f"{cadence_guard['maximumCandidateToBaselineRatio']:.2f}`).\n"
                      )

          if cadence_guard["evaluated"] and not cadence_guard["passed"]:
              raise SystemExit(
                  "block-active candidate checks/tick escaped the expected mixed-type cadence: "
                  f"ratio={cadence_guard['candidateToBaselineRatio']:.6f}"
              )
          PY
          fi

      - name: Publish runtime evidence
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: phase2-${{ env.CAMPAIGN_EVIDENCE_KIND }}-runtime-${{ env.CAMPAIGN_SCENARIO }}-${{ env.CAMPAIGN_AB_FACTOR }}-${{ github.sha }}-${{ github.run_id }}
          path: |
            phase2-results/${{ env.CAMPAIGN_EVIDENCE_KIND }}
            phase2-dependencies/protocol-client/client-build-manifest.json
            phase2-dependencies/protocol-client/client-files.sha256
            phase2-dependencies/protocol-client/node-minecraft-protocol/package-lock.json
            phase2-dependencies/protocol-client/node-minecraft-protocol/production-lock-inventory.json
          if-no-files-found: warn
          retention-days: 30