SOLIDWORKS Bridge

#!/usr/bin/env python3
"""drain-gate.py — prove the DRAIN's close key on the real machine.

RUN THIS ONLY ON A BRIDGE-LAUNCHED SESSION DREW HAS SAID IS FREE. It creates its
own documents and closes ONE of them. It never closes a document it did not
create, never answers a dialog, never kills anything, and never touches a file
outside the scratch directory it makes.

WHY THIS EXISTS
───────────────
The drain identifies a leaked import temporary by its path under the resolved
`%TEMP%` and then closes it with `ISldWorks::CloseDoc(<that full path>)`. The
documented help text for `CloseDoc` says only *"name of the document to close"*.
**That a full path is accepted is this bridge's premise and has never been run
against a real SOLIDWORKS.** Every other property of the drain — the anchor, the
bounded walk, the collision refusal — rests on it:

  * if the path form works, the drain closes exactly the document it chose;
  * if it does not, 0.5.7 reports `close_by_path_unsupported` and refuses rather
    than retrying by title (a title is not an identity: on a session holding
    leaked temps, several documents usually share one);
  * and the case that must be impossible — closing a DIFFERENT document — is
    what the count discriminator and this gate are here to rule out.

THE EXPERIMENT, on documents the bridge makes itself
────────────────────────────────────────────────────
  1. BASELINE      swx_docs + swx_get_app_state. Records the whole open set, so
                   step 5 can prove nothing else moved.
  2. TWO TEMPS     swx_import the same STEP fixture twice. SOLIDWORKS titles both
                   from the file stem, so their titles COLLIDE with each other.
  3. A SAVED TWIN  swx_save_as one of them into the scratch dir under the SAME
                   title. Now a saved user-style document shares a temp's title —
                   which is Drew's live session in miniature, and the exact shape
                   that made closing by title a data-loss path.
  4. ONE BATCH     swx_close_import_temps {batch:1}. Then assert:
                     - exactly one document is gone, and it is gone BY PATH;
                     - the SAVED twin is still open;
                     - `documentCount` dropped by exactly 1;
                     - `titleCollisions` names the colliding pair;
                     - `closedDocs[0].path` is the temp path, not a title.
  5. UNTOUCHED     every document open at baseline that we did not create is
                   still open.
  6. CLEANUP       the bridge's own documents are closed and the scratch dir is
                   removed. If cleanup cannot finish it says so LOUDLY rather
                   than leaving the user to find the mess.

    tools/drain-gate.py                 # the gate (STEP fixture)
    tools/drain-gate.py --fixture new   # same proof without importing anything
    tools/drain-gate.py --baseline      # step 1 only. Reads. Creates nothing.
    tools/drain-gate.py --keep          # skip step 6 (leave the evidence open)

WHY `--fixture new` EXISTS, and what it proves instead
──────────────────────────────────────────────────────
Measured 2026-09-17 on SOLIDWORKS Connected 2026: with
`swMultiCAD_Enable3DInterconnect` TRUE — Drew's setting, which this bridge reads
and never changes — there is NO programmatic neutral-file import route at all
(LoadFile4 errors=1 with Nothing and with an ImportFileData object, LoadFile2
false, OpenDoc6 swFileRequiresRepairError). So the STEP fixture cannot create
anything to drain on that machine, and a gate that cannot create its own evidence
cannot run.

MEASURED LIVE, 2026-09-17: SOLIDWORKS REFUSED to open a second document with the
same title from another directory — `errors = 65536`
(`swFileWithSameTitleAlreadyOpen`). So a title collision between SAVED documents
is IMPOSSIBLE on this edition, and only PATHLESS documents can ever collide —
which is precisely the state identity-at-birth removes, and it means the drain's
collision guard protects a case that can only arise before a document is given a
file.

`--fixture new` uses `swx_new_document` twice instead. THE OWNERSHIP PROOF IS THE
REGISTRY, NOT THE VERB THAT MADE THE DOCUMENT: both routes go through the same
identity-at-birth and `register_created`, so the same ownership question is asked
of the same registry.

What changes is WHICH DRAIN VERB the gate can use, and this is a real difference
worth stating rather than papering over: `swx_close_import_temps` identifies its
victims BY PATH — a `...\\Temp\\swx<pid>\\...\\IC~~\\...` conversion temp — and that is
deliberate, because the path is the ownership proof that survives a pid change
(it is what made the 1342-temp leak drainable at all). A `new_document` part has
no such path and is CORRECTLY not drained by it. So the `new` fixture drains with
`swx_reset_workspace`, which closes documents the bridge created according to the
ownership REGISTRY. The same-titled-collision half is unchanged: a saved twin
still shares a title with a bridge-created document, and must still survive.

EXIT CODES
    0  the premise holds and every claim held
    1  a claim did not hold
    2  refused to run (no session, a dialog on screen, not bridge-launched)
    3  a probe could not be READ — unproven, never a pass
"""
import argparse
import atexit
import os
import time
import base64
import json
import subprocess
import sys

# THE ROUTE: verbs go through adom-bridge's verb routing to the bridge it hosts
# (see `verb`); SWX_BRIDGE_URL forces a loopback /command URL instead. The old
# default, http://127.0.0.1:24695/command, is dead since adom-bridge 2.1.163.
THREAD = "SolidWorks Bridge"
PS_FAST_EMPTY_RETRIES = 3
PS_RETRY_BACKOFF_SECS = 2
FIXTURE_STEM = "DrainGate"

FAILURES = []
NOTES = []


class Unreadable(Exception):
    """A probe could not be ANSWERED. Not an empty answer — no answer."""


class Refused(Exception):
    """The preconditions were not met. Nothing was created or closed."""


# EVERY SCRIPT SAYS WHY IT FAILED (2026-10-01). A shell_execute that exits 1 with
# empty stdout AND stderr was read for two days as a "transient" and retried; it
# was never transient — on 09-30 it was curl.exe against the dead loopback port
# (curl exit 7 surfacing as an empty exit 1), and on 10-01 it was the launch
# gate's UIA probe, every single time. So every script now runs inside a wrapper
# that PRINTS its own failure to stdout: a trap for terminating errors, and on a
# non-zero native exit the exit code and every error PowerShell recorded. Empty
# stderr cannot hide a failure again. ($ErrorActionPreference is deliberately left
# at its default: making it Stop would turn the non-terminating errors the
# measured scripts already tolerate into new failures.)
def _wrap(script):
    return (
        "$Error.Clear(); "
        "trap { [Console]::Out.WriteLine('PS-ERROR: ' + ($_ | Out-String).Trim()); exit 1 }; "
        + script +
        "\n$__code = $LASTEXITCODE; "
        "if ($__code) { [Console]::Out.WriteLine('PS-NATIVE-EXIT: ' + $__code); "
        "foreach ($__e in $Error) { [Console]::Out.WriteLine('PS-ERROR: ' + ($__e | Out-String).Trim()) }; "
        "exit $__code }"
    )


