"""Queued #304 producers. Each is one attempt, no retry; run explicitly by label."""
import json
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import lane_run as L  # noqa: E402

TREE = r"C:\Users\decid\Documents\projects\spt-core\.worktrees\hertz-304"

RC = [
    "rc::tests::viewer_byte_detach_keeps_controller_and_child",
    "rc::tests::viewer_event_detach_keeps_controller_and_child",
    "rc::tests::detach_keybind_semantics",
    "rc::tests::detach_prefix_spans_chunks",
    "rc::tests::key_event_step_detach_sm",
]

WEBSERVE = [
    "webserve::tests::a_scoped_entry_is_still_served_on_loopback",
    "webserve::tests::the_adapter_docs_segment_serves_docs_dir_and_never_falls_through",
    "webserve::tests::the_adapter_facet_root_still_serves_the_core_owned_web_root",
    "webserve::tests::node_routing_reserves_facets_without_registry_fallback",
    "webserve::tests::docs_aliases_preserve_existing_and_new_bundle_roots",
    "webserve::tests::references_observe_edits_deletion_and_registry_removal",
    "webserve::tests::directory_references_reject_traversal_and_track_current_children",
    "webserve::tests::adapter_facet_and_disambiguated_alias_share_one_contained_reference",
    "webserve::tests::node_roots_redirect_to_docs_without_exposing_registry_entries",
    "webserve::tests::corrupt_registry_is_loud_but_does_not_disable_unrelated_facets",
    "webserve::tests::range_grammar_partial_suffix_clamp_and_unsatisfiable",
    "webserve::tests::materialize_reads_a_file_plan_through_the_range",
    "webserve::tests::peer_paths_proxy_by_facet_and_unknown_labels_stay_docs_404",
    "webserve::tests::served_subject_is_the_entrys_origin_or_none",
    # Restored at 9f09db4e: enumerated by the list but NOT selected by the
    # 20-term filter, so it would have compiled and never run — a new case that
    # the admitted subset does not name is inventory, not coverage.
    "webserve::tests::hostile_text_is_html_escaped_before_it_reaches_a_rendered_page",
    "shellchan::tests::vocab_check_bounds_ops_and_args",
]

# Authored 2026-09-11, hermetic (no process spawn). Admitted only on doyle's word.
FIREWALL_297 = [
    "bootstrap_firewall::tests::the_opt_out_disables_mutation_for_every_value_including_empty_and_zero",
    "bootstrap_firewall::tests::the_opt_out_refuses_before_the_argument_guard_and_before_any_host_command",
    "bootstrap_firewall::tests::verify_refuses_a_relative_binder_and_port_zero_under_either_opt_out_state",
    "bootstrap_firewall::tests::the_residual_cleanup_command_is_a_self_verifying_removal_that_creates_nothing",
    "bootstrap_firewall::windows::tests::binder_text_accepts_only_an_absolute_nul_free_path",
    "bootstrap_firewall::windows::tests::encoded_scripts_round_trip_as_utf16le_and_carry_the_ownership_identifiers",
]

DOCS_E2E = ["serves_published_surface_and_refuses_escapes_and_writes"]

# Producers 3-5, authored 2026-09-11 at tip 61bfd85c. Predicted in
# predictions-61bfd85c.txt BEFORE this file named them.
WEBSERVE_E2E = [
    "production_listener_routes_nodes_docs_and_live_registry_references",
    "subnet_node_labels_shadow_only_the_docs_compat_alias_while_members",
    "production_listener_exposes_only_the_adapter_web_root_through_both_urls",
    "production_listener_redirects_directory_roots_before_serving_relative_links",
    "the_adapter_docs_facet_serves_over_the_production_listener",
]
ATTACH_PUSH = ["attachment_frames_reach_a_linked_shell_through_the_real_daemon"]
CROSS_NODE = ["a_peers_url_is_served_by_its_owner_through_the_local_listener"]


