diff --git a/.github/workflows/accuracy-digest.yml b/.github/workflows/accuracy-digest.yml new file mode 100644 index 000000000..03aa0ffc0 --- /dev/null +++ b/.github/workflows/accuracy-digest.yml @@ -0,0 +1,103 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +name: Accuracy Slack Daily + +on: + workflow_run: + workflows: [E2E Accuracy Matrix, FPM Accuracy Matrix] + types: [completed] + schedule: + # 09:00 Los Angeles is 16:00 UTC in summer, 17:00 in winter. + # The script checks the actual local deadline. Rechecks resume interrupted delivery. + - cron: "0,30 16-18 * * *" + workflow_dispatch: + inputs: + mode: + description: Preview only, or send one test report to the configured channel with a TEST label + type: choice + options: [dry-run, test] + default: dry-run + e2e_run_id: + description: Completed E2E run ID (manual runs are allowed for testing) + type: string + required: true + fpm_run_id: + description: Completed FPM run ID + type: string + required: true + +permissions: + contents: read + actions: read + +concurrency: + group: accuracy-slack-delivery + cancel-in-progress: false + +jobs: + report: + if: >- + github.repository == 'ai-dynamo/aisimulate' && + ((github.event_name == 'workflow_dispatch' && + (inputs.mode == 'dry-run' || github.ref == 'refs/heads/main')) || + (vars.SLACK_ACCURACY_ENABLED == 'true' && github.ref == 'refs/heads/main' && + (github.event_name == 'schedule' || + (github.event.workflow_run.event == 'schedule' && + github.event.workflow_run.head_branch == 'main' && + github.event.workflow_run.head_repository.full_name == github.repository)))) + runs-on: ubuntu-latest + timeout-minutes: 30 + env: + GH_TOKEN: ${{ github.token }} + MODE: ${{ github.event_name == 'workflow_dispatch' && inputs.mode || 'daily' }} + E2E_RUN_ID: ${{ inputs.e2e_run_id }} + FPM_RUN_ID: ${{ inputs.fpm_run_id }} + steps: + # Never check out the triggering accuracy run's code in this privileged consumer. + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + with: + persist-credentials: false + fetch-depth: 0 + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7 + with: + python-version: '3.13' + - name: Prepare report and preview + env: + SLACK_ACCURACY_BOT_TOKEN: ${{ env.MODE != 'dry-run' && secrets.SLACK_ACCURACY_BOT_TOKEN || '' }} + SLACK_ACCURACY_CHANNEL_ID: ${{ vars.SLACK_ACCURACY_CHANNEL_ID }} + run: | + args=(--mode "$MODE" --prepare-only) + if [[ "$MODE" != daily ]]; then + args+=(--e2e-run-id "$E2E_RUN_ID" --fpm-run-id "$FPM_RUN_ID") + fi + python scripts/notify_accuracy.py "${args[@]}" + - name: Preserve frozen report before any Slack write + if: hashFiles('accuracy-preview/report.json') != '' + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 + with: + name: accuracy-report + path: accuracy-preview/report.json + overwrite: true + retention-days: 90 + - name: Preserve readable preview + if: hashFiles('accuracy-preview/preview.md') != '' + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 + with: + name: accuracy-preview + path: accuracy-preview/preview.md + overwrite: true + retention-days: 30 + - name: Deliver main message and detail thread + if: env.MODE != 'dry-run' && hashFiles('accuracy-preview/report.json') != '' + env: + SLACK_ACCURACY_BOT_TOKEN: ${{ secrets.SLACK_ACCURACY_BOT_TOKEN }} + SLACK_ACCURACY_CHANNEL_ID: ${{ vars.SLACK_ACCURACY_CHANNEL_ID }} + run: python scripts/notify_accuracy.py --mode "$MODE" --deliver accuracy-preview/report.json + - name: Preserve delivered baseline and recovery state + if: hashFiles('accuracy-preview/delivery/state.json') != '' + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 + with: + name: accuracy-delivery-state + path: accuracy-preview/delivery/state.json + overwrite: true + retention-days: 90 diff --git a/.github/workflows/fpm-accuracy-branch.yml b/.github/workflows/fpm-accuracy-branch.yml index 7eb75b7f1..f989221ad 100644 --- a/.github/workflows/fpm-accuracy-branch.yml +++ b/.github/workflows/fpm-accuracy-branch.yml @@ -156,7 +156,8 @@ jobs: --commit "${EXPECTED_SHA}" --branch "${EVALUATED_BRANCH}" \ --evaluator-sha "${GITHUB_SHA}" --hf-revision "${HF_REVISION}" \ --run-id "${GITHUB_RUN_ID}" --run-attempt "${GITHUB_RUN_ATTEMPT}" \ - --output "${RUNNER_TEMP}/accuracy-public" + --output "${RUNNER_TEMP}/accuracy-public" \ + --comparison-output "${RUNNER_TEMP}/accuracy-comparison/comparison.json" - name: Report advisory accuracy and coverage run: | python - <<'PYCODE' >> "${GITHUB_STEP_SUMMARY}" @@ -179,3 +180,12 @@ jobs: path: ${{ runner.temp }}/accuracy-public/ if-no-files-found: error retention-days: 90 + + - name: Preserve point evidence for notification comparisons + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 + with: + name: fpm-accuracy-comparison-${{ inputs.artifact_key }} + overwrite: true + path: ${{ runner.temp }}/accuracy-comparison/comparison.json + if-no-files-found: error + retention-days: 90 diff --git a/docs/accuracy-slack.md b/docs/accuracy-slack.md new file mode 100644 index 000000000..c7de6a63e --- /dev/null +++ b/docs/accuracy-slack.md @@ -0,0 +1,170 @@ +# Daily accuracy Slack report + +`Accuracy Slack Daily` posts one daily top-level message to +**#swdl-dynamo-aisim-daily** (`C0BULBSTXJ6`). Alerts and results share that message; +E2E per-model and per-GPU tables are included in the main message. No mentions are +sent. The bot does not accept commands or start evaluations. + +The message title is `:rainbow: *Accuracy Daily · YYYY-MM-DD*`, with a rainbow +emoji and bold text, without a timezone suffix. Immediately below it, a quote +block shows run links with duration and attempt, overview links, and short +alert/recovery summaries. The alert count and comparison-note count appear at +the end of the main message. + +## Delivery policy + +- Select today's scheduled E2E and FPM runs by their creation date in + `America/Los_Angeles`. A rerun uses its latest attempt; successful branch + artifacts retained from an earlier attempt are validated against that attempt. +- Send when both runs finish, including failures/cancellations. Completion of + either producer wakes the notifier. A scheduled fallback checks at 09:00 local + time (UTC 16:00 in summer, 17:00 in winter) and reports missing/unfinished runs. + GitHub schedules can be delayed; 09:00 is the intended cutoff, not a wall-clock SLA. +- Each main message includes both pipeline links. An unstarted pipeline links to + its workflow. Failed jobs include their job links and failing step names. +- Include every branch evaluated in those runs. Do not substitute yesterday's + data for a missing branch. Manual campaigns never enter production daily reports. +- After delivery, late completions do not update the frozen report or create + another daily message. Subsequent daily reports can announce recovery. +- Serialize notifier runs. Identify production messages using Slack metadata + keyed by repository and local date. Persist the frozen report before posting; + on retry, restore it and send only missing thread parts. Rechecks at 30-minute + intervals through 18:30 UTC can resume an interrupted delivery. Do not delete + the bot's messages or its retained report artifacts; they are delivery records. +- History/thread reads use GET query parameters and omit empty pagination cursors. +- Slack timeouts and `ok:false` fail the notification job. An uncertain POST is + not blindly retried: the next run checks Slack history/thread metadata first. + An extended Slack/GitHub outage may prevent delivery; no system can guarantee + a notification while its delivery service is unavailable. + +## Report contents and comparison + +The E2E table has six columns: Branch, Overall, vLLM, SGLang, TRT-LLM, Coverage. +Each accuracy cell is **TPOT%/TTFT% MAPE** (for example `24.32%/43.67%`) +for AISimulate, not legacy AIC. +Main-message detail tables independently group by model and by GPU, within each branch, +combining frameworks. They average successful point errors directly; they do +not average already-aggregated MAPEs. Coverage is predicted / eligible points, +not all original source measurements. FPM shows KV-warmup on/off and online +regression MAPE, weighted by successful prediction counts. FPM coverage is in a +thread reply to keep the main table narrow. The parent uses single-line spacing, +shows at most three short alert summaries, and puts comparison notes and full +alert details in the thread. Every numerical MAPE value includes `%`. + +For each branch: + +- Start with the most recent earlier qualified scheduled campaign in the retained + 90-day history. Download only the requested branch's artifact and comparison + evidence during this search. Without one, explicitly report an initial baseline. +- Require identical measurement revision/content and evaluation-rule fingerprints. + The rule fingerprint covers evaluator source and its branch workflow at the + producer revision, not the evaluated AISimulate package revision. The FPM + fingerprint includes `scripts/accuracy_digest.py`, which supplies its point + codec. A change + establishes a new baseline and reports comparison unavailable, not recovery. +- Compare each E2E topology/configuration and FPM configuration/predictor on the + intersection of successfully predicted points. Alert when MAPE increases by + **at least 2 percentage points AND at least 10% relative**. From zero, an + increase of at least 2 points qualifies. +- A previously successful point becoming failed, unavailable, or absent alerts + separately, even if total coverage is unchanged because other points improved. +- Keep the pre-alert baseline while the regression persists. Do not silently + accept yesterday's regression as today's normal. Successful comparable results + against that baseline can be marked recovered in a later daily message. +- Initial prediction failures are shown as attention notes; without prior evidence + they are not labeled new regressions. A green pipeline does not imply every + prediction succeeded. + +FPM previously retained only aggregates, which cannot detect exchanged successful +and failed points. The producer now writes a separate +`fpm-accuracy-comparison-` artifact containing a hash of observation order +and compressed little-endian float64 percentage-error sequences (-1 for unsuccessful predictions). It binds +to the public summary checksum and exact producer snapshot. Reruns overwrite +both public and comparison artifacts so their attempt identities stay aligned. The notifier verifies +counts and means against that qualified summary. No raw measured/predicted +latencies are exported. Existing Pages artifacts remain unchanged. Old FPM runs +without point evidence can be displayed, but point regression checks are explicitly +unavailable until a comparable pair with evidence exists. + +## Slack setup + +GitHub Actions runs the notifier; the Slack app supplies its bot identity. No +server, Event Subscriptions, Socket Mode, or slash commands are needed. + +At https://api.slack.com/apps, choose **Create New App > From scratch**, name it +`AISimulate Daily`, and select the workspace containing the daily channel. +Under **OAuth & Permissions > Bot Token Scopes**, add: + +- `chat:write` to post the main message and replies. +- `channels:history` to check prior messages and resume threads in this public + channel. If using a private channel later, use `groups:history` as well. +- `metadata.message:read` to read daily-report and thread-part metadata. + +Click **Install to Workspace** (or request workspace approval), then copy the +**Bot User OAuth Token** beginning with `xoxb-`. After changing scopes on an +installed app, reinstall it to apply the new permissions. + +In **#swdl-dynamo-aisim-daily**, open **Channel details > Integrations > Add apps** +and add `AISimulate Daily`. Store its bot token only as repository +secret `SLACK_ACCURACY_BOT_TOKEN`. Configure repository variables: + +| Variable | Value | +| --- | --- | +| `SLACK_ACCURACY_CHANNEL_ID` | `C0BULBSTXJ6` | +| `SLACK_ACCURACY_ENABLED` | `true` only after the test below passes | + +The previous `SLACK_ACCURACY_WEBHOOK_URL` and Slack Workflow Builder trigger +are not used by this version. Messages use Slack `mrkdwn`: bold headings, +monospaced tables, and named links. This is not GitHub-flavored Markdown; tables +are rendered inside code blocks rather than as Markdown table syntax. + +Both test and production use this same channel. Test messages are labeled +`[TEST]` and use separate metadata, so they do not consume the daily production +slot or update its baseline. Automatic delivery is disabled unless explicitly +enabled. GitHub permissions are read-only (`contents:read`, `actions:read`). +The notifier checks out its own trusted workflow revision; it never executes +code from downloaded artifacts. Pages validation remains strict; explicit manual +preview/test alone permits same-repository producer runs from a non-main branch. + +## Dry run and one real Slack test + +Open **Actions → Accuracy Slack Daily → Run workflow**. Choose `dry-run` +(the default), and provide completed `e2e_run_id` and `fpm_run_id` values. Manual +accuracy runs are allowed here. For example: + +```bash +gh workflow run accuracy-digest.yml --ref main \ + -f mode=dry-run -f e2e_run_id=36643620072 -f fpm_run_id=36558333115 +``` + +Before the workflow is merged/registered, run the same dependency-free Python +entrypoint from its checkout with a read-only GitHub credential: + +```bash +GH_TOKEN="$(gh auth token)" python3 scripts/notify_accuracy.py \ + --e2e-run-id 36643620072 --fpm-run-id 36558333115 \ + --output /tmp/accuracy-preview +``` + +Inspect the Actions Summary and `accuracy-preview` / `accuracy-report` artifacts, +or local `preview.md` / `report.json`. The JSON includes alert reasons, comparison +limitations, and the exact baseline snapshots. Dry run does not read Slack +credentials, post messages, or write production baseline state. + +After merging and configuring the bot/channel, run once with `mode=test`: + +```bash +gh workflow run accuracy-digest.yml --ref main \ + -f mode=test -f e2e_run_id=36643620072 -f fpm_run_id=36558333115 +``` + +This sends a real `[TEST]` parent message plus its detail thread to the daily +channel. Check desktop/mobile table readability, links, and per-model/per-GPU +sections, then set `SLACK_ACCURACY_ENABLED=true`. Test sending is restricted to +`main`; a branch dry run receives no Slack token. No manual mode sends a +production daily report. + +A workflow dry run does not rerun predictors. Use the deterministic tests in +`tests/fpm_accuracy/test_digest.py` to exercise regressions, coverage losses, +changed datasets, missing runs, daylight-saving cutoffs, deduplication, and +interrupted thread delivery without sending fake alarms to Slack. diff --git a/docs/ci.md b/docs/ci.md index 40273afac..c1c21d0b7 100644 --- a/docs/ci.md +++ b/docs/ci.md @@ -528,6 +528,16 @@ snapshot until that release has a qualified campaign. See the [accuracy campaign contract](../pages/e2e-accuracy/README.md) for pinned scheduler settings, measurement selection, and provenance. +### Daily accuracy Slack report + +[Accuracy Slack Daily](../.github/workflows/accuracy-digest.yml) combines the day's +scheduled E2E and FPM results in one message in **#swdl-dynamo-aisim-daily**. +Failures and comparable regressions appear in that message; separate per-model +and per-GPU E2E tables appear in the main message; FPM coverage and comparison +details remain in its thread. Delivery waits for both pipelines, +with a 09:00 America/Los_Angeles fallback. It is opt-in and has a default dry-run +mode plus an explicit test-send mode. See [setup, comparisons, and testing](accuracy-slack.md). + ## Reading results and troubleshooting | What you see | Meaning and next check | diff --git a/scripts/accuracy_digest.py b/scripts/accuracy_digest.py new file mode 100644 index 000000000..b8295f64e --- /dev/null +++ b/scripts/accuracy_digest.py @@ -0,0 +1,303 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Pure reductions and Slack rendering for the daily accuracy report.""" + +from __future__ import annotations + +import base64 +import hashlib +import html +import json +import sys +import zlib +from array import array +from collections import defaultdict +from statistics import mean + +METHODS = ("warmup", "nowarmup", "regression") + + +def encode_points(values): + """Little-endian float64 percentage errors; -1 denotes an unsuccessful prediction.""" + packed = array("d", (-1 if value is None else value for value in values)) + if sys.byteorder != "little": + packed.byteswap() + return {"count": len(packed), "data": base64.b64encode(zlib.compress(packed.tobytes())).decode()} + + +def decode_points(packed): + count = packed["count"] + if type(count) is not int or not 0 <= count <= 10_000_000: + raise ValueError("invalid point sequence length") + decoder = zlib.decompressobj() + raw = decoder.decompress(base64.b64decode(packed["data"], validate=True), count * 8 + 1) + if len(raw) != count * 8 or not decoder.eof or decoder.unused_data or decoder.unconsumed_tail: + raise ValueError("invalid compressed point sequence") + result = array("d") + result.frombytes(raw) + if sys.byteorder != "little": + result.byteswap() + return result + + +def identity(value): + return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest() + + +def e2e_snapshot(summary, rules, url): + campaign = summary["snapshot"]["campaign"] + groups = {} + for model in summary["models"]: + for workload in model["workloads"]: + for gpu in workload["gpus"]: + for topology in gpu["topologies"]: + key = identity( + [ + model["model"], + workload["identity"], + gpu["gpu"], + topology["id"], + ] + ) + framework = topology["framework"].removeprefix("dynamo-") + framework = "trtllm" if framework == "trt" else framework + points = {} + for point in topology["points"]: + point_key = str(point["concurrency"]) + if point_key in points: + raise ValueError("duplicate E2E point identity") + points[point_key] = ( + [abs(point["aisimulate"][metric + "_error_pct"]) for metric in ("tpot", "ttft")] + if point["status"] == "success" + else None + ) + if key in groups: + raise ValueError("duplicate E2E topology identity") + groups[key] = { + "label": f"{model['model']} / {gpu['gpu']} / {framework} / {topology['id']}", + "model": model["model"], + "gpu": gpu["gpu"], + "framework": framework, + "points": points, + } + return { + "kind": "e2e", + "branch": campaign["branch"], + "commit": campaign["commit_sha"], + "dataset": campaign["measurement_sha256"], + "rules": rules, + "url": url, + "groups": groups, + } + + +def fpm_snapshot(summary, rules, url, evidence=None): + groups = {} + for row in summary["rows"]: + row_key = row["configuration_id"] + "/" + row["snapshot_id"] + for method, result in row["results"].items(): + key = row_key + "/" + method + samples = evidence[row_key]["methods"].get(method) if evidence is not None else None + groups[key] = { + "label": f"{row['model']} / {row['gpu']} / {row['configuration_id']} / {method}", + "model": row["model"], + "gpu": row["gpu"], + "method": method, + "membership": row["membership_sha256"], + "protocol": [ + row["protocol_id"], + row["parser_policy_id"], + row["ordering"], + ], + "points": samples, + "point_order": evidence[row_key]["order_sha256"] if evidence is not None else None, + "metric": result["metrics"]["all"], + } + snapshot = summary["snapshot"] + return { + "kind": "fpm", + "branch": snapshot["branch"], + "commit": snapshot["commit_sha"], + "dataset": snapshot["hf_revision"], + "rules": rules, + "url": url, + "groups": groups, + } + + +def compare(current, previous): + """Compare successful common points; lost successes alert independently of MAPE.""" + if previous is None: + return [], ["Initial baseline; no earlier qualified comparison."] + if current["dataset"] != previous["dataset"] or current["rules"] != previous["rules"]: + return [], ["Dataset or evaluation rules changed; baseline reset, no regression conclusion."] + alerts, notes = [], [] + fpm = current["kind"] == "fpm" + metrics = ("forward-pass",) if fpm else ("TPOT", "TTFT") + for key, old in previous["groups"].items(): + new = current["groups"].get(key) + if old["points"] is None or (new is not None and new["points"] is None): + notes.append("Point evidence unavailable; coverage and common-point regression comparison unavailable.") + continue + before_points = decode_points(old["points"]) if fpm else old["points"] + successful = ( + {i for i, v in enumerate(before_points) if v >= 0} + if fpm + else {k for k, v in before_points.items() if v is not None} + ) + if new is None: + if successful: + alerts.append(f"{old['label']}: {len(successful)} previously predicted points missing.") + continue + if old.get("protocol") != new.get("protocol") or ( + fpm and (old.get("membership"), old.get("point_order")) != (new.get("membership"), new.get("point_order")) + ): + notes.append(f"{new['label']}: measurement membership/order or protocol changed; not comparable.") + continue + after_points = decode_points(new["points"]) if fpm else new["points"] + after_success = ( + {i for i, v in enumerate(after_points) if v >= 0} + if fpm + else {k for k, v in after_points.items() if v is not None} + ) + lost = successful - after_success + if lost: + alerts.append(f"{new['label']}: {len(lost)} previously predicted points now missing/unavailable/failed.") + common = successful & after_success + if not common: + continue + for index, metric in enumerate(metrics): + before = mean(before_points[k] if fpm else before_points[k][index] for k in common) + after = mean(after_points[k] if fpm else after_points[k][index] for k in common) + delta = after - before + if delta >= 2 - 1e-9 and delta >= before * 0.10 - 1e-9: + relative = f"{delta / before * 100:.1f}%" if before else "from zero" + alerts.append( + f"{new['label']}: {metric} MAPE {before:.2f}% -> {after:.2f}% " + f"(+{delta:.2f} pp, {relative}; {len(common)} common points)." + ) + return alerts, sorted(set(notes)) + + +def reduce_e2e(groups): + points = [point for group in groups for point in group["points"].values()] + success = [point for point in points if point is not None] + pair = "/".join(f"{mean(p[i] for p in success):.2f}%" for i in (0, 1)) if success else "N/A" + return pair, f"{len(success)}/{len(points)}" + + +def reduce_fpm(groups, *, coverage=True): + metrics = [group["metric"] for group in groups] + count = sum(m["predicted_count"] for m in metrics) + measured = sum(m["measured_count"] for m in metrics) + mape = sum((m["mape_pct"] or 0) * m["predicted_count"] for m in metrics) / count if count else None + value = f"{mape:.2f}%" if mape is not None else "N/A" + return f"{value} ({count}/{measured})" if coverage else value + + +def table(headers, rows): + # Keep arbitrary artifact strings from breaking code fences or generating mentions. + def clean(value): + return str(value).replace("`", "'").replace("\n", " ").replace("\r", " ") + + rows = [[clean(c) for c in row] for row in [headers, *rows]] + widths = [max(len(row[i]) for row in rows) for i in range(len(headers))] + return ( + "```\n" + + "\n".join(" ".join(cell.ljust(width) for cell, width in zip(row, widths, strict=True)) for row in rows) + + "\n```" + ) + + +def escape(text): + return html.escape(str(text), quote=False) + + +def messages(day, pipelines, snapshots, alerts, notes, recovered=()): + run_links = [] + for kind, pipeline in pipelines.items(): + status = pipeline["status"] + detail = status.removeprefix("success").strip() if status.startswith("success") else f"({status})" + run_links.append(f"<{pipeline['url']}|{kind.upper()} run>{escape(detail)}") + links = " · ".join(run_links) + summary = f"> *:alert: {len(alerts)} alert(s)*" + if notes: + summary += f" · {len(notes)} comparison note(s) in thread" + lines = [ + f":rainbow: *Accuracy Daily · {day}*", + "> " + links + " · " + " · ", + ] + lines.extend("> " + escape("• " + (alert if len(alert) <= 240 else alert[:237] + "...")) for alert in alerts[:3]) + if len(alerts) > 3: + lines.append(f"> {len(alerts) - 3} additional alerts in thread.") + lines.extend("> " + escape("• Recovered: " + item) for item in recovered) + rows = [] + for branch, snapshot in sorted(snapshots.get("e2e", {}).items()): + groups = list(snapshot["groups"].values()) + overall, coverage = reduce_e2e(groups) + row = [branch, overall] + row.extend(reduce_e2e([g for g in groups if g["framework"] == f])[0] for f in ("vllm", "sglang", "trtllm")) + rows.append([*row, coverage]) + lines += [ + "*E2E · TPOT/TTFT MAPE*", + escape(table(["Branch", "Overall", "vLLM", "SGLang", "TRT-LLM", "Coverage"], rows)) + if rows + else "No qualified E2E results.", + ] + for branch, snapshot in sorted(snapshots.get("e2e", {}).items()): + for dimension in ("model", "gpu"): + buckets = defaultdict(list) + for group in snapshot["groups"].values(): + buckets[group[dimension]].append(group) + entries = [[label, *reduce_e2e(groups)] for label, groups in sorted(buckets.items())] + lines.extend( + [ + f"*E2E · {escape(branch)} · per {dimension}*", + escape(table([dimension.title(), "TPOT/TTFT MAPE", "Coverage"], entries)), + ] + ) + rows = [] + for branch, snapshot in sorted(snapshots.get("fpm", {}).items()): + rows.append( + [ + branch, + *[ + reduce_fpm([g for g in snapshot["groups"].values() if g["method"] == m], coverage=False) + for m in METHODS + ], + ] + ) + lines += [ + "*FPM · MAPE*", + escape(table(["Branch", "KV on", "KV off", "Regression"], rows)) if rows else "No qualified FPM results.", + ] + lines.append(summary) + replies = [] + coverage_rows = [] + for branch, snapshot in sorted(snapshots.get("fpm", {}).items()): + cells = [] + for method in METHODS: + metrics = [g["metric"] for g in snapshot["groups"].values() if g["method"] == method] + cells.append(f"{sum(m['predicted_count'] for m in metrics)}/{sum(m['measured_count'] for m in metrics)}") + coverage_rows.append([branch, *cells]) + if coverage_rows: + replies.append( + "*FPM coverage · predicted/eligible*\n" + + links + + "\n" + + escape(table(["Branch", "KV on", "KV off", "Regression"], coverage_rows)) + + "\nCoverage counts eligible points. Run artifacts are authoritative; Pages may lag." + ) + for offset in range(0, len(alerts) + len(notes), 8): + replies.append( + "*Comparison details*\n" + + links + + "\n" + + "\n".join(escape("• " + line) for line in (alerts + notes)[offset : offset + 8]) + ) + root = "\n".join(lines) + # Slack hard limit is 40k; refuse instead of silently losing branches or evidence. + if any(len(message) > 35000 for message in [root, *replies]): + raise ValueError("Slack message too large; review report size before delivery") + return root, replies diff --git a/scripts/fpm_accuracy/README.md b/scripts/fpm_accuracy/README.md index 7be594895..0a20393ae 100644 --- a/scripts/fpm_accuracy/README.md +++ b/scripts/fpm_accuracy/README.md @@ -14,7 +14,9 @@ removed. Unused presentation metadata and its conversion helpers are omitted; the worker schema and MoE mapping required for evaluation are retained. Unsupported measurement protocols remain visible as unsupported configurations; missing protocol identities and corrupt inputs still fail closed. `evaluate.py` reduces each shared measurement stream directly into -overview aggregates, without local reports, raw result exports, or history. +overview aggregates. Optional notification evidence retains observation-order hashes +and compressed percentage-error sequences separately from public Pages artifacts; it does not retain +raw measured or predicted latencies. See [daily reporting](../../docs/accuracy-slack.md). The public `skipped_count` combines excluded and unavailable source observations; the Overview labels this count “excluded or unavailable.” It does not mean that all of these observations were deliberately filtered out. diff --git a/scripts/fpm_accuracy/evaluate.py b/scripts/fpm_accuracy/evaluate.py index 4c75ae424..9b97753a2 100644 --- a/scripts/fpm_accuracy/evaluate.py +++ b/scripts/fpm_accuracy/evaluate.py @@ -4,16 +4,24 @@ from __future__ import annotations +import hashlib +import json import math +from array import array from collections.abc import Callable from dataclasses import asdict, dataclass from typing import Any +from accuracy_digest import encode_points + from fpm_accuracy.exceptions import DependencyError from fpm_accuracy.hf.models import MeasurementCase from fpm_accuracy.models.aic_predictors import AicFpmPredictor, AicRegressionPredictor from fpm_accuracy.models.fpt_predictor import ForwardPassTimePredictor, PredictorContext -from fpm_accuracy.models.worker_regression import WorkerRegressionPredictor, infer_worker_roles +from fpm_accuracy.models.worker_regression import ( + WorkerRegressionPredictor, + infer_worker_roles, +) WORKLOADS = ("all", "prefill", "decode", "mixed") METHODS = ("warmup", "nowarmup", "regression") @@ -60,8 +68,10 @@ def score( artifact: Any = None, *, factory: Callable = create_predictor, + include_points: bool = False, ) -> dict: """Keep every eligible outcome; targets reach regression only after scoring.""" + points = array("d") metrics = {workload: Metric() for workload in WORKLOADS} status = "evaluated" predictor = None @@ -71,7 +81,9 @@ def score( status = "no_fpm_input" else: context = PredictorContext( - worker=case.configuration.worker_config_record, worker_role=case.worker_role, fpm_artifact=artifact + worker=case.configuration.worker_config_record, + worker_role=case.worker_role, + fpm_artifact=artifact, ) try: if method == "regression": @@ -81,7 +93,10 @@ def score( predictor = factory("aic-fpm", context) except Exception as exc: # Raw diagnostics stay in Actions logs, never in the public JSON. - print(f"{case.configuration_id} {method}: construction failed: {exc}", flush=True) + print( + f"{case.configuration_id} {method}: construction failed: {exc}", + flush=True, + ) construction_failed = not isinstance(exc, DependencyError) status = "predictor_error" if construction_failed else "unsupported_predictor" try: @@ -100,6 +115,8 @@ def score( raise ValueError("invalid prediction") except Exception: predicted, error = None, True + if include_points: + points.append(-1 if error or predicted is None else abs(predicted - actual) / actual * 100) # Score before the model sees this target. for key in ("all", workload): metrics[key].add(actual, predicted, error, False) @@ -115,6 +132,7 @@ def score( if predictor is not None: predictor.close() return { + **({"_points": encode_points(points)} if include_points else {}), "status": status, "artifact": None if artifact is None @@ -134,12 +152,21 @@ def rank(result): metric = result["metrics"]["all"] coverage = metric["predicted_count"] / metric["measured_count"] if metric["measured_count"] else 0 mape = metric["mape_pct"] - return (-coverage, mape if mape is not None else math.inf, result["artifact"]["id"]) + return ( + -coverage, + mape if mape is not None else math.inf, + result["artifact"]["id"], + ) return min(results, key=rank) -def evaluate_case(case: MeasurementCase, *, factory: Callable = create_predictor) -> dict: +def evaluate_case( + case: MeasurementCase, + *, + factory: Callable = create_predictor, + comparison: dict | None = None, +) -> dict: config = case.configuration ids = [item.observation_id for item in case.observations] orders = [item.order for item in case.observations] @@ -147,7 +174,7 @@ def evaluate_case(case: MeasurementCase, *, factory: Callable = create_predictor raise ValueError("measurement stream must have unique IDs and strictly increasing order") results = {} if str(case.status) == "ready": - results["regression"] = score(case, "regression", factory=factory) + results["regression"] = score(case, "regression", factory=factory, include_points=comparison is not None) for mode in METHODS[:2]: variants = [ item @@ -155,10 +182,26 @@ def evaluate_case(case: MeasurementCase, *, factory: Callable = create_predictor if ("nowarmup" if ".kv-off." in item.path.lower() else "warmup") == mode ] results[mode] = ( - choose_variant([score(case, mode, item, factory=factory) for item in variants]) + choose_variant( + [ + score( + case, + mode, + item, + factory=factory, + include_points=comparison is not None, + ) + for item in variants + ] + ) if variants - else score(case, mode, factory=factory) + else score(case, mode, factory=factory, include_points=comparison is not None) ) + if comparison is not None: + comparison[config.configuration_id + "/" + config.snapshot_id] = { + "order_sha256": hashlib.sha256(json.dumps(ids).encode()).hexdigest(), + "methods": {method: result.pop("_points") for method, result in results.items()}, + } return { "configuration_id": config.configuration_id, "configuration_path": config.configuration_path, diff --git a/scripts/notify_accuracy.py b/scripts/notify_accuracy.py new file mode 100644 index 000000000..5ffd24856 --- /dev/null +++ b/scripts/notify_accuracy.py @@ -0,0 +1,561 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Preview or deliver one daily accuracy message plus its detail thread.""" + +from __future__ import annotations + +import argparse +import functools +import hashlib +import io +import json +import math +import os +import re +import subprocess +import urllib.error +import urllib.request +import uuid +import zipfile +import zlib +from datetime import UTC, date, datetime, time, timedelta +from pathlib import Path +from urllib.parse import urlencode +from zoneinfo import ZoneInfo + +from accuracy_digest import compare, decode_points, e2e_snapshot, fpm_snapshot, messages +from prepare_e2e_accuracy_pages import ( + api, + api_items, + artifact_key, + strict_json, + unpack_artifact, + validate_artifact, +) +from prepare_fpm_accuracy_pages import unpack as unpack_fpm +from prepare_fpm_accuracy_pages import validate_artifact as validate_fpm + +REPO = "ai-dynamo/aisimulate" +LA = ZoneInfo("America/Los_Angeles") +WORKFLOWS = {"e2e": "e2e-accuracy.yml", "fpm": "fpm-accuracy.yml"} +REPORT_WORKFLOW = ".github/workflows/accuracy-digest.yml" + + +def parse_time(value): + return datetime.fromisoformat(value.replace("Z", "+00:00")) + + +def report_day(run): + return parse_time(run["created_at"]).astimezone(LA).date() + + +def scheduled_runs(kind, since, until): + page = 1 + while True: + path = ( + f"actions/workflows/{WORKFLOWS[kind]}/runs?event=schedule&branch=main" + f"&created={since}..{until}&per_page=100&page={page}" + ) + batch = api(path)["workflow_runs"] + yield from batch + if len(batch) < 100: + return + page += 1 + + +def select_run(kind, day): + candidates = [r for r in scheduled_runs(kind, day, day + timedelta(days=1)) if report_day(r) == day] + return max(candidates, key=lambda r: (r["created_at"], r["id"]), default=None) + + +def ready(runs, day, now): + return all(r and r["status"] == "completed" for r in runs.values()) or ( + now.astimezone(LA) >= datetime.combine(day, time(9), LA) + ) + + +def validate_run(run, kind, *, allow_manual_branch=False): + if not ( + run["path"] == ".github/workflows/" + WORKFLOWS[kind] + and (run["head_branch"] == "main" or (allow_manual_branch and run["event"] == "workflow_dispatch")) + and run["event"] in {"schedule", "workflow_dispatch"} + and run["repository"]["full_name"] == REPO + and run["head_repository"]["full_name"] == REPO + ): + raise ValueError("unexpected accuracy producer") + + +@functools.lru_cache(maxsize=32) +def rules_digest(kind, sha): + paths = ( + [ + "scripts/run_e2e_accuracy.py", + "scripts/build_e2e_accuracy_overview.py", + "scripts/fetch_accuracy_measurements.py", + ] + if kind == "e2e" + else ["scripts/run_fpm_accuracy.py", "scripts/fpm_accuracy", "scripts/accuracy_digest.py"] + ) + paths += [f".github/workflows/{kind}-accuracy-branch.yml"] + exists = subprocess.run(["git", "cat-file", "-e", sha], capture_output=True) + if exists.returncode: + subprocess.run( + ["git", "fetch", "--no-tags", "origin", sha], + check=True, + capture_output=True, + ) + tree = subprocess.check_output(["git", "ls-tree", "-r", sha, "--", *paths]) + if not tree: + raise ValueError("missing evaluator source") + return hashlib.sha256(tree).hexdigest() + + +def json_zip(archive, name, limit=256 * 1024 * 1024): + with zipfile.ZipFile(io.BytesIO(archive)) as bundle: + if bundle.namelist() != [name] or bundle.getinfo(name).file_size > limit: + raise ValueError("unexpected notification artifact") + return strict_json(bundle.read(name)) + + +def fpm_points(archive, summary, public_archive): + data = json_zip(archive, "comparison.json") + with zipfile.ZipFile(io.BytesIO(public_archive)) as bundle: + digest = hashlib.sha256(bundle.read("summary.json")).hexdigest() + if data["schema_version"] != 1 or data["snapshot"] != summary["snapshot"] or data["summary_sha256"] != digest: + raise ValueError("FPM point evidence identity/checksum mismatch") + expected = {row["configuration_id"] + "/" + row["snapshot_id"]: row for row in summary["rows"]} + if set(data["points"]) != set(expected): + raise ValueError("FPM point evidence configuration mismatch") + for key, row in expected.items(): + evidence = data["points"][key] + if not re.fullmatch(r"[0-9a-f]{64}", evidence["order_sha256"]): + raise ValueError("invalid FPM observation order hash") + if set(evidence["methods"]) != set(row["results"]): + raise ValueError("FPM point evidence predictor mismatch") + for method, result in row["results"].items(): + points = decode_points(evidence["methods"][method]) + metric = result["metrics"]["all"] + if any(not math.isfinite(v) or (v < 0 and v != -1) for v in points): + raise ValueError("invalid FPM point error") + values = [v for v in points if v >= 0] + if len(points) != metric["measured_count"] or len(values) != metric["predicted_count"]: + raise ValueError("FPM point evidence coverage mismatch") + if values and not math.isclose(sum(values) / len(values), metric["mape_pct"], rel_tol=1e-9, abs_tol=1e-9): + raise ValueError("FPM point evidence aggregate mismatch") + return data["points"] + + +def load_snapshots(kind, run, *, allow_manual_branch=False, only_branch=None): + """Validate each artifact against its original attempt, including preserved retry successes.""" + if not run or run["status"] != "completed": + return {}, [] + validate_run(run, kind, allow_manual_branch=allow_manual_branch) + artifacts = api_items(f"actions/runs/{run['id']}/artifacts", "artifacts") + companions = {a["name"]: a for a in artifacts if not a["expired"]} + snapshots, warnings, attempts = {}, [], {} + wanted = None if only_branch is None else f"{kind}-accuracy-web-{artifact_key(only_branch)}" + for artifact in artifacts: + if wanted is not None and artifact["name"] != wanted: + continue + if not re.fullmatch(kind + r"-accuracy-web-[0-9a-f]{16}", artifact["name"]): + continue + try: + if artifact["expired"]: + raise ValueError("artifact expired") + archive = api(f"actions/artifacts/{artifact['id']}/zip", binary=True) + summary = unpack_artifact(archive) if kind == "e2e" else unpack_fpm(archive) + snapshot = summary["snapshot"]["campaign"] if kind == "e2e" else summary["snapshot"] + number = int(snapshot["run_attempt"]) + if snapshot["run_id"] != str(run["id"]) or not 1 <= number <= run["run_attempt"]: + raise ValueError("artifact run attempt mismatch") + if number not in attempts: + endpoint = f"actions/runs/{run['id']}/attempts/{number}" + attempt = run if number == run["run_attempt"] else api(endpoint) + attempts[number] = attempt, api_items(endpoint + "/jobs", "jobs") + attempt, jobs = attempts[number] + if attempt["head_sha"] != run["head_sha"] or attempt["id"] != run["id"]: + raise ValueError("attempt source mismatch") + if kind == "e2e": + validate_artifact( + archive, attempt, artifact_name=artifact["name"], jobs=jobs, allow_manual_branch=allow_manual_branch + ) + normalized = e2e_snapshot(summary, rules_digest(kind, run["head_sha"]), run["html_url"]) + else: + validate_fpm( + archive, + attempt, + artifact["name"], + jobs, + allow_manual_branch=allow_manual_branch, + ) + evidence = None + name = artifact["name"].replace("-web-", "-comparison-") + if name in companions: + evidence = fpm_points( + api( + f"actions/artifacts/{companions[name]['id']}/zip", + binary=True, + ), + summary, + archive, + ) + normalized = fpm_snapshot( + summary, + rules_digest(kind, run["head_sha"]), + run["html_url"], + evidence, + ) + branch = normalized["branch"] + if branch in snapshots: + raise ValueError("duplicate qualified branch") + snapshots[branch] = normalized + except (ValueError, KeyError, TypeError, zipfile.BadZipFile, zlib.error) as exc: + warnings.append(f"{kind.upper()} {artifact['name']}: invalid or unavailable evidence ({exc}).") + except urllib.error.HTTPError as exc: + if exc.code not in {404, 410}: + raise + warnings.append(f"{kind.upper()} {artifact['name']}: evidence unavailable (HTTP {exc.code}).") + return snapshots, warnings + + +def pipeline_status(kind, run, now, *, allow_manual_branch=False): + url = f"https://github.com/{REPO}/actions/workflows/{WORKFLOWS[kind]}" + if not run: + return {"url": url, "status": "NOT STARTED"}, [f"{kind.upper()}: scheduled run missing."] + validate_run(run, kind, allow_manual_branch=allow_manual_branch) + url = run["html_url"] + status = run.get("conclusion") or "still running at deadline" + problems = [] if status == "success" else [f"{kind.upper()}: {status} ({url})."] + if run["status"] == "completed": + jobs = api_items(f"actions/runs/{run['id']}/attempts/{run['run_attempt']}/jobs", "jobs") + for job in jobs: + if job["conclusion"] in { + "failure", + "cancelled", + "timed_out", + "action_required", + }: + steps = ", ".join(s["name"] for s in job.get("steps", []) if s["conclusion"] == "failure") + problems.append(f"{kind.upper()} {job['name']}: {steps or job['conclusion']} ({job['html_url']}).") + end = max( + (parse_time(j["completed_at"]) for j in jobs if j["completed_at"]), + default=now, + ) + minutes = (end - parse_time(run["run_started_at"])).total_seconds() / 60 + status += f" ({minutes:.0f} min, attempt {run['run_attempt']})" + return {"url": url, "status": status}, problems + + +def prior_state(day): + # Successful scheduled deliveries only. Dry runs and test-channel sends cannot change production baselines. + page = 1 + while page <= 10: + try: + runs = api( + f"actions/workflows/accuracy-digest.yml/runs?status=success&branch=main&per_page=100&page={page}" + )["workflow_runs"] + except urllib.error.HTTPError as exc: + if exc.code == 404: + return {"baselines": {}, "active": []} + raise + for run in runs: + if run["event"] not in {"schedule", "workflow_run"} or run["path"] != REPORT_WORKFLOW: + continue + for item in api_items(f"actions/runs/{run['id']}/artifacts", "artifacts"): + if item["name"] != "accuracy-delivery-state" or item["expired"]: + continue + state = json_zip( + api(f"actions/artifacts/{item['id']}/zip", binary=True), + "state.json", + ) + if ( + state.get("schema_version") == 1 + and state.get("production_sent") is True + and state["day"] < str(day) + ): + return state + if len(runs) < 100: + break + page += 1 + return {"baselines": {}, "active": []} + + +def initial_baseline(kind, branch, day): + # Bounded to the producers' 90-day artifact retention. No manual baseline substitution. + for listed in scheduled_runs(kind, day - timedelta(days=90), day): + if report_day(listed) >= day or listed["status"] != "completed": + continue + run = api(f"actions/runs/{listed['id']}") + found, _ = load_snapshots(kind, run, only_branch=branch) + if branch in found: + return found[branch] + return None + + +def build_report(day, runs, state, now, historical=True, allow_manual_branch=False): + pipelines, snapshots, alerts, notes, active = {}, {}, [], [], [] + baselines = dict(state.get("baselines", {})) + comparable = set() + for kind, run in runs.items(): + pipelines[kind], failures = pipeline_status(kind, run, now, allow_manual_branch=allow_manual_branch) + alerts.extend(failures) + if failures: + active.append(kind + ":pipeline") + snapshots[kind], warnings = load_snapshots(kind, run, allow_manual_branch=allow_manual_branch) + alerts.extend(warnings) + if warnings: + active.append(kind + ":evidence") + # Discover evaluated branches from jobs, rather than reporting a newly created branch as missing. + if run and run["status"] == "completed": + jobs = api_items(f"actions/runs/{run['id']}/jobs", "jobs") + branches = {m.group(1) for j in jobs if (m := re.match(r"(?:E2E|FPM) accuracy \((.+)\) /", j["name"]))} + if not branches and run["conclusion"] == "success": + alerts.append(f"{kind.upper()}: completed run has no evaluated branch jobs.") + active.append(kind + ":evidence") + for branch in sorted(branches - snapshots[kind].keys()): + alerts.append(f"{kind.upper()} {branch}: no qualified result in today's run.") + active.append(kind + ":" + branch) + for branch, current in snapshots[kind].items(): + key = kind + ":" + branch + previous = baselines.get(key) + if previous is None and historical: + previous = initial_baseline(kind, branch, day) + regressions, limitations = compare(current, previous) + if previous is not None and not limitations: + comparable.add(key) + if kind == "fpm" and any(g["points"] is None for g in current["groups"].values()): + limitations.append("Point evidence unavailable in this run; point regression checks unavailable.") + alerts.extend(f"{kind.upper()} {branch}: {a}" for a in regressions) + notes.extend(f"{kind.upper()} {branch}: {n}" for n in limitations) + if regressions: + active.append(key) + baselines[key] = previous # Keep the pre-alert baseline until recovery. + else: + baselines[key] = current + if kind == "e2e": + lost = sum(v is None for g in current["groups"].values() for v in g["points"].values()) + if lost: + notes.append(f"E2E {branch}: {lost} unsuccessful eligible predictions (not necessarily new).") + recovered = [] + for key in state.get("active", []): + if key in active: + continue + kind, branch = key.split(":", 1) + if (branch == "pipeline" and runs[kind] and runs[kind]["conclusion"] == "success") or (key in comparable): + recovered.append(key) + root, replies = messages(str(day), pipelines, snapshots, alerts, notes, recovered) + return { + "schema_version": 1, + "day": str(day), + "root": root, + "replies": replies, + "alerts": alerts, + "notes": notes, + "pipelines": pipelines, + "state": { + "schema_version": 1, + "day": str(day), + "production_sent": False, + "baselines": baselines, + "active": active, + }, + } + + +class Slack: + def __init__(self, token, channel): + if not token or not re.fullmatch(r"[CG][A-Z0-9]+", channel): + raise ValueError("configure a Slack bot token and channel ID") + self.token, self.channel = token, channel + + def call(self, method, **payload): + url = "https://slack.com/api/" + method + data = json.dumps(payload).encode() + if method in {"conversations.history", "conversations.replies"}: + query = { + key: str(value).lower() if isinstance(value, bool) else value + for key, value in payload.items() + if value != "" + } + url += "?" + urlencode(query) + data = None + request = urllib.request.Request( + url, + data=data, + headers={ + "Authorization": "Bearer " + self.token, + "Content-Type": "application/json; charset=utf-8", + }, + ) + # Do not blindly retry POST: a timeout can occur after Slack accepted the message. + with urllib.request.urlopen(request, timeout=30) as response: + result = json.load(response) + if not result.get("ok"): + raise RuntimeError(f"Slack {method}: {result.get('error', 'unknown error')}") + return result + + def history(self, day): + cursor = "" + oldest = str(datetime.combine(day, time(), LA).timestamp()) + latest = str(datetime.combine(day + timedelta(days=1), time(), LA).timestamp()) + while True: + result = self.call( + "conversations.history", + channel=self.channel, + oldest=oldest, + latest=latest, + limit=100, + cursor=cursor, + include_all_metadata=True, + ) + yield from result["messages"] + cursor = result.get("response_metadata", {}).get("next_cursor", "") + if not cursor: + return + + def existing(self, day): + for message in self.history(day): + meta = message.get("metadata", {}) + if ( + meta.get("event_type") == "aisim_accuracy_daily" + and meta.get("event_payload", {}).get("day") == str(day) + and meta.get("event_payload", {}).get("repo") == REPO + ): + return message + return None + + def send(self, report, test=False): + day = report["day"] + existing = None if test else self.existing(date.fromisoformat(day)) + key = f"{REPO}:{self.channel}:{day}" + ( + ":" + os.environ.get("GITHUB_RUN_ID", str(uuid.uuid4())) if test else "" + ) + payload = { + "channel": self.channel, + "text": ("[TEST] " if test else "") + report["root"], + "unfurl_links": False, + "unfurl_media": False, + "parse": "none", + "mrkdwn": True, + "link_names": False, + "client_msg_id": str(uuid.uuid5(uuid.NAMESPACE_URL, key)), + "metadata": { + "event_type": "aisim_accuracy_test" if test else "aisim_accuracy_daily", + "event_payload": { + "day": day, + "repo": REPO, + "report_run": os.environ.get("GITHUB_RUN_ID", "local"), + }, + }, + } + root = existing or self.call("chat.postMessage", **payload) + sent, cursor = set(), "" + if existing: + while True: + result = self.call( + "conversations.replies", + channel=self.channel, + ts=root["ts"], + cursor=cursor, + limit=100, + include_all_metadata=True, + ) + for message in result["messages"]: + meta = message.get("metadata", {}) + if meta.get("event_type") == "aisim_accuracy_detail": + sent.add(meta.get("event_payload", {}).get("part")) + cursor = result.get("response_metadata", {}).get("next_cursor", "") + if not cursor: + break + for index, reply in enumerate(report["replies"]): + if str(index) in sent: + continue + self.call( + "chat.postMessage", + channel=self.channel, + thread_ts=root["ts"], + text=reply, + metadata={ + "event_type": "aisim_accuracy_detail", + "event_payload": {"part": str(index)}, + }, + unfurl_links=False, + unfurl_media=False, + parse="none", + mrkdwn=True, + link_names=False, + client_msg_id=str(uuid.uuid5(uuid.NAMESPACE_URL, f"{key}:{index}")), + ) + return True + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--prepare-only", action="store_true") + parser.add_argument("--deliver", type=Path, help="Deliver a previously persisted report") + parser.add_argument("--mode", choices=("dry-run", "test", "daily"), default="dry-run") + parser.add_argument("--e2e-run-id", type=int) + parser.add_argument("--fpm-run-id", type=int) + parser.add_argument("--output", type=Path, default=Path("accuracy-preview")) + args = parser.parse_args() + now = datetime.now(UTC) + day = now.astimezone(LA).date() + if args.mode == "daily" and (args.e2e_run_id or args.fpm_run_id): + parser.error("daily mode accepts scheduled runs only") + if not args.deliver and args.mode != "daily" and not (args.e2e_run_id and args.fpm_run_id): + parser.error("dry-run/test requires both explicit run IDs") + slack = None + if args.mode != "dry-run": + channel = os.environ.get("SLACK_ACCURACY_CHANNEL_ID", "") + slack = Slack(os.environ.get("SLACK_ACCURACY_BOT_TOKEN", ""), channel) + existing = slack.existing(day) if slack and args.mode == "daily" else None + if args.deliver: + report = strict_json(args.deliver.read_text()) + if report["day"] != str(day): + raise ValueError("do not send a report for a previous local day") + elif existing: + # Resume the original frozen thread, never recompute after late pipeline results. + producer = existing["metadata"]["event_payload"]["report_run"] + if not str(producer).isdigit(): + raise ValueError("cannot resume report without Actions provenance") + run = api(f"actions/runs/{producer}") + if ( + run["path"] != REPORT_WORKFLOW + or run["head_branch"] != "main" + or run["event"] not in {"schedule", "workflow_run"} + or run["head_repository"]["full_name"] != REPO + ): + raise ValueError("untrusted notification producer") + artifacts = api_items(f"actions/runs/{producer}/artifacts", "artifacts") + artifact = next(a for a in artifacts if a["name"] == "accuracy-report" and not a["expired"]) + report = json_zip(api(f"actions/artifacts/{artifact['id']}/zip", binary=True), "report.json") + if report["day"] != str(day): + raise ValueError("restored report date mismatch") + else: + runs = { + kind: api(f"actions/runs/{run_id}") if run_id else select_run(kind, day) + for kind, run_id in (("e2e", args.e2e_run_id), ("fpm", args.fpm_run_id)) + } + if args.mode == "daily" and not ready(runs, day, now): + print("Waiting for both scheduled pipelines; no message sent.") + return + report = build_report(day, runs, prior_state(day), now, allow_manual_branch=args.mode != "daily") + args.output.mkdir(parents=True, exist_ok=True) + (args.output / "report.json").write_text(json.dumps(report, allow_nan=False, indent=2) + "\n") + preview = report["root"] + "\n\n" + "\n\n--- Thread reply ---\n\n".join(report["replies"]) + (args.output / "preview.md").write_text(preview + "\n") + summary = os.environ.get("GITHUB_STEP_SUMMARY") + if summary: + with open(summary, "a") as stream: + stream.write(preview[:900000] + "\n") + if slack and not args.prepare_only and slack.send(report, test=args.mode == "test") and args.mode == "daily": + report["state"]["production_sent"] = True + state_dir = args.output / "delivery" + state_dir.mkdir(exist_ok=True) + (state_dir / "state.json").write_text(json.dumps(report["state"], allow_nan=False) + "\n") + print(f"{args.mode}: {len(report['alerts'])} alert(s); preview: {args.output / 'preview.md'}") + + +if __name__ == "__main__": + main() diff --git a/scripts/prepare_e2e_accuracy_pages.py b/scripts/prepare_e2e_accuracy_pages.py index 078097f31..a88e786de 100644 --- a/scripts/prepare_e2e_accuracy_pages.py +++ b/scripts/prepare_e2e_accuracy_pages.py @@ -281,10 +281,10 @@ class UnqualifiedBranch(ValueError): """A branch without a successful qualification job cannot publish.""" -def trusted_run(run: dict) -> bool: +def trusted_run(run: dict, *, allow_manual_branch=False) -> bool: return ( run.get("event") in {"schedule", "workflow_dispatch"} - and run.get("head_branch") == "main" + and (run.get("head_branch") == "main" or (allow_manual_branch and run.get("event") == "workflow_dispatch")) and run.get("path") == WORKFLOW and run.get("conclusion") in {"success", "failure"} and run.get("repository", {}).get("full_name") == REPO @@ -292,8 +292,15 @@ def trusted_run(run: dict) -> bool: ) -def validate_artifact(archive: bytes, run: dict, *, artifact_name="e2e-accuracy-web", jobs=None) -> dict: - if not trusted_run(run): +def validate_artifact( + archive: bytes, + run: dict, + *, + artifact_name="e2e-accuracy-web", + jobs=None, + allow_manual_branch=False, +) -> dict: + if not trusted_run(run, allow_manual_branch=allow_manual_branch): raise ValueError("untrusted accuracy workflow run") summary = unpack_artifact(archive) q = summary["snapshot"]["campaign"] @@ -356,7 +363,8 @@ def api_items(path: str, key: str) -> list: items = [] page = 1 while True: - batch = api(f"{path}?per_page=100&page={page}")[key] + separator = "&" if "?" in path else "?" + batch = api(f"{path}{separator}per_page=100&page={page}")[key] items.extend(batch) if len(batch) < 100: return items diff --git a/scripts/prepare_fpm_accuracy_pages.py b/scripts/prepare_fpm_accuracy_pages.py index 20e2704f8..fc669b218 100755 --- a/scripts/prepare_fpm_accuracy_pages.py +++ b/scripts/prepare_fpm_accuracy_pages.py @@ -54,10 +54,10 @@ def unpack(archive: bytes) -> dict: return summary -def trusted_run(run): +def trusted_run(run, *, allow_manual_branch=False): return ( run.get("event") in {"schedule", "workflow_dispatch"} - and run.get("head_branch") == "main" + and (run.get("head_branch") == "main" or (allow_manual_branch and run.get("event") == "workflow_dispatch")) and run.get("path") == WORKFLOW and run.get("status") == "completed" and run.get("conclusion") in {"success", "failure"} @@ -66,8 +66,8 @@ def trusted_run(run): ) -def validate_artifact(archive, run, name, jobs): - require(trusted_run(run), "untrusted FPM campaign") +def validate_artifact(archive, run, name, jobs, *, allow_manual_branch=False): + require(trusted_run(run, allow_manual_branch=allow_manual_branch), "untrusted FPM campaign") summary = unpack(archive) snapshot = summary["snapshot"] require( diff --git a/scripts/run_fpm_accuracy.py b/scripts/run_fpm_accuracy.py index 53ef3f539..c0186a1d4 100755 --- a/scripts/run_fpm_accuracy.py +++ b/scripts/run_fpm_accuracy.py @@ -12,7 +12,13 @@ from datetime import UTC, datetime from pathlib import Path -from fpm_accuracy.contract import HF_REPO, METHODS, eligible_branch, sha, validate_summary +from fpm_accuracy.contract import ( + HF_REPO, + METHODS, + eligible_branch, + sha, + validate_summary, +) from fpm_accuracy.evaluate import evaluate_case from fpm_accuracy.hf import HfDataset @@ -28,6 +34,11 @@ def main(): parser.add_argument("--run-attempt", required=True) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--local-dataset", type=Path) + parser.add_argument( + "--comparison-output", + type=Path, + help="Separate notification evidence; not a Pages artifact", + ) parser.add_argument( "--configuration", action="append", @@ -58,11 +69,12 @@ def main(): configurations = tuple(item for item in configurations if item.configuration_path in selected) if {item.configuration_path for item in configurations} != selected: raise ValueError("unknown smoke configuration") + comparison = {} if args.comparison_output else None rows = [] for configuration in configurations: print(f"Evaluating {configuration.configuration_path}", flush=True) case = dataset.measurement_case(configuration.configuration_path, snapshot_id=configuration.snapshot_id) - rows.append(evaluate_case(case)) + rows.append(evaluate_case(case, comparison=comparison)) summary = { "schema_version": 1, "snapshot": { @@ -96,6 +108,20 @@ def main(): "summary_sha256": hashlib.sha256(data).hexdigest(), } (args.output / "qualification.json").write_text(json.dumps(qualification, allow_nan=False, indent=2) + "\n") + if args.comparison_output: + args.comparison_output.parent.mkdir(parents=True, exist_ok=True) + args.comparison_output.write_text( + json.dumps( + { + "schema_version": 1, + "snapshot": summary["snapshot"], + "summary_sha256": hashlib.sha256(data).hexdigest(), + "points": comparison, + }, + allow_nan=False, + ) + + "\n" + ) print(f"Completed {len(rows)} configurations; accuracy is advisory.", flush=True) diff --git a/tests/fpm_accuracy/test_digest.py b/tests/fpm_accuracy/test_digest.py new file mode 100644 index 000000000..d53157244 --- /dev/null +++ b/tests/fpm_accuracy/test_digest.py @@ -0,0 +1,523 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Notification policy, reductions, timing, and delivery with deterministic fakes.""" + +import hashlib +import io +import json +import sys +import zipfile +from datetime import UTC, date, datetime +from pathlib import Path +from urllib.parse import parse_qs, urlsplit + +import accuracy_digest as digest +import notify_accuracy as notify +import prepare_e2e_accuracy_pages as pages +import pytest +import yaml + + +def snapshot(errors, *, kind="e2e", dataset="data", rules="rules"): + return { + "kind": kind, + "branch": "main", + "commit": "a" * 40, + "dataset": dataset, + "rules": rules, + "url": "https://github.com/ai-dynamo/aisimulate/actions/runs/1", + "groups": { + "config": { + "label": "model / GPU / config", + "model": "model", + "gpu": "GPU", + "framework": "vllm", + "points": digest.encode_points(None if v is None else v[0] for v in errors.values()) + if kind == "fpm" and errors is not None + else errors, + } + }, + } + + +@pytest.mark.parametrize( + "before,after,alert", + [ + (10, 12, True), + (30, 32, False), + (10, 11.99, False), + (20, 22, True), + (0, 2, True), + (0, 1, False), + (12, 10, False), + ], +) +def test_regression_requires_both_thresholds(before, after, alert): + previous = snapshot({"p": [before, 0]}) + current = snapshot({"p": [after, 0]}) + alerts, notes = digest.compare(current, previous) + assert bool(alerts) is alert + assert not notes + + +def test_only_successful_common_points_compared_and_lost_successes_alert(): + previous = snapshot({"same": [10, 10], "lost": [1000, 1000]}) + current = snapshot({"same": [15, 15], "lost": None, "new": [0, 0]}) + alerts, _ = digest.compare(current, previous) + assert len(alerts) == 3 + assert "1 previously predicted" in alerts[0] + assert "10.00% -> 15.00%" in alerts[1] + assert "1 common points" in alerts[1] + + +def test_exchanging_failed_points_still_alerts_with_unchanged_coverage(): + previous = snapshot({"a": [1], "b": None}, kind="fpm") + current = snapshot({"a": None, "b": [1]}, kind="fpm") + alerts, _ = digest.compare(current, previous) + assert len(alerts) == 1 and "previously predicted" in alerts[0] + + +def test_missing_configuration_alerts(): + previous = snapshot({"a": [1, 1]}) + current = snapshot({}) + current["groups"] = {} + assert "previously predicted points missing" in digest.compare(current, previous)[0][0] + + +@pytest.mark.parametrize("field", ["dataset", "rules"]) +def test_changed_inputs_are_not_regressions(field): + previous = snapshot({"a": [1, 1]}) + current = snapshot({"a": [99, 99]}) + current[field] = "different" + alerts, notes = digest.compare(current, previous) + assert not alerts and "baseline reset" in notes[0] + + +def test_legacy_fpm_without_points_does_not_claim_comparable(): + previous = snapshot(None, kind="fpm") + current = snapshot({"a": [99]}, kind="fpm") + alerts, notes = digest.compare(current, previous) + assert not alerts and "Point evidence unavailable" in notes[0] + + +def test_reduction_uses_points_not_mean_of_means(): + groups = [ + {"points": {"1": [0, 10]}}, + {"points": {"2": [30, 40], "3": [30, 40], "4": None}}, + ] + assert digest.reduce_e2e(groups) == ("20.00%/30.00%", "3/4") + assert ( + digest.reduce_fpm( + [ + {"metric": {"predicted_count": 1, "measured_count": 1, "mape_pct": 0}}, + {"metric": {"predicted_count": 2, "measured_count": 3, "mape_pct": 30}}, + ] + ) + == "20.00% (3/4)" + ) + + +def test_two_separate_main_message_dimensions_and_framework_columns(): + current = snapshot({"a": [24, 43]}) + current["groups"]["config2"] = { + **current["groups"]["config"], + "gpu": "GPU2", + "points": {"b": [26, 45]}, + } + pipelines = {kind: {"url": "https://example.com/" + kind, "status": "success"} for kind in ("e2e", "fpm")} + root, replies = digest.messages("2026-09-29", pipelines, {"e2e": {"main": current}}, [], []) + assert all(name in root for name in ("Overall", "vLLM", "SGLang", "TRT-LLM", "Coverage")) + assert "https://example.com/e2e" in root and "https://example.com/fpm" in root + assert not replies + assert "per model" in root and "25.00%/44.00%" in root + assert "per gpu" in root and "GPU2" in root + assert root.index("per model") < root.index("per gpu") < root.index("*FPM") + + +@pytest.mark.parametrize( + "now,expected", + [ + ("2026-09-29T15:59:59+00:00", False), + ("2026-09-29T16:00:00+00:00", True), + ("2026-12-29T16:59:59+00:00", False), + ("2026-12-29T17:00:00+00:00", True), + ], +) +def test_deadline_follows_los_angeles_dst(now, expected): + now = datetime.fromisoformat(now) + assert notify.ready({"e2e": {"status": "completed"}, "fpm": None}, now.date(), now) is expected + + +def test_failures_are_finished_and_do_not_delay_digest(): + now = datetime(2026, 9, 29, 12, tzinfo=UTC) + assert notify.ready( + { + "e2e": {"status": "completed", "conclusion": "failure"}, + "fpm": {"status": "completed", "conclusion": "success"}, + }, + now.date(), + now, + ) + + +def test_select_scheduled_run_by_local_day(monkeypatch): + runs = [ + {"id": 1, "created_at": "2026-09-29T00:00:00Z"}, + {"id": 2, "created_at": "2026-09-29T10:30:00Z", "run_attempt": 2}, + ] + monkeypatch.setattr(notify, "scheduled_runs", lambda *args: iter(runs)) + assert notify.select_run("e2e", date(2026, 9, 29))["id"] == 2 + assert notify.report_day(runs[0]) == date(2026, 9, 28) + + +def test_pipeline_failure_reports_step_and_link(monkeypatch): + run = { + "id": 7, + "path": ".github/workflows/e2e-accuracy.yml", + "event": "schedule", + "head_branch": "main", + "repository": {"full_name": notify.REPO}, + "head_repository": {"full_name": notify.REPO}, + "status": "completed", + "conclusion": "failure", + "run_attempt": 2, + "html_url": "https://github.com/run/7", + "run_started_at": "2026-09-29T10:00:00Z", + } + monkeypatch.setattr( + notify, + "api_items", + lambda *args: [ + { + "name": "Qualify", + "conclusion": "failure", + "html_url": "https://github.com/job/8", + "completed_at": "2026-09-29T10:10:00Z", + "steps": [{"name": "Install", "conclusion": "failure"}], + } + ], + ) + status, alerts = notify.pipeline_status("e2e", run, datetime.now(UTC)) + assert "10 min" in status["status"] + assert "Install" in alerts[1] and "https://github.com/job/8" in alerts[1] + + +def test_recovery_compares_against_pre_alert_baseline(monkeypatch): + day, now = date(2026, 9, 29), datetime(2026, 9, 29, 16, tzinfo=UTC) + healthy = snapshot({"a": [10, 10], "existing-failure": None}) + regressed = snapshot({"a": [20, 20]}) + monkeypatch.setattr( + notify, + "pipeline_status", + lambda *args, **kwargs: ( + {"url": "https://example.com", "status": "success"}, + [], + ), + ) + monkeypatch.setattr(notify, "load_snapshots", lambda *args, **kwargs: ({"main": regressed}, [])) + state = {"baselines": {"e2e:main": healthy}, "active": []} + report = notify.build_report(day, {"e2e": None}, state, now, historical=False) + assert report["state"]["baselines"]["e2e:main"] == healthy + assert report["state"]["active"] == ["e2e:main"] + monkeypatch.setattr(notify, "load_snapshots", lambda *args, **kwargs: ({"main": healthy}, [])) + recovered = notify.build_report(day, {"e2e": None}, report["state"], now, historical=False) + assert "Recovered: e2e:main" in recovered["root"] + + +def test_new_dataset_is_not_announced_as_recovery(monkeypatch): + current = snapshot({"a": [1, 1]}, dataset="new") + monkeypatch.setattr( + notify, + "pipeline_status", + lambda *args, **kwargs: ( + {"url": "https://example.com", "status": "success"}, + [], + ), + ) + monkeypatch.setattr(notify, "load_snapshots", lambda *args, **kwargs: ({"main": current}, [])) + state = { + "baselines": {"e2e:main": snapshot({"a": [10, 10]})}, + "active": ["e2e:main"], + } + report = notify.build_report(date(2026, 9, 29), {"e2e": None}, state, datetime.now(UTC), historical=False) + assert "Recovered" not in report["root"] and report["notes"] + + +class FakeSlack(notify.Slack): + def __init__(self): + super().__init__("fake", "C123") + self.posts = [] + self.fail_part = None + + def call(self, method, **payload): + if method == "conversations.history": + return {"messages": [p for p in self.posts if "thread_ts" not in p]} + if method == "conversations.replies": + return {"messages": self.posts} + assert method == "chat.postMessage" + self.posts.append({**payload, "ts": str(len(self.posts) + 1)}) + part = payload.get("metadata", {}).get("event_payload", {}).get("part") + if part is not None and part == self.fail_part: + self.fail_part = None + raise TimeoutError("response lost after acceptance") + return {"ts": self.posts[-1]["ts"]} + + +def test_one_root_and_resume_partial_thread_after_ambiguous_timeout(): + slack = FakeSlack() + slack.fail_part = "0" + report = {"day": "2026-09-29", "root": "Daily", "replies": ["models", "GPUs"]} + with pytest.raises(TimeoutError): + slack.send(report) + slack.send(report) + slack.send(report) + assert len(slack.posts) == 3 + assert sum("thread_ts" not in p for p in slack.posts) == 1 + assert all(p.get("link_names") is False and p["parse"] == "none" and p["mrkdwn"] is True for p in slack.posts) + + +def test_test_message_not_production_dedup(): + slack = FakeSlack() + report = {"day": "2026-09-29", "root": "Daily", "replies": []} + slack.send(report, test=True) + slack.send(report) + assert len(slack.posts) == 2 and slack.posts[0]["text"].startswith("[TEST]") + + +def test_slack_api_ok_false_is_failure(monkeypatch): + response = io.BytesIO(b'{"ok": false, "error": "not_in_channel"}') + monkeypatch.setattr(notify.urllib.request, "urlopen", lambda *args, **kwargs: response) + with pytest.raises(RuntimeError, match="not_in_channel"): + notify.Slack("fake", "C123").call("chat.postMessage", text="test") + + +def archive(name, data): + target = io.BytesIO() + with zipfile.ZipFile(target, "w") as bundle: + bundle.writestr(name, json.dumps(data)) + return target.getvalue() + + +def test_point_sidecar_must_match_qualified_summary(): + row = { + "configuration_id": "config", + "snapshot_id": "s", + "results": {"warmup": {"metrics": {"all": {"measured_count": 2, "predicted_count": 1, "mape_pct": 10}}}}, + } + summary = {"snapshot": {"run_id": "1"}, "rows": [row]} + public = archive("summary.json", summary) + evidence = { + "schema_version": 1, + "snapshot": summary["snapshot"], + "summary_sha256": hashlib.sha256(json.dumps(summary).encode()).hexdigest(), + "points": {"config/s": {"order_sha256": "a" * 64, "methods": {"warmup": digest.encode_points([10, None])}}}, + } + assert notify.fpm_points(archive("comparison.json", evidence), summary, public) == evidence["points"] + evidence["points"]["config/s"]["methods"]["warmup"] = digest.encode_points([99, None]) + with pytest.raises(ValueError, match="aggregate mismatch"): + notify.fpm_points(archive("comparison.json", evidence), summary, public) + + +def test_read_only_dry_run_and_guarded_secret_scope(): + workflow = yaml.safe_load((Path(__file__).parents[2] / ".github/workflows/accuracy-digest.yml").read_text()) + triggers = workflow.get("on", workflow.get(True)) + assert triggers["workflow_dispatch"]["inputs"]["mode"]["default"] == "dry-run" + assert triggers["workflow_dispatch"]["inputs"]["mode"]["options"] == [ + "dry-run", + "test", + ] + assert workflow["permissions"] == {"contents": "read", "actions": "read"} + assert workflow["concurrency"]["cancel-in-progress"] is False + steps = workflow["jobs"]["report"]["steps"] + assert "workflow_run.head_sha" not in json.dumps(steps) + prepare = next(s for s in steps if s.get("name") == "Prepare report and preview") + assert "env.MODE != 'dry-run'" in prepare["env"]["SLACK_ACCURACY_BOT_TOKEN"] + + +def test_corrupt_or_oversized_compressed_points_rejected(): + packed = digest.encode_points([1, None, 2]) + assert list(digest.decode_points(packed)) == [1, -1, 2] + with pytest.raises(ValueError): + digest.decode_points({**packed, "count": 1}) + with pytest.raises(ValueError): + digest.decode_points({"count": 10_000_001, "data": ""}) + + +def test_fpm_observation_order_change_does_not_compare_positions(): + old = snapshot({"a": [1]}, kind="fpm") + new = snapshot({"a": [90]}, kind="fpm") + old["groups"]["config"]["point_order"] = "old" + new["groups"]["config"]["point_order"] = "new" + alerts, notes = digest.compare(new, old) + assert not alerts and "not comparable" in notes[0] + + +def test_explicit_manual_branch_only_allowed_for_preview_and_test(): + run = { + "path": ".github/workflows/e2e-accuracy.yml", + "head_branch": "simonec/fix", + "event": "workflow_dispatch", + "repository": {"full_name": notify.REPO}, + "head_repository": {"full_name": notify.REPO}, + } + with pytest.raises(ValueError): + notify.validate_run(run, "e2e") + notify.validate_run(run, "e2e", allow_manual_branch=True) + run["head_repository"]["full_name"] = "someone/fork" + with pytest.raises(ValueError): + notify.validate_run(run, "e2e", allow_manual_branch=True) + + +def test_dry_run_never_instantiates_slack_or_writes_delivery_state(monkeypatch, tmp_path): + report = {"root": "preview", "replies": [], "alerts": [], "state": {"production_sent": False}} + monkeypatch.setattr( + sys, "argv", ["notify_accuracy.py", "--e2e-run-id", "1", "--fpm-run-id", "2", "--output", str(tmp_path)] + ) + monkeypatch.setattr(notify, "api", lambda path: {"id": path.rsplit("/", 1)[-1]}) + monkeypatch.setattr(notify, "prior_state", lambda day: {}) + monkeypatch.setattr(notify, "build_report", lambda *args, **kwargs: report) + + def no_slack(*args, **kwargs): + pytest.fail("dry run must not instantiate Slack") + + monkeypatch.setattr(notify, "Slack", no_slack) + notify.main() + assert (tmp_path / "report.json").exists() + assert not (tmp_path / "delivery").exists() + + +def test_artifact_pagination_preserves_name_filter(monkeypatch): + paths = [] + + def fake_api(path): + paths.append(path) + return {"artifacts": [{}] * 100 if len(paths) == 1 else []} + + monkeypatch.setattr(pages, "api", fake_api) + assert len(pages.api_items("actions/artifacts?name=accuracy-attempt-2026-09-29", "artifacts")) == 100 + assert paths == [f"actions/artifacts?name=accuracy-attempt-2026-09-29&per_page=100&page={page}" for page in (1, 2)] + + +@pytest.mark.parametrize("method", ["conversations.history", "conversations.replies"]) +def test_slack_reads_use_query_parameters_without_empty_cursor(monkeypatch, method): + calls = [] + + def respond(request, **kwargs): + calls.append(request) + return io.BytesIO(b'{"ok":true,"messages":[]}') + + monkeypatch.setattr(notify.urllib.request, "urlopen", respond) + notify.Slack("fake", "C123").call(method, channel="C123", cursor="", include_all_metadata=True) + assert calls[0].get_method() == "GET" + assert calls[0].data is None + assert parse_qs(urlsplit(calls[0].full_url).query) == {"channel": ["C123"], "include_all_metadata": ["true"]} + + +@pytest.mark.parametrize("kind", ["e2e", "fpm"]) +@pytest.mark.parametrize("has_target", [False, True]) +def test_baseline_branch_filter_skips_unrelated_downloads(monkeypatch, kind, has_target): + target = f"{kind}-accuracy-web-{pages.artifact_key('main')}" + artifacts = [{"id": 1, "name": f"{kind}-accuracy-web-{pages.artifact_key('release/0.12.0')}", "expired": False}] + if has_target: + artifacts.append({"id": 2, "name": target, "expired": False}) + downloads = [] + monkeypatch.setattr(notify, "validate_run", lambda *args, **kwargs: None) + monkeypatch.setattr(notify, "api_items", lambda *args: artifacts) + + def download(path, **kwargs): + downloads.append(path) + raise ValueError("target artifact validation failed") + + monkeypatch.setattr(notify, "api", download) + found, warnings = notify.load_snapshots(kind, {"id": 7, "status": "completed"}, only_branch="main") + assert not found + assert downloads == (["actions/artifacts/2/zip"] if has_target else []) + assert bool(warnings) is has_target + + +def test_initial_baseline_filters_branch_and_stops_at_newest_match(monkeypatch): + runs = [ + {"id": 3, "status": "completed", "created_at": "2026-09-28T10:00:00Z"}, + {"id": 2, "status": "completed", "created_at": "2026-09-27T10:00:00Z"}, + {"id": 1, "status": "completed", "created_at": "2026-09-26T10:00:00Z"}, + ] + calls = [] + monkeypatch.setattr(notify, "scheduled_runs", lambda *args: iter(runs)) + monkeypatch.setattr(notify, "api", lambda path: next(r for r in runs if path.endswith(str(r["id"])))) + + def snapshots(kind, run, *, only_branch): + calls.append((run["id"], only_branch)) + return ({"main": "latest qualified"} if run["id"] == 2 else {}), [] + + monkeypatch.setattr(notify, "load_snapshots", snapshots) + assert notify.initial_baseline("e2e", "main", date(2026, 9, 29)) == "latest qualified" + assert calls == [(3, "main"), (2, "main")] + + +def test_slack_history_follows_populated_cursor(monkeypatch): + queries = [] + + def respond(request, **kwargs): + assert request.get_method() == "GET" + queries.append(parse_qs(urlsplit(request.full_url).query)) + result = {"ok": True, "messages": [{"ts": str(len(queries))}]} + if len(queries) == 1: + result["response_metadata"] = {"next_cursor": "history+cursor="} + return io.BytesIO(json.dumps(result).encode()) + + monkeypatch.setattr(notify.urllib.request, "urlopen", respond) + messages = list(notify.Slack("fake", "C123").history(date(2026, 9, 29))) + assert [message["ts"] for message in messages] == ["1", "2"] + assert "cursor" not in queries[0] + assert queries[1]["cursor"] == ["history+cursor="] + assert queries[0]["oldest"] == queries[1]["oldest"] + + +def test_slack_resume_reads_all_reply_pages_before_posting(monkeypatch): + queries = [] + slack = notify.Slack("fake", "C123") + monkeypatch.setattr(slack, "existing", lambda day: {"ts": "1"}) + + def respond(request, **kwargs): + assert request.get_method() == "GET", "both existing replies must be discovered without reposting" + assert urlsplit(request.full_url).path.endswith("conversations.replies") + queries.append(parse_qs(urlsplit(request.full_url).query)) + part = str(len(queries) - 1) + result = { + "ok": True, + "messages": [{"metadata": {"event_type": "aisim_accuracy_detail", "event_payload": {"part": part}}}], + } + if len(queries) == 1: + result["response_metadata"] = {"next_cursor": "reply+cursor="} + return io.BytesIO(json.dumps(result).encode()) + + monkeypatch.setattr(notify.urllib.request, "urlopen", respond) + assert slack.send({"day": "2026-09-29", "root": "Daily", "replies": ["first", "second"]}) + assert len(queries) == 2 + assert "cursor" not in queries[0] + assert queries[1]["cursor"] == ["reply+cursor="] + assert queries[1]["ts"] == ["1"] + + +def test_actual_reusable_job_names_discover_missing_branch(monkeypatch): + # Names observed on runs 36643620072 and 36558333115 include the called job suffix. + monkeypatch.setattr( + notify, "pipeline_status", lambda *args, **kwargs: ({"url": "https://example.com", "status": "success"}, []) + ) + monkeypatch.setattr(notify, "load_snapshots", lambda *args, **kwargs: ({"main": snapshot({"a": [1, 1]})}, [])) + monkeypatch.setattr( + notify, + "api_items", + lambda *args: [ + {"name": "E2E accuracy (main) / Prepare exact accuracy wheel"}, + {"name": "E2E accuracy (release/0.12.0) / Qualify E2E accuracy (cb21c32489602c02)"}, + ], + ) + report = notify.build_report( + date(2026, 9, 29), + {"e2e": {"id": 1, "status": "completed", "conclusion": "success"}}, + {}, + datetime.now(UTC), + historical=False, + ) + assert report["alerts"] == ["E2E release/0.12.0: no qualified result in today's run."] diff --git a/tests/fpm_accuracy/test_evaluation.py b/tests/fpm_accuracy/test_evaluation.py index d3813ef69..c132ee84b 100644 --- a/tests/fpm_accuracy/test_evaluation.py +++ b/tests/fpm_accuracy/test_evaluation.py @@ -8,6 +8,7 @@ import pyarrow as pa import pyarrow.parquet as pq import pytest +from accuracy_digest import decode_points from fpm_accuracy.evaluate import Metric, choose_variant, evaluate_case from fpm_accuracy.exceptions import ConfigurationError, DependencyError from fpm_accuracy.models import aic_predictors @@ -465,3 +466,26 @@ def result(identity, predicted, mape): def test_bad_measurement_fails_campaign(case): with pytest.raises(ValueError, match="unique IDs"): evaluate_case(replace(case, observations=(case.observations[0], case.observations[0]))) + + +def test_notification_evidence_keeps_point_failures_without_changing_public_result(case): + """The notification sidecar is opt-in and must not change prediction/tuning order.""" + + def factory(method, context): + return Predictor(method, context, []) + + expected = evaluate_case(case, factory=factory) + evidence = {} + actual = evaluate_case(case, factory=factory, comparison=evidence) + assert actual == expected + points = evidence[case.configuration_id + "/" + case.configuration.snapshot_id]["methods"] + assert set(points) == {"warmup", "nowarmup", "regression"} + for method, packed in points.items(): + samples = decode_points(packed) + metric = actual["results"][method]["metrics"]["all"] + assert len(samples) == metric["measured_count"] + values = [v for v in samples if v >= 0] + assert len(values) == metric["predicted_count"] + if values: + assert sum(values) / len(values) == pytest.approx(metric["mape_pct"]) + assert "_points" not in json.dumps(actual)