def ps(script, reason, timeout=120, _attempt=0):
    enc = base64.b64encode(_wrap(script).encode("utf-16le")).decode()
    r = subprocess.run(
        [
            "adom-bridge", "--target", "adom_drew2", "--ai-thread", THREAD, "shell_execute",
            json.dumps({
                "command": "powershell -NoProfile -EncodedCommand " + enc,
                "timeoutSeconds": timeout + 30,
                "reason": reason,
            }),
        ],
        capture_output=True, text=True,
    )
    try:
        payload = json.loads(r.stdout)
    except Exception:
        raise Unreadable("adom-bridge did not return JSON: %s%s"
                         % (r.stdout[:200], (" / " + r.stderr[:200]) if r.stderr else ""))
    if payload.get("success") is False:
        # A FAST, EMPTY exit 1 (under 5 s, nothing on stdout or stderr). It was
        # called "transient" and it was not: on 2026-09-30 it was curl.exe against
        # the dead loopback port (exit 7, surfacing as an empty exit 1), and on
        # 2026-10-01 it was the launch gate's UIA probe, deterministically — four
        # attempts in a row. Since `_wrap`, a script that fails prints WHY, so a
        # truly empty failure means the script never ran at all; the bounded
        # retry (three times, 2 s apart, each printed) stays for that case only.
        # Any other failure, or a fourth, is Unreadable — with the script's own
        # error text.
        fast_empty = (payload.get("exitCode") == 1 and not (payload.get("stdout") or "")
                      and not (payload.get("stderr") or "") and (payload.get("durationMs") or 99999) < 5000)
        if fast_empty and _attempt < PS_FAST_EMPTY_RETRIES:
            print("  (shell_execute failed fast with no output; retry %d of %d in %ds)"
                  % (_attempt + 1, PS_FAST_EMPTY_RETRIES, PS_RETRY_BACKOFF_SECS))
            time.sleep(PS_RETRY_BACKOFF_SECS)
            return ps(script, reason, timeout, _attempt=_attempt + 1)
        said = "\n".join(l for l in (payload.get("stdout") or "").splitlines()
                         if l.startswith("PS-ERROR: ") or l.startswith("PS-NATIVE-EXIT: "))
        raise Unreadable("shell_execute failed (exit %s): %s%s"
                         % (payload.get("exitCode"), said or "no error text at all — the script never ran",
                            "" if said else " / " + json.dumps(payload)[:200]))
    out = payload.get("stdout")
    if out is None:
        raise Unreadable("shell_execute returned no stdout: %s" % json.dumps(payload)[:300])
    return out.strip()


# ── LOCAL MODE ──────────────────────────────────────────────────────────────
#
# `--local http://127.0.0.1:PORT/command` points every call at a bridge running
# HERE instead of tunnelling PowerShell to the desktop. It exists because this
# gate shipped four bugs of its own that only a live run could find — it asked
# for a pathless document and then failed the bridge for not giving it a path; it
# saved a twin to a path with no extension and reported the format refusal as a
# consent-gate bug; it tried to close a document that had never been created. All
# four are reproducible against the simulator in seconds. A gate that can only be
# tested on the machine it is meant to protect gets debugged in production.
LOCAL = None


def local_verb(name, args=None, timeout=180):
    import urllib.request
    body = json.dumps({"command": name, "args": args or {}}).encode()
    req = urllib.request.Request(LOCAL, data=body,
                                 headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read().decode())


def verb(name, args=None, timeout=180):
    """Call a bridge verb THROUGH the desktop bridge's verb routing (adom-bridge
    routes swx_* to the bridge instance it hosts). Raises `Unreadable` rather
    than inventing an answer.

    MEASURED 2026-09-30: adom-bridge 2.1.163 hosts the released bridge itself
    (bridges-cache\\solidworks, its own port) and terminates a side-loaded copy
    on 24695 within 30 s, so the loopback route this helper used until 0.8.9 is
    dead. Set SWX_BRIDGE_URL to a loopback /command URL to force the old route.
    """
    if LOCAL:
        return local_verb(name, args, timeout)
    import os
    routed = not os.environ.get("SWX_BRIDGE_URL")
    if not routed:
        body = json.dumps({"command": name, "args": args or {}})
        script = (
            "$b=@'\n" + body + "\n'@; $p=\"$env:TEMP\\swx_gate.json\"; "
            "[IO.File]::WriteAllText($p,$b); "
            f"curl.exe -s --max-time {timeout} -H \"Content-Type: application/json\" "
            f"--data-binary \"@$p\" {os.environ['SWX_BRIDGE_URL']}"
        )
        out = ps(script, f"SOLIDWORKS gate: calling {name}", timeout)
    else:
        # A VERB THAT DOES NOT ANSWER IS 'could not ask', not a crash. Without
        # this, subprocess.TimeoutExpired escaped as a traceback — a gate dying
        # on a slow verb before it could print a verdict or clean up.
        try:
            r = subprocess.run(
                ["adom-bridge", "--target", "adom_drew2", "--ai-thread", THREAD, name,
                 json.dumps(args or {})],
                capture_output=True, text=True, timeout=timeout + 60,
            )
        except subprocess.TimeoutExpired:
            raise Unreadable("%s did not answer through adom-bridge within %ds" % (name, timeout + 60))
        out = r.stdout
        i = out.find("{")
        if i < 0:
            raise Unreadable("%s did not answer with JSON: %r%s" % (name, out[:200], (" / " + r.stderr[:200]) if r.stderr else ""))
        out = out[i:]
    try:
        d = json.loads(out)
    except Exception:
        raise Unreadable("%s did not answer with JSON: %r" % (name, out[:300]))
    # adom-bridge wraps a ROUTED reply as {"status":..,"result":{...}} (and merges
    # top-level fields and adds an `output` string); the bridge's own reply is the
    # `result`. Only the routed reply is unwrapped: on the loopback route the JSON
    # IS the bridge's reply, and a verb field that happens to be called `result`
    # is not a wrapper.
    if routed and isinstance(d, dict) and isinstance(d.get("result"), dict):
        return d["result"]
    return d


def check(ok, claim, detail=""):
    print(("  PASS  " if ok else "  FAIL  ") + claim + (f"   [{detail}]" if detail else ""))
    if not ok:
        FAILURES.append(claim + (f" [{detail}]" if detail else ""))
    return ok


def docs():
    r = verb("swx_docs", {})
    if r.get("truncated"):
        raise Refused("the session has more documents open than can be enumerated "
                      "(count=%r) — run the drain in anger first; this gate needs to see the whole "
                      "open set to prove nothing else moved" % r.get("count"))
    rows = r.get("docs")
    if not isinstance(rows, list):
        raise Unreadable("swx_docs returned no document list: %s" % json.dumps(r)[:300])
    return rows


