fix: make Q1-Q8 matrix and HF aggregation fail closed

.github/workflows/ci-benchmarks.yml

@@ -15,6 +15,22 @@ on:
         description: "Experiments to run (comma-separated Q-IDs or 'all')"
         required: false
         default: "all"
+      backend:
+        description: "Runtime backend"
+        required: false
+        default: "sage"
+        type: choice
+        options:
+          - sage
+          - ray
+      seed:
+        description: "Random seed"
+        required: false
+        default: "42"
+      parallelism:
+        description: "Single-node operator parallelism"
+        required: false
+        default: "2"
   schedule:
     - cron: "0 2 * * *"
 
@@ -23,7 +39,7 @@ jobs:
     runs-on: ubuntu-latest
     timeout-minutes: 90
     env:
-      HF_ENDPOINT: https://hf-mirror.com
+      HF_ENDPOINT: https://huggingface.co
       SAGELLM_PORT: 8888
       SAGELLM_EMBED_PORT: 8890
     steps:
@@ -42,8 +58,11 @@ jobs:
       - name: Install Python dependencies
         run: |
           python -m pip install --upgrade pip
-          python -m pip install "isagellm>=0.5.1.9"
-          python -m pip install -e .
+          if [ "${{ github.event.inputs.backend || 'sage' }}" = "ray" ]; then
+            python -m pip install -e ".[ray-baseline]"
+          else
+            python -m pip install -e .
+          fi
 
       - name: Start sagellm full stack (CPU + embedding, port ${{ env.SAGELLM_PORT }})
         run: |
@@ -63,6 +82,11 @@ jobs:
             if curl -sf "http://localhost:${SAGELLM_PORT}/health" > /dev/null 2>&1; then
               echo "Gateway healthy after ~$((i * 5))s"; break
             fi
+            if ! kill -0 "$SAGELLM_PID" 2>/dev/null; then
+              echo "::error::sagellm exited before the gateway became healthy"
+              cat /tmp/sagellm.log
+              exit 1
+            fi
             [ "$i" -eq 72 ] && { cat /tmp/sagellm.log; exit 1; }
             sleep 5
           done
@@ -86,20 +110,31 @@ jobs:
             | python -c "import sys,json; d=json.load(sys.stdin); print('LLM ok')"
           curl -sf -X POST "http://localhost:${SAGELLM_PORT}/v1/embeddings" \
             -H "Content-Type: application/json" \
-            -d '{"model":"BAAI/bge-small-zh-v1.5","input":["hello"]}' \
+            -d '{"model":"sentence-transformers/all-MiniLM-L6-v2","input":["hello"]}' \
             | python -c "import sys,json; d=json.load(sys.stdin); print('Embed ok, dim='+str(len(d['data'][0]['embedding'])))"
 
       - name: Run Q1-Q8 benchmarks
         run: |
           QUICK_FLAG=""
           [ "${{ github.event.inputs.quick || 'true' }}" = "true" ] && QUICK_FLAG="--quick"
           EXPERIMENTS="${{ github.event.inputs.experiments || 'all' }}"
+          COMMON_ARGS=(
+            --backend "${{ github.event.inputs.backend || 'sage' }}"
+            --seed "${{ github.event.inputs.seed || '42' }}"
+            --nodes 1
+            --parallelism "${{ github.event.inputs.parallelism || '2' }}"
+            --continue-on-error
+            --output-dir results/ci_run
+          )
           if [ "$EXPERIMENTS" = "all" ]; then
-            python __main__.py --all $QUICK_FLAG --output-dir results/ci_run
+            python __main__.py --all $QUICK_FLAG "${COMMON_ARGS[@]}"
           else
             IFS=',' read -ra EXP_LIST <<< "$EXPERIMENTS"
             for EXP in "${EXP_LIST[@]}"; do
-              python __main__.py --experiment "$(echo "$EXP" | tr -d ' ')" $QUICK_FLAG --output-dir results/ci_run
+              python __main__.py \
+                --experiment "$(echo "$EXP" | tr -d ' ')" \
+                $QUICK_FLAG \
+                "${COMMON_ARGS[@]}"
             done
           fi
 
@@ -111,7 +146,10 @@ jobs:
         uses: actions/upload-artifact@v4
         with:
           name: benchmark-results-${{ github.run_id }}
-          path: results/ci_run
+          path: |
+            results/ci_run
+            hf_data/benchmark_results.json
+            hf_data/benchmark_summary.json
           if-no-files-found: warn
 
       - name: Upload sagellm logs (on failure)
