From a89634489666692b221ff443ecee315d3c5183e1 Mon Sep 17 00:00:00 2001 From: Oseltamivir <58582368+Oseltamivir@users.noreply.github.com> Date: Sun, 25 Oct 2026 17:38:11 +0800 Subject: [PATCH] feat(ingest): comment app ingest reports on the merged PR MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add ingest-results-callback.yml, which receives ingest-results-completed from InferenceX-app, resolves the PR merged by the Merge Ingest run's commit and posts the per-run ingest counts as one marker comment per merge run, updated in place on reruns and failures. 中文：在已合并的 PR 上评论 InferenceX-app 的 ingest 报告。新增 ingest-results-callback.yml，接收 InferenceX-app 发送的 ingest-results-completed，根据 Merge Ingest 运行的提交解析已合并的 PR，并以每个合并运行一条带标记的评论发布本次 ingest 的统计；重跑或失败时原地更新。 --- .github/workflows/README.md | 7 + .github/workflows/ingest-results-callback.yml | 33 +++ .../workflows/test_ingest_results_callback.py | 118 ++++++++++ .../infx/workflows/ingest_results_callback.py | 218 ++++++++++++++++++ 4 files changed, 376 insertions(+) create mode 100644 .github/workflows/ingest-results-callback.yml create mode 100644 inferencex-e2e/infx/tests/workflows/test_ingest_results_callback.py create mode 100644 inferencex-e2e/infx/workflows/ingest_results_callback.py diff --git a/.github/workflows/README.md b/.github/workflows/README.md index cfe8867fe3..2888402746 100644 --- a/.github/workflows/README.md +++ b/.github/workflows/README.md @@ -308,6 +308,13 @@ normal ingestion code skips failed benchmark rows. Benchmark rows and public links retain source-run provenance. Source coverage is authoritative, so later matrix/eval policy changes do not invalidate reuse. +When the app's ingest finishes, on success or failure, it sends +`ingest-results-completed` back with per-run counts (benchmark, eval, trace and +power rows: new, duplicate, skipped, unmapped, failed). `ingest-results-callback.yml` +checks that `merge-run-id` is a Merge Ingest push to `main`, resolves the PR merged +by that commit, and posts one comment per merge run, updated in place on reruns +(marker ``). + Reuse fails closed when authorized but ineligible or invalid. Pushes to `main` never run a sweep: `run-sweep.yml` is PR-only, and without reuse authorization the Merge Ingest run fails and nothing is benchmarked or ingested. diff --git a/.github/workflows/ingest-results-callback.yml b/.github/workflows/ingest-results-callback.yml new file mode 100644 index 0000000000..7e347edb44 --- /dev/null +++ b/.github/workflows/ingest-results-callback.yml @@ -0,0 +1,33 @@ +name: Report Ingest Result + +run-name: Report ingest of merge run ${{ github.event.client_payload.merge-run-id }} + +on: + repository_dispatch: + types: [ingest-results-completed] + +permissions: {} + +concurrency: + group: ingest-results-callback-${{ github.event.client_payload.merge-run-id }} + cancel-in-progress: false + +jobs: + comment: + name: Comment on merged pull request + runs-on: ubuntu-latest + permissions: + actions: read # Resolve the Merge Ingest run to its commit + contents: read # Read trusted callback implementation and commit PRs + pull-requests: write # Post the ingest report as github-actions + steps: + - name: Checkout trusted callback code + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.sha }} + persist-credentials: false + - name: Validate callback and comment + working-directory: inferencex-e2e + env: + GH_TOKEN: ${{ github.token }} + run: python3 -m infx.workflows.ingest_results_callback diff --git a/inferencex-e2e/infx/tests/workflows/test_ingest_results_callback.py b/inferencex-e2e/infx/tests/workflows/test_ingest_results_callback.py new file mode 100644 index 0000000000..1b6edd3884 --- /dev/null +++ b/inferencex-e2e/infx/tests/workflows/test_ingest_results_callback.py @@ -0,0 +1,118 @@ +import copy + +import pytest + +from infx import github +from infx.workflows import ingest_results_callback as callback + + +def counts(**overrides): + rows = {c: dict.fromkeys(callback.OUTCOMES, 0) for c in callback.CATEGORIES} + for category, values in overrides.items(): + rows[category].update(values) + return rows + + +@pytest.fixture +def case(monkeypatch): + state = { + "event": { + "action": "ingest-results-completed", + "client_payload": { + "source-repository": "SemiAnalysisAI/InferenceX-app", + "merge-run-id": "40178629864", + "source-run-id": "40141873883", + "app-run-id": "40178653878", + "app-run-attempt": "1", + "database-target": "production", + "ingest-succeeded": "true", + "report": { + "counts": counts(benchmark={"new": 1}), + "detail": {"benchmarkFiles": 2, "changelogEntriesWritten": 1}, + "unmapped": {"models": [], "hardware": ["mi999x"]}, + }, + }, + }, + "responses": { + "/actions/runs/40178629864": { + "path": ".github/workflows/merge-ingest.yml", "event": "push", + "head_branch": "main", "head_sha": "abc", + }, + "/commits/abc/pulls": [ + {"number": 4456, "merged_at": "2026-10-24T00:00:00Z", "merge_commit_sha": "abc"}, + ], + "/issues/4456/comments": [], + }, + "writes": [], + } + + def api(repo, path, token=None, params=None, *, method="GET", data=None, paginate=False): + if method != "GET": + state["writes"].append((method, path, data["body"])) + return {"id": 1} + response = state["responses"][path] + return copy.deepcopy([response] if paginate else response) + + monkeypatch.setattr(github, "api", api) + return state + + +def run(state): + callback.report("SemiAnalysisAI/InferenceX", state["event"], "t", "https://github.com") + + +def test_posts_counts_on_the_merged_pr(case): + run(case) + [(method, path, body)] = case["writes"] + assert (method, path) == ("POST", "/issues/4456/comments") + assert body.startswith("") + assert "| benchmark | 1 | 0 | 0 | 0 | 0 |" in body + assert "benchmarkFiles 2, changelogEntriesWritten 1" in body + assert "Unmapped hardware: `mi999x`" in body + assert "中文" in body + + +def test_updates_existing_comment_in_place_on_failure(case): + case["responses"]["/issues/4456/comments"] = [ + {"id": 9, "user": {"login": "someone"}, + "body": ""}, + {"id": 10, "user": {"login": "github-actions[bot]"}, + "body": "\nold"}, + ] + case["event"]["client_payload"]["ingest-succeeded"] = "false" + case["event"]["client_payload"]["report"] = None + run(case) + [(method, path, body)] = case["writes"] + assert (method, path) == ("PATCH", "/issues/comments/10") + assert "failed" in body + assert "No ingest report was produced" in body + + +def test_no_merged_pr_posts_nothing(case): + case["responses"]["/commits/abc/pulls"] = [{"number": 1, "merged_at": None}] + run(case) + assert case["writes"] == [] + + +@pytest.mark.parametrize( + ("key", "value"), + [ + ("source-repository", "someone/fork"), + ("merge-run-id", "1; rm"), + ("ingest-succeeded", "yes"), + ("database-target", "prod`uction"), + ("report", {"counts": counts(eval={"new": -1})}), + ("report", {"counts": counts(), "unmapped": {"models": ["[x](http://e)"]}}), + ], +) +def test_rejects_malformed_payload(case, key, value): + case["event"]["client_payload"][key] = value + with pytest.raises(RuntimeError): + run(case) + assert case["writes"] == [] + + +def test_rejects_runs_other_than_merge_ingest(case): + case["responses"]["/actions/runs/40178629864"]["path"] = ".github/workflows/run-sweep.yml" + with pytest.raises(RuntimeError): + run(case) diff --git a/inferencex-e2e/infx/workflows/ingest_results_callback.py b/inferencex-e2e/infx/workflows/ingest_results_callback.py new file mode 100644 index 0000000000..514cfeb4b6 --- /dev/null +++ b/inferencex-e2e/infx/workflows/ingest_results_callback.py @@ -0,0 +1,218 @@ +"""Report an InferenceX-app ingest outcome on the PR whose changelog entry it published. + +InferenceX-app sends ``ingest-results-completed`` after every dispatched merge +ingest, on success and failure. The PR is resolved here from the Merge Ingest +run's commit, never taken from the payload, and one comment per merge run is +updated in place. +""" + +from __future__ import annotations + +import json +import os +import re +import sys +from pathlib import Path +from typing import Any + +from infx import github + +APP_REPOSITORY = "SemiAnalysisAI/InferenceX-app" +MERGE_WORKFLOW = ".github/workflows/merge-ingest.yml" +BOT_LOGIN = "github-actions[bot]" +CATEGORIES = ("benchmark", "eval", "trace", "power") +OUTCOMES = ("new", "duplicate", "skipped", "unmapped", "failed") +DETAIL_KEYS = ( + "benchmarkFiles", + "rawBenchmarkRows", + "coveredByPerJobFiles", + "purgedPoints", + "failedRunRows", + "runStatsNew", + "runStatsDuplicate", + "evalSamplesNew", + "gpuMetricSamplesNew", + "powerPublicationPoints", + "changelogEntriesWritten", +) +NAME = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:/+-]{0,79}") +MAX_NAMES = 20 + + +def _count(value: Any) -> int: + """Return ``value`` as a non-negative int count; raise RuntimeError otherwise.""" + if type(value) is not int or value < 0: + raise RuntimeError("Report counts must be non-negative integers") + return value + + +def render_report(report: Any) -> list[str]: + """Return markdown lines for the payload's count report, validating every value. + + Raises RuntimeError on a malformed report. A missing report renders a note. + """ + if report is None: + return ["No ingest report was produced; the ingest stopped before counting rows."] + if not isinstance(report, dict) or not isinstance(report.get("counts"), dict): + raise RuntimeError("Callback report must be an object with counts") + counts = report["counts"] + lines = [ + f"| Category | {' | '.join(OUTCOMES)} |", + f"| --- | {' | '.join('---:' for _ in OUTCOMES)} |", + ] + for category in CATEGORIES: + row = counts.get(category) + if not isinstance(row, dict): + raise RuntimeError(f"Callback report is missing {category} counts") + lines.append(f"| {category} | {' | '.join(str(_count(row.get(o))) for o in OUTCOMES)} |") + detail = report.get("detail") or {} + if not isinstance(detail, dict): + raise RuntimeError("Callback report detail must be an object") + extras = [f"{key} {_count(detail[key])}" for key in DETAIL_KEYS if key in detail] + if extras: + lines += ["", f"{', '.join(extras)}"] + unmapped = report.get("unmapped") or {} + if not isinstance(unmapped, dict): + raise RuntimeError("Callback report unmapped must be an object") + for kind in ("models", "hardware", "precisions", "datasets"): + names = unmapped.get(kind) or [] + if not isinstance(names, list): + raise RuntimeError("Unmapped names must be lists") + shown = [f"`{n}`" for n in names[:MAX_NAMES] if isinstance(n, str) and NAME.fullmatch(n)] + if len(shown) != min(len(names), MAX_NAMES): + raise RuntimeError("Unmapped names must be plain identifiers") + if shown: + more = f" (+{len(names) - MAX_NAMES} more)" if len(names) > MAX_NAMES else "" + lines.append(f"\n⚠️ Unmapped {kind}: {', '.join(shown)}{more}") + return lines + + +def merged_pr(repo: str, merge_run_id: str, token: str) -> int | None: + """Return the PR merged by the Merge Ingest run ``merge_run_id``, or None if none matches. + + Raises RuntimeError when the run is not a Merge Ingest push to main. + """ + run = github.api(repo, f"/actions/runs/{merge_run_id}", token) + if ( + str(run.get("path", "")).split("@")[0] != MERGE_WORKFLOW + or run.get("event") != "push" + or run.get("head_branch") != "main" + ): + raise RuntimeError("Callback merge run is not a Merge Ingest push to main") + sha = str(run.get("head_sha", "")) + pulls = github.api(repo, f"/commits/{sha}/pulls", token) + merged = [p for p in pulls if p.get("merged_at") and p.get("merge_commit_sha") == sha] + return int(merged[0]["number"]) if merged else None + + +def report(repo: str, event: dict[str, Any], token: str, server_url: str) -> None: + """Post or update the ingest comment for one ``ingest-results-completed`` event. + + Raises RuntimeError if the payload is malformed or from an unexpected source. + """ + if event.get("action") != "ingest-results-completed": + raise RuntimeError("Unsupported ingest callback type") + payload = event["client_payload"] + if payload.get("source-repository") != APP_REPOSITORY: + raise RuntimeError("Unsupported ingest callback source") + merge_run_id, source_run_id, app_run_id, app_attempt = ( + str(payload.get(key, "")) + for key in ("merge-run-id", "source-run-id", "app-run-id", "app-run-attempt") + ) + if not all( + re.fullmatch(r"[0-9]+", v) for v in (merge_run_id, source_run_id, app_run_id, app_attempt) + ): + raise RuntimeError("Callback run identifiers must be numeric") + succeeded = payload.get("ingest-succeeded") + if succeeded not in {"true", "false"}: + raise RuntimeError("Callback ingest-succeeded value must be true or false") + target = str(payload.get("database-target", "")) + if not re.fullmatch(r"[a-z-]+", target): + raise RuntimeError("Callback database target must be a plain name") + table = render_report(payload.get("report")) + + pr_number = merged_pr(repo, merge_run_id, token) + if pr_number is None: + print(f"No merged PR found for merge run {merge_run_id}; nothing to report.") + return + + app_url = f"{server_url}/{APP_REPOSITORY}/actions/runs/{app_run_id}/attempts/{app_attempt}" + merge_url = f"{server_url}/{repo}/actions/runs/{merge_run_id}" + source_url = f"{server_url}/{repo}/actions/runs/{source_run_id}" + marker = f"" + if succeeded == "true": + status_en = f"✅ Results ingested into the `{target}` database." + status_zh = f"✅ 结果已写入 `{target}` 数据库。" + else: + status_en = ( + f"❌ Ingest into the `{target}` database failed. " + f"[Inspect the ingest workflow]({app_url})." + ) + status_zh = ( + f"❌ 写入 `{target}` 数据库的 ingest 失败，请[查看 ingest workflow]({app_url})。" + ) + links = ( + f"[Ingest workflow]({app_url}) · [Merge Ingest run]({merge_url}) · " + f"[Benchmark source run]({source_url})" + ) + body = "\n".join( + [ + marker, + f"**Ingest report** — {status_en}", + "", + links, + "", + *table, + "", + ( + "Counts cover only this run. Rows: benchmark_results, eval_results, agentic rows " + "linked to a trace, gpu_metrics series. This comment updates if the ingest is rerun." + ), + "", + "
", + "中文", + "", + f"**Ingest 报告** — {status_zh}", + "", + ( + "上表仅统计本次运行写入的数据，各行单位依次为 benchmark_results 行、eval_results 行、" + "已关联 trace 的 agentic 行和 gpu_metrics 序列。重新运行 ingest 时本评论会原地更新。" + ), + "", + "
", + ] + ) + comments = github.paginate(repo, f"/issues/{pr_number}/comments", token, "") + existing = next( + ( + c + for c in comments + if c.get("user", {}).get("login") == BOT_LOGIN and marker in str(c.get("body")) + ), + None, + ) + if existing: + github.api( + repo, f"/issues/comments/{existing['id']}", token, method="PATCH", data={"body": body} + ) + else: + github.api(repo, f"/issues/{pr_number}/comments", token, method="POST", data={"body": body}) + + +def main() -> None: + """Report the ingest callback from the GitHub Actions environment.""" + event = json.loads(Path(os.environ["GITHUB_EVENT_PATH"]).read_text(encoding="utf-8")) + report( + os.environ["GITHUB_REPOSITORY"], + event, + os.environ["GH_TOKEN"], + os.environ["GITHUB_SERVER_URL"], + ) + + +if __name__ == "__main__": + try: + main() + except RuntimeError as exc: + print(f"error: {exc}", file=sys.stderr) + sys.exit(1)