#!/usr/bin/env python3 """Benchmark bounded spreadsheet workflows through LibreOffice and WolfXL. The producer makes an immutable, quality-pack-shaped receipt bundle. It measures only the registered spreadsheet workflows; it does not make general Excel or cross-product performance claims. """ from __future__ import annotations import argparse import csv import hashlib import json import os import platform import re import shutil import signal import subprocess import sys import threading import time import tempfile import uuid from concurrent.futures import ThreadPoolExecutor from threading import BoundedSemaphore import xml.etree.ElementTree as ET from dataclasses import asdict, dataclass from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from io import BytesIO from pathlib import Path from typing import Any, Iterable, Mapping, Sequence from zipfile import ZipFile ROOT = Path(__file__).resolve().parents[1] _ABSOLUTE_PATH_TOKEN_RE = re.compile( r"""(?])(?:/[^\s"'<>]+|[A-Za-z]:[\\/][^\s"'<>]+)""" ) _FILE_URI_RE = re.compile(r"""file://(?P/[^\s"'<>]+)""") _PATH_TRAILING_PUNCTUATION = ".,;:)]}" SCHEMA_VERSION = 1 REGISTRY_VERSION = "libreoffice-vs-wolfxl-v1" BUNDLE_FILES = ( "manifest.json", "raw.jsonl", "failures.jsonl", "summary.json", "status.json", "bundle.sha256.json", ) HARNESS_BUNDLE_PATH = "harness/benchmark_libreoffice_vs_wolfxl.py" DEFAULT_ROWS = 10 DEFAULT_COLS = 4 DEFAULT_ITERATIONS = 1 DEFAULT_WARMUPS = 1 DEFAULT_TIMEOUT_SECONDS = 30.0 OUTPUT_CLIP_BYTES = 16_384 SHA256_RE = re.compile(r"^[0-9a-f]{64}$") SAFE_RUN_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") @dataclass(frozen=True) class Scenario: """One user-visible family of comparable spreadsheet work.""" name: str description: str tasks: tuple[str, ...] modes: tuple[str, ...] TASK_REGISTRY = { "xlsx_to_csv_extract": { "description": "Headless XLSX load followed by CSV value extraction.", "libreoffice": "supported", "wolfxl": "supported", }, "surgical_edit_save": { "description": "Change one bounded cell and save an XLSX preserving the workbook semantics.", "libreoffice": "unsupported", "wolfxl": "supported", }, "recalc_extract": { "description": "Recalculate a bounded formula workbook and extract saved answers.", "libreoffice": "supported", "wolfxl": "supported", }, "roundtrip_preservation": { "description": "XLSX open/save round trip with semantic preservation checks.", "libreoffice": "supported", "wolfxl": "supported", }, "xlsx_to_pdf_export": { "description": "Headless XLSX-to-PDF export with PDF artifact validation.", "libreoffice": "supported", "wolfxl": "supported", }, } SCENARIOS = ( Scenario( "cold-start", "Fresh-process XLSX-to-CSV extraction latency for a bounded workbook.", ("xlsx_to_csv_extract",), ("cold",), ), Scenario( "warm-latency", "Repeated XLSX-to-CSV extraction latency after explicit warmups.", ("xlsx_to_csv_extract",), ("warm",), ), Scenario( "footprint-rss", "Runtime footprint and a fresh process-tree peak-RSS observation.", ("footprint_rss",), ("footprint",), ), Scenario( "agent-tasks", "Bounded agent workflows with explicit unsupported engine/task pairs.", ( "xlsx_to_csv_extract", "surgical_edit_save", "recalc_extract", "roundtrip_preservation", "xlsx_to_pdf_export", ), ("cold",), ), Scenario( "scale-frontier", "Generated small and optional explicit large scale workbooks.", ("xlsx_to_csv_extract", "roundtrip_preservation"), ("cold",), ), ) SCENARIO_BY_NAME = {item.name: item for item in SCENARIOS} HARD_FAILURE_STATUSES = frozenset({"failed", "timed_out", "semantic_mismatch"}) NON_SUCCESS_STATUSES = frozenset({"failed", "timed_out", "semantic_mismatch", "unsupported", "unavailable"}) class BenchmarkError(RuntimeError): """A bounded benchmark setup or receipt failure.""" def _utc_now() -> str: return datetime.now(timezone.utc).isoformat(timespec="microseconds").replace("+00:00", "Z") def _canonical_json(value: Any) -> str: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) def _sha256_bytes(payload: bytes) -> str: return hashlib.sha256(payload).hexdigest() def _sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _json_write(path: Path, value: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps(_sanitize_value(value), sort_keys=True, indent=2) + "\n", encoding="utf-8", ) def _jsonl_write(path: Path, rows: Iterable[Mapping[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8") as handle: for row in rows: handle.write(_canonical_json(_sanitize_value(dict(row))) + "\n") def _jsonl_read(path: Path) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for line in path.read_text(encoding="utf-8").splitlines(): if line.strip(): value = json.loads(line) if not isinstance(value, dict): raise ValueError(f"{path}: JSONL rows must be objects") rows.append(value) return rows def _display_path(path: str | Path | None) -> str | None: """Preserve an actionable path while keeping the local home prefix private.""" if path is None: return None value = Path(path).expanduser() try: home_relative = value.resolve(strict=False).relative_to(Path.home().resolve()) except ValueError: return str(value) return "~" if not home_relative.parts else f"~/{home_relative.as_posix()}" def _sanitizer_roots() -> list[tuple[str, str]]: roots: list[tuple[str, str]] = [] for path, label in ( (ROOT, ""), (Path.home(), ""), (Path(tempfile.gettempdir()), ""), ): for candidate in (path.expanduser(), path.expanduser().resolve(strict=False)): value = str(candidate).rstrip("/\\") if value and all(value != existing for existing, _ in roots): roots.append((value, label)) roots.sort(key=lambda item: len(item[0]), reverse=True) return roots def _sanitize_path_token(value: str, *, roots: list[tuple[str, str]]) -> str: trailing = "" while value and value[-1] in _PATH_TRAILING_PUNCTUATION: trailing = value[-1] + trailing value = value[:-1] normalized = value.replace("\\", "/") for root, label in roots: normalized_root = root.replace("\\", "/") if normalized == normalized_root or normalized.startswith(f"{normalized_root}/"): suffix = normalized[len(normalized_root) :].lstrip("/") return f"{label}/{suffix}{trailing}" if suffix else f"{label}{trailing}" parts = normalized.split("/") if normalized.startswith(("/Users/", "/home/")) and len(parts) > 3: suffix = "/".join(parts[3:]) return f"/{suffix}{trailing}" if suffix else f"{trailing}" if normalized.startswith(("/private/", "/tmp/", "/var/folders/")): name = Path(normalized).name return f"/{name}{trailing}" if name else f"{trailing}" if re.match(r"^[A-Za-z]:/", normalized): drive = normalized[0].upper() suffix = normalized[3:] return f"/{suffix}{trailing}" if suffix else f"{trailing}" if normalized.startswith("/"): suffix = normalized.lstrip("/") return f"/{suffix}{trailing}" if suffix else f"{trailing}" return f"{normalized}{trailing}" def _sanitize_text(value: str) -> str: roots = _sanitizer_roots() text = _FILE_URI_RE.sub( lambda match: "" + _sanitize_path_token(match.group("path"), roots=roots), value, ) return _ABSOLUTE_PATH_TOKEN_RE.sub( lambda match: _sanitize_path_token(match.group(0), roots=roots), text, ) def _sanitize_value(value: Any) -> Any: if isinstance(value, str): return _sanitize_text(value) if isinstance(value, list): return [_sanitize_value(item) for item in value] if isinstance(value, tuple): return [_sanitize_value(item) for item in value] if isinstance(value, Mapping): return {str(key): _sanitize_value(item) for key, item in value.items()} return value def _display_command(command: Sequence[str]) -> list[str]: return [_sanitize_text(str(part)) for part in command] def _clip_output(value: str) -> dict[str, Any]: encoded = value.encode("utf-8", errors="replace") clipped = len(encoded) > OUTPUT_CLIP_BYTES payload = encoded[:OUTPUT_CLIP_BYTES] return { "text": _sanitize_text(payload.decode("utf-8", errors="replace")), "truncated": clipped, "byte_count": len(encoded), } def _portable_path(path: Path, root: Path) -> str: try: return path.relative_to(root).as_posix() except ValueError as error: raise BenchmarkError(f"bundle artifact is outside the bundle root: {path}") from error def _safe_run_id(value: str) -> str: if not SAFE_RUN_ID_RE.fullmatch(value) or value in {".", ".."}: raise ValueError("run ID must be a safe 1-128 character filename") return value def _source_identity() -> dict[str, Any]: receipts: list[dict[str, Any]] = [] def git(*args: str) -> str | None: command = ["git", *args] started_at_utc = _utc_now() try: completed = subprocess.run( command, cwd=ROOT, capture_output=True, text=True, timeout=15, check=False, ) receipt = { "command": command, "cwd": str(ROOT), "terminal_status": "completed", "returncode": completed.returncode, "stdout": completed.stdout, "stderr": completed.stderr, "started_at_utc": started_at_utc, "completed_at_utc": _utc_now(), } except (OSError, subprocess.TimeoutExpired) as error: receipt = { "command": command, "cwd": str(ROOT), "terminal_status": "failed", "returncode": None, "stdout": "", "stderr": f"{type(error).__name__}: {error}", "started_at_utc": started_at_utc, "completed_at_utc": _utc_now(), } receipts.append(receipt) return receipt["stdout"].strip() if receipt["returncode"] == 0 else None source_hash = git("rev-parse", "HEAD") source_tree = git("rev-parse", "HEAD^{tree}") dirty = git("status", "--porcelain") return { "source_hash": source_hash or "unknown", "source_tree": source_tree, "source_dirty": dirty != "" if dirty is not None else None, "source_command_receipts": receipts, } def _directory_size(path: Path) -> int | None: try: if path.is_file(): return path.stat().st_size total = 0 for entry in path.rglob("*"): if entry.is_file() and not entry.is_symlink(): total += entry.stat().st_size return total except OSError: return None def _file_hash_or_none(path: Path) -> str | None: try: return _sha256_file(path) if path.is_file() else None except OSError: return None def _runtime_json_for(path: Path) -> Path | None: """Find the nearest optional runtime.json without assuming any runtime layout.""" current = path.resolve(strict=False) for parent in (current.parent, *current.parents): candidate = parent / "runtime.json" if candidate.is_file(): return candidate if parent == parent.parent: break return None def _runtime_manifest(runtime_json: Path | None) -> tuple[str | None, str | None, dict[str, Any]]: if runtime_json is None: return None, None, {"status": "missing"} digest = _file_hash_or_none(runtime_json) try: payload = json.loads(runtime_json.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: return _display_path(runtime_json), digest, {"status": "unreadable", "error": type(error).__name__} if not isinstance(payload, dict): return _display_path(runtime_json), digest, {"status": "invalid", "reason": "JSON root is not an object"} return _display_path(runtime_json), digest, {"status": "present", "fields": _sanitize_value(payload)} def _libreoffice_install_root(executable: Path, runtime_json: Path | None) -> Path: """Choose the bounded installation tree that owns a LibreOffice executable.""" if runtime_json is not None: native_root = runtime_json.parent / "dependencies/native/libreoffice-headless" app_root = native_root / "libreoffice/LibreOfficeDev.app" if app_root.is_dir(): return app_root if native_root.is_dir(): return native_root app_root = next((parent for parent in executable.parents if parent.suffix == ".app"), None) if app_root is not None: return app_root suite_root = next( (parent for parent in executable.parents if "libreoffice" in parent.name.lower()), None, ) return suite_root or executable.parent def _empty_rss(reason: str) -> dict[str, Any]: return {"status": "unavailable", "reason": reason, "peak_process_tree_rss_bytes": None} def _process_tree_rss_bytes(root_pid: int) -> int | None: """Sum an observed POSIX process tree RSS sample, returning None if unsupported.""" if os.name == "nt": return None try: completed = subprocess.run( ["ps", "-axo", "pid=,ppid=,rss="], capture_output=True, text=True, timeout=2, check=False, ) except (OSError, subprocess.TimeoutExpired): return None if completed.returncode != 0: return None table: dict[int, tuple[int, int]] = {} for line in completed.stdout.splitlines(): fields = line.split() if len(fields) != 3: continue try: pid, parent, rss_kib = (int(field) for field in fields) except ValueError: continue table[pid] = (parent, rss_kib) if root_pid not in table: return None pending = [root_pid] included: set[int] = set() while pending: pid = pending.pop() if pid in included: continue included.add(pid) pending.extend(child for child, (parent, _rss) in table.items() if parent == pid) return sum(table[pid][1] for pid in included) * 1024 def _terminate_process_group(process: subprocess.Popen[str]) -> None: if process.poll() is not None: return try: if os.name == "nt": process.terminate() else: os.killpg(process.pid, signal.SIGTERM) process.wait(timeout=2) return except (OSError, subprocess.TimeoutExpired): pass try: if os.name == "nt": process.kill() else: os.killpg(process.pid, signal.SIGKILL) except OSError: return def run_process( command: Sequence[str], *, timeout_seconds: float, input_text: str | None = None, cwd: Path | None = None, env: Mapping[str, str] | None = None, ) -> dict[str, Any]: """Run one bounded command with captured streams and process-tree RSS samples.""" started_utc = _utc_now() started = time.monotonic() display_command = _display_command(command) try: process = subprocess.Popen( list(command), cwd=cwd, env=dict(env) if env is not None else None, stdin=subprocess.PIPE if input_text is not None else None, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=os.name != "nt", ) except OSError as error: completed_utc = _utc_now() return { "command": display_command, "terminal_status": "failed", "exit_code": None, "started_at_utc": started_utc, "completed_at_utc": completed_utc, "wall_time_seconds": time.monotonic() - started, "stdout": _clip_output(""), "stderr": _clip_output(""), "error": _sanitize_text(f"{type(error).__name__}: {error}"), "rss_measurement": _empty_rss("process_start_failed"), } sampled_rss: list[int] = [] sampling_stop = threading.Event() def sample() -> None: while not sampling_stop.wait(0.05): observed = _process_tree_rss_bytes(process.pid) if observed is not None: sampled_rss.append(observed) sampler = threading.Thread(target=sample, daemon=True) sampler.start() timed_out = False try: stdout, stderr = process.communicate(input=input_text, timeout=timeout_seconds) except subprocess.TimeoutExpired: timed_out = True _terminate_process_group(process) stdout, stderr = process.communicate() finally: sampling_stop.set() sampler.join(timeout=1) final_rss = _process_tree_rss_bytes(process.pid) if final_rss is not None: sampled_rss.append(final_rss) completed_utc = _utc_now() terminal_status = "timed_out" if timed_out else "ok" if process.returncode == 0 else "failed" rss = ( { "status": "observed", "method": "sampled_posix_process_tree_rss", "sample_count": len(sampled_rss), "peak_process_tree_rss_bytes": max(sampled_rss), } if sampled_rss else _empty_rss("process_tree_sampling_unavailable") ) return { "command": display_command, "terminal_status": terminal_status, "exit_code": process.returncode, "started_at_utc": started_utc, "completed_at_utc": completed_utc, "wall_time_seconds": time.monotonic() - started, "stdout": _clip_output(stdout), "stderr": _clip_output(stderr), "error": None, "rss_measurement": rss, } def _number_token(value: str) -> str: try: number = Decimal(value) except InvalidOperation: return f"string:{value}" if not number.is_finite(): return f"string:{value}" normalized = number.normalize() if normalized == normalized.to_integral(): normalized = normalized.quantize(Decimal(1)) return f"number:{normalized}" def _xml_local_name(tag: str) -> str: return tag.rsplit("}", 1)[-1] def _first_child(element: ET.Element, name: str) -> ET.Element | None: return next((child for child in element if _xml_local_name(child.tag) == name), None) def _shared_strings(archive: ZipFile) -> list[str]: try: data = archive.read("xl/sharedStrings.xml") except KeyError: return [] values: list[str] = [] for element in ET.fromstring(data): if _xml_local_name(element.tag) != "si": continue values.append("".join(node.text or "" for node in element.iter() if _xml_local_name(node.tag) == "t")) return values def xlsx_semantics(path: Path) -> dict[str, Any]: """Return a streaming, format-independent spreadsheet value/formula digest.""" digest = hashlib.sha256() cell_count = 0 surgical_edit_marker = False formula_count = 0 sheet_count = 0 with ZipFile(path) as archive: strings = _shared_strings(archive) sheet_names = sorted( name for name in archive.namelist() if name.startswith("xl/worksheets/") and name.endswith(".xml") ) for sheet_path in sheet_names: sheet_count += 1 digest.update(_canonical_json(["sheet", sheet_path]).encode() + b"\n") for _event, element in ET.iterparse(BytesIO(archive.read(sheet_path)), events=("end",)): if _xml_local_name(element.tag) != "c": continue reference = element.attrib.get("r", "") cell_type = element.attrib.get("t") formula = _first_child(element, "f") raw_value = _first_child(element, "v") inline = _first_child(element, "is") if cell_type == "s" and raw_value is not None and raw_value.text is not None: try: token = f"string:{strings[int(raw_value.text)]}" except (IndexError, ValueError): token = "invalid_shared_string" elif cell_type == "inlineStr" and inline is not None: token = "string:" + "".join( node.text or "" for node in inline.iter() if _xml_local_name(node.tag) == "t" ) elif cell_type == "b" and raw_value is not None: token = f"bool:{raw_value.text == '1'}".lower() elif raw_value is None or raw_value.text is None: token = "blank" elif cell_type == "str": token = f"string:{raw_value.text}" else: token = _number_token(raw_value.text) surgical_edit_marker = surgical_edit_marker or ( reference == "B2" and token == "string:WOLFXL-SURGICAL-EDIT" ) record = [reference, formula.text if formula is not None else None, token] digest.update(_canonical_json(record).encode() + b"\n") cell_count += 1 formula_count += formula is not None element.clear() return { "semantic_digest": digest.hexdigest(), "cell_count": cell_count, "formula_count": formula_count, "sheet_count": sheet_count, "surgical_edit_marker": surgical_edit_marker, } def csv_value_semantics(path: Path) -> dict[str, Any]: digest = hashlib.sha256() cell_count = 0 row_count = 0 with path.open("r", encoding="utf-8-sig", newline="") as handle: for row_count, row in enumerate(csv.reader(handle), 1): for column, value in enumerate(row, 1): digest.update(_canonical_json([row_count, column, _number_token(value)]).encode() + b"\n") cell_count += 1 return {"semantic_digest": digest.hexdigest(), "cell_count": cell_count, "row_count": row_count} def pdf_semantics(path: Path) -> dict[str, Any]: payload = path.read_bytes() return { "pdf_signature": payload.startswith(b"%PDF-"), "semantic_digest": _sha256_bytes(payload), "cell_count": 0, "byte_count": len(payload), } def _basic_xlsx_value_semantics(path: Path) -> dict[str, Any]: digest = hashlib.sha256() cell_count = 0 row_count = 0 with ZipFile(path) as archive: strings = _shared_strings(archive) sheet_path = "xl/worksheets/sheet1.xml" for _event, element in ET.iterparse(BytesIO(archive.read(sheet_path)), events=("end",)): if _xml_local_name(element.tag) != "c": continue reference = element.attrib.get("r", "") match = re.match(r"([A-Z]+)([0-9]+)$", reference) if match is None: element.clear() continue letters, row_text = match.groups() column = 0 for letter in letters: column = column * 26 + ord(letter) - ord("A") + 1 row = int(row_text) row_count = max(row_count, row) raw_value = _first_child(element, "v") cell_type = element.attrib.get("t") if cell_type == "s" and raw_value is not None and raw_value.text is not None: token = f"string:{strings[int(raw_value.text)]}" elif raw_value is None or raw_value.text is None: token = "blank" else: token = _number_token(raw_value.text) digest.update(_canonical_json([row, column, token]).encode() + b"\n") cell_count += 1 element.clear() return {"semantic_digest": digest.hexdigest(), "cell_count": cell_count, "row_count": row_count} def _semantic_check(task: str, expected: Mapping[str, Any], observed: Mapping[str, Any]) -> dict[str, Any]: if task == "surgical_edit_save": compared = ("surgical_edit_marker",) elif task == "xlsx_to_pdf_export": compared = ("pdf_signature",) else: compared = ("semantic_digest", "cell_count") mismatch = { key: {"expected": expected.get(key), "observed": observed.get(key)} for key in compared if expected.get(key) != observed.get(key) } return { "status": "match" if not mismatch else "mismatch", "expected": dict(expected), "observed": dict(observed), "mismatch": mismatch, "task": task, } def _output_identity(path: Path, root: Path) -> dict[str, Any]: if not path.is_file(): return {"path": _portable_path(path, root), "exists": False, "size_bytes": None, "sha256": None} return { "path": _portable_path(path, root), "exists": True, "size_bytes": path.stat().st_size, "sha256": _sha256_file(path), } def _profile_path(work_dir: Path) -> Path: profile = work_dir / "libreoffice-profile" profile.mkdir(parents=True, exist_ok=True) return profile def _soffice_command( soffice: Path, *, profile: Path, convert_to: str | None = None, out_dir: Path | None = None, input_path: Path | None = None, ) -> list[str]: command = [ str(soffice), "--headless", "--nologo", "--nodefault", "--nolockcheck", "--nofirststartwizard", "--norestore", f"-env:UserInstallation={profile.resolve().as_uri()}", ] if convert_to is not None: assert out_dir is not None and input_path is not None command.extend(["--convert-to", convert_to, "--outdir", str(out_dir), str(input_path)]) return command def libreoffice_runtime_identity(soffice: Path, *, timeout_seconds: float, scratch: Path) -> tuple[dict[str, Any], dict[str, Any]]: requested = soffice.expanduser() resolved = requested.resolve(strict=False) runtime_json = _runtime_json_for(resolved) runtime_json_path, runtime_json_sha256, runtime_fields = _runtime_manifest(runtime_json) profile = _profile_path(scratch / f"runtime-identity-{uuid.uuid4().hex}") version_process = run_process( _soffice_command(requested, profile=profile) + ["--version"], timeout_seconds=timeout_seconds ) version_stdout = version_process["stdout"]["text"] install_root = _libreoffice_install_root(resolved, runtime_json) identity = { "requested_path": _display_path(requested), "resolved_path": _display_path(resolved), "wrapper_sha256": _file_hash_or_none(requested), "executable_path": _display_path(resolved), "executable_sha256": _file_hash_or_none(resolved), "version_stdout": version_stdout, "runtime_json_path": runtime_json_path, "runtime_json_sha256": runtime_json_sha256, "runtime_manifest_fields": runtime_fields, "platform": sys.platform, "arch": platform.machine(), "observed_at_utc": _utc_now(), "runtime_footprint": { "executable_bytes": _directory_size(resolved), "installation_root": _display_path(install_root), "installation_bytes": _directory_size(install_root), }, "availability": "available" if version_process["terminal_status"] == "ok" else "unavailable", } return identity, version_process def _wolfxl_identity() -> dict[str, Any]: import wolfxl module_path = Path(getattr(wolfxl, "__file__", "")).resolve() extension_path: Path | None = None try: import wolfxl._rust as rust raw_extension_path = getattr(rust, "__file__", None) extension_path = Path(raw_extension_path).resolve() if raw_extension_path else None except ImportError: extension_path = None executable = extension_path or module_path runtime_root = module_path.parent return { "requested_path": _display_path(sys.executable), "resolved_path": _display_path(Path(sys.executable).resolve()), "wrapper_sha256": _file_hash_or_none(Path(sys.executable).resolve()), "executable_path": _display_path(executable), "executable_sha256": _file_hash_or_none(executable), "version_stdout": f"wolfxl {getattr(wolfxl, '__version__', 'unknown')}", "runtime_json_path": None, "runtime_json_sha256": None, "runtime_manifest_fields": {"status": "missing", "reason": "WolfXL Python runtime has no runtime.json contract"}, "platform": sys.platform, "arch": platform.machine(), "observed_at_utc": _utc_now(), "runtime_footprint": { "executable_bytes": _directory_size(executable), "installation_root": _display_path(runtime_root), "installation_bytes": _directory_size(runtime_root), }, "availability": "available", } def _close_workbook(workbook: Any) -> None: close = getattr(workbook, "close", None) if callable(close): close() def _worker_fixture(request: Mapping[str, Any]) -> dict[str, Any]: import wolfxl path = Path(str(request["path"])) rows = int(request["rows"]) cols = int(request["cols"]) kind = str(request["kind"]) path.parent.mkdir(parents=True, exist_ok=True) workbook = wolfxl.Workbook() worksheet = workbook.active if worksheet is None: raise RuntimeError("new WolfXL workbook has no active worksheet") worksheet.title = "Sheet" if kind == "basic": for row in range(1, rows + 1): for column in range(1, cols + 1): worksheet.cell(row=row, column=column, value=(row * 1_000_000) + column) elif kind == "formula": worksheet["A1"] = 7 worksheet["B1"] = 5 worksheet["C1"] = "=A1+B1" worksheet["D1"] = "=SUM(A1:C1)" workbook.calculate() else: raise ValueError(f"unknown fixture kind: {kind}") workbook.save(path) _close_workbook(workbook) return {"path": str(path), "xlsx_semantics": xlsx_semantics(path)} def _worker_execute(request: Mapping[str, Any]) -> dict[str, Any]: import wolfxl task = str(request["task"]) input_path = Path(str(request["input_path"])) output_path = Path(str(request["output_path"])) output_path.parent.mkdir(parents=True, exist_ok=True) if task == "xlsx_to_csv_extract": workbook = wolfxl.load_workbook( str(input_path), read_only=True, data_only=True, ) try: worksheet = workbook.active if worksheet is None: raise RuntimeError("loaded WolfXL workbook has no active worksheet") with output_path.open("w", encoding="utf-8", newline="") as handle: csv.writer(handle, lineterminator="\n").writerows( worksheet.iter_rows(values_only=True) ) finally: _close_workbook(workbook) return {"semantic": None, "output_path": str(output_path)} if task == "xlsx_to_pdf_export": options = wolfxl.ConversionOptions(destination_format="pdf") wolfxl.convert(str(input_path), str(output_path), options=options) return {"semantic": None, "output_path": str(output_path)} modify = task in {"surgical_edit_save", "recalc_extract"} workbook = wolfxl.load_workbook( str(input_path), modify=modify, data_only=False, read_only=task == "roundtrip_preservation", ) try: worksheet = workbook.active if worksheet is None: raise RuntimeError("loaded WolfXL workbook has no active worksheet") if task == "surgical_edit_save": worksheet["B2"] = "WOLFXL-SURGICAL-EDIT" elif task == "recalc_extract": workbook.calculate() elif task != "roundtrip_preservation": raise ValueError(f"unknown worker task: {task}") workbook.save(output_path) finally: _close_workbook(workbook) return {"semantic": None, "output_path": str(output_path)} def worker(operation: str) -> int: try: request = json.loads(sys.stdin.read() or "{}") if not isinstance(request, dict): raise ValueError("worker request must be a JSON object") if operation == "identity": response = _wolfxl_identity() elif operation == "fixture": response = _worker_fixture(request) elif operation == "execute": response = _worker_execute(request) else: raise ValueError(f"unknown worker operation: {operation}") print(_canonical_json(response)) return 0 except Exception as error: # Worker error details are captured by the parent receipt. print(_canonical_json({"error": f"{type(error).__name__}: {error}"})) return 1 def _worker_command(python: str, operation: str) -> list[str]: return [python, str(Path(__file__).resolve()), "worker", operation] def _run_worker( python: str, operation: str, request: Mapping[str, Any], *, timeout_seconds: float ) -> tuple[dict[str, Any], dict[str, Any] | None]: process = run_process( _worker_command(python, operation), timeout_seconds=timeout_seconds, input_text=_canonical_json(dict(request)), cwd=ROOT, ) if process["terminal_status"] != "ok": return process, None try: value = json.loads(process["stdout"]["text"]) except json.JSONDecodeError: return {**process, "terminal_status": "failed", "error": "worker stdout was not JSON"}, None if not isinstance(value, dict) or "error" in value: return {**process, "terminal_status": "failed", "error": "worker returned an error payload"}, None return process, value def _libreoffice_recalc_profile(profile: Path) -> None: user = profile / "user" user.mkdir(parents=True, exist_ok=True) (user / "registrymodifications.xcu").write_text( """ 0 """, encoding="utf-8", ) def _run_libreoffice_task( task: str, *, soffice: Path, fixture: Path, work_dir: Path, timeout_seconds: float, ) -> tuple[dict[str, Any], dict[str, Any] | None, Path | None]: if task == "surgical_edit_save": return ( { "command": [], "terminal_status": "unsupported", "exit_code": None, "started_at_utc": _utc_now(), "completed_at_utc": _utc_now(), "wall_time_seconds": 0.0, "stdout": _clip_output(""), "stderr": _clip_output(""), "error": "LibreOffice headless CLI has no bounded surgical cell-edit command without an external macro or UNO program", "rss_measurement": _empty_rss("unsupported_task"), }, None, None, ) output_dir = work_dir / "output" output_dir.mkdir(parents=True, exist_ok=True) profile = _profile_path(work_dir) if task == "recalc_extract": _libreoffice_recalc_profile(profile) if task == "xlsx_to_csv_extract": output_path = output_dir / f"{fixture.stem}.csv" convert_to = "csv" elif task in {"recalc_extract", "roundtrip_preservation"}: output_path = output_dir / fixture.name convert_to = "xlsx" elif task == "xlsx_to_pdf_export": output_path = output_dir / f"{fixture.stem}.pdf" convert_to = "pdf" else: raise ValueError(f"unknown LibreOffice task: {task}") command = _soffice_command( soffice, profile=profile, convert_to=convert_to, out_dir=output_dir, input_path=fixture, ) process = run_process(command, timeout_seconds=timeout_seconds, cwd=ROOT) if process["terminal_status"] != "ok": return process, None, output_path if not output_path.is_file(): return { **process, "terminal_status": "failed", "error": "LibreOffice completed without the expected output artifact", }, None, output_path semantics = _task_output_semantics(task, output_path) return process, semantics, output_path def _task_output_semantics(task: str, output_path: Path) -> dict[str, Any]: if task == "xlsx_to_csv_extract": return csv_value_semantics(output_path) if task == "xlsx_to_pdf_export": return pdf_semantics(output_path) return xlsx_semantics(output_path) def scenario_registry() -> tuple[Scenario, ...]: return SCENARIOS def cache_state_for(mode: str, engine: str) -> str: if mode == "cold": return "cold-fresh-process" if mode == "warm" and engine == "wolfxl": return "warm-filesystem-cache-fresh-library-subprocess" if mode == "warm" and engine == "libreoffice": return "warm-filesystem-cache-fresh-headless-process" if mode == "footprint": return "not-applicable-runtime-identity" raise ValueError(f"unknown cache state mode/engine: {mode}/{engine}") def _task_fixture_kind(task: str) -> str: return "formula" if task == "recalc_extract" else "basic" def _bounded_worker( args: argparse.Namespace, operation: str, request: Mapping[str, Any], ) -> tuple[dict[str, Any], dict[str, Any] | None]: semaphore = getattr(args, "_wolfxl_worker_slots", None) if semaphore is None: return _run_worker(args.python, operation, request, timeout_seconds=args.timeout_seconds) with semaphore: return _run_worker(args.python, operation, request, timeout_seconds=args.timeout_seconds) def _expected_semantics(task: str, fixture: Path) -> dict[str, Any]: if task == "xlsx_to_csv_extract": return _basic_xlsx_value_semantics(fixture) if task == "surgical_edit_save": return {"surgical_edit_marker": True} if task == "xlsx_to_pdf_export": return {"pdf_signature": True} return xlsx_semantics(fixture) def _artifact_row_path(bundle_root: Path, row_key: str) -> Path: return bundle_root / "artifacts" / f"{row_key}.json" def _row_metadata( *, source: Mapping[str, Any], built_hash: str | None, build_mode: str, scenario: str, iterations: int, warmups: int, timeout_seconds: float, cache_state: str, artifact_path: str, command: Sequence[str], started_at_utc: str, completed_at_utc: str, parallel_jobs: int, wolfxl_worker_jobs: int, ) -> dict[str, Any]: return { "source_hash": source["source_hash"], "source_tree": source.get("source_tree"), "source_dirty": source["source_dirty"], "built_hash_or_baked_constant": built_hash or "unavailable", "build_mode": build_mode, "scenario": scenario, "iteration_count": {"measured": iterations, "warmups": warmups}, "timeout_seconds": timeout_seconds, "cache_state": cache_state, "artifact_path": artifact_path, "command": list(command), "started_at_utc": started_at_utc, "completed_at_utc": completed_at_utc, "concurrency": { "parallel_jobs": parallel_jobs, "wolfxl_worker_jobs": wolfxl_worker_jobs, }, } def _unsupported_row( *, root: Path, row_key: str, engine: str, task: str, scenario: str, mode: str, phase: str, iteration: int, source: Mapping[str, Any], engine_identity: Mapping[str, Any], args: argparse.Namespace, reason: str, ) -> dict[str, Any]: started = _utc_now() artifact_path = _artifact_row_path(root, row_key) portable_artifact = _portable_path(artifact_path, root) process = { "command": [], "terminal_status": "unsupported", "exit_code": None, "started_at_utc": started, "completed_at_utc": _utc_now(), "wall_time_seconds": 0.0, "stdout": _clip_output(""), "stderr": _clip_output(""), "error": reason, "rss_measurement": _empty_rss("unsupported_task"), } row = { "schema_version": SCHEMA_VERSION, "row_key": row_key, "engine": engine, "task": task, "scenario": scenario, "mode": mode, "phase": phase, "iteration": iteration, "terminal_status": "unsupported", "reason": reason, "process": process, "metrics": {}, "output": None, "semantic_check": {"status": "not_applicable", "task": task}, "engine_identity": dict(engine_identity), } row["metadata"] = _row_metadata( source=source, built_hash=engine_identity.get("executable_sha256"), build_mode=args.build_mode, scenario=scenario, iterations=args.iterations, warmups=args.warmups, timeout_seconds=args.timeout_seconds, cache_state=cache_state_for(mode, engine), artifact_path=portable_artifact, command=[], started_at_utc=process["started_at_utc"], completed_at_utc=process["completed_at_utc"], parallel_jobs=getattr(args, "parallel_jobs", 1), wolfxl_worker_jobs=getattr(args, "wolfxl_worker_jobs", 1), ) _json_write(artifact_path, row) row["artifact_sha256"] = _sha256_file(artifact_path) return row def _observe_task( *, root: Path, row_key: str, engine: str, task: str, scenario: str, mode: str, phase: str, iteration: int, fixture: Path, source: Mapping[str, Any], engine_identity: Mapping[str, Any], args: argparse.Namespace, ) -> dict[str, Any]: if engine_identity.get("availability") != "available": return _unsupported_row( root=root, row_key=row_key, engine=engine, task=task, scenario=scenario, mode=mode, phase=phase, iteration=iteration, source=source, engine_identity=engine_identity, args=args, reason="engine runtime identity could not execute its bounded version probe", ) if (engine, task) == ("libreoffice", "surgical_edit_save"): reason = ( "LibreOffice headless CLI has no bounded surgical cell-edit command " "without an external macro or UNO program" ) return _unsupported_row( root=root, row_key=row_key, engine=engine, task=task, scenario=scenario, mode=mode, phase=phase, iteration=iteration, source=source, engine_identity=engine_identity, args=args, reason=reason, ) work_dir = root / "work" / row_key output_path: Path | None = None worker_payload: dict[str, Any] | None = None if engine == "wolfxl": suffix = { "xlsx_to_csv_extract": ".csv", "xlsx_to_pdf_export": ".pdf", }.get(task, ".xlsx") output_path = work_dir / f"output{suffix}" process, worker_payload = _bounded_worker( args, "execute", {"task": task, "input_path": str(fixture), "output_path": str(output_path)}, ) observed = ( _task_output_semantics(task, output_path) if worker_payload is not None and output_path.is_file() else None ) elif engine == "libreoffice": process, observed, output_path = _run_libreoffice_task( task, soffice=args.soffice, fixture=fixture, work_dir=work_dir, timeout_seconds=args.timeout_seconds, ) else: raise ValueError(f"unknown engine: {engine}") expected = _expected_semantics(task, fixture) if process["terminal_status"] == "ok" and isinstance(observed, Mapping): semantic_check = _semantic_check(task, expected, observed) terminal_status = "ok" if semantic_check["status"] == "match" else "semantic_mismatch" else: semantic_check = {"status": "not_checked", "task": task, "expected": expected, "observed": observed} terminal_status = process["terminal_status"] artifact_path = _artifact_row_path(root, row_key) portable_artifact = _portable_path(artifact_path, root) output = _output_identity(output_path, root) if output_path is not None else None row = { "schema_version": SCHEMA_VERSION, "row_key": row_key, "engine": engine, "task": task, "scenario": scenario, "mode": mode, "phase": phase, "iteration": iteration, "terminal_status": terminal_status, "fixture": _portable_path(fixture, root), "reason": process.get("error"), "process": process, "metrics": { "wall_time_seconds": process["wall_time_seconds"], "peak_process_tree_rss_bytes": process["rss_measurement"].get("peak_process_tree_rss_bytes"), "rss_measurement": process["rss_measurement"], }, "output": output, "semantic_check": semantic_check, "engine_identity": dict(engine_identity), } row["metadata"] = _row_metadata( source=source, built_hash=engine_identity.get("executable_sha256"), build_mode=args.build_mode, scenario=scenario, iterations=args.iterations, warmups=args.warmups, timeout_seconds=args.timeout_seconds, cache_state=cache_state_for(mode, engine), artifact_path=portable_artifact, command=process["command"], started_at_utc=process["started_at_utc"], completed_at_utc=process["completed_at_utc"], parallel_jobs=getattr(args, "parallel_jobs", 1), wolfxl_worker_jobs=getattr(args, "wolfxl_worker_jobs", 1), ) _json_write(artifact_path, row) row["artifact_sha256"] = _sha256_file(artifact_path) return row def _observe_footprint( *, root: Path, row_key: str, engine: str, source: Mapping[str, Any], engine_identity: Mapping[str, Any], version_process: Mapping[str, Any], args: argparse.Namespace, ) -> dict[str, Any]: artifact_path = _artifact_row_path(root, row_key) portable_artifact = _portable_path(artifact_path, root) terminal_status = "ok" if engine_identity.get("availability") == "available" else "unavailable" footprint = engine_identity.get("runtime_footprint", {}) row = { "schema_version": SCHEMA_VERSION, "row_key": row_key, "engine": engine, "task": "footprint_rss", "scenario": "footprint-rss", "mode": "footprint", "phase": "measured", "iteration": 1, "terminal_status": terminal_status, "reason": version_process.get("error"), "process": dict(version_process), "metrics": { "runtime_executable_bytes": footprint.get("executable_bytes"), "runtime_installation_bytes": footprint.get("installation_bytes"), "peak_process_tree_rss_bytes": version_process["rss_measurement"].get("peak_process_tree_rss_bytes"), "rss_measurement": version_process["rss_measurement"], "wall_time_seconds": version_process["wall_time_seconds"], }, "output": None, "semantic_check": {"status": "not_applicable", "task": "footprint_rss"}, "engine_identity": dict(engine_identity), } row["metadata"] = _row_metadata( source=source, built_hash=engine_identity.get("executable_sha256"), build_mode=args.build_mode, scenario="footprint-rss", iterations=1, warmups=0, timeout_seconds=args.timeout_seconds, cache_state=cache_state_for("footprint", engine), artifact_path=portable_artifact, command=version_process["command"], started_at_utc=version_process["started_at_utc"], completed_at_utc=version_process["completed_at_utc"], parallel_jobs=getattr(args, "parallel_jobs", 1), wolfxl_worker_jobs=getattr(args, "wolfxl_worker_jobs", 1), ) _json_write(artifact_path, row) row["artifact_sha256"] = _sha256_file(artifact_path) return row def _prepare_fixture( root: Path, *, kind: str, rows: int, cols: int, args: argparse.Namespace ) -> tuple[Path, dict[str, Any]]: name = f"{kind}-{rows}x{cols}.xlsx" destination = root / "fixtures" / name process, payload = _bounded_worker( args, "fixture", {"path": str(destination), "kind": kind, "rows": rows, "cols": cols}, ) if process["terminal_status"] != "ok" or payload is None or not destination.is_file(): reason = ( process.get("error") or process["stderr"]["text"] or f"{process['terminal_status']} (exit code {process['exit_code']})" ) raise BenchmarkError(f"fixture preparation failed for {name}: {reason}") return destination, { "path": _portable_path(destination, root), "sha256": _sha256_file(destination), **payload, } def _bundle_evidence_file(path: Path, root: Path) -> bool: relative = path.relative_to(root) return ( path.is_file() and relative.as_posix() != "bundle.sha256.json" and all(part != ".DS_Store" and not part.startswith("._") for part in relative.parts) ) def _write_bundle_index(root: Path) -> None: files = sorted(path for path in root.rglob("*") if _bundle_evidence_file(path, root)) _json_write( root / "bundle.sha256.json", { "schema_version": SCHEMA_VERSION, "files": {_portable_path(path, root): _sha256_file(path) for path in files}, "self_hash_excluded": True, }, ) def _stability_label(values: list[str]) -> str: if not values: return "not_applicable" if len(values) < 2: return "insufficient_samples" return "stable" if len(set(values)) == 1 else "unstable" def determinism_metrics(rows: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]: """Summarize repeated output-byte and semantic-hash observations without averaging them.""" groups: dict[tuple[str, str, str, str, str], list[Mapping[str, Any]]] = {} for row in rows: if row.get("terminal_status") != "ok" or row.get("phase") != "measured": continue fixture = row.get("fixture") if not isinstance(fixture, str): continue key = ( str(row.get("scenario")), str(row.get("engine")), str(row.get("task")), str(row.get("mode")), fixture, ) groups.setdefault(key, []).append(row) result: list[dict[str, Any]] = [] for key, samples in sorted(groups.items()): output_hashes = [ str(output["sha256"]) for sample in samples if isinstance((output := sample.get("output")), Mapping) and isinstance(output.get("sha256"), str) ] semantic_hashes = [ str(observed["semantic_digest"]) for sample in samples if isinstance((check := sample.get("semantic_check")), Mapping) and isinstance((observed := check.get("observed")), Mapping) and isinstance(observed.get("semantic_digest"), str) ] result.append( { "scenario": key[0], "engine": key[1], "task": key[2], "mode": key[3], "fixture": key[4], "sample_count": len(samples), "output_byte_hash_stability": _stability_label(output_hashes), "semantic_hash_stability": _stability_label(semantic_hashes), "output_byte_hashes": sorted(set(output_hashes)), "semantic_hashes": sorted(set(semantic_hashes)), } ) return result def _finalize_bundle( root: Path, *, run_id: str, source: Mapping[str, Any], args: argparse.Namespace, engine_identities: Mapping[str, Mapping[str, Any]], fixtures: Mapping[str, Mapping[str, Any]], rows: list[dict[str, Any]], ) -> None: failures = [row for row in rows if row["terminal_status"] in NON_SUCCESS_STATUSES] hard_failures = [row for row in rows if row["terminal_status"] in HARD_FAILURE_STATUSES] determinism = determinism_metrics(rows) _jsonl_write(root / "raw.jsonl", rows) _jsonl_write(root / "failures.jsonl", failures) harness = root / HARNESS_BUNDLE_PATH _json_write( root / "manifest.json", { "schema_version": SCHEMA_VERSION, "registry_version": REGISTRY_VERSION, "bundle_kind": "libreoffice-vs-wolfxl-benchmark", "run_id": run_id, "run_root": ".", "created_at_utc": _utc_now(), "source": dict(source), "build": {"build_mode": args.build_mode, "harness_sha256": _sha256_file(harness)}, "engines": {name: dict(identity) for name, identity in engine_identities.items()}, "fixtures": {name: dict(value) for name, value in fixtures.items()}, "parameters": { "rows": args.rows, "cols": args.cols, "large_rows": args.large_rows, "large_cols": args.large_cols, "iterations": args.iterations, "warmups": args.warmups, "timeout_seconds": args.timeout_seconds, "selected_scenarios": list(args.scenario or SCENARIO_BY_NAME), "parallel_jobs": args.parallel_jobs, "wolfxl_worker_jobs": args.wolfxl_worker_jobs, }, "scenario_registry": [asdict(item) for item in SCENARIOS], "task_registry": TASK_REGISTRY, "concurrency": { "parallel_jobs": args.parallel_jobs, "wolfxl_worker_jobs": args.wolfxl_worker_jobs, "libreoffice_profile_isolation": "one distinct profile directory per LibreOffice observation", "wolfxl_worker_execution": "bounded by wolfxl_worker_jobs", }, "determinism": determinism, "harness": {"path": HARNESS_BUNDLE_PATH, "sha256": _sha256_file(harness)}, "non_claims": [ "This bundle covers only the registered spreadsheet workflows and generated scenarios.", "Runtime footprint and process-tree RSS are distinct metrics and are not interchangeable.", "Unsupported engine/task pairs remain evidence rows; they are not replaced with easier work.", ], }, ) _json_write( root / "summary.json", { "schema_version": SCHEMA_VERSION, "run_id": run_id, "row_count": len(rows), "ok_count": sum(row["terminal_status"] == "ok" for row in rows), "unsupported_count": sum(row["terminal_status"] == "unsupported" for row in rows), "unavailable_count": sum(row["terminal_status"] == "unavailable" for row in rows), "hard_failure_count": len(hard_failures), "failure_visibility_count": len(failures), "determinism": determinism, }, ) _json_write( root / "status.json", { "schema_version": SCHEMA_VERSION, "run_id": run_id, "state": "failed" if hard_failures else "passed", "completed_at_utc": _utc_now(), "hard_failure_count": len(hard_failures), "unsupported_count": sum(row["terminal_status"] == "unsupported" for row in rows), }, ) _write_bundle_index(root) def validate_bundle(root: Path) -> dict[str, Any]: root = root.resolve() errors: list[str] = [] missing = [name for name in BUNDLE_FILES if not (root / name).is_file()] if missing: return {"valid": False, "errors": ["missing bundle files: " + ", ".join(missing)]} try: manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8")) summary = json.loads((root / "summary.json").read_text(encoding="utf-8")) status = json.loads((root / "status.json").read_text(encoding="utf-8")) index = json.loads((root / "bundle.sha256.json").read_text(encoding="utf-8")) rows = _jsonl_read(root / "raw.jsonl") failures = _jsonl_read(root / "failures.jsonl") except (OSError, ValueError, json.JSONDecodeError) as error: return {"valid": False, "errors": [str(error)]} if not isinstance(manifest, dict) or manifest.get("schema_version") != SCHEMA_VERSION: errors.append("invalid manifest schema") if not isinstance(index, dict) or not isinstance(index.get("files"), dict): errors.append("invalid bundle checksum index") else: if "bundle.sha256.json" in index["files"]: errors.append("bundle checksum index must exclude its own file") actual = { _portable_path(path, root): _sha256_file(path) for path in root.rglob("*") if _bundle_evidence_file(path, root) } if index["files"] != actual: errors.append("bundle checksum index is incomplete or mismatched") expected_failures = [row for row in rows if row.get("terminal_status") in NON_SUCCESS_STATUSES] if failures != expected_failures: errors.append("failures.jsonl does not preserve every non-success observation") if not isinstance(summary, dict) or summary.get("row_count") != len(rows): errors.append("summary row count does not match raw observations") if not isinstance(status, dict) or status.get("state") not in {"passed", "failed"}: errors.append("status receipt has invalid terminal state") if isinstance(status, dict): expected_state = "failed" if any(row.get("terminal_status") in HARD_FAILURE_STATUSES for row in rows) else "passed" if status.get("state") != expected_state: errors.append("status receipt does not reflect hard failures") required_metadata = { "source_hash", "source_tree", "source_dirty", "built_hash_or_baked_constant", "build_mode", "scenario", "iteration_count", "timeout_seconds", "cache_state", "artifact_path", "command", "started_at_utc", "completed_at_utc", } for number, row in enumerate(rows, 1): metadata = row.get("metadata") if not isinstance(metadata, dict) or not required_metadata.issubset(metadata): errors.append(f"raw row {number} lacks required provenance metadata") continue artifact_name = metadata["artifact_path"] if not isinstance(artifact_name, str) or artifact_name.startswith("/") or ".." in Path(artifact_name).parts: errors.append(f"raw row {number} has unsafe artifact path") continue artifact = root / artifact_name if not artifact.is_file(): errors.append(f"raw row {number} artifact is missing") elif row.get("artifact_sha256") != _sha256_file(artifact): errors.append(f"raw row {number} artifact hash mismatches") return {"valid": not errors, "errors": errors, "manifest": manifest, "rows": rows, "status": status} def _selected_scenarios(values: Sequence[str] | None) -> tuple[Scenario, ...]: if not values: return SCENARIOS selected: list[Scenario] = [] seen: set[str] = set() for value in values: if value not in SCENARIO_BY_NAME: raise ValueError(f"unknown scenario: {value}") if value not in seen: selected.append(SCENARIO_BY_NAME[value]) seen.add(value) return tuple(selected) def run_benchmark(args: argparse.Namespace) -> int: _safe_run_id(args.run_id) if args.rows <= 0 or args.cols <= 0: raise ValueError("--rows and --cols must be positive") if args.large_rows is not None and args.large_rows <= 0: raise ValueError("--large-rows must be positive") if args.large_cols <= 0 or args.iterations <= 0 or args.warmups < 0 or args.timeout_seconds <= 0: raise ValueError("numeric benchmark limits must be positive (warmups may be zero)") args._wolfxl_worker_slots = BoundedSemaphore(args.wolfxl_worker_jobs) output_parent = args.output_dir.expanduser().resolve() root = output_parent / args.run_id if root.exists(): raise ValueError(f"immutable bundle root already exists: {root}") selected = _selected_scenarios(args.scenario) root.mkdir(parents=True) (root / "harness").mkdir() shutil.copy2(Path(__file__), root / HARNESS_BUNDLE_PATH) source = _source_identity() runtime_scratch = root / "work" / "runtime-identities" libreoffice_identity, libreoffice_version_process = libreoffice_runtime_identity( args.soffice, timeout_seconds=args.timeout_seconds, scratch=runtime_scratch ) wolfxl_version_process, wolfxl_identity_payload = _run_worker( args.python, "identity", {}, timeout_seconds=args.timeout_seconds ) if wolfxl_identity_payload is None: wolfxl_identity = { "requested_path": _display_path(args.python), "resolved_path": _display_path(Path(args.python).resolve(strict=False)), "wrapper_sha256": _file_hash_or_none(Path(args.python).resolve(strict=False)), "executable_path": None, "executable_sha256": None, "version_stdout": wolfxl_version_process["stdout"]["text"], "runtime_json_path": None, "runtime_json_sha256": None, "runtime_manifest_fields": {"status": "missing", "reason": "identity worker unavailable"}, "platform": sys.platform, "arch": platform.machine(), "observed_at_utc": _utc_now(), "runtime_footprint": {"executable_bytes": None, "installation_root": None, "installation_bytes": None}, "availability": "unavailable", } else: wolfxl_identity = wolfxl_identity_payload engine_identities = {"libreoffice": libreoffice_identity, "wolfxl": wolfxl_identity} engine_processes = {"libreoffice": libreoffice_version_process, "wolfxl": wolfxl_version_process} fixtures: dict[str, dict[str, Any]] = {} prepared: dict[tuple[str, int, int], Path] = {} def fixture_for(task: str, rows: int, cols: int) -> Path: kind = _task_fixture_kind(task) key = (kind, rows, cols) if key not in prepared: path, receipt = _prepare_fixture(root, kind=kind, rows=rows, cols=cols, args=args) prepared[key] = path fixtures[f"{kind}-{rows}x{cols}"] = receipt return prepared[key] rows: list[dict[str, Any]] = [] for scenario in selected: if scenario.name == "footprint-rss": for engine in ("libreoffice", "wolfxl"): row_key = f"{scenario.name}-{engine}-measured-1" rows.append( _observe_footprint( root=root, row_key=row_key, engine=engine, source=source, engine_identity=engine_identities[engine], version_process=engine_processes[engine], args=args, ) ) continue dimensions = [("smoke", args.rows, args.cols)] if scenario.name == "scale-frontier" and args.large_rows is not None: dimensions.append(("large", args.large_rows, args.large_cols)) for dimension_name, row_count, col_count in dimensions: for task in scenario.tasks: fixture = fixture_for(task, row_count, col_count) for mode in scenario.modes: warmup_count = args.warmups if mode == "warm" else 0 for phase, count in (("warmup", warmup_count), ("measured", args.iterations)): jobs = [ { "root": root, "row_key": f"{scenario.name}-{dimension_name}-{task}-{mode}-{phase}-{engine}-{index}", "engine": engine, "task": task, "scenario": scenario.name, "mode": mode, "phase": phase, "iteration": index, "fixture": fixture, "source": source, "engine_identity": engine_identities[engine], "args": args, } for index in range(1, count + 1) for engine in ("libreoffice", "wolfxl") ] with ThreadPoolExecutor(max_workers=args.parallel_jobs) as executor: futures = [executor.submit(_observe_task, **job) for job in jobs] rows.extend(future.result() for future in futures) _finalize_bundle( root, run_id=args.run_id, source=source, args=args, engine_identities=engine_identities, fixtures=fixtures, rows=rows, ) result = validate_bundle(root) if not result["valid"]: raise BenchmarkError("bundle validation failed: " + "; ".join(result["errors"])) print(json.dumps({"bundle": str(root), "status": result["status"]["state"], "rows": len(rows)}, sort_keys=True)) return 0 if result["status"]["state"] == "passed" else 1 def _positive_int(value: str) -> int: parsed = int(value) if parsed <= 0: raise argparse.ArgumentTypeError("must be positive") return parsed def _nonnegative_int(value: str) -> int: parsed = int(value) if parsed < 0: raise argparse.ArgumentTypeError("must be non-negative") return parsed def _positive_float(value: str) -> float: parsed = float(value) if parsed <= 0: raise argparse.ArgumentTypeError("must be positive") return parsed def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) subparsers = parser.add_subparsers(required=True) run_parser = subparsers.add_parser("run", help="run registered scenarios into a new immutable receipt bundle") run_parser.add_argument("--soffice", type=Path, required=True, help="explicit LibreOffice soffice executable") run_parser.add_argument("--python", default=sys.executable, help="Python executable used for WolfXL worker subprocesses") run_parser.add_argument("--output-dir", type=Path, required=True, help="parent directory for the immutable bundle") run_parser.add_argument("--run-id", required=True, help="new immutable bundle directory name") run_parser.add_argument("--scenario", action="append", choices=sorted(SCENARIO_BY_NAME)) run_parser.add_argument("--rows", type=_positive_int, default=DEFAULT_ROWS) run_parser.add_argument("--cols", type=_positive_int, default=DEFAULT_COLS) run_parser.add_argument("--large-rows", type=_positive_int, help="explicit large scale row count, e.g. 200000 on a VPS") run_parser.add_argument("--large-cols", type=_positive_int, default=DEFAULT_COLS) run_parser.add_argument("--iterations", type=_positive_int, default=DEFAULT_ITERATIONS) run_parser.add_argument("--warmups", type=_nonnegative_int, default=DEFAULT_WARMUPS) run_parser.add_argument("--timeout-seconds", type=_positive_float, default=DEFAULT_TIMEOUT_SECONDS) run_parser.add_argument("--build-mode", default="local-source", help="recorded build/runtime mode label") run_parser.set_defaults(func=run_benchmark) worker_parser = subparsers.add_parser("worker", help=argparse.SUPPRESS) worker_parser.add_argument("operation", choices=("identity", "fixture", "execute")) worker_parser.set_defaults(func=lambda args: worker(args.operation)) validate_parser = subparsers.add_parser("validate", help="validate an existing immutable receipt bundle") run_parser.add_argument( "--parallel-jobs", type=_positive_int, default=1, help="maximum concurrent benchmark observations; each LibreOffice job gets a distinct profile", ) run_parser.add_argument( "--wolfxl-worker-jobs", type=_positive_int, default=1, help="maximum concurrent WolfXL library worker subprocesses within --parallel-jobs", ) validate_parser.add_argument("bundle", type=Path) def validate_command(args: argparse.Namespace) -> int: result = validate_bundle(args.bundle) print(json.dumps(result, sort_keys=True)) return 0 if result["valid"] else 1 validate_parser.set_defaults(func=validate_command) return parser def main(argv: Sequence[str] | None = None) -> int: args = build_parser().parse_args(argv) return args.func(args) if __name__ == "__main__": # pragma: no cover - exercised by CLI smoke runs. raise SystemExit(main())