def expr(names):
    return " | ".join("test(=%s)" % n for n in names)


def nextest_all(label, scope, deadline, profile="ci-windows"):
    """Enumerate a target's WHOLE population - no -E, nothing to be narrower than."""
    argv = ["cargo", "nextest", "list"] + scope + [
        "--locked", "--build-jobs", "2", "--profile", profile,
        "--message-format", "json",
    ]
    return L.run(label, argv, TREE, deadline)


def nextest(verb, label, scope, names, deadline, profile="ci-windows"):
    argv = ["cargo", "nextest", verb] + scope + [
        "--locked", "--build-jobs", "2", "--profile", profile,
        "-E", expr(names),
    ]
    if verb == "list":
        argv += ["--message-format", "json"]
    else:
        # One attempt, no retry - but a red in case 1 must not CANCEL cases 2..n
        # and read back as coverage. Fail-fast hid 1 of 5 at 9f09db4e.
        argv += ["--no-fail-fast"]
    return L.run(label, argv, TREE, deadline)


def matched(stdout_path):
    """Selected cases actually present in a nextest list, per suite."""
    data = json.load(open(stdout_path))
    out = {}
    for suite, body in data["rust-suites"].items():
        hit = sorted(n for n, t in body["testcases"].items()
                     if t["filter-match"]["status"] == "matches")
        out[suite] = {"binary": body.get("binary-path"), "matched": hit}
    return {"enumerated": data["test-count"], "suites": out}


PRODUCERS = {
    "spt-bins-list": lambda: nextest("list", "spt-bins-list", ["-p", "spt", "--bins"], RC, 2400),
    "docs-e2e-list": lambda: nextest(
        "list", "docs-e2e-list", ["-p", "spt-daemon", "--test", "docs_server_e2e"], DOCS_E2E, 1800),
    "spt-daemon-lib-297-list": lambda: nextest(
        "list", "spt-daemon-lib-297-list", ["-p", "spt-daemon", "--lib"],
        WEBSERVE + FIREWALL_297, 1800),
    # The admitted 21: doyle admitted the six hermetic #297 cases into the subset
    # (msg VUWZZ5LC), so the run expression matches the 21-term inventory exactly.
    "spt-daemon-lib-run": lambda: nextest(
        "run", "spt-daemon-lib-run", ["-p", "spt-daemon", "--lib"],
        WEBSERVE + FIREWALL_297, 480),
    "spt-bins-run": lambda: nextest("run", "spt-bins-run", ["-p", "spt", "--bins"], RC, 420),
    "docs-e2e-run": lambda: nextest(
        "run", "docs-e2e-run", ["-p", "spt-daemon", "--test", "docs_server_e2e"], DOCS_E2E, 300),

    # --- window at 61bfd85c -------------------------------------------------
    # Inventories enumerate the WHOLE target; the diff below is what discharges
    # them, in both directions.
    "spt-bins-inv": lambda: nextest_all("spt-bins-inv", ["-p", "spt", "--bins"], 2400),
    "webserve-e2e-inv": lambda: nextest_all(
        "webserve-e2e-inv", ["-p", "spt-daemon", "--test", "webserve_e2e"], 1800),
    # One build, both heavy binaries. Profile `default` because BOTH are in this
    # repo's <HEAVY> kind(test) set, which ci-windows deliberately does not carry.
    "spt-e2e-inv": lambda: nextest_all(
        "spt-e2e-inv",
        ["-p", "spt", "--test", "attach_link_push_e2e",
         "--test", "webserve_cross_node_e2e"], 2400, profile="default"),

    "webserve-e2e-run": lambda: nextest(
        "run", "webserve-e2e-run", ["-p", "spt-daemon", "--test", "webserve_e2e"],
        WEBSERVE_E2E, 600),
    # attach_link_push_e2e needs mock-session, and mock-adapter is a DIFFERENT
    # package: cargo guarantees a package's own bins to ITS integration tests
    # and nothing about a sibling's, so nothing built it and the case failed on
    # the missing fixture without ever reaching its subject. Pre-build it.
    "mock-session-fixture": lambda: L.run(
        "mock-session-fixture",
        ["cargo", "build", "--locked", "-p", "mock-adapter", "--bin", "mock-session"],
        TREE, 600),
    "attach-push-run": lambda: nextest(
        "run", "attach-push-run", ["-p", "spt", "--test", "attach_link_push_e2e"],
        ATTACH_PUSH, 600, profile="default"),
    "cross-node-run": lambda: nextest(
        "run", "cross-node-run", ["-p", "spt", "--test", "webserve_cross_node_e2e"],
        CROSS_NODE, 600, profile="default"),
}