@@ -120,3 +158,11 @@ jobs:
         with:
           name: sagellm-logs-${{ github.run_id }}
           path: /tmp/sagellm.log
+
+      - name: Stop sagellm
+        if: always()
+        run: |
+          if [ -n "${SAGELLM_PID:-}" ]; then
+            kill "$SAGELLM_PID" 2>/dev/null || true
+            wait "$SAGELLM_PID" 2>/dev/null || true
+          fi

.github/workflows/upload-to-hf.yml

@@ -1,20 +1,21 @@
 name: Upload to Hugging Face
 
-# 当 hf_data/ 下有 JSON 文件变动时自动触发（仅 main-dev 分支）
+# 当 main 分支的 hf_data/ 下有 JSON 文件变动时自动触发
 # 用户工作流：
 #   1. python scripts/aggregate_for_hf.py   （本地聚合）
-#   2. git add hf_data/ && git commit && git push origin main-dev  （触发此 workflow）
+#   2. git add hf_data/ && git commit && git push origin main  （触发此 workflow）
 #   3. Actions 自动：并发安全合并 → 上传 HF → 清理 hf_data/
-#
-# 注意：main 分支不触发此 workflow，benchmark results 不应直接推送到 main。
 
 on:
   push:
     branches:
-      - main-dev
+      - main
     paths:
       - "hf_data/**/*.json"
 
+permissions:
+  contents: write
+
 jobs:
   upload-to-hf:
     runs-on: ubuntu-latest
@@ -44,7 +45,7 @@ jobs:
       - name: Upload to Hugging Face
         env:
           HF_TOKEN: ${{ secrets.HF_TOKEN }}
-          HF_ENDPOINT: https://hf-mirror.com
+          HF_ENDPOINT: https://huggingface.co
         run: |
           python scripts/upload_to_hf.py
 

README.md

@@ -92,6 +92,10 @@ The `quickstart.sh` script will automatically:
 
 ## ⚡ One-Click Full Pipeline
 
