#!/usr/bin/env python3
"""twice.py: ask the same question twice and diff the answers.

Every other sweep in this directory reads each surface ONCE. So none of them
can see a value that MOVES, and a value that moves is not a fact about the
vendor -- it is a nonce. Wake dc7d9337 found the first instance by accident:
Allium's x402 challenge names a different destination address on two requests
four seconds apart, on two of its four rails, at the same amount. A filed probe
record had compared two readings NINE DAYS apart and called them "identical
shape", which was true and blind: a comparison of shapes cannot see a value
that rotates.

This is the general form of that question, and the reasoning that made the
Allium finding sound is the part worth mechanising. A field that differs
between two readings has two possible causes, and they are not equally
interesting:

  the SERVER rotates it          -> a finding about the vendor
  MY OWN client saw two          -> a finding about my network, worth nothing
  different edges / caches
  / a load balancer

Nothing in a single differing field tells those apart. What tells them apart is
the CONTROL INSIDE THE SAME RESPONSE: if two members of a group move while the
other two hold still, no story about edge nodes survives. So this instrument
never reports "field X moved". It reports, for every group of comparable
fields, how many moved and how many held, and it grades:

  STABLE         nothing differed at all.
  VOLATILE-ONLY  only fields HTTP itself says move (Date, request ids, a
                 remaining-quota counter) differed. Listed anyway: "not
                 counted" is not "cleared".
  ROTATING       a field moved while a comparable field HELD. The control is
                 inside the response. This is a finding.
  ALL-MOVED      every comparable field moved together. Suspicious, and NOT a
                 finding on its own: this is exactly what a client reaching two
                 different edge nodes looks like. Reported for a human.
  NO-ANSWER      one or both readings failed.

Two live controls run before any verdict prints, because an instrument whose
failure mode looks like success is worse than none (corpus.py's lesson):

  POSITIVE: across the whole run, the `date` response header MUST differ
            somewhere. If no surface's clock moved between two readings, then
            the two "readings" are one response served twice out of something
            in front of me, and every STABLE verdict below is manufactured.
  NEGATIVE: a static document (the registry's own published schema) MUST come
            back byte-identical twice. If a file that cannot change differs,
            the diff machinery or the path to the network is noisy, and every
            ROTATING verdict is manufactured.

Either control failing REFUSES the run rather than annotating it.

    python3 tools/twice.py --clone /tmp/reg          # every surfaces.mcp
    python3 tools/twice.py --url https://a/x ...     # ad-hoc, no clone
    python3 tools/twice.py --selftest                # no network, no clone
"""

import sys as _sys

# tools/queue.py shadows the stdlib `queue`, which concurrent.futures imports.
_here = __file__.rsplit("/", 1)[0]
_sys.path[:] = [p for p in _sys.path if p not in ("", ".", _here)]

import argparse
import concurrent.futures
import datetime
import glob
import hashlib
import json
import os
import re
import time
import urllib.error
import urllib.request
import importlib.util as _ilu

_cspec = _ilu.spec_from_file_location("corpus", os.path.join(_here, "corpus.py"))
corpus = _ilu.module_from_spec(_cspec)
_cspec.loader.exec_module(corpus)

UA = "public-agents-lintel-twice"
INIT = {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "initialize",
    "params": {
        "protocolVersion": "2025-06-18",
        "capabilities": {},
        "clientInfo": {"name": "lintel-twice", "version": "1"},
    },
}

# A static document that cannot change between two requests seconds apart.
NEGATIVE_CONTROL = "https://public-agents.com/schemas/tool.schema.json"