def paths(rows):
    return {str(d.get("path") or "") for d in rows if d.get("path")}


def titles(rows):
    return [str(d.get("title") or "") for d in rows]


# ── WHAT THIS RUN CREATED ────────────────────────────────────────────────────
#
# Written down BEFORE each creation, so cleanup knows what is the gate's own on
# EVERY exit path — including the ones that used to skip it.
#
# Two live defects made this necessary. Cleanup ran only on the success path, so
# a Refused (exit 2) or a crash left this gate's documents OPEN IN DREW'S SESSION
# and its files on disk — and on that machine every run currently refuses behind
# the License Manager dialog, so the cleanup step had effectively never run. And
# the leftover check called the gate's own files foreign, because the only record
# of them was a variable assigned in a branch that the measured behaviour does
# not take.
GATE_PARENT = "swx-drain-gate"
MANIFEST = {"documents": [], "files": [], "dirs": []}
MANIFEST_PATH = None
_CLEANED = False


def _cleanup_at_exit():
    """Returns the list of refusals, so the caller can raise the exit code."""
    """A Ctrl-C or an unhandled exception skips `finally` in some shells; this
    does not. Idempotent, so the normal path's cleanup is not repeated."""
    global _CLEANED
    if _CLEANED or MANIFEST_PATH is None:
        return []
    _CLEANED = True
    try:
        root = os.path.dirname(MANIFEST_PATH)
        return cleanup(_ARGS, root) or []
    except Exception as e:
        print("  cleanup at exit failed: %s" % str(e)[:200])
    return []


atexit.register(_cleanup_at_exit)


def _solidworks_session():
    """The running SOLIDWORKS process id, or None when it cannot be read.

    `None` is UNKNOWN and is treated as "not proven to be the same session" —
    the opt-in is withheld rather than granted on a guess.
    """
    try:
        st = verb("swx_status", {})
    except Exception:
        return None
    com = st.get("com") or {}
    for key in ("processId", "pid", "solidworksPid"):
        v = com.get(key) or st.get(key)
        if isinstance(v, int):
            return v
    return None


def remember(kind, item):
    """Record a creation BEFORE making it, and persist immediately."""
    if item and item not in MANIFEST[kind]:
        MANIFEST[kind].append(item)
    if MANIFEST_PATH:
        try:
            with open(MANIFEST_PATH, "w") as fh:
                json.dump(MANIFEST, fh, indent=2)
        except OSError:
            pass
    return item


def gate_parent(scratch_root):
    """The fixed directory this gate owns. Every run lives in a subdirectory of
    it, and NOTHING outside it is ever closed or deleted (see `owned`)."""
    return os.path.dirname(scratch_root.rstrip("/\\")) or scratch_root


def owned(parent, path):
    """IS THIS PATH INSIDE THE DIRECTORY THIS GATE OWNS?

    Cleanup closes documents with `allowUserWrites` + `confirm`, which is a
    user-write that closes a document the bridge did not create. That is correct
    for the twin the gate itself opened and WRONG for anything else — and once a
    run can adopt another run's manifest, a `run-*.json` that a person or another
    tool had left in %TEMP% would feed arbitrary paths straight into it. So every
    path is checked for containment first, and anything outside is refused and
    reported rather than acted on.
    """
    if not path or not parent:
        return False
    try:
        a = os.path.realpath(str(path))
        b = os.path.realpath(str(parent))
        return os.path.commonpath([a, b]) == b and a != b
    except (ValueError, OSError):
        # Different drives, or an unreadable path: NOT ours. Refusing is the safe
        # direction for a check that guards a write.
        return False


def prior_manifests(scratch_root):
    """Manifests left by EARLIER RUNS, found in the parent this gate owns.

    A run that died hard — SIGKILL, a crash, a reboot — cannot run its own
    cleanup, and `atexit` does not fire. The next run collects it.
    """
    parent = gate_parent(scratch_root)
    out = []
    try:
        for entry in sorted(os.listdir(parent)):
            run_dir = os.path.join(parent, entry)
            if not os.path.isdir(run_dir):
                continue
            for name in sorted(os.listdir(run_dir)):
                if name.startswith("run-") and name.endswith(".json"):
                    path = os.path.join(run_dir, name)
                    try:
                        with open(path) as fh:
                            out.append((path, json.load(fh)))
                    except (OSError, ValueError):
                        pass
    except OSError:
        pass
    return out