+For the Q1–Q8 runner contract, output schema, CI inputs, aggregation flow, and
+failure troubleshooting, see the
+[Q1–Q8 single-node benchmark guide](docs/q-matrix-benchmark.md).
+
 Run the end-to-end benchmark pipeline in one command:
 
 ```bash

__main__.py

@@ -15,9 +15,13 @@
 from __future__ import annotations
 
 import argparse
+import json
 import sys
 import uuid
+from contextlib import redirect_stderr, redirect_stdout
+from datetime import datetime, timezone
 from pathlib import Path
+from typing import TextIO
 
 WORKLOAD_CATALOG = {
     "Q1": {
@@ -65,6 +69,22 @@
 VALID_EXPERIMENTS = tuple(WORKLOAD_CATALOG.keys())
 
 
+class _Tee:
+    """Write workload output to both the terminal and its per-run log."""
+
+    def __init__(self, *streams: TextIO):
+        self.streams = streams
+
+    def write(self, data: str) -> int:
+        for stream in self.streams:
+            stream.write(data)
+        return len(data)
+
+    def flush(self) -> None:
+        for stream in self.streams:
+            stream.flush()
+
+
 def _workload_label(exp_q: str) -> str:
     meta = WORKLOAD_CATALOG[exp_q]
     return f"{exp_q} ({meta['name']}, {meta['entry']})"
@@ -86,6 +106,23 @@ def _resolve_default_config_path(base_dir: Path, canonical_q: str) -> Path | Non
     return None
 
 
+def _write_matrix_status(output_dir: Path, attempts: list[dict[str, object]]) -> None:
+    """Persist a machine-readable status summary, including partial runs."""
+    output_dir.mkdir(parents=True, exist_ok=True)
+    payload = {
+        "schema_version": 1,
+        "updated_at": datetime.now(timezone.utc).isoformat(),
+        "total": len(attempts),
+        "passed": sum(item["status"] == "passed" for item in attempts),
+        "failed": sum(item["status"] == "failed" for item in attempts),
+        "attempts": attempts,
+    }
+    target = output_dir / "matrix_status.json"
+    temporary = target.with_suffix(".json.tmp")
+    temporary.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8")
+    temporary.replace(target)
+
+
 def main() -> int:
     # Import here so --help is fast even without heavy deps installed.
     from experiments.common.cli_args import (
@@ -119,28 +156,37 @@ def main() -> int:
 
     # ── Workload selection (suite-level) ────────────────────────────────────
     selection_grp = parser.add_argument_group("workload selection")
-    selection_grp.add_argument(
+    selection = selection_grp.add_mutually_exclusive_group(required=True)
+    selection.add_argument(
         "--experiment",
         "-e",
         type=str,
         help="Workload to run (Q1..Q8).",
     )
-    selection_grp.add_argument(
-        "--all", "-a", action="store_true", help="Run all workloads in catalog."
-    )
+    selection.add_argument("--all", "-a", action="store_true", help="Run all workloads in catalog.")
     selection_grp.add_argument(
         "--config", "-c", type=str, help="Path to a custom config YAML file."
     )
 
     # ── Standardised benchmark flags (shared across all workloads) ──────────
     add_common_benchmark_args(parser, include_quick=True, include_dry_run=True)
+    failure_grp = parser.add_mutually_exclusive_group()
+    failure_grp.add_argument(
+        "--continue-on-error",
+        dest="fail_fast",
+        action="store_false",
+        help="Run the remaining workloads after a failure (default).",
+    )
+    failure_grp.add_argument(
+        "--fail-fast",
+        dest="fail_fast",
+        action="store_true",
+        help="Stop the matrix immediately after the first failed repetition.",
+    )
+    parser.set_defaults(fail_fast=False)
 
     args = parser.parse_args()
 
-    if not args.experiment and not args.all:
-        parser.print_help()
-        return 1
-
     validate_benchmark_args(args)
 
     # Import here to avoid slow startup for --help
@@ -179,6 +225,9 @@ def main() -> int:
     output_dir.mkdir(parents=True, exist_ok=True)
 
     results: dict[str, object] = {}
+    attempts: list[dict[str, object]] = []
+    failure_count = 0
+    stop_matrix = False
 
     for exp_q in experiments_to_run:
         run_cfg = build_run_config(args, workload=exp_q)
@@ -236,22 +285,63 @@ def main() -> int:
             experiment.nodes = int(args.nodes)
             experiment.parallelism = int(args.parallelism)
             experiment.run_id = f"{exp_q.lower()}-{args.backend}-{rep}-{uuid.uuid4().hex[:8]}"
+            setup_started = False
             try:
-                experiment.setup()
-                result = experiment.run()
-                experiment.teardown()
+                rep_output.mkdir(parents=True, exist_ok=True)
+                with (rep_output / "run.log").open("a", encoding="utf-8") as log_handle:
+                    with redirect_stdout(_Tee(sys.stdout, log_handle)):
+                        with redirect_stderr(_Tee(sys.stderr, log_handle)):
+                            setup_started = True
+                            experiment.setup()
+                            result = experiment.run()
+                            setup_started = False
+                            experiment.teardown()
                 results.setdefault(exp_q, []).append(result)  # type: ignore[union-attr]
+                attempts.append(
+                    {
+                        "workload": exp_q.lower(),
+                        "repeat": rep,
+                        "run_id": experiment.run_id,
+                        "status": "passed",
+                        "output_dir": str(rep_output),
+                    }
+                )
+                _write_matrix_status(output_dir, attempts)
                 print(
                     f"  Workload {_workload_label(exp_q)}{rep_label} completed. "
                     f"Results saved to {rep_output}"
                 )
             except Exception as exc:  # noqa: BLE001
+                if setup_started:
+                    try:
+                        experiment.teardown()
+                    except Exception as teardown_exc:  # noqa: BLE001
+                        exc = RuntimeError(f"{exc}; teardown also failed: {teardown_exc}")
                 print(f"Error running workload {_workload_label(exp_q)}{rep_label}: {exc}")
+                with (rep_output / "run.log").open("a", encoding="utf-8") as log_handle:
+                    log_handle.write(f"ERROR: {exc}\n")
                 if args.verbose:
                     import traceback
 
                     traceback.print_exc()
                 results.setdefault(exp_q, []).append({"error": str(exc)})  # type: ignore[union-attr]
+                failure_count += 1
+                attempts.append(
+                    {
+                        "workload": exp_q.lower(),
+                        "repeat": rep,
+                        "run_id": experiment.run_id,
+                        "status": "failed",
+                        "error": str(exc),
+                        "output_dir": str(rep_output),
+                    }
+                )
+                _write_matrix_status(output_dir, attempts)
+                if args.fail_fast:
+                    stop_matrix = True
+                    break
+        if stop_matrix:
+            break
 
     if not args.dry_run:
         print(f"\n{'=' * 60}")
@@ -270,7 +360,7 @@ def main() -> int:
                 print(f"  {_workload_label(exp_q)}: COMPLETED")
         print(f"\nResults saved to: {output_dir.absolute()}")
 
-    return 0
+    return 1 if failure_count else 0
 
 
 if __name__ == "__main__":  # pragma: no cover