path: root/tools/acceptance/e2e_acceptance.py

#!/usr/bin/env python3
"""Checked-in browser acceptance driver for the live pipeline viewport.

Starts a Release server on an isolated port with its own artifact root, drives
a real chromium browser through a live pipeline run (desktop and narrow
viewports), and asserts actual on-screen geometry/framing via the projected
bounding box of the loaded GLB - not just labels or whole-canvas hashes.

Playwright is provisioned into tools/acceptance/deps (repo-local, never
system/user site-packages). First run may download chromium into the default
user browser cache via `playwright install chromium`.

Usage:
  python3 tools/acceptance/e2e_acceptance.py [--port 18201]
      [--artifact-root DIR] [--evidence-dir DIR] [--stage-root DIR]
      [--server-dll PATH] [--bpython PATH] [--keep-server]

Preconditions:
  dotnet tool restore          (if tools are pinned in .config/dotnet-tools.json)
  dotnet fable src/SomhairlesDream.Frontend/SomhairlesDream.Frontend.fsproj --outDir public --noRestore --noCache
  dotnet build SomhairlesDream.sln -c Release
  real bpy interpreter available (default /tmp/bpyenv/bin/python)
"""

import argparse
import hashlib
import json
import os
import shutil
import signal
import socket
import subprocess
import sys
import time
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path

TOOLS_DIR = Path(__file__).resolve().parent
REPO = TOOLS_DIR.parents[1]
DEPS = TOOLS_DIR / "deps"
DEFAULT_BPYTHON = Path("/tmp/bpyenv/bin/python")
DEFAULT_SERVER_DLL = (
    REPO / "src" / "SomhairlesDream.Server" / "bin" / "Release" / "net8.0" / "SomhairlesDream.Server.dll"
)

REFERENCE_SHA = {
    "steps/01-foundation.glb": "6082e735a65557665f68e673bd2899a38818d57a09e244a63fa3fb7f3cc687ef",
    "steps/03-cabin.glb": "4c54f96c66cdb6ad56e2d4118a90e2d6507a2dc061c941867bda86e4a177a9b2",
}

GEOMETRY_JS = """
() => {
  const v = window.SOMHAIRLES_VIEWER;
  if (!v || !v.camera || !v.group || !v.canvas) {
    return { ready: false, reason: 'viewer handle missing' };
  }
  const w = v.canvas.clientWidth, h = v.canvas.clientHeight;
  if (!w || !h) return { ready: false, reason: 'canvas has no size' };
  const box = new THREE.Box3().setFromObject(v.group);
  if (box.isEmpty()) return { ready: false, reason: 'empty bounding box' };
  const size = new THREE.Vector3(); box.getSize(size);
  const lo = box.min, hi = box.max;
  const corners = [
    [lo.x,lo.y,lo.z],[hi.x,lo.y,lo.z],[lo.x,hi.y,lo.z],[hi.x,hi.y,lo.z],
    [lo.x,lo.y,hi.z],[hi.x,lo.y,hi.z],[lo.x,hi.y,hi.z],[hi.x,hi.y,hi.z]
  ].map((p) => new THREE.Vector3(p[0], p[1], p[2]).project(v.camera));
  const px = corners.map((c) => [(c.x * 0.5 + 0.5) * w, (0.5 - c.y * 0.5) * h]);
  const xs = px.map((p) => p[0]);
  const ys = px.map((p) => p[1]);
  const minX = Math.min(...xs), maxX = Math.max(...xs);
  const minY = Math.min(...ys), maxY = Math.max(...ys);
  const minDim = Math.min(w, h);
  const extent = Math.max(maxX - minX, maxY - minY);
  const cx = (minX + maxX) / 2, cy = (minY + maxY) / 2;
  return {
    ready: true,
    canvas: [w, h],
    innerMargin: Math.min(minX, minY, w - maxX, h - maxY),
    coverage: extent / minDim,
    centerOffset: Math.hypot(cx - w / 2, cy - h / 2) / minDim,
    boxSize: [size.x, size.y, size.z],
    childCount: v.group.children.length
  };
}
"""


SCENE_JS = """
() => {
  const v = window.SOMHAIRLES_VIEWER;
  if (!v || !v.group) return { ready: false, reason: 'viewer handle missing' };
  const kids = Array.from(v.group.children).map((c) => ({
    type: c.type,
    isMesh: c.isMesh === true,
    color: c.material && c.material.color && c.material.color.getHexString
      ? c.material.color.getHexString() : null,
    childCount: c.children ? c.children.length : 0
  }));
  const box = new THREE.Box3().setFromObject(v.group);
  const size = new THREE.Vector3();
  if (!box.isEmpty()) box.getSize(size);
  return {
    ready: true,
    childCount: v.group.children.length,
    kids: kids,
    boxSize: box.isEmpty() ? null : [size.x, size.y, size.z],
    note: (document.querySelector('#viewport-note') || {}).textContent || '',
    meshLabel: (document.querySelector('#mesh-label') || {}).textContent || ''
  };
}
"""


