Bytecode parity - boolop, comprehension, CFG passes (#7631)

* Bytecode parity - boolop, comprehension, CFG passes

- Flatten nested same-op BoolOp and add IfExpr to jump_if
- Simplify is_name_imported to module-level only
- Enable inlined comprehensions in module/class scope
- Add TweakInlinedComprehensionScopes with fast_hidden tracking
- Track fb_range in FBlockInfo for with-statement line info
- Add constant subscript folding and unary Not folding
- Add emit_return_const_no_location for implicit returns
- Use JumpNoInterrupt for ternary/except jumps
- Add STACK_USE_GUIDELINE threshold for collection building
- Reorder CFG pipeline: inline small blocks earlier, resolve
  line numbers before cold block extraction
- Add redirect_empty_unconditional_jump_targets,
  reorder_conditional_chain_and_jump_back_blocks,
  materialize_empty_conditional_exit_targets,
  duplicate_shared_jump_back_targets passes
- Add borrow deoptimization for multi-handler, named-except,
  protected conditional tail, and protected import joins
- Run folding/optimization passes twice around peephole

* Bytecode parity - slice augassign, async comp inline

- Augmented assignment with two-part slices uses BINARY_SLICE/STORE_SLICE
- Inline async comprehensions (remove await/async guards)
- Inlined comprehension cleanup jump uses JumpNoInterrupt
- Class firstlineno uses first decorator line when decorated
- Recurse into nested functions for static attribute collection
- Fold unary positive complex constants (+0.0j)
- Add deoptimize_borrow_for_folded_nonliteral_exprs pass
- Add inline_single_predecessor_artificial_expr_exit_blocks pass
- Skip shared artificial expr-stmt exit blocks in small-block inlining
- Mark folded boolop tail as folded_from_nonliteral_expr

* apply reviews

scripts/compare_bytecode.py

@@ -25,6 +25,7 @@ SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
 PROJECT_ROOT = os.path.dirname(SCRIPT_DIR)
 DIS_DUMP = os.path.join(SCRIPT_DIR, "dis_dump.py")
 DEFAULT_REPORT = os.path.join(PROJECT_ROOT, "compare_bytecode.report")
+DUMP_TIMEOUT = 600
 
 
 def find_rustpython():
@@ -69,14 +70,20 @@ def _start_one(interpreter, targets, base_dir):
     if interpreter != sys.executable:
         env["RUSTPYTHONPATH"] = base_dir
 
-    files_file = tempfile.NamedTemporaryFile(
-        mode="w", encoding="utf-8", delete=False, dir=PROJECT_ROOT
-    )
+    files_file = None
+    output_file = None
     try:
+        files_file = tempfile.NamedTemporaryFile(
+            mode="w", encoding="utf-8", delete=False, dir=PROJECT_ROOT
+        )
+        output_file = tempfile.NamedTemporaryFile(
+            mode="w", encoding="utf-8", delete=False, dir=PROJECT_ROOT
+        )
         for _, path in targets:
             files_file.write(path)
             files_file.write("\n")
         files_file.close()
+        output_file.close()
         cmd = [
             interpreter,
             DIS_DUMP,
@@ -84,6 +91,8 @@ def _start_one(interpreter, targets, base_dir):
             base_dir,
             "--files-from",
             files_file.name,
+            "--output",
+            output_file.name,
             "--progress",
             "10",
         ]
@@ -94,44 +103,162 @@ def _start_one(interpreter, targets, base_dir):
             env=env,
             cwd=PROJECT_ROOT,
         )
-        return proc, files_file.name
+        return {
+            "proc": proc,
+            "files_file": files_file.name,
+            "output_file": output_file.name,
+            "targets": targets,
+            "interpreter": interpreter,
+            "base_dir": base_dir,
+        }
     except Exception:
-        os.unlink(files_file.name)
+        for handle in (files_file, output_file):
+            if handle is None:
+                continue
+            try:
+                handle.close()
+            finally:
+                if os.path.exists(handle.name):
+                    os.unlink(handle.name)
         raise
 
 
-def _finish_one(proc, files_file):
-    """Wait for a single dis_dump.py process and return parsed JSON."""
+def _load_dump_output(output_file):
     try:
-        stdout = proc.communicate(timeout=600)[0]
-    except subprocess.TimeoutExpired:
-        proc.kill()
-        proc.communicate()
-        print("  Timeout (600s)", file=sys.stderr)
-        os.unlink(files_file)
-        return {}
-    finally:
-        if os.path.exists(files_file):
-            os.unlink(files_file)
-
-    if proc.returncode != 0:
-        print("  Warning: exited with code %d" % proc.returncode, file=sys.stderr)
-
-    content = stdout.decode(errors="replace").strip()
+        with open(output_file, encoding="utf-8") as f:
+            content = f.read().strip()
+    except OSError as e:
+        print("  Failed to read dump output: %s" % e, file=sys.stderr)
+        return None
     if not content:
         return {}
     try:
         return json.loads(content)
     except json.JSONDecodeError as e:
         print("  JSON parse error: %s" % e, file=sys.stderr)