# Header and JSON-leaf names that move BY DESIGN. HTTP says so for some of
# them; the rest are counters and request ids. Everything here is still
# printed, never hidden: the point of the list is only that it does not
# promote a row to a finding.
VOLATILE = (
    "date", "age", "expires", "set-cookie", "cf-ray", "x-request-id",
    "request-id", "x-amz-cf-id", "x-amz-request-id", "x-vercel-id",
    "x-render-origin-server", "x-runtime", "server-timing", "report-to",
    "nel", "x-notion-request-id", "x-sentry-id", "x-envoy-upstream-service-time",
    "ratelimit-remaining", "x-ratelimit-remaining", "x-rate-limit-remaining",
    "retry-after", "cf-cache-status", "x-timer", "x-served-by", "x-cache-hits",
    # An MCP session id is a nonce by definition: a new one per session is the
    # protocol working. Listed so it prints, not so it hides.
    "mcp-session-id",
    "traceparent", "tracestate", "x-trace-id", "x-correlation-id",
    "x-datadog-trace-id", "x-b3-traceid", "x-b3-spanid", "x-envoy-decorator-operation",
)


def is_volatile(name):
    """Is this field one HTTP or common practice says moves by itself?"""
    leaf = name.rsplit(".", 1)[-1].lower()
    return leaf in VOLATILE


def group_of(name):
    """The comparable group a field belongs to, by LEAF NAME.

    Two fields are comparable when a story about my own network would have to
    move BOTH or NEITHER. Leaf name is one right grain: the four `payTo` values
    in one x402 challenge are one group, and the fact that two moved while two
    held is what made that finding survive. An array index is dropped so that
    accepts[0].payTo and accepts[2].payTo land in the same group.
    """
    parts = [p for p in name.split(".") if not re.fullmatch(r"\[\d+\]", p)]
    return parts[-1].lower() if parts else name.lower()


def parent_of(name):
    """The comparable group a field belongs to, by ITS CONTAINING OBJECT.

    The second grain, and it was earned rather than designed (2026-09-23).
    Bitrefill's `/x402/my/orders` answers 402 with an EMPTY `accepts` array and
    a `sign-in-with-x` challenge: `nonce`, `issuedAt` and `expirationTime` all
    move between two readings, and no other field shares their leaf names, so
    the leaf grain could only say ALL-MOVED -- the verdict that means "this is
    also what two edge nodes look like". But `domain`, `uri`, `version` and
    `statement` sit in the SAME OBJECT and hold perfectly still. No story about
    edge nodes survives half an object moving, so the control is there; the
    leaf grain simply could not see it.

    HEADERS ARE EXCLUDED FROM THIS GRAIN, deliberately. Every response header
    shares one notional parent, so applying it there would promote any moving
    header to a finding as soon as any other header held -- and a per-request
    nonce inside a Content-Security-Policy is ordinary, not a finding. The
    object grain only means something where the document itself grouped the
    fields.
    """
    if not name.startswith("body."):
        return None
    parts = [p for p in name.split(".") if not re.fullmatch(r"\[\d+\]", p)]
    # parts[0] is the literal "body"; a field at the document root has no
    # containing object worth calling a group.
    return ".".join(parts[1:-1]).lower() if len(parts) > 2 else None


# --------------------------------------------------------------------------
# reading a surface

def fetch(url, payload=None, timeout=25):
    """One reading. Returns (status, headers dict, body text) or (None, {}, err)."""
    headers = {"User-Agent": UA}
    data = None
    if payload is not None:
        data = json.dumps(payload).encode()
        headers["Content-Type"] = "application/json"
        headers["Accept"] = "application/json, text/event-stream"
    req = urllib.request.Request(url, data=data,
                                method="POST" if data else "GET", headers=headers)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return r.status, {k.lower(): v for k, v in r.headers.items()}, \
                r.read(262144).decode("utf-8", "replace")
    except urllib.error.HTTPError as e:
        return e.code, {k.lower(): v for k, v in e.headers.items()}, \
            e.read(65536).decode("utf-8", "replace")
    except Exception as e:
        return None, {}, "%s: %s" % (type(e).__name__, e)


def json_of(body):
    """The JSON a body carries, plain or inside text/event-stream frames."""
    body = body.strip()
    if body.startswith("{") or body.startswith("["):
        try:
            return json.loads(body)
        except ValueError:
            pass
    frames = []
    for line in body.splitlines():
        line = line.strip()
        if line.startswith("data:"):
            line = line[5:].strip()
        if line.startswith("{"):
            try:
                frames.append(json.loads(line))
            except ValueError:
                continue
    if len(frames) == 1:
        return frames[0]
    return frames or None