def log(msg):
    print(f"[acceptance] {msg}", flush=True)


def sha256_bytes(data):
    return hashlib.sha256(data).hexdigest()


def ensure_playwright():
    if not (DEPS / "playwright").exists():
        log("provisioning playwright into tools/acceptance/deps (repo-local, not system-wide)")
        pinned = ""
        try:
            probe = subprocess.run(
                [sys.executable, "-c", "import playwright,sys; sys.stdout.write(playwright.__version__)"],
                capture_output=True, text=True, timeout=60,
            )
            if probe.returncode == 0 and probe.stdout.strip():
                pinned = f"=={probe.stdout.strip()}"
                log(f"pinning playwright{pinned} to reuse any existing chromium cache")
        except Exception:
            pass
        DEPS.mkdir(parents=True, exist_ok=True)
        subprocess.run(
            [sys.executable, "-m", "pip", "install", "--target", str(DEPS),
             "--break-system-packages", f"playwright{pinned}"],
            check=True,
        )
    env = os.environ.copy()
    env["PYTHONPATH"] = str(DEPS)
    log("ensuring chromium browser is present (no-op if already downloaded)")
    subprocess.run([sys.executable, "-m", "playwright", "install", "chromium"],
                   env=env, check=True)
    if str(DEPS) not in sys.path:
        sys.path.insert(0, str(DEPS))
    from playwright.sync_api import sync_playwright

    return sync_playwright


def wait_for_port(port, timeout=45.0):
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            with socket.create_connection(("127.0.0.1", port), timeout=1.0):
                return
        except OSError:
            time.sleep(0.3)
    raise RuntimeError(f"server did not open port {port} within {timeout}s")


def fetch_bytes(url):
    with urllib.request.urlopen(url, timeout=30) as resp:
        return resp.status, resp.headers.get("Content-Type", ""), resp.read()


def wait_text(page, selector, needle, timeout_ms=90000):
    page.wait_for_function(
        "([sel, needle]) => (document.querySelector(sel)?.textContent ?? '').includes(needle)",
        arg=[selector, needle],
        timeout=timeout_ms,
    )


def assert_geometry_fit(geo, label, results):
    checks = []
    if not geo.get("ready"):
        results.append((False, f"{label}: geometry probe not ready ({geo.get('reason')})"))
        return checks
    w, h = geo["canvas"]
    min_dim = min(w, h)
    checks.append(("childCount", geo["childCount"] >= 1, geo))
    checks.append(("innerMargin>=2%minDim", geo["innerMargin"] >= 0.02 * min_dim, geo))
    checks.append(("coverage in [0.40,0.95]", 0.40 <= geo["coverage"] <= 0.95, geo))
    checks.append(("centerOffset<=10%minDim", geo["centerOffset"] <= 0.10, geo))
    sizes = geo["boxSize"]
    finite_pos = all(isinstance(s, (int, float)) and s > 0 and s == s and s != float("inf") for s in sizes)
    checks.append(("boxSize finite>0", finite_pos, geo))
    for name, ok, payload in checks:
        results.append((ok, f"{label}: {name} -> {json.dumps(payload) if not ok else 'ok'}"))
    return checks