def cleanup(a, scratch_root):
    """CLOSE AND REMOVE EXACTLY WHAT THIS RUN CREATED — on every exit path."""
    if a.keep:
        print("\nSTEP 6 — CLEANUP SKIPPED (--keep): this run's documents and files are left "
              "in place. Its manifest is at %s" % MANIFEST_PATH)
        return []
    print("\nSTEP 6 — CLEANUP (only what this gate created, from its own manifest)")
    refused = []
    current_session = _solidworks_session()
    parent = gate_parent(scratch_root)
    runs = [(MANIFEST_PATH, MANIFEST)] + [
        (p, m) for (p, m) in prior_manifests(scratch_root) if p != MANIFEST_PATH
    ]
    # THE TWIN IS THE ONLY DOCUMENT THAT NEEDS THE OPT-IN. Everything else this
    # gate made, the bridge made, and the default posture closes those.
    twin_paths = set()
    for _p, man in runs:
        for f in man.get("files", []):
            if "twin-" in str(f):
                twin_paths.add(str(f))
    for path, man in runs:
        for doc in man.get("documents", []):
            if not str(doc or "").strip():
                continue
            # CONTAINMENT GATES THE OPT-IN, not the close.
            #
            # Two different documents are in here. The TWIN is a file this gate
            # copied and opened, so the bridge sees a user document and closing it
            # needs allowUserWrites + confirm — a confirmed user-write, which must
            # never be pointed at an arbitrary path a manifest happened to name.
            # That one is checked for containment: it lives under the directory
            # this gate owns, and anything outside is refused and named.
            #
            # The others the BRIDGE created, and they live in the BRIDGE's scratch
            # area, not the gate's. Those are closed in the DEFAULT POSTURE — and
            # the default posture is itself the safety proof: the bridge refuses
            # to close a document it did not create. The gate does not assert that
            # ownership; the bridge decides it. Requiring containment here instead
            # left the gate's own documents open, which is the harm this whole
            # cleanup exists to prevent.
            inside = owned(parent, doc)
            if doc in twin_paths and not inside:
                refused.append(
                    "a manifest named a TWIN document OUTSIDE this gate's own directory and it "
                    "was NOT closed: %r is not under %r. Closing a twin is a confirmed "
                    "user-write, so it only ever touches what this gate could have written."
                    % (doc, parent))
                continue
            try:
                # THE OPT-IN, NAMED, FOR THE GATE'S OWN DOCUMENTS ONLY.
                #
                # Most of these the bridge created and can close in the default
                # posture. The TWIN cannot: the gate copied a file and opened it,
                # so from the bridge's point of view it is a user document, and
                # closing one needs allowUserWrites + confirm + a reason. Without
                # it the twin stayed open in the session — which is the very harm
                # this cleanup exists to prevent, one layer down.
                #
                # The scope is the manifest: only documents THIS RUN created.
                # THE OPT-IN IS SCOPED TO THE TWIN. Most of these the BRIDGE
                # created and closes in the default posture; only the twin — a
                # file this gate copied and opened, which the bridge therefore
                # sees as a user document — needs allowUserWrites + confirm.
                # Asking for it on every document would mean the widest possible
                # permission on every path a manifest happened to name.
                args = {"path": doc,
                        "reason": "drain gate cleanup: closing a document this gate created "
                                  "during a run, inside the directory it owns"}
                # THE SESSION MUST MATCH. A twin recorded against another
                # SOLIDWORKS is not this session's document, whatever its path
                # says, and the opt-in is the one authority this gate holds that
                # the bridge does not check for it.
                same_session = (man.get("sessionPid") is not None
                                and man.get("sessionPid") == current_session)
                if doc in twin_paths and inside and same_session:
                    args.update({
                        "allowUserWrites": True,
                        "confirm": True,
                        "reason": "drain gate cleanup: closing the same-title TWIN this gate "
                                  "copied and opened — the bridge did not create it, so closing "
                                  "it is a user write, and it is inside the gate's own directory",
                    })
                elif doc in twin_paths:
                    refused.append(
                        "a TWIN from another SOLIDWORKS session (manifest sessionPid=%r, running "
                        "now=%r) was NOT force-closed: %r. Its file is still removed — containment "
                        "proves this gate wrote it — but the write opt-in is withheld, because a "
                        "path is not proof of which session has the document open."
                        % (man.get("sessionPid"), current_session, doc))
                    continue
                c = verb("swx_close_doc", args)
            except Exception as e:          # a probe that cannot be read is not a reason to stop
                NOTES.append("could not close %s: %s" % (doc, str(e)[:120]))
                continue
            code = c.get("errorCode")
            if c.get("success") is not True and code not in ("document_not_open",
                                                             "no_target_document"):
                NOTES.append("could not close %s (%s) — it is still open on the user's screen"
                             % (doc, code))
        for f in man.get("files", []):
            if not owned(parent, f):
                refused.append(
                    "a manifest named a file OUTSIDE this gate's own directory and it was NOT "
                    "removed: %r is not under %r." % (f, parent))
                continue
            _remove_file(f)
        for d in sorted(man.get("dirs", []), key=len, reverse=True):
            if not owned(parent, d):
                refused.append(
                    "a manifest named a directory OUTSIDE this gate's own directory and it was "
                    "NOT removed: %r is not under %r." % (d, parent))
                continue
            _remove_dir_if_empty(d)
        if path and path != MANIFEST_PATH and owned(parent, path):
            _remove_file(path)
            _remove_dir_if_empty(os.path.dirname(path))
    _remove_file(MANIFEST_PATH)
    _remove_dir_if_empty(scratch_root)
    left = _listing(scratch_root)
    if left:
        NOTES.append(
            f"{scratch_root} still holds files this gate has no record of creating, so the "
            f"directory was left alone: {json.dumps(left)[:300]}")
    print("  cleanup: %d item(s) left in the scratch dir" % len(left))
    # REFUSALS ARE PRINTED HERE, not folded into a verdict that has already been
    # computed: cleanup runs from a `finally`, AFTER the exit code is decided, so
    # a FAILURES entry added here would never be seen. The caller raises the exit
    # code instead.
    for r in refused:
        print("  CLEANUP REFUSED: %s" % r)
    return refused


def _remove_file(path):
    if not path:
        return
    if LOCAL:
        try:
            os.remove(path)
        except OSError:
            pass
        return
    try:
        ps("$f='%s'; if (Test-Path -LiteralPath $f) { Remove-Item -Force -LiteralPath $f } ; "
           "Write-Output ok" % str(path).replace("'", "''"),
           "SOLIDWORKS drain gate: removing one file it wrote", 60)
    except Unreadable:
        pass


def _remove_dir_if_empty(path):
    if not path:
        return
    if LOCAL:
        try:
            os.rmdir(path)
        except OSError:
            pass
        return
    try:
        ps("$d='%s'; if (Test-Path -LiteralPath $d) { "
           "if (@(Get-ChildItem -Force -LiteralPath $d).Count -eq 0) { Remove-Item -Force "
           "-LiteralPath $d } } ; Write-Output ok" % str(path).replace("'", "''"),
           "SOLIDWORKS drain gate: removing its scratch directory if it is empty", 60)
    except Unreadable:
        pass


def _listing(path):
    if LOCAL:
        try:
            return sorted(os.listdir(path))
        except OSError:
            return []
    try:
        out = ps("$d='%s'; if (Test-Path -LiteralPath $d) { "
                 "@(Get-ChildItem -Force -LiteralPath $d | ForEach-Object { $_.Name }) } "
                 "| ConvertTo-Json -Compress" % str(path).replace("'", "''"),
                 "SOLIDWORKS drain gate: listing what is left in its scratch directory", 60)
        v = json.loads(out) if out.strip() else []
        return v if isinstance(v, list) else [v]
    except (Unreadable, ValueError):
        return []


def make_fixture(_unused=None):
    """A minimal STEP file, written by us, in a directory we made."""
    step = (
        "ISO-10303-21;\nHEADER;\nFILE_DESCRIPTION((''),'2;1');\n"
        "FILE_NAME('%s','2026-01-01T00:00:00',(''),(''),'','','');\n"
        "FILE_SCHEMA(('AUTOMOTIVE_DESIGN'));\nENDSEC;\nDATA;\nENDSEC;\nEND-ISO-10303-21;\n"
        % FIXTURE_STEM
    )
    if LOCAL:
        import tempfile
        # ONE FIXED PARENT, PER-RUN SUBDIRECTORIES UNDER IT.
        #
        # 0.8.5 made the whole scratch root unique per run, which fixed "earlier
        # runs' files look foreign" by making earlier runs INVISIBLE: the manifest
        # scan looked inside the fresh directory, found only its own, and adopted
        # nothing. So a SIGKILL, a hard crash or a reboot — exactly what `atexit`
        # cannot cover — left a whole directory behind on every occurrence, with
        # no run that would ever collect them.
        #
        # A fixed PARENT that the gate owns, with a unique subdirectory per run,
        # gives both: no run can mistake another's files for a stranger's, and any
        # run can find and clean up what a dead one left.
        parent = os.path.join(tempfile.gettempdir(), GATE_PARENT)
        os.makedirs(parent, exist_ok=True)
        d = os.path.join(parent, "run-%d-%d" % (os.getpid(), int(time.time())))
        os.makedirs(d, exist_ok=False)
        with open(os.path.join(d, FIXTURE_STEM + ".step"), "w") as fh:
            fh.write(step)
        return d
    script = (
        f"$p=Join-Path $env:TEMP '{GATE_PARENT}'; "
        "New-Item -ItemType Directory -Force -Path $p | Out-Null; "
        f"$d=Join-Path $p 'run-{os.getpid()}-{int(time.time())}'; "
        "New-Item -ItemType Directory -Path $d -ErrorAction Stop | Out-Null; "
        f"$f=Join-Path $d '{FIXTURE_STEM}.step'; "
        "[IO.File]::WriteAllText($f, @'\n" + step + "\n'@); Write-Output $d"
    )
    d = ps(script, "SOLIDWORKS drain gate: writing its own STEP fixture into a scratch dir under "
                   "%TEMP% (nothing outside it is touched)", 60)
    if not d:
        raise Unreadable("the scratch directory was not reported back")
    return d.strip().splitlines()[-1].strip()