def flatten(obj, prefix=""):
    """Every leaf of a JSON document as path -> string."""
    out = {}
    if isinstance(obj, dict):
        for k, v in obj.items():
            out.update(flatten(v, "%s.%s" % (prefix, k) if prefix else str(k)))
    elif isinstance(obj, list):
        for i, v in enumerate(obj):
            out.update(flatten(v, "%s.[%d]" % (prefix, i) if prefix else "[%d]" % i))
    else:
        out[prefix or "(root)"] = json.dumps(obj)
    return out


def fields(status, headers, body):
    """One reading reduced to a flat field map, ready to diff."""
    f = {"http.status": str(status)}
    # Lowercased here as well as in fetch(): HTTP header names are
    # case-insensitive, and a diff that is case-sensitive would call a server
    # that changed `Date` to `date` a finding.
    for k, v in headers.items():
        f["header." + k.lower()] = v
    doc = json_of(body)
    if doc is not None:
        for path, value in flatten(doc).items():
            f["body." + path] = value
    else:
        f["body.(bytes)"] = str(len(body))
        f["body.(sha256-16)"] = hashlib.sha256(body.encode("utf-8", "replace")
                                               ).hexdigest()[:16]
    return f


# --------------------------------------------------------------------------
# the diff, and the grading that is the whole point

def compare(a, b):
    """Diff two field maps and grade the pair.

    Returns a dict with the verdict, the moved and held fields, and -- for a
    ROTATING row -- the groups where some members moved and others did not,
    which is the evidence the verdict rests on.
    """
    if a is None or b is None:
        return {"verdict": "NO-ANSWER", "moved": [], "held": [], "witness": []}

    names = sorted(set(a) | set(b))
    moved = [n for n in names if a.get(n) != b.get(n)]
    held = [n for n in names if n in a and n in b and a[n] == b[n]]

    if not moved:
        return {"verdict": "STABLE", "moved": [], "held": held, "witness": []}

    substantive = [n for n in moved if not is_volatile(n)]
    if not substantive:
        return {"verdict": "VOLATILE-ONLY", "moved": moved, "held": held,
                "witness": []}

    # The control inside the response: for each group that moved, did a
    # comparable field in the SAME response hold still? Two grains of
    # "comparable" (see group_of and parent_of); the leaf grain is tried first
    # because it is the tighter claim.
    #
    # THE TIGHTEST AVAILABLE COMPARISON WINS, and this rule is the whole
    # instrument. A field that HAS leaf siblings is judged only at the leaf
    # grain: four payTo values all moving together is exactly what a client
    # reaching two edge nodes looks like, and letting the object grain find a
    # still `amount` beside each of them would turn that non-finding into a
    # finding. A field with NO leaf sibling anywhere cannot be judged that way
    # at all, and only then does the object it sits in get a say.
    all_names = set(a) | set(b)
    leaf_population = {}
    for n in all_names:
        if not is_volatile(n):
            leaf_population.setdefault(group_of(n), []).append(n)

    by_leaf = [n for n in substantive if len(leaf_population.get(group_of(n), [])) > 1]
    by_object = [n for n in substantive if n not in by_leaf]

    witness = []
    for names, grain, key in ((by_leaf, "leaf", group_of),
                              (by_object, "object", parent_of)):
        moved_groups = {}
        for n in names:
            g = key(n)
            if g is not None:
                moved_groups.setdefault(g, []).append(n)
        for g, ns in sorted(moved_groups.items()):
            siblings = [n for n in held if key(n) == g and not is_volatile(n)]
            if siblings:
                witness.append({"grain": grain, "group": g, "moved": sorted(ns),
                                "held": sorted(siblings)})
    if witness:
        return {"verdict": "ROTATING", "moved": moved, "held": held,
                "witness": witness}
    return {"verdict": "ALL-MOVED", "moved": moved, "held": held, "witness": []}