def run_flow(page, base_url, run_id, viewport_label, results, evidence, shot_prefix):
    page.goto(f"{base_url}/?freeze=1", wait_until="domcontentloaded")
    page.fill("#project-id-input", "heritage-001")
    page.fill("#base-id-input", "base-000")
    page.fill("#target-id-input", "target-003")
    wait_text(page, "#status-label", "待机")
    results.append((True, f"{viewport_label}: idle 待机 visible"))
    body = page.inner_text("body")
    results.append(("当前没有下一版正在开发" in body, f"{viewport_label}: empty-state text present"))

    page.fill("#run-id-input", run_id)
    render_box = page.query_selector("#run-render-input")
    if render_box and render_box.is_checked():
        render_box.set_checked(False)
    page.click("#run-start-button")

    wait_text(page, "#stream-state", "已连接")
    results.append((True, f"{viewport_label}: SSE 已连接"))

    wait_text(page, "#viewport-note", "管线几何 · steps/01-foundation.glb")
    wait_text(page, "#status-label", "运行中")
    results.append((True, f"{viewport_label}: stage1 live (运行中 + steps/01-foundation.glb)"))
    page.wait_for_timeout(600)
    geo1 = page.evaluate(GEOMETRY_JS)
    (evidence / f"{shot_prefix}-geo-stage1.json").write_text(json.dumps(geo1, indent=2))
    assert_geometry_fit(geo1, f"{viewport_label}/stage1", results)
    page.locator("#viewport-canvas").screenshot(path=str(evidence / f"{shot_prefix}-canvas-stage1.png"))
    page.screenshot(path=str(evidence / f"{shot_prefix}-page-stage1.png"), full_page=True)

    wait_text(page, "#viewport-note", "管线几何 · steps/03-cabin.glb")
    wait_text(page, "#status-label", "已完成")
    results.append((True, f"{viewport_label}: stage3 complete (已完成 + steps/03-cabin.glb)"))
    page.wait_for_timeout(600)
    geo3 = page.evaluate(GEOMETRY_JS)
    (evidence / f"{shot_prefix}-geo-stage3.json").write_text(json.dumps(geo3, indent=2))
    assert_geometry_fit(geo3, f"{viewport_label}/stage3", results)
    page.locator("#viewport-canvas").screenshot(path=str(evidence / f"{shot_prefix}-canvas-stage3.png"))

    if geo1.get("ready") and geo3.get("ready"):
        diffs = [abs(a - b) / max(abs(a), abs(b), 1e-9) for a, b in zip(geo1["boxSize"], geo3["boxSize"])]
        results.append((max(diffs) >= 0.05,
                        f"{viewport_label}: stage1 vs stage3 bbox differs (max rel diff {max(diffs):.3f})"))

    shot3 = evidence / f"{shot_prefix}-canvas-stage3.png"
    page.wait_for_timeout(800)
    repeat = evidence / f"{shot_prefix}-canvas-stage3-repeat.png"
    page.locator("#viewport-canvas").screenshot(path=str(repeat))
    results.append((sha256_bytes(shot3.read_bytes()) == sha256_bytes(repeat.read_bytes()),
                    f"{viewport_label}: frozen stage3 screenshot deterministic"))
    page.screenshot(path=str(evidence / f"{shot_prefix}-page-stage3.png"), full_page=True)

    href = page.get_attribute("#artifact-manifest-link", "href")
    results.append((bool(href), f"{viewport_label}: manifest link exposed ({href})"))