def presence_verdict(reply):
    """THE USER-PRESENCE GUARD (0.9.9), as a gate reads it. Returns
    ("absent" | "present", message). A gate NEVER passes userPresentOverride:
    the session belongs to the person using it, and a gate is not that person.
    An unreadable reply is "present" — fail closed."""
    if not isinstance(reply, dict) or reply.get("success") is False and reply.get("errorCode") != "user_is_active":
        return ("present", "the bridge could not say whether the user is working in SOLIDWORKS (%s) — "
                           "the gate does not run on a guess" % (reply.get("error") if isinstance(reply, dict) else reply))
    if reply.get("errorCode") == "user_is_active":
        ev = reply.get("userPresence") or {}
        return ("present", "the bridge refused: the user is working in this SOLIDWORKS (%s)" % ev.get("basis"))
    if reply.get("present") is False:
        return ("absent", "no sign of the user: %s" % ((reply.get("evidence") or {}).get("basis")))
    ev = reply.get("evidence") or {}
    return ("present", "THE USER IS WORKING IN THIS SOLIDWORKS (%s; foreground=%s, %ss since last input, "
                       "%s document(s) the bridge did not open). This gate does not run on a session "
                       "somebody is using, and never overrides that — wait until they are done."
            % (ev.get("basis"), (ev.get("foregroundWindow") or {}).get("title"),
               ev.get("secondsSinceLastInput"), (ev.get("foreignDocuments") or {}).get("count")))


def main():
    """Run the gate, and CLEAN UP ON EVERY EXIT PATH.

    Cleanup used to live at the end of the happy path, so a Refused (exit 2), an
    Unreadable (exit 3), a crash or a Ctrl-C skipped it — leaving this gate's own
    documents OPEN in the user's session. On Drew's machine every run currently
    refuses behind the License Manager dialog, so it had effectively never run.
    """
    scratch_root = None
    a = None
    try:
        code, scratch_root = _run()
        return code
    finally:
        # THE SAME IDEMPOTENT CLEANUP the atexit hook calls, so the two cannot
        # both run — which they did, printing STEP 6 twice.
        if scratch_root is not None:
            refused = _cleanup_at_exit()
            if refused:
                print("\nGATE EXIT RAISED TO 1 — cleanup refused %d path(s) that a manifest "
                      "named outside this gate's own directory. Nothing outside it was closed "
                      "or deleted; somebody should look at why a manifest said that."
                      % len(refused))
                # A RETURN IN `finally` IS WHAT ACTUALLY CHANGES THE EXIT CODE.
                # Assigning to `code` here does nothing: the `return` in the try
                # block has already fixed the value, so the refusal printed and
                # the gate still exited 0.
                return 1


class _Args:
    baseline = False
    keep = False
    fixture = "step"
    local = None


_ARGS = _Args()