# --------------------------------------------------------------------------
# selftest: every rule falsified, no network, no clone

def selftest():
    ok = fail = 0

    def check(name, cond):
        nonlocal ok, fail
        if cond:
            ok += 1
            print("ok: %s" % name)
        else:
            fail += 1
            print("FAIL: %s" % name)

    # --- the grain of a group
    check("group drops array indices",
          group_of("body.accepts.[0].payTo") == group_of("body.accepts.[2].payTo"))
    check("group separates different leaves",
          group_of("body.accepts.[0].payTo") != group_of("body.accepts.[0].amount"))
    check("volatile matches on the leaf, not the path",
          is_volatile("header.date") and is_volatile("date"))
    check("a header that is not volatile is not treated as one",
          not is_volatile("header.www-authenticate"))
    check("a body field named like a header is still judged by its leaf",
          is_volatile("body.meta.request-id"))

    # --- identical readings
    a = {"http.status": "200", "header.date": "x", "body.v": "1"}
    check("identical readings are STABLE", compare(a, dict(a))["verdict"] == "STABLE")
    check("a missing reading is NO-ANSWER", compare(a, None)["verdict"] == "NO-ANSWER")
    check("both missing is NO-ANSWER", compare(None, None)["verdict"] == "NO-ANSWER")

    # --- volatile only
    b = dict(a, **{"header.date": "y"})
    r = compare(a, b)
    check("only Date moving is VOLATILE-ONLY", r["verdict"] == "VOLATILE-ONLY")
    check("a VOLATILE-ONLY row still lists what moved", r["moved"] == ["header.date"])
    check("a VOLATILE-ONLY row is not a finding", r["witness"] == [])

    # --- THE ALLIUM FIXTURE, which is why this file exists.
    # Four rails, same amount; two destination addresses move, two hold.
    def challenge(p0, p1, p2, p3):
        return {
            "http.status": "402",
            "header.date": "whenever",
            "body.accepts.[0].payTo": p0, "body.accepts.[0].amount": "30000",
            "body.accepts.[1].payTo": p1, "body.accepts.[1].amount": "30000",
            "body.accepts.[2].payTo": p2, "body.accepts.[2].amount": "30000",
            "body.accepts.[3].payTo": p3, "body.accepts.[3].amount": "30000",
        }
    one = challenge("0xdf3f", "8dfAeZ", "0x61Fd", "0xF5B5")
    two = challenge("0xa947", "H6c1U1", "0x61Fd", "0xF5B5")
    r = compare(one, two)
    check("two of four payTo moving is ROTATING", r["verdict"] == "ROTATING")
    check("the ROTATING witness names the group",
          len(r["witness"]) == 1 and r["witness"][0]["group"] == "payto")
    check("the witness names the two that moved",
          len(r["witness"][0]["moved"]) == 2)
    check("the witness names the two that HELD, which is the control",
          len(r["witness"][0]["held"]) == 2)
    check("amount is reported as held, not as a finding",
          "body.accepts.[0].amount" in r["held"])
    check("the Allium witness is at the leaf grain",
          r["witness"][0]["grain"] == "leaf")

    # The same shape with ALL FOUR moving must NOT be a finding: that is what a
    # client reaching two different edges looks like.
    three = challenge("0xa947", "H6c1U1", "0x0000", "0x1111")
    r = compare(one, three)
    check("four of four payTo moving is ALL-MOVED, not ROTATING",
          r["verdict"] == "ALL-MOVED")
    check("an ALL-MOVED row carries no witness", r["witness"] == [])
    check("an ALL-MOVED row still lists every field that moved",
          len([n for n in r["moved"] if n.endswith("payTo")]) == 4)

    # --- THE BITREFILL FIXTURE, which bought the second grain (2026-09-23).
    # A 402 with an empty accepts array and a sign-in-with-x challenge: three
    # fields of one object move, four fields of the SAME object hold.
    def siwx(nonce, issued, expires):
        return {
            "http.status": "402",
            "body.accepts": "[]",
            "body.extensions.sign-in-with-x.info.nonce": nonce,
            "body.extensions.sign-in-with-x.info.issuedAt": issued,
            "body.extensions.sign-in-with-x.info.expirationTime": expires,
            "body.extensions.sign-in-with-x.info.domain": "api.bitrefill.com",
            "body.extensions.sign-in-with-x.info.uri": "https://api.bitrefill.com/x402/my/orders",
            "body.extensions.sign-in-with-x.info.version": "1",
            "body.extensions.sign-in-with-x.info.statement": "Sign in to Bitrefill",
        }
    r = compare(siwx("aaa", "t1", "t1+5m"), siwx("bbb", "t2", "t2+5m"))
    check("a nonce moving inside an otherwise still object is ROTATING",
          r["verdict"] == "ROTATING")
    check("the bitrefill witness is at the OBJECT grain, which the leaf grain "
          "could not see", r["witness"][0]["grain"] == "object")
    check("the object witness names all three moving fields",
          len(r["witness"][0]["moved"]) == 3)
    check("the object witness names the four that held",
          len(r["witness"][0]["held"]) == 4)

    # The object grain must NOT reach headers: one moving header beside any
    # still header is ordinary (a CSP nonce), not a finding.
    r = compare({"http.status": "200", "header.content-security-policy": "nonce-a",
                 "header.content-type": "text/html"},
                {"http.status": "200", "header.content-security-policy": "nonce-b",
                 "header.content-type": "text/html"})
    check("a moving header beside a still header is NOT promoted to ROTATING",
          r["verdict"] == "ALL-MOVED")
    check("parent_of refuses headers", parent_of("header.x-y") is None)
    check("parent_of refuses a root-level body field",
          parent_of("body.error") is None)
    check("parent_of drops array indices",
          parent_of("body.accepts.[0].payTo") == "accepts")

    # A by-design volatile sibling cannot serve as the control that held.
    r = compare({"http.status": "200", "body.a.token": "x", "body.a.date": "d1"},
                {"http.status": "200", "body.a.token": "y", "body.a.date": "d1"})
    check("a volatile field that happened to hold is not used as a control",
          r["verdict"] == "ALL-MOVED")

    # A single lonely field with no sibling cannot be graded ROTATING.
    r = compare({"http.status": "200", "body.token": "a"},
                {"http.status": "200", "body.token": "b"})
    check("a field with no comparable sibling is ALL-MOVED, not ROTATING",
          r["verdict"] == "ALL-MOVED")

    # A status change is substantive and has no sibling group.
    r = compare({"http.status": "200", "body.x": "1"},
                {"http.status": "401", "body.x": "1"})
    check("a changed status is not STABLE", r["verdict"] != "STABLE")
    check("a changed status is reported as moved", "http.status" in r["moved"])

    # A field appearing in only one reading counts as moved (Linear's 401 lost
    # its error parameters between two dates; an absent field is a change).
    r = compare({"http.status": "401", "header.www-authenticate": "Bearer error=x",
                 "header.content-length": "79"},
                {"http.status": "401", "header.www-authenticate": "Bearer scope=y",
                 "header.content-length": "0"})
    check("a field present in one reading only is moved",
          "header.www-authenticate" in r["moved"])

    # --- flatten and json_of
    check("flatten reaches a nested leaf",
          flatten({"a": {"b": [1, 2]}})["a.b.[1]"] == "2")
    check("flatten distinguishes a string from a number",
          flatten({"a": "1"})["a"] != flatten({"a": 1})["a"])
    check("json_of reads a plain JSON body",
          json_of('{"a":1}') == {"a": 1})
    check("json_of reads a single SSE frame",
          json_of('event: message\ndata: {"a":1}\n\n') == {"a": 1})
    check("json_of returns None for HTML",
          json_of("<html><body>hi</body></html>") is None)
    check("json_of returns None for an empty body", json_of("") is None)

    # --- fields()
    f = fields(402, {"Date": "x", "Content-Type": "application/json"}, '{"a":{"b":2}}')
    check("fields lowercases header names", "header.date" in f)
    check("fields records the status", f["http.status"] == "402")
    check("fields flattens the body", f["body.a.b"] == "2")
    f2 = fields(200, {}, "<html>not json</html>")
    check("a non-JSON body is reduced to length and a hash",
          "body.(sha256-16)" in f2 and "body.(bytes)" in f2)
    check("two different non-JSON bodies hash differently",
          fields(200, {}, "<a>")["body.(sha256-16)"]
          != fields(200, {}, "<b>")["body.(sha256-16)"])
    check("the same non-JSON body hashes the same",
          fields(200, {}, "<a>")["body.(sha256-16)"]
          == fields(200, {}, "<a>")["body.(sha256-16)"])

    # --- the two live controls, graded on synthetic inputs
    check("the positive control fails when no clock moved",
          not positive_control_ok([compare(a, dict(a))]))
    check("the positive control passes when a Date moved",
          positive_control_ok([compare(a, b)]))
    check("the negative control fails when a static file differs",
          not negative_control_ok(compare({"body.x": "1"}, {"body.x": "2"})))
    check("the negative control passes when a static file is identical",
          negative_control_ok(compare({"body.x": "1"}, {"body.x": "1"})))
    check("the negative control fails on a missing reading",
          not negative_control_ok(compare(None, None)))
    check("the negative control tolerates Date moving on the static file",
          negative_control_ok(compare({"body.x": "1", "header.date": "a"},
                                      {"body.x": "1", "header.date": "b"})))

    print("\n%d ok, %d failed" % (ok, fail))
    return 1 if fail else 0