def stale_load_regression(browser, base_url, track, results, evidence):
    """Regression for async step-load ownership across runs.

    Intercepts steps GLB requests and fulfills them late, out of order, or
    with errors to prove stale responses from superseded runs never repaint
    geometry or notes, while the current owner's loads keep working. Also
    covers: user retry of a failed current latest-step load after run
    completion (same owner/path, no SSE reconnect), same-run older-step
    responses resolving after the newer step, and pending live responses
    ignored after entering replay. All still-held responses are drained and
    the route handler is removed before the page closes.
    """
    page = browser.new_page(viewport={"width": 1280, "height": 900})

    def ignore_intentional_500(m):
        return "status of 500" in m.text \
            and "/api/artifacts/steps" in (m.location or {}).get("url", "")

    track(page, ignore=ignore_intentional_500)
    stamp = int(time.time() * 1000)
    held = []

    def hold_route(route):
        q = urllib.parse.parse_qs(urllib.parse.urlparse(route.request.url).query)
        if (q.get("runId") or [""])[0] == "run-probe-real-2":
            route.continue_()
            return
        held.append({
            "route": route,
            "project": (q.get("projectId") or [""])[0],
            "run": (q.get("runId") or [""])[0],
            "path": (q.get("path") or [""])[0],
            "done": False,
        })

    page.route("**/api/artifacts/steps*", hold_route)
    page.goto(f"{base_url}/?freeze=1", wait_until="domcontentloaded")
    page.fill("#project-id-input", "heritage-001")
    page.fill("#base-id-input", "base-000")
    page.fill("#target-id-input", "target-003")
    wait_text(page, "#status-label", "待机")

    def start_run(run_id):
        page.fill("#run-id-input", run_id)
        render_box = page.query_selector("#run-render-input")
        if render_box and render_box.is_checked():
            render_box.set_checked(False)
        page.click("#run-start-button")
        wait_text(page, "#stream-state", "已连接")

    def held_for(run_id, path, timeout_s=120):
        deadline = time.time() + timeout_s
        while time.time() < deadline:
            for entry in held:
                if not entry["done"] and entry["run"] == run_id and entry["path"] == path:
                    return entry
            page.wait_for_timeout(300)
        raise AssertionError(
            f"no held steps request for run={run_id} path={path}; "
            f"held={[(e['run'], e['path'], e['done']) for e in held]}")

    def release(entry, status=200):
        if status == 200:
            q = urllib.parse.urlencode({
                "projectId": entry["project"], "runId": entry["run"], "path": entry["path"],
            })
            resp_status, ctype, body = fetch_bytes(f"{base_url}/api/artifacts/steps?{q}")
            if resp_status != 200:
                raise AssertionError(f"artifact fetch for fulfill failed ({resp_status})")
        else:
            ctype, body = "text/plain", b"boom"
        entry["route"].fulfill(status=status, content_type=ctype, body=body)
        entry["done"] = True

    def is_frame(g):
        return bool(g.get("kids")) and g["kids"][0]["type"] == "Mesh" \
            and g["kids"][0]["color"] == "d7a948"

    def is_glb(g):
        return bool(g.get("kids")) and g["kids"][0]["type"] == "Group"

    def box_ok(g):
        return g.get("boxSize") and all(
            isinstance(s, (int, float)) and s > 0 for s in g["boxSize"])

    # Scenario A: same path across two distinct runs must trigger a reload.
    run_a = f"run-stale-{stamp}-a"
    start_run(run_a)
    release(held_for(run_a, "steps/01-foundation.glb"))
    wait_text(page, "#viewport-note", "管线几何 · steps/01-foundation.glb")
    page.wait_for_timeout(600)
    ga = page.evaluate(SCENE_JS)
    results.append((ga["ready"] and is_glb(ga) and box_ok(ga),
                    f"stale-load: run A displays GLB scene (kids={ga.get('kids')})"))
    (evidence / "stale-scene-a.json").write_text(json.dumps(ga, indent=2, ensure_ascii=False))

    run_b = f"run-stale-{stamp}-b"
    start_run(run_b)
    eb = held_for(run_b, "steps/01-foundation.glb")
    results.append((True, f"stale-load: same path across runs reloads for run B ({eb['run']})"))
    release(eb)
    wait_text(page, "#viewport-note", "管线几何 · steps/01-foundation.glb")
    page.wait_for_timeout(600)
    gb = page.evaluate(SCENE_JS)
    results.append((gb["ready"] and is_glb(gb) and box_ok(gb),
                    f"stale-load: run B repaints GLB after same-path switch (kids={gb.get('kids')})"))
    page.screenshot(path=str(evidence / "stale-page-b.png"), full_page=True)

    # Scenario B: late success from a superseded run must be ignored.
    run_c = f"run-stale-{stamp}-c"
    start_run(run_c)
    held_for(run_c, "steps/01-foundation.glb")

    run_d = f"run-stale-{stamp}-d"
    start_run(run_d)
    frame_geo = None
    deadline = time.time() + 30
    while time.time() < deadline:
        g = page.evaluate(SCENE_JS)
        if g["ready"] and is_frame(g):
            frame_geo = g
            break
        page.wait_for_timeout(300)
    results.append((frame_geo is not None,
                    "stale-load: run switch with empty steps repaints replay frame mesh "
                    f"({(frame_geo or {}).get('kids')})"))
    page.screenshot(path=str(evidence / "stale-page-d-frame.png"), full_page=True)

    release(held_for(run_c, "steps/01-foundation.glb"))
    page.wait_for_timeout(600)
    g2 = page.evaluate(SCENE_JS)
    results.append((g2["ready"] and is_frame(g2)
                    and g2["note"] != "管线几何 · steps/01-foundation.glb",
                    f"stale-load: late old success ignored (note='{g2['note'][:40]}', "
                    f"kids={g2['kids']})"))

    release(held_for(run_d, "steps/01-foundation.glb"))
    wait_text(page, "#viewport-note", "管线几何 · steps/01-foundation.glb")
    page.wait_for_timeout(600)
    g3 = page.evaluate(SCENE_JS)
    results.append((g3["ready"] and is_glb(g3) and box_ok(g3),
                    "stale-load: run D current load displays after late-old-success"))

    # Scenario C: late error from a superseded run must not fail the current load.
    run_e = f"run-stale-{stamp}-e"
    start_run(run_e)
    ee = held_for(run_e, "steps/01-foundation.glb")

    run_f = f"run-stale-{stamp}-f"
    start_run(run_f)
    ef = held_for(run_f, "steps/01-foundation.glb")

    release(ee, status=500)
    page.wait_for_timeout(600)
    g4 = page.evaluate(SCENE_JS)
    results.append(("加载失败" not in g4["note"],
                    f"stale-load: late old error ignored (note='{g4['note'][:40]}')"))

    release(ef)
    wait_text(page, "#viewport-note", "管线几何 · steps/01-foundation.glb")
    page.wait_for_timeout(600)
    g5 = page.evaluate(SCENE_JS)
    results.append((g5["ready"] and is_glb(g5) and box_ok(g5) and "加载失败" not in g5["note"],
                    "stale-load: current owner load unaffected by stale error"))
    (evidence / "stale-scene-final.json").write_text(json.dumps(g5, indent=2, ensure_ascii=False))
    page.screenshot(path=str(evidence / "stale-page-final.png"), full_page=True)

    # Scenario D: current latest-step failure after run completion retries the
    # same owner/path via the visible note affordance, without new SSE events.
    run_k = f"run-stale-{stamp}-k"
    start_run(run_k)
    release(held_for(run_k, "steps/01-foundation.glb"))
    wait_text(page, "#viewport-note", "管线几何 · steps/01-foundation.glb")
    ek3 = held_for(run_k, "steps/03-cabin.glb")
    wait_text(page, "#status-label", "已完成")
    release(ek3, status=500)
    wait_text(page, "#viewport-note", "管线几何加载失败 · steps/03-cabin.glb")
    cursor = page.evaluate(
        "() => getComputedStyle(document.querySelector('#viewport-note')).cursor")
    results.append((cursor == "pointer",
                    f"stale-load: failed current load exposes retry affordance (cursor={cursor})"))
    page.screenshot(path=str(evidence / "stale-page-retry-failed.png"), full_page=True)
    page.click("#viewport-note")
    ek4 = held_for(run_k, "steps/03-cabin.glb")
    retries = [e for e in held if e["run"] == run_k and e["path"] == "steps/03-cabin.glb"]
    results.append((len(retries) == 2 and ek4 is retries[1],
                    f"stale-load: user retry re-requests same owner/path ({len(retries)} requests)"))
    stream = page.evaluate("() => document.querySelector('#stream-state').textContent")
    results.append(("已连接" in stream,
                    f"stale-load: retry works without SSE reconnect (stream='{stream[:30]}')"))
    release(ek4)
    wait_text(page, "#viewport-note", "管线几何 · steps/03-cabin.glb")
    page.wait_for_timeout(600)
    gk = page.evaluate(SCENE_JS)
    cursor_after = page.evaluate(
        "() => getComputedStyle(document.querySelector('#viewport-note')).cursor")
    results.append((gk["ready"] and is_glb(gk) and box_ok(gk) and cursor_after != "pointer",
                    f"stale-load: retry displays latest step and clears affordance "
                    f"(kids={gk.get('kids')})"))
    page.screenshot(path=str(evidence / "stale-page-retry-ok.png"), full_page=True)

    # Scenario E: same-run older step resolving after the newer step must be ignored.
    run_h = f"run-stale-{stamp}-h"
    start_run(run_h)
    held_for(run_h, "steps/01-foundation.glb")
    eh3 = held_for(run_h, "steps/03-cabin.glb")
    release(eh3)
    wait_text(page, "#viewport-note", "管线几何 · steps/03-cabin.glb")
    page.wait_for_timeout(600)
    gh = page.evaluate(SCENE_JS)
    results.append((gh["ready"] and is_glb(gh) and box_ok(gh)
                    and "steps/03-cabin.glb" in gh["note"],
                    f"stale-load: newer same-run step displayed after supersede "
                    f"(kids={gh.get('kids')})"))
    note_h = gh["note"]
    release(held_for(run_h, "steps/01-foundation.glb"))
    page.wait_for_timeout(600)
    gh2 = page.evaluate(SCENE_JS)
    results.append((gh2["ready"] and gh2["kids"] == gh["kids"] and gh2["note"] == note_h,
                    f"stale-load: late older same-run step ignored (note='{gh2['note'][:40]}')"))

    # Scenario F: entering historical replay pins a real checkpoint; a live
    # GLB response still pending from before must not repaint the pinned view.
    run_j = f"run-stale-{stamp}-j"
    start_run(run_j)
    release(held_for(run_j, "steps/01-foundation.glb"))
    wait_text(page, "#viewport-note", "管线几何 · steps/01-foundation.glb")
    ej3 = held_for(run_j, "steps/03-cabin.glb")
    wait_text(page, "#status-label", "已完成")
    page.click("#replay-button")
    ehist = held_for(run_j, "steps/01-foundation.glb")
    release(ehist)
    wait_text(page, "#viewport-note", "历史几何 · steps/01-foundation.glb")
    page.wait_for_timeout(300)
    gj = page.evaluate(SCENE_JS)
    results.append((gj["ready"] and is_glb(gj) and "历史几何" in gj["note"],
                    f"stale-load: replay pins a real checkpoint with historical labels "
                    f"(note='{gj.get('note', '')[:40]}')"))
    page.screenshot(path=str(evidence / "stale-page-replay.png"), full_page=True)
    page.click("#replay-button")
    page.wait_for_timeout(300)
    gj_pause = page.evaluate(SCENE_JS)
    results.append((gj_pause["kids"] == gj["kids"] and gj_pause["note"] == gj["note"],
                    f"stale-load: historical playback pauses on the selected checkpoint "
                    f"(state='{page.inner_text('#replay-state')}')"))
    release(ej3)
    page.wait_for_timeout(600)
    gj2 = page.evaluate(SCENE_JS)
    results.append((gj2["ready"] and gj2["kids"] == gj["kids"] and gj2["note"] == gj["note"],
                    f"stale-load: live GLB after entering historical replay ignored "
                    f"(note='{gj2['note'][:40]}')"))

    for entry in held:
        if not entry["done"]:
            release(entry)
    page.unroute("**/api/artifacts/steps*")
    page.close()