def _run():
    ap = argparse.ArgumentParser(description="prove the drain's close key on the real machine")
    ap.add_argument("--baseline", action="store_true", help="step 1 only; creates nothing")
    ap.add_argument("--keep", action="store_true", help="leave this gate's documents open")
    ap.add_argument("--local", metavar="URL",
                    help="run against a bridge on THIS machine (e.g. a mock on "
                         "http://127.0.0.1:PORT/command) instead of the desktop — the only way to "
                         "test the gate itself without touching the machine it protects")
    ap.add_argument("--fixture", choices=("step", "new"), default="step",
                    help="what the gate creates to drain: 'step' imports a STEP twice (needs 3D "
                         "Interconnect OFF), 'new' makes two documents with swx_new_document "
                         "(works regardless of that setting)")
    a = ap.parse_args()
    global LOCAL, _ARGS, MANIFEST_PATH
    LOCAL = a.local
    _ARGS = a

    print("=" * 72)
    print("STEP 1 — BASELINE")
    st = verb("swx_status", {})
    if (st.get("com") or {}).get("running") is not True:
        raise Refused("SOLIDWORKS is not attached (swx_status.com.running is not true). This gate "
                      "runs on a bridge-launched session; start one with swx_start first.")
    who, why_present = presence_verdict(verb("swx_user_presence", {}))
    print("  user presence: %s — %s" % (who, why_present))
    if who != "absent":
        raise Refused(why_present)
    launch = st.get("launch") or {}
    if launch.get("attributed") is not True:
        NOTES.append("launch.attributed is %r — this bridge cannot prove it started the running "
                     "SOLIDWORKS, so treat everything below as measured on the USER's session and "
                     "read the cleanup step twice" % launch.get("attributed"))
    state = verb("swx_get_app_state", {})
    if state.get("documentsTruncated"):
        raise Refused("app_state reports the document set as truncated — see swx_docs above")
    base_rows = docs()
    base_paths = paths(base_rows)
    print("  %d document(s) open at baseline" % len(base_rows))
    for d in base_rows:
        print("    %-40s %s" % (str(d.get("title"))[:40], d.get("path")))
    if a.baseline:
        print("\nBASELINE ONLY — nothing was created. Re-run without --baseline for the experiment.")
        return (1 if FAILURES else 0), None

    scratch = make_fixture(None)
    # THE MANIFEST, opened before the first creation. `run-<pid>-<stamp>.json`
    # so concurrent or earlier runs each have their own and cleanup can adopt
    # the ones they left behind.
    MANIFEST_PATH = (scratch.rstrip("/\\") + (os.sep if LOCAL else "\\")
                     + "run-%d-%d.json" % (os.getpid(), int(time.time())))
    remember("files", scratch.rstrip("/\\") + (os.sep if LOCAL else "\\")
             + FIXTURE_STEM + ".step")
    # WHICH SOLIDWORKS THIS RUN'S DOCUMENTS BELONG TO.
    #
    # The twin close is the ONE place this tool overrides the bridge's ownership
    # answer — allowUserWrites + confirm on a document the bridge did not create.
    # Path containment says the file is one this gate could have written; it says
    # nothing about WHICH SOLIDWORKS has it open. A twin path from a dead run
    # (pid + epoch second) could name a document in a session that started
    # afterwards, and the opt-in would force it closed. So the session is recorded
    # and the opt-in is withheld when it differs — the FILE is still removed,
    # because that part is proven by containment.
    MANIFEST["sessionPid"] = _solidworks_session()
    remember("files", None)      # persist the session id immediately
    fixture = scratch.rstrip("\\") + "\\" + FIXTURE_STEM + ".step"
    if a.fixture == "new":
        print("\nSTEP 2 — TWO BRIDGE-CREATED DOCUMENTS (swx_new_document)")
    else:
        print("\nSTEP 2 — TWO IMPORT TEMPORARIES from %s" % fixture)
    imported = []
    for i in (1, 2):
        if a.fixture == "new":
            # THE SAME OWNERSHIP PATH, A DIFFERENT MAKER. new_document goes
            # through identity-at-birth and register_created exactly as import
            # does, so the registry this gate is really testing is the same one.
            # NO `keepUnsaved`. The gate asserts below that every document it
            # made can be addressed BY PATH, and `keepUnsaved:true` asks for the
            # opposite — a pathless document. The 2026-09-17 run failed itself on
            # exactly that: it requested no file and then reported the bridge for
            # not giving it one. A plain new_document gets identity-at-birth and
            # a scratch file, which is what this gate is here to exercise.
            r = verb("swx_new_document", {"type": "part",
                                          "reason": "drain gate: creating a bridge-owned document"})
        else:
            r = verb("swx_import", {"filePath": fixture,
                                    "reason": "drain gate: creating a bridge-owned import temporary"})
        if r.get("success") is not True:
            if a.fixture != "new" and r.get("errorCode") == "interconnect_blocks_import":
                raise Refused(
                    "swx_import is blocked by the user's 3D Interconnect setting, which this "
                    "bridge does not change. Re-run with --fixture new, which proves the same "
                    "ownership and collision properties using swx_new_document: %s"
                    % str(r.get("error"))[:300])
            raise Refused("%s failed, so this gate created nothing to drain: %s"
                          % ("swx_new_document" if a.fixture == "new" else "swx_import",
                             json.dumps(r)[:400]))
        imp = r.get("imported") or r.get("created") or r.get("document") or {}
        ident = r.get("documentIdentity") or {}
        # RECORDED THE MOMENT IT EXISTS, before anything can fail.
        remember("documents", str(imp.get("path") or "") or str(imp.get("title") or ""))
        imported.append(imp)
        print("  %s %d -> title=%r path=%r identity=%r"
              % ("document" if a.fixture == "new" else "import",
                 i, imp.get("title"), imp.get("path"), ident.get("identity")))
        # ADDRESS BY IDENTITY, NOT BY TITLE. Live on 2026-09-11 both imports came
        # back with path:"" and the SAME title, and every verb that had to name
        # one of them refused — correctly. 0.6.0 gives each one a file at birth;
        # if that did not happen, this gate says so rather than proceeding to
        # address documents it cannot name.
        if not str(imp.get("path") or "").strip():
            FAILURES.append(
                "document %d came back with NO PATH (identity=%r). Two documents from one fixture "
                "then share a title and neither can be addressed by any verb — which is the state "
                "0.6.0's identity-at-birth exists to prevent. Everything below this line is "
                "unreliable." % (i, ident.get("identity")))
    temp_paths = [str(d.get("path") or "") for d in imported]
    temp_title = str(imported[0].get("title") or "")

    # EVERY FILE THIS GATE WRITES, relative to the scratch dir, so cleanup removes
    # exactly those and nothing else. `twin_name` used to be defined in one branch
    # and read in both — which crashed cleanup on the other, again.
    twin_name = None
    if a.fixture == "new":
        # ── UNIQUENESS BY DESIGN, stated rather than worked around.
        #
        # Live, 2026-09-17: swx_save_as on a bridge-created document RE-PATHS
        # THAT DOCUMENT rather than producing a second one, so the twin step
        # produced no collision at all — "3 document(s) open, 1 of them titled
        # Part7-…". That is not a bug to route around: under identity-at-birth
        # two bridge-created documents CANNOT share a title, because each is
        # given its own file the moment it exists. So the `new` fixture asserts
        # the property instead of faking its opposite.
        print("\nSTEP 3 — UNIQUENESS BY DESIGN (no twin: two bridge-created documents cannot collide)")
        check(len(set(t for t in titles(docs()) if t in [str(d.get("title") or "") for d in imported])) ==
              len([d for d in imported if str(d.get("title") or "").strip()]),
              "the two bridge-created documents have DISTINCT titles",
              json.dumps([d.get("title") for d in imported]))
        check(len(set(temp_paths)) == len([p for p in temp_paths if p.strip()]),
              "...and DISTINCT paths, each its own file at birth",
              json.dumps(temp_paths))

        # ── AND A MEASUREMENT, because whether SOLIDWORKS permits a collision at
        #    all for SAVED documents is not something this gate should assume.
        #    Copy one document's file to a second directory under the SAME
        #    basename and open it: if SOLIDWORKS allows it, the collision case is
        #    real for saved documents; if it refuses (swFileWithSameTitleAlreadyOpen
        #    = 65536), then only PATHLESS documents can ever collide — which is
        #    worth knowing and worth writing down.
        print("\nSTEP 3b — MEASUREMENT: can two SAVED documents share a title?")
        # A FRESH SUBDIRECTORY, never reused and never overwritten: -Force and
        # exist_ok=True both accept somebody else's directory, and Copy-Item
        # -Force overwrites somebody else's file.
        second_dir = (scratch.rstrip("/\\") + (os.sep if LOCAL else "\\")
                      + "twin-%d" % int(time.time() * 1000))
        src = temp_paths[0]
        base = src.replace("\\", "/").rsplit("/", 1)[-1]
        dup = second_dir + (os.sep if LOCAL else "\\") + base
        made = False
        if LOCAL:
            try:
                os.makedirs(second_dir, exist_ok=False)
                remember("dirs", second_dir)
                remember("files", dup)
                import shutil
                if os.path.exists(dup):
                    raise OSError("refusing to overwrite an existing file at the twin path")
                shutil.copyfile(src, dup)
                made = True
            except OSError as e:
                NOTES.append("could not stage the same-title copy locally: %s" % e)
        else:
            try:
                remember("dirs", second_dir)
                remember("files", dup)
                ps("$d='%s'; New-Item -ItemType Directory -Path $d -ErrorAction Stop | Out-Null; "
                   "Copy-Item -LiteralPath '%s' -Destination '%s'; Write-Output ok"
                   % (second_dir.replace("'", "''"), src.replace("'", "''"), dup.replace("'", "''")),
                   "SOLIDWORKS drain gate: copying its own document to a second path with the same "
                   "basename, to measure whether SOLIDWORKS allows two open documents to share a "
                   "title", 60)
                made = True
            except Unreadable as e:
                NOTES.append("could not stage the same-title copy: %s" % e)
        twin_open = False
        if made:
            opened = verb("swx_open", {"path": dup,
                                       "reason": "drain gate: measuring whether SOLIDWORKS permits "
                                                 "two open documents with the same title"})
            twin_open = opened.get("success") is True
            if twin_open:
                # AN OPENED DOCUMENT IS A CREATION TOO. Recording only the FILE
                # left the probe's document open in the user's session — the same
                # defect one layer down from the one this manifest exists to fix.
                remember("documents", dup)
            code = opened.get("errorCode")
            err = str(opened.get("error") or "")
            if twin_open:
                NOTES.append(
                    "MEASURED: SOLIDWORKS DID open a second document with the same basename (%s) "
                    "from a different directory — so a title collision between SAVED documents is "
                    "possible on this edition, and the drain's collision guard is exercised below."
                    % base)
                saved_path = dup
                twin_name = "twin" + (os.sep if LOCAL else "\\") + base
            elif "65536" in err or "same title" in err.lower() or "already open" in err.lower():
                NOTES.append(
                    "MEASURED: SOLIDWORKS REFUSED to open a second document with the same title. "
                    "Measured on date %s. Invocation: swx_open {path: %r} after copying a document's own "
                    "file to a second directory under the same basename. Result: %s / %s. "
                    "swFileWithSameTitleAlreadyOpen = 65536 — so for SAVED documents a title "
                    "collision is IMPOSSIBLE on this edition, and only PATHLESS documents can "
                    "ever collide. This is written down beside the guard it justifies, in "
                    "src/comhost/com.rs (the drain's title-collision check) — a note printed at "
                    "the end of a run is a note that is gone."
                    % (time.strftime("%Y-%m-%d"), dup, code, err[:200]))
            else:
                NOTES.append("the same-title probe was refused for another reason (%s: %s) — the "
                             "collision question is UNMEASURED this run" % (code, err[:200]))
        before_rows = docs()
        before_count = len(before_rows)
        same_titled = [d for d in before_rows if str(d.get("title") or "") == temp_title]
        print("  %d document(s) open, %d of them titled %r"
              % (before_count, len(same_titled), temp_title))
    else:
        print("\nSTEP 3 — A SAVED TWIN sharing that title")
        # SaveAs3 DISPATCHES ITS WRITER ON THE EXTENSION, so a destination with none
        # is refused with `swFileSaveError_e 256` (format not supported) — which is
        # not a consent decision at all. A `new_document` title is bare ("Part2"),
        # where an import's carried `.SLDPRT` already, so the extension is added when
        # the title does not have one.
        twin_name = temp_title if "." in temp_title else temp_title + ".SLDPRT"
        # The separator belongs to the machine the BRIDGE runs on: a backslash on the
        # desktop, and the host's own in local mode — where a Windows path would make
        # one long filename in /tmp instead of a file in the scratch directory, and
        # cleanup would then honestly report nothing left while a file sat outside it.
        sep = "\\" if not LOCAL else os.sep
        saved_path = scratch.rstrip("/\\") + sep + twin_name
        # DEFAULT POSTURE FIRST, and the result is a FINDING of this gate.
        #
        # The document is one the BRIDGE created and the destination does not exist,
        # inside a directory the gate made — the choke-point rule is supposed to allow
        # exactly that with no `allowUserWrites`, because nothing of the user's can be
        # lost. If it declines, the live proof is unreachable on any bridge that is not
        # running wide open, and THAT IS A BRIDGE BUG worth naming rather than working
        # around silently. (It does not reproduce in the mock: see
        # `the_default_posture_lets_the_bridge_save_its_own_document_to_a_new_path`.)
        save_args = {
            # BY PATH. `name` is kept only as a fallback for a bridge older than
            # 0.6.0, where the import may have no path to name it by.
            "name": temp_title, "path": temp_paths[1], "out": saved_path, "copy": True,
            "reason": "drain gate: a saved document sharing a temporary's title, which is the shape "
                      "that made closing by title a data-loss path",
        }
        saved = verb("swx_save_as", dict(save_args))
        default_allowed = saved.get("success") is True
        print("  defaultGateAllowedScratchSave: %s%s"
              % (default_allowed, "" if default_allowed else
                 "   [%s: %s]" % (saved.get("errorCode"), str(saved.get("error"))[:160])))
        if not default_allowed:
            code = saved.get("errorCode") or "?"
            # TWO VERY DIFFERENT REFUSALS, and the 2026-09-11 run reported the wrong
            # one. `ambiguous_target` is not the write gate saying no — it is the
            # bridge saying it cannot tell WHICH document was meant, which is a
            # property of the session (two open documents sharing a title and having
            # no path), not of the consent rules. Calling it a write-gate bug sent
            # the reader looking in the wrong module.
            # A REFUSAL IS NOT A REFUSAL BY THE GATE. Classify by CODE, never by
            # "it failed": the 2026-09-17 run reported a FORMAT refusal
            # (`swFileSaveError_e 256`, a destination with no extension — the gate's
            # own bug) as "BRIDGE GATE BUG (write gate) ... refused in the default
            # posture", which is an accusation about the consent rules on the
            # strength of an error that has nothing to do with them. A gate that
            # cannot tell its own bug from the bridge's is worse than no gate.
            FORMAT_OR_IO = ("format_not_supported", "save_failed", "export_failed",
                            "invalid_path", "path_not_writable", "destination_unverifiable")
            IDENTITY = ("ambiguous_target", "document_not_open", "document_list_unavailable")
            if code in FORMAT_OR_IO or "format" in str(saved.get("error") or "").lower():
                FAILURES.append(
                    "THE GATE ASKED FOR SOMETHING SOLIDWORKS CANNOT WRITE (%s: %s). This is the "
                    "gate's bug, not the bridge's — SaveAs3 dispatches on the destination extension, "
                    "so the twin path must end in .SLDPRT. Fix the gate."
                    % (code, str(saved.get("error"))[:200]))
            elif code in IDENTITY:
                NOTES.append(
                    "the twin save was refused with `%s` — this is an IDENTITY problem, not the write "
                    "gate: the document could not be named, because a pathless document's only name is "
                    "its title and another open document shares it. From 0.6.0 every bridge-created "
                    "document is given a file at birth, so this should not recur; if it does, the "
                    "identity step is what failed, not the consent rules." % code)
            else:
                FAILURES.append(
                    "BRIDGE GATE BUG (write gate): saving a BRIDGE-CREATED document to a NON-EXISTING "
                    "path inside a bridge-created scratch dir was refused in the default posture (%s). "
                    "Nothing of the user's could have been lost by it, and the rule is pinned in the "
                    "suite — so this is a divergence between the live gate and the mock, not a policy "
                    "decision." % code)
            # ...and then, having RECORDED it, retry with the opt-in so the rest of the
            # gate can still run. The finding above stands either way.
            retry = dict(save_args)
            retry.update({"allowUserWrites": True, "confirm": True})
            saved = verb("swx_save_as", retry)
            if saved.get("success") is True:
                NOTES.append("the twin was saved only WITH the write opt-in; the default-posture "
                             "refusal above is the finding")
        twin_open = False
        if saved.get("success") is not True:
            NOTES.append("swx_save_as declined even with the opt-in (%s) — the SAVED-twin half of the "
                         "premise is untested this run" % (saved.get("errorCode") or "?"))
        else:
            opened = verb("swx_open", {"path": saved_path,
                                       "reason": "drain gate: opening the saved twin so its title "
                                                 "collides with the temporary's"})
            twin_open = opened.get("success") is True
            if not twin_open:
                NOTES.append("the saved twin did not open (%s); the collision half is untested"
                             % (opened.get("errorCode") or "?"))

        before_rows = docs()
        before_count = len(before_rows)
        same_titled = [d for d in before_rows if str(d.get("title") or "") == temp_title]
        print("  %d document(s) open, %d of them titled %r" % (before_count, len(same_titled), temp_title))

    # WHICH DRAIN, AND WHY. `swx_close_import_temps` identifies victims BY PATH
    # (a `IC~~` conversion temp), which is the ownership proof that survives a pid
    # change — so it CORRECTLY does not touch a bridge-created new document, which
    # has no such path. The ownership registry is what both fixtures share, so the
    # `new` fixture drains through the verb that reads it.
    if a.fixture == "new":
        print("\nSTEP 4 — THE REGISTRY DRAIN: swx_reset_workspace")
        print("  (swx_close_import_temps is IMPORT-TEMP-BY-PATH by definition and correctly does")
        print("   not drain a pathless bridge-created document; the registry is the ownership")
        print("   proof this fixture exercises)")
        r = verb("swx_reset_workspace", {"reason": "drain gate: closing the documents this bridge "
                                                   "created, by the ownership registry"})
    else:
        print("\nSTEP 4 — ONE BATCH: swx_close_import_temps {batch:1}")
        r = verb("swx_close_import_temps", {"batch": 1, "reason": "drain gate: the live proof of the "
                                                                  "close key, one document"})
    print("  " + json.dumps({k: r.get(k) for k in
                             ("success", "errorCode", "closed", "closedDocs", "done", "walkStop",
                              "titleCollisions", "titleCollisionsCheckedOver", "closeIgnoredPath",
                              "documentCountBefore", "documentCountAfter", "tempRoot")},
                            indent=2).replace("\n", "\n  "))

    if r.get("errorCode") == "close_outcome_unknown":
        check(False, "UNDETERMINED: the count could not be read after a close",
              "a human must check whether %s or another document closed — see closeFailed"
              % json.dumps(r.get("closeFailed"))[:200])
    elif r.get("errorCode") == "close_by_path_unsupported":
        check(False, "THE PREMISE FAILED: CloseDoc(<full path>) closed nothing on this build",
              "the bridge refused rather than retrying by title, which is correct — but the drain "
              "cannot work until the close key is settled another way")
    elif a.fixture == "new":
        # THE SAME TWO PROPERTIES, asked of the registry drain: OUR documents go,
        # and the twin sharing a title does NOT.
        after_rows = docs()
        after_paths = paths(after_rows)
        after_titles = titles(after_rows)
        check(r.get("success") is True, "the registry drain ran", json.dumps(r)[:300])
        closed = r.get("closed") or []
        check(len(closed) >= 2,
              "both bridge-created documents were closed", json.dumps(closed)[:300])
        if twin_open:
            check(saved_path in after_paths,
                  "the SAVED twin sharing the title is STILL OPEN", saved_path)
            check(temp_title in after_titles,
                  "...so that title is still present on a document this gate did not create",
                  temp_title)
        check(base_paths <= after_paths,
              "every document open before this gate ran is still open",
              json.dumps(sorted(base_paths - after_paths))[:300])
    else:
        after_rows = docs()
        after_paths = paths(after_rows)
        gone = base_paths | set(temp_paths) | {saved_path}
        closed_docs = r.get("closedDocs") or []
        check(r.get("closed") == 1 and len(closed_docs) == 1,
              "exactly ONE document was closed", json.dumps(r.get("closed")))
        closed_path = str((closed_docs[0] if closed_docs else {}).get("path") or "")
        check("IC~~" in closed_path,
              "...and it was named by its TEMP PATH, not by a title", closed_path)
        check(closed_path not in after_paths,
              "...and that exact path is gone from the open set", closed_path)
        check(len(after_rows) == before_count - 1,
              "documentCount dropped by exactly 1", "%d -> %d" % (before_count, len(after_rows)))
        if twin_open:
            check(saved_path in after_paths,
                  "the SAVED twin sharing the title is STILL OPEN", saved_path)
            cols = r.get("titleCollisions") or []
            check(any(str(c.get("title")) == temp_title for c in cols),
                  "...and the collision was reported, with its paths", json.dumps(cols)[:300])
        check(base_paths <= after_paths,
              "every document open before this gate ran is still open",
              json.dumps(sorted(base_paths - after_paths))[:300])
        _ = gone

    _verdict_marker = None
    print("\n" + "=" * 72)
    for n in NOTES:
        print("NOTE: " + n)
    if FAILURES:
        print("GATE FAILED — %d claim(s) did not hold:" % len(FAILURES))
        for f in FAILURES:
            print("  - " + f)
        return 1, scratch
    if not twin_open:
        print("GATE PARTIAL — the CLOSE KEY is proven: CloseDoc(<full path>) closed exactly the "
              "document it was given and nothing else moved. The COLLISION half is UNTESTED this "
              "run (the saved twin was never open), so nothing here says what happens when two "
              "open documents share a title. Re-run once swx_save_as succeeds.")
        # EXIT 3 — UNPROVEN, never 0. A caller reading the status could not tell
        # this from a full pass, and one of the two properties this gate exists
        # to prove is untested. 3 is this suite's code for "a probe could not be
        # READ", which is exactly what an untested half is.
        return 3, scratch
    print("GATE PASSED — CloseDoc(<full path>) closes exactly the document it was given, the "
          "same-titled saved document survived, and nothing else moved.")
    return 0, scratch


if __name__ == "__main__":
    try:
        sys.exit(main())
    except Refused as e:
        print("\n" + "=" * 72)
        print("GATE REFUSED TO RUN — nothing was created or closed.")
        print("  %s" % e)
        sys.exit(2)
    except Unreadable as e:
        print("\n" + "=" * 72)
        print("GATE ABORTED — a probe could not be read, so nothing below it is proven.")
        print("  %s" % e)
        print("This is 'could not ask', not 'nothing found'.")
        sys.exit(3)