def population(stdout_path, prefix=""):
    """Every case the target CONTAINS, per suite. No filter, so no clean zero."""
    data = json.load(open(stdout_path))
    out = {}
    for suite, body in data["rust-suites"].items():
        out[suite] = sorted(n for n in body["testcases"] if n.startswith(prefix))
    return out


def diff(pop, selected):
    """Both directions. Named-but-absent is a broken filter; present-but-unnamed
    is INVENTORY, not coverage."""
    flat = sorted({n for names in pop.values() for n in names})
    sel = sorted(set(selected))
    return {"population": len(flat), "selected": len(sel),
            "named_but_absent": [n for n in sel if n not in flat],
            "present_but_unnamed": [n for n in flat if n not in sel]}


if __name__ == "__main__":
    label = sys.argv[1]
    obs = PRODUCERS[label]()
    report = {k: obs[k] for k in
              ("label", "producer_exit", "stop_reason", "elapsed_seconds",
               "start_free", "min_free", "end_free", "growth_bytes", "survivors", "utc")}
    if label.endswith("-list") and obs["producer_exit"] == 0:
        report["inventory"] = matched(os.path.join(L.EVID, label + ".stdout"))
    print(json.dumps(report, indent=2))


# --- WINDOW 5 at 69438c01 (the merged tip) -------------------------------
# UNFILTERED on purpose. The stock `spt-daemon-lib-run` selects 21 exact
# names via test(=...), and doyle's END for this window asks two questions a
# filtered run CANNOT answer:
#   1. index_escapes_html_encodes_urls_and_reports_real_source_paths must be
#      shown ABSENT FROM THE TREE. It is not in the 21-term list, so under the
#      filter its absence is over-determined - not selected and not present
#      read identically, and "absent" would be a claim the run never made.
#   2. the six LEAKY cases from todlando's run (brainproc x4, broker, livehost)
#      are not in the list either, so whether they leak here is unaskable.
# The filter is also exact-match, so a cell this lane RENAMED drops out of
# selection silently - which is how index_escapes left the selected set in the
# first place. Same class as the 9f09db4e note above: a case the admitted
# subset does not name is inventory, not coverage.
# --success-output immediate is doyle's instruction and my own need: P3 is
# about the VALUE of a Location header, and nextest prints a passing case's
# output nowhere.
def nextest_full(label, scope, deadline, profile="ci-windows"):
    argv = ["cargo", "nextest", "run"] + scope + [
        "--locked", "--build-jobs", "2", "--profile", profile,
        "--no-fail-fast", "--success-output", "immediate",
    ]
    return L.run(label, argv, TREE, deadline)


PRODUCERS["w5-lib-full"] = lambda: nextest_full(
    "w5-lib-full", ["-p", "spt-daemon", "--lib"], 420)
# One #[test] in the whole file, so unfiltered == the population and no filter
# can mismatch the name.
PRODUCERS["w5-cross-node"] = lambda: nextest_full(
    "w5-cross-node", ["-p", "spt", "--test", "webserve_cross_node_e2e"], 420,
    profile="default")