+        return None
+
+
+def _run_sync_dump(interpreter, targets, base_dir, timeout=DUMP_TIMEOUT):
+    job = _start_one(interpreter, targets, base_dir)
+    proc = job["proc"]
+    stdout = b""
+    timed_out = False
+    try:
+        stdout = proc.communicate(timeout=timeout)[0]
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        proc.communicate()
+        print(f"  Timeout ({timeout}s)", file=sys.stderr)
+        timed_out = True
+
+    try:
+        data = _load_dump_output(job["output_file"])
+    finally:
+        for path in (job["files_file"], job["output_file"]):
+            if os.path.exists(path):
+                os.unlink(path)
+
+    if timed_out:
         return {}
+    if proc.returncode != 0:
+        print("  Warning: exited with code %d" % proc.returncode, file=sys.stderr)
+    stray = stdout.decode(errors="replace").strip()
+    if stray:
+        print("  Warning: unexpected stdout from dump helper", file=sys.stderr)
+    return data
+
+
+def _rerun_missing_targets(interpreter, targets, base_dir):
+    recovered = {}
+    failed = []
+    empty = []
+    for target in targets:
+        relpath = target[0]
+        data = _run_sync_dump(interpreter, [target], base_dir)
+        if data is None:
+            failed.append(relpath)
+            recovered[relpath] = {
+                "status": "error",
+                "error": "dump helper failed while rerunning target",
+            }
+        elif data:
+            recovered.update(data)
+        else:
+            empty.append(relpath)
+            recovered[relpath] = {
+                "status": "error",
+                "error": "dump helper produced no data while rerunning target",
+            }
+    if failed:
+        print(
+            "  Warning: rerun failed for %d file(s): %s"
+            % (len(failed), ", ".join(failed[:5])),
+            file=sys.stderr,
+        )
+    if empty:
+        print(
+            "  Warning: rerun produced no data for %d file(s): %s"
+            % (len(empty), ", ".join(empty[:5])),
+            file=sys.stderr,
+        )
+    return recovered
+
+
+def _finish_one(job, timeout=DUMP_TIMEOUT):
+    """Wait for a single dis_dump.py process and return parsed JSON."""
+    proc = job["proc"]
+    expected = {relpath for relpath, _ in job["targets"]}
+    stdout = b""
+    try:
+        stdout = proc.communicate(timeout=timeout)[0]
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        proc.communicate()
+        print(
+            f"  Timeout ({timeout}s), retrying {len(job['targets'])} file(s) serially",
+            file=sys.stderr,
+        )
+        data = None
+    else:
+        data = _load_dump_output(job["output_file"])
+    finally:
+        for path in (job["files_file"], job["output_file"]):
+            if os.path.exists(path):
+                os.unlink(path)
+
+    if proc.returncode != 0:
+        print("  Warning: exited with code %d" % proc.returncode, file=sys.stderr)
+
+    stray = stdout.decode(errors="replace").strip()
+    if stray:
+        print("  Warning: unexpected stdout from dump helper", file=sys.stderr)
+
+    if data is None:
+        return _rerun_missing_targets(
+            job["interpreter"], job["targets"], job["base_dir"]
+        )
+
+    missing = [
+        target
+        for target in job["targets"]
+        if target[0] not in data and target[0] in expected
+    ]
+    if missing:
+        print(
+            "  Re-running %d missing file(s) serially" % len(missing),
+            file=sys.stderr,
+        )
+        data.update(
+            _rerun_missing_targets(job["interpreter"], missing, job["base_dir"])
+        )
+    return data
 
 
 def start_dump(interpreter, targets, base_dir, num_workers=1):
     """Start dis_dump.py under the given interpreter, split across workers."""
     if num_workers <= 1 or len(targets) <= num_workers:
-        proc, ff = _start_one(interpreter, targets, base_dir)
-        return [(proc, ff)]
+        return [_start_one(interpreter, targets, base_dir)]
 
     chunks = [[] for _ in range(num_workers)]
     for i, t in enumerate(targets):
@@ -143,8 +270,8 @@ def start_dump(interpreter, targets, base_dir, num_workers=1):
 def finish_dump(procs):
     """Wait for all dis_dump.py processes and merge results."""
     merged = {}
-    for proc, files_file in procs:
-        merged.update(_finish_one(proc, files_file))
+    for job in procs:
+        merged.update(_finish_one(job))
     return merged