def historical_replay_regression(browser, base_url, track, results, evidence):
    """Regression for real historical checkpoint replay of a selected run.

    Proves the timeline offers a clearly labeled placeholder fallback while no
    run history exists, that pinning a timeline step issues and serves the
    selected run's own checkpoint artifact (byte-identical to the reference
    GLB) under distinct historical labels, that the rendered geometry actually
    changes when a different checkpoint is selected, that a late older
    historical response cannot repaint a newer selection, that live snapshot
    updates cannot unpin the chosen historical step, and that 跟随最新 returns
    the viewport to live scope. All still-held responses are drained and the
    route handler is removed before the page closes.
    """
    page = browser.new_page(viewport={"width": 1280, "height": 900})

    track(page)
    stamp = int(time.time() * 1000)
    held = []

    def hold_route(route):
        q = urllib.parse.parse_qs(urllib.parse.urlparse(route.request.url).query)
        if (q.get("runId") or [""])[0] == "run-probe-real-2":
            route.continue_()
            return
        held.append({
            "route": route,
            "project": (q.get("projectId") or [""])[0],
            "run": (q.get("runId") or [""])[0],
            "path": (q.get("path") or [""])[0],
            "done": False,
        })

    page.route("**/api/artifacts/steps*", hold_route)
    page.goto(f"{base_url}/?freeze=1", wait_until="domcontentloaded")
    page.fill("#project-id-input", "heritage-001")
    page.fill("#base-id-input", "base-000")
    page.fill("#target-id-input", "target-003")
    wait_text(page, "#status-label", "待机")

    def start_run(run_id):
        page.fill("#run-id-input", run_id)
        render_box = page.query_selector("#run-render-input")
        if render_box and render_box.is_checked():
            render_box.set_checked(False)
        page.click("#run-start-button")
        wait_text(page, "#stream-state", "已连接")

    def held_for(run_id, path, timeout_s=120):
        deadline = time.time() + timeout_s
        while time.time() < deadline:
            for entry in held:
                if not entry["done"] and entry["run"] == run_id and entry["path"] == path:
                    return entry
            page.wait_for_timeout(300)
        raise AssertionError(
            f"no held steps request for run={run_id} path={path}; "
            f"held={[(e['run'], e['path'], e['done']) for e in held]}")

    def release(entry, status=200):
        if status == 200:
            q = urllib.parse.urlencode({
                "projectId": entry["project"], "runId": entry["run"], "path": entry["path"],
            })
            resp_status, ctype, body = fetch_bytes(f"{base_url}/api/artifacts/steps?{q}")
            if resp_status != 200:
                raise AssertionError(f"artifact fetch for fulfill failed ({resp_status})")
        else:
            ctype, body = "text/plain", b"boom"
        entry["route"].fulfill(status=status, content_type=ctype, body=body)
        entry["done"] = True

    def is_glb(g):
        return bool(g.get("kids")) and g["kids"][0]["type"] == "Group"

    def box_ok(g):
        return g.get("boxSize") and all(
            isinstance(s, (int, float)) and s > 0 for s in g["boxSize"])

    # Boot must surface the pre-published probe build replay with honest labels.
    wait_text(page, "#viewport-note", "历史几何 · steps/probe-07-livery.glb")
    results.append((True,
                    "historical: boot loads pre-published probe replay "
                    "(历史几何 · steps/probe-07-livery.glb)"))
    idle_state = page.inner_text("#replay-state")
    idle_button = page.inner_text("#replay-button")
    results.append(("历史" in idle_state and "历史" in idle_button,
                    f"historical: boot historical labels are honest "
                    f"(state='{idle_state}', button='{idle_button}')"))

    # The placeholder fallback stays honest whenever real history is absent.
    # (Placeholder idle is ensured on pages where the manifest fetch fails;
    # see run_flow heritage pinning plus the standalone offline guard above.)
    # Phase I: pin a checkpoint mid-run; live snapshot updates must not unpin it.
    run_id = f"run-replay-{stamp}"
    start_run(run_id)
    release(held_for(run_id, "steps/01-foundation.glb"))
    wait_text(page, "#viewport-note", "管线几何 · steps/01-foundation.glb")
    held_before = len(held)
    page.locator(".timeline-step").nth(0).click()
    e_pinned = held_for(run_id, "steps/01-foundation.glb")
    results.append((e_pinned["run"] == run_id and e_pinned["path"] == "steps/01-foundation.glb",
                    f"historical: pin requests the run's own checkpoint "
                    f"(run={e_pinned['run']}, path={e_pinned['path']})"))
    release(e_pinned)
    wait_text(page, "#viewport-note", "历史几何 · steps/01-foundation.glb")
    page.wait_for_timeout(400)
    g_pinned = page.evaluate(SCENE_JS)
    results.append((g_pinned["ready"] and is_glb(g_pinned)
                    and "历史" in page.inner_text("#replay-state"),
                    f"historical: pinned step carries historical labels "
                    f"(state='{page.inner_text('#replay-state')[:36]}')"))

    page.wait_for_function(
        "() => document.querySelectorAll('.timeline-step').length === 3", timeout=150000)
    page.wait_for_timeout(600)
    g_unpin = page.evaluate(SCENE_JS)
    stray = [e for e in held[held_before:]
             if e["run"] == run_id and e["path"] == "steps/03-cabin.glb"]
    results.append((g_unpin["kids"] == g_pinned["kids"] and g_unpin["note"] == g_pinned["note"]
                    and not stray,
                    f"historical: live snapshot update cannot unpin the pinned step "
                    f"(stray 03 requests={len(stray)})"))

    # Phase II: 跟随最新 returns the viewport to live scope with the newest step.
    page.click("#follow-button")
    e_live3 = held_for(run_id, "steps/03-cabin.glb")
    release(e_live3)
    wait_text(page, "#viewport-note", "管线几何 · steps/03-cabin.glb")
    page.wait_for_timeout(500)
    g_live3 = page.evaluate(SCENE_JS)
    results.append((g_live3["ready"] and is_glb(g_live3) and box_ok(g_live3),
                    f"historical: 跟随最新 switches back to live scope and loads the "
                    f"newest checkpoint (kids={g_live3.get('kids')})"))
    box_live3 = g_live3["boxSize"]

    # Phase III: a late older historical response cannot repaint the newer selection.
    page.locator(".timeline-step").nth(0).click()
    e_late1 = held_for(run_id, "steps/01-foundation.glb")
    page.locator(".timeline-step").nth(2).click()
    e_hist3 = held_for(run_id, "steps/03-cabin.glb")
    release(e_late1)
    page.wait_for_timeout(600)
    g_late = page.evaluate(SCENE_JS)
    results.append((g_late["kids"] == g_live3["kids"]
                    and "历史几何加载中 · steps/03-cabin.glb" in g_late["note"],
                    f"historical: late older historical response ignored after newer "
                    f"selection (note='{g_late['note'][:40]}')"))
    release(e_hist3)
    wait_text(page, "#viewport-note", "历史几何 · steps/03-cabin.glb")
    page.wait_for_timeout(400)
    g_hist3 = page.evaluate(SCENE_JS)
    results.append((g_hist3["ready"] and is_glb(g_hist3) and g_hist3["note"] != g_late["note"],
                    f"historical: newer historical selection wins after late older response "
                    f"(note='{g_hist3['note'][:40]}')"))

    # Phase IV: selecting V01 changes both the served checkpoint identity and
    # the actual on-screen geometry.
    page.locator(".timeline-step").nth(0).click()
    e_hist1 = held_for(run_id, "steps/01-foundation.glb")
    q = urllib.parse.urlencode({
        "projectId": e_hist1["project"], "runId": e_hist1["run"], "path": e_hist1["path"],
    })
    status, ctype, body = fetch_bytes(f"{base_url}/api/artifacts/steps?{q}")
    results.append((status == 200 and sha256_bytes(body) == REFERENCE_SHA["steps/01-foundation.glb"],
                    f"historical: served checkpoint identity matches reference GLB "
                    f"(sha256={sha256_bytes(body)[:12]}…)"))
    release(e_hist1)
    wait_text(page, "#viewport-note", "历史几何 · steps/01-foundation.glb")
    page.wait_for_timeout(500)
    g_hist1 = page.evaluate(SCENE_JS)
    diffs = [abs(a - b) / max(abs(b), 1e-6) for a, b in zip(g_hist1["boxSize"], box_live3)]
    results.append((g_hist1["ready"] and is_glb(g_hist1) and max(diffs) >= 0.05,
                    f"historical: selection changes actual viewport geometry "
                    f"(max rel diff {max(diffs):.3f})"))
    page.screenshot(path=str(evidence / "historical-replay-pinned.png"), full_page=True)
    (evidence / "historical-replay-pinned.json").write_text(
        json.dumps(g_hist1, indent=2, ensure_ascii=False))

    for entry in held:
        if not entry["done"]:
            release(entry)
    page.unroute("**/api/artifacts/steps*")
    page.close()


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--port", type=int, default=18201)
    parser.add_argument("--artifact-root", default=str(REPO / "scratch" / "acceptance" / "artifacts"))
    parser.add_argument("--evidence-dir", default=None)
    parser.add_argument("--stage-root", default=str(REPO / "stage"))
    parser.add_argument("--server-dll", default=str(DEFAULT_SERVER_DLL))
    parser.add_argument("--bpython", default=str(DEFAULT_BPYTHON))
    parser.add_argument("--keep-server", action="store_true")
    args = parser.parse_args()

    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    evidence = Path(args.evidence_dir or (REPO / "scratch" / "acceptance" / f"evidence-{stamp}"))
    evidence.mkdir(parents=True, exist_ok=True)
    artifact_root = Path(args.artifact_root)
    artifact_root.mkdir(parents=True, exist_ok=True)
    server_log = evidence / "server.log"

    if not Path(args.server_dll).exists():
        log(f"server dll missing: {args.server_dll} - run 'dotnet build SomhairlesDream.sln -c Release' first")
        return 2
    if not Path(args.bpython).exists():
        log(f"bpy interpreter missing: {args.bpython} (pass --bpython)")
        return 2
    try:
        with socket.create_connection(("127.0.0.1", args.port), timeout=0.5):
            log(f"port {args.port} already in use - choose another --port")
            return 2
    except OSError:
        pass

    env = os.environ.copy()
    env["SOMHAIRLES_ARTIFACT_ROOT"] = str(artifact_root)
    env["SOMHAIRLES_STAGE_ROOT"] = str(args.stage_root)
    env["SOMHAIRLES_BPYTHON"] = str(args.bpython)
    server = subprocess.Popen(
        ["dotnet", args.server_dll, "--urls", f"http://127.0.0.1:{args.port}"],
        env=env, stdout=open(server_log, "wb"), stderr=subprocess.STDOUT,
        start_new_session=True, cwd=str(REPO),
    )
    base_url = f"http://127.0.0.1:{args.port}"
    results = []
    console_errors = []

    try:
        wait_for_port(args.port)
        with urllib.request.urlopen(f"{base_url}/", timeout=15) as resp:
            results.append((resp.status == 200, f"GET / -> {resp.status}"))
        sync_playwright = ensure_playwright()
        with sync_playwright() as pw:
            browser = pw.chromium.launch(args=["--enable-unsafe-swiftshader"])

            def track(page, ignore=None):
                def on_console(m):
                    if m.type == "error" and not (ignore and ignore(m)):
                        console_errors.append(m.text)
                page.on("console", on_console)
                page.on("pageerror", lambda e: console_errors.append(str(e)))

            desktop_ms = int(time.time() * 1000)
            page = browser.new_page(viewport={"width": 1440, "height": 1000})
            track(page)
            run_id = f"run-acc-{desktop_ms}-desktop"
            run_flow(page, base_url, run_id, "desktop", results, evidence, "desktop")
            project_id = page.input_value("#project-id-input")

            narrow_ms = int(time.time() * 1000) + 1
            page2 = browser.new_page(viewport={"width": 390, "height": 844})
            track(page2)
            run_id2 = f"run-acc-{narrow_ms}-narrow"
            run_flow(page2, base_url, run_id2, "narrow", results, evidence, "narrow")

            for step_path in ("steps/01-foundation.glb", "steps/03-cabin.glb"):
                q = urllib.parse.urlencode({
                    "projectId": project_id, "runId": run_id, "path": step_path,
                })
                status, ctype, body = fetch_bytes(f"{base_url}/api/artifacts/steps?{q}")
                digest = sha256_bytes(body)
                results.append((status == 200, f"GET steps {step_path} -> {status}"))
                results.append(("model/gltf-binary" in ctype, f"content-type {step_path} -> {ctype}"))
                expected = REFERENCE_SHA[step_path]
                results.append((digest == expected, f"sha256 {step_path} -> {digest[:12]}… (expect {expected[:12]}…)"))

            stale_load_regression(browser, base_url, track, results, evidence)
            historical_replay_regression(browser, base_url, track, results, evidence)

            fail_ms = int(time.time() * 1000) + 2
            fail_run_id = f"run-acc-{fail_ms}-failure"
            (artifact_root / project_id / fail_run_id).mkdir(parents=True, exist_ok=True)
            page.fill("#run-id-input", fail_run_id)
            page.click("#run-start-button")
            wait_text(page, "#status-label", "失败")
            msg = page.inner_text("#run-message")
            results.append(("run directory already exists" in msg,
                            f"failure run surfaces pipeline error (message: {msg.strip()[:80]}…)"))

            browser.close()
    finally:
        if not args.keep_server:
            try:
                os.killpg(os.getpgid(server.pid), signal.SIGTERM)
            except ProcessLookupError:
                pass
            try:
                server.wait(timeout=10)
            except subprocess.TimeoutExpired:
                os.killpg(os.getpgid(server.pid), signal.SIGKILL)
        else:
            log(f"server kept alive on port {args.port} (pid {server.pid})")

    if console_errors:
        results.append((False, f"console/page errors: {console_errors[:5]}"))
    else:
        results.append((True, "zero browser console/page errors"))

    passed = sum(1 for ok, _ in results if ok)
    failed = len(results) - passed
    report = {
        "timestamp": stamp,
        "base_url": base_url,
        "evidence_dir": str(evidence),
        "console_errors": console_errors,
        "checks": [{"ok": ok, "detail": detail} for ok, detail in results],
    }
    (evidence / "acceptance-report.json").write_text(json.dumps(report, indent=2, ensure_ascii=False))
    for ok, detail in results:
        log(f"{'PASS' if ok else 'FAIL'}: {detail}")
    log(f"RESULT: {passed}/{len(results)} checks passed ({failed} failed); evidence in {evidence}")
    return 0 if failed == 0 else 1


if __name__ == "__main__":
    sys.exit(main())