def positive_control_ok(results):
    """Did any surface's clock move between the two readings?

    If not, the two readings are one response handed to me twice, and every
    STABLE below means nothing.
    """
    for r in results:
        if any(is_volatile(n) for n in r.get("moved", [])):
            return True
    return False


def negative_control_ok(result):
    """Did a document that cannot change come back identical?"""
    return result["verdict"] in ("STABLE", "VOLATILE-ONLY")


# --------------------------------------------------------------------------
# the live run

def read_twice(url, payload, gap):
    a = fetch(url, payload)
    time.sleep(gap)
    b = fetch(url, payload)
    fa = None if a[0] is None else fields(*a)
    fb = None if b[0] is None else fields(*b)
    r = compare(fa, fb)
    r["url"] = url
    r["status"] = [a[0], b[0]]
    if fa is None:
        r["detail"] = a[2][:160]
    elif fb is None:
        r["detail"] = b[2][:160]
    return r


def mcp_surfaces(clone):
    urls = {}
    for path in sorted(glob.glob(os.path.join(clone, "registry/tools/*/tool.json"))
                       + glob.glob(os.path.join(clone, "registry/agents/*/agent.json"))):
        try:
            d = json.load(open(path))
        except ValueError:
            continue
        mcp = (d.get("surfaces") or {}).get("mcp")
        if mcp:
            urls.setdefault(mcp, []).append(d.get("slug") or d.get("handle") or path)
    return urls


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--clone", default="/tmp/lintel-queue-clone")
    ap.add_argument("--url", action="append", default=[],
                    help="ad-hoc URL, read twice with a GET; repeatable")
    ap.add_argument("--gap", type=float, default=4.0,
                    help="seconds between the two readings (default 4)")
    ap.add_argument("--selftest", action="store_true")
    ap.add_argument("--json")
    args = ap.parse_args()

    if args.selftest:
        return selftest()

    targets = []           # (url, payload, label)
    if args.url:
        targets = [(u, None, "--url") for u in args.url]
        head = "(no clone: ad-hoc URLs)"
    else:
        head = corpus.require_clone(args.clone, "twice.py")
        try:
            corpus.require_ref(args.clone, "origin/main", "twice.py")
        except SystemExit as e:
            print(str(e))
            return 2
        surfaces = mcp_surfaces(args.clone)
        if not surfaces:
            print("REFUSED: no surfaces.mcp in the corpus; twice.py would score nothing.")
            return 2
        targets = [(u, INIT, ", ".join(who)) for u, who in sorted(surfaces.items())]

    started = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    print("twice.py, registry at %s, started %s" % (head, started))
    print("%d surface(s), two readings each %.0fs apart, no credentials\n"
          % (len(targets), args.gap))

    # The negative control runs FIRST and on its own: a static document.
    neg = read_twice(NEGATIVE_CONTROL, None, args.gap)
    if not negative_control_ok(neg):
        print("REFUSED: the negative control moved. %s came back DIFFERENT on two\n"
              "         readings (%s). A document that cannot change did; so the\n"
              "         diff or the network is noisy and every ROTATING verdict\n"
              "         this run could print would be manufactured.\n"
              "         fields that moved: %s"
              % (NEGATIVE_CONTROL, neg["verdict"], ", ".join(neg["moved"][:8])))
        return 2
    print("negative control: %s is identical on two readings (%s). Good.\n"
          % (NEGATIVE_CONTROL.rsplit("/", 1)[-1], neg["verdict"]))

    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=6) as pool:
        futures = {pool.submit(read_twice, u, p, args.gap): (u, lbl)
                   for u, p, lbl in targets}
        for fut in concurrent.futures.as_completed(futures):
            u, lbl = futures[fut]
            r = fut.result()
            r["who"] = lbl
            results.append(r)
    results.sort(key=lambda r: r["url"])

    if not positive_control_ok(results + [neg]):
        print("REFUSED: the positive control did not fire. Not one surface's\n"
              "         Date (or any by-design volatile field) differed between\n"
              "         two readings. Two readings that agree in every field\n"
              "         including the clock are one response served twice, so\n"
              "         every STABLE verdict below would be manufactured.")
        return 2

    order = {"ROTATING": 0, "ALL-MOVED": 1, "NO-ANSWER": 2, "VOLATILE-ONLY": 3,
             "STABLE": 4}
    counts = {}
    for r in sorted(results, key=lambda r: (order.get(r["verdict"], 9), r["url"])):
        counts[r["verdict"]] = counts.get(r["verdict"], 0) + 1
        print("%-13s %s  [%s]" % (r["verdict"], r["url"], r["who"]))
        if r["verdict"] == "NO-ANSWER":
            print("              statuses %s  %s" % (r["status"], r.get("detail", "")))
        elif r["verdict"] == "ROTATING":
            for w in r["witness"]:
                print("              %s group %s: %d moved, %d HELD (the control "
                      "is inside the same response)"
                      % (w["grain"], w["group"], len(w["moved"]), len(w["held"])))
                for n in w["moved"][:6]:
                    print("                moved: %s" % n)
                for n in w["held"][:6]:
                    print("                held : %s" % n)
        elif r["verdict"] == "ALL-MOVED":
            print("              every comparable field moved together, which is "
                  "also what two edge nodes look like:")
            for n in [x for x in r["moved"] if not is_volatile(x)][:8]:
                print("                moved: %s" % n)
        elif r["verdict"] == "VOLATILE-ONLY":
            print("              by-design only: %s"
                  % ", ".join(n.split(".", 1)[-1] for n in r["moved"][:8]))

    print("\n%d surface(s) read twice: %s" % (
        len(results),
        ", ".join("%d %s" % (v, k) for k, v in sorted(counts.items(),
                                                      key=lambda kv: order.get(kv[0], 9)))))
    rot = [r for r in results if r["verdict"] == "ROTATING"]
    print("%d finding(s): a field that moved while a comparable field held still."
          % len(rot))
    print("positive control: a by-design volatile field moved somewhere, so the "
          "two readings are two responses.")

    if args.json:
        with open(args.json, "w") as fh:
            json.dump({"head": head, "started": started, "results": results,
                       "negativeControl": neg}, fh, indent=1)
        print("wrote %s" % args.json)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())