"""WINDOW 6 at cba18cd5: the BOUND-PORT cell, its two red-first arms, and the
TMP A/B on the two pre-existing reds, then the unfiltered lib for Q4.

The mutation arms are why this is a driver rather than a label list handed to
window3.py: R-GATE and R-DROP need a source edit BETWEEN producer runs, and a
window that edits product source has to revert it and PROVE the revert by oid
before the next arm, or every later arm is measuring an unknown tree.

Reported against predictions-cba18cd5.txt (Q1-Q6), filed before any execution.
"""
import io, json, os, subprocess, sys, time
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import lane_producers as P
import lane_run as L

TAG = "cba18cd5"
REPORT = os.path.join(L.EVID, "window6-%s.json" % TAG)
SRC = os.path.join(P.TREE, "crates", "spt-daemon", "src", "docshost.rs")
REL = "crates/spt-daemon/src/docshost.rs"
CAP_S = 900

CELL = ("docshost::tests::the_bound_docs_port_publishes_only_for_"
        "a_broker_backed_listener_and_retires_on_drop")
PRE = [
    "registryhost::tests::recent_projects_for_dedups_newest_first_excludes_spt_internal",
    "projwriter::tests::batched_complexity_counters_hold",
]

# START rides the launch, same guard as window3: the sequencer's only view of
# the box is the START message, and a short window is exactly the one that gets
# announced late and never noticed.
START_FILE = os.environ.get("HERTZ_START_FILE")
PEERS = ("doyle", "todlando")


def send_start():
    if not START_FILE or not os.path.isfile(START_FILE):
        raise SystemExit(
            "REFUSING TO RUN: set HERTZ_START_FILE to the START body.")
    body = io.open(START_FILE, "rb").read()
    out = {}
    for peer in PEERS:
        proc = subprocess.run(["spt", "send", peer], input=body,
                              capture_output=True)
        out[peer] = (proc.stdout or proc.stderr).decode("utf-8", "replace").strip()
    for peer, line in out.items():
        if not line.startswith(("SENT", "QUEUED")):
            raise SystemExit("START not delivered to %s: %s" % (peer, line))
    return out


def git(*args):
    return subprocess.run(["git"] + list(args), cwd=P.TREE,
                          capture_output=True, text=True)


def head_blob():
    return git("rev-parse", "HEAD:" + REL).stdout.strip()


def work_blob():
    return git("hash-object", SRC).stdout.strip()


# --- the two product mutations, exact anchors copied from the file ---------
GATE_OLD = (
    "                        let _published = broker.as_ref().map(|_| {\n"
    "                            BOUND_DOCS_PORT.store(bound, Ordering::Release);\n"
    "                            BoundDocsPort\n"
    "                        });\n"
)
GATE_NEW = (
    "                        BOUND_DOCS_PORT.store(bound, Ordering::Release);\n"
    "                        let _published = broker.as_ref().map(|_| {\n"
    "                            BoundDocsPort\n"
    "                        });\n"
)
DROP_OLD = (
    "    fn drop(&mut self) {\n"
    "        BOUND_DOCS_PORT.store(0, Ordering::Release);\n"
    "    }\n"
)
DROP_NEW = (
    "    fn drop(&mut self) {\n"
    "        // R-DROP mutation: retirement removed on purpose.\n"
    "    }\n"
)


def mutate(old, new):
    """Apply one anchored edit, with the anchor translated to the file's ACTUAL
    line endings.

    Window 6 lost two arms to this: the anchors are written LF, the file was LF
    when the driver started, and `git checkout --` in revert() re-materialized
    it as CRLF (git applies its eol conversion on checkout). The DROP anchor
    then counted 0 and the driver refused mid-window. Worse, the revert's own
    verification could not see it: `git hash-object` and `git diff --quiet`
    both NORMALIZE line endings, so they reported an identical blob and a clean
    tree over bytes that had changed. So the anchor is now built from the file
    in hand rather than assumed, and revert() records the byte shape too.
    """
    src = io.open(SRC, encoding="utf-8", newline="").read()
    eol = "\r\n" if "\r\n" in src else "\n"
    old = old.replace("\n", eol)
    new = new.replace("\n", eol)
    n = src.count(old)
    if n != 1:
        raise SystemExit("REFUSING: mutation anchor count %d (eol=%r), expected 1"
                         % (n, eol))
    io.open(SRC, "w", encoding="utf-8", newline="").write(src.replace(old, new))


def revert():
    """Restore the COMMITTED blob and prove it by oid, not by eye.

    `git checkout --` is safe here only because the sole edit in this tree is
    the mutation this driver just made; the cell itself is committed at cba18cd5.
    """
    git("checkout", "--", REL)
    raw = io.open(SRC, "rb").read()
    # The oid and the diff both NORMALIZE line endings, so they are blind to a
    # checkout that rewrote LF as CRLF - which is exactly what cost window 6
    # its last two arms. Record the byte shape beside them so "clean" is a
    # statement about bytes as well as content.
    return {"work_blob": work_blob(), "head_blob": head_blob(),
            "clean": git("diff", "--quiet", "--", REL).returncode == 0,
            "bytes": len(raw), "crlf": raw.count(b"\r\n"),
            "bare_lf": raw.count(b"\n") - raw.count(b"\r\n")}


def nextest(label, names, deadline, env_extra=None):
    argv = ["cargo", "nextest", "run", "-p", "spt-daemon", "--lib",
            "--locked", "--build-jobs", "2", "--profile", "ci-windows",
            "--no-fail-fast", "--success-output", "immediate"]
    if names:
        argv += ["-E", " | ".join("test(=%s)" % n for n in names)]
    return L.run(label, argv, P.TREE, deadline, env_extra=env_extra)


def row(obs):
    return {"exit": obs["producer_exit"], "stop_reason": obs["stop_reason"],
            "elapsed": round(obs["elapsed_seconds"], 1),
            "survivors": obs["survivors"],
            "pool_growth_bytes": obs["pool_growth_bytes"]}


def main():
    state = {"tip": TAG, "start_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
             "runs": {}, "reverts": {}, "cap_s": CAP_S}
    state["start_dispatch"] = send_start()
    state["pre_blob"] = {"work": work_blob(), "head": head_blob()}
    emit(state)

    t0 = time.monotonic()

    def budget_left():
        return CAP_S - (time.monotonic() - t0)

    def step(label, fn, need_s):
        """Run one arm, or leave a LABELLED HOLE rather than overrun the cap."""
        if budget_left() < need_s:
            state["runs"][label] = {"cancelled": "cap: %.0f s left, %d s needed"
                                    % (budget_left(), need_s)}
            emit(state)
            return None
        obs = fn()
        state["runs"][label] = row(obs)
        emit(state)
        return obs

    # Q1 — the cell, unmutated.
    step("w6-cell", lambda: nextest("w6-cell", [CELL], 240), 60)

    # Q2 — R-GATE: make a brokerless listener publish. Predicted RED at the gate.
    if budget_left() > 120:
        mutate(GATE_OLD, GATE_NEW)
        state["mutations"] = {"r_gate_applied_blob": work_blob()}
        emit(state)
        step("w6-rgate", lambda: nextest("w6-rgate", [CELL], 240), 60)
        state["reverts"]["after_rgate"] = revert()
        emit(state)

    # Q3 — the TMP A/B on the two pre-existing cells. Ahead of R-DROP on
    # purpose: these confirm a mechanism and a withdrawn warning, so if the cap
    # squeezes anything it should be the second red-first arm, not these.
    base_a = L.priv_for("w6-armA")
    step("w6-armA", lambda: nextest("w6-armA", PRE, 240), 60)

    # ARM B's switch must live in the PARENT env: lane_run.priv_for reads
    # os.environ to pick the TMP base, and window 6 drove it through env_extra,
    # which only reaches the CHILD. Arm B therefore ran arm A's conditions and
    # its PASS read as a refutation of the TMP mechanism when it was a
    # non-measurement. The driver now sets it here AND records the base each arm
    # resolved, refusing to call arm B a measurement when the two bases match.
    os.environ["HERTZ_RIG_TMP_IN_REPO"] = "1"
    try:
        base_b = L.priv_for("w6-armB")
        state["tmp_bases"] = {"armA": base_a, "armB": base_b,
                              "differ": base_a != base_b}
        emit(state)
        if base_a == base_b:
            state["runs"]["w6-armB"] = {
                "invalid": "arm B resolved arm A's TMP base; the switch is not wired"}
            emit(state)
        else:
            step("w6-armB", lambda: nextest("w6-armB", PRE, 240), 60)
    finally:
        os.environ.pop("HERTZ_RIG_TMP_IN_REPO", None)

    # Q2 — R-DROP: remove the retirement. Predicted RED at the final assert.
    if budget_left() > 120:
        mutate(DROP_OLD, DROP_NEW)
        state.setdefault("mutations", {})["r_drop_applied_blob"] = work_blob()
        emit(state)
        step("w6-rdrop", lambda: nextest("w6-rdrop", [CELL], 240), 60)
        state["reverts"]["after_rdrop"] = revert()
        emit(state)

    # Q4 — the whole lib, unfiltered.
    step("w6-lib-full", lambda: nextest("w6-lib-full", None, 420), 90)

    state["end_utc"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    state["post_census"] = L.builder_census()["by_owner"]
    state["final_blob"] = {"work": work_blob(), "head": head_blob()}
    state["tree_clean"] = git("status", "--porcelain").stdout.strip()
    state["elapsed_total_s"] = round(time.monotonic() - t0, 1)
    emit(state)
    print(json.dumps(state, indent=2))


def emit(state):
    json.dump(state, open(REPORT, "w"), indent=2)


if __name__ == "__main__":
    main()