worker.js

// Pyodide web worker: loads omnibenchmark and converts YAML → Snakefile preview

importScripts("https://cdn.jsdelivr.net/pyodide/v0.27.3/full/pyodide.js");

let pyodide = null;

// Python code stored as a regular (non-template) string to avoid escaping hell.
// Each \n is a real JS escape → newline.  Each \\ is a literal backslash.
// Python sees exactly what's written here.
const PREVIEW_PY = [
  "import io, json, warnings, re, hashlib, gzip, base64",
  "import yaml as _yaml",
  "from pathlib import PurePosixPath",
  "warnings.filterwarnings('ignore')",
  "",
  "def _repo_name(url):",
  "    name = PurePosixPath(url.rstrip('/')).name",
  "    return name[:-4] if name.endswith('.git') else name",
  "",
  "def _rulename(node_id):",
  "    name = node_id.replace('-', '_').replace('.', '_')",
  "    return ('rule_' + name) if name and not name[0].isalpha() else name",
  "",
  "def _ikey(entry):",
  "    return re.sub(r'[^a-zA-Z0-9_]', '_', entry)",
  "",
  "def _env_dir(bench, module):",
  "    eid = module.software_environment",
  "    if not eid: return ''",
  "    env = next((e for e in bench.software_environments if e.id == eid), None)",
  "    if not env: return ''",
  "    b = bench.software_backend.value",
  "    if b == 'conda' and env.conda:",
  "        return f'    conda: \"{env.conda}\"\\n'",
  "    if b == 'apptainer' and env.apptainer:",
  "        return f'    container: \"{env.apptainer}\"\\n'",
  "    if b == 'envmodules' and env.envmodule:",
  "        return f'    envmodules: \"{env.envmodule}\"\\n'",
  "    return ''",
  "",
  "def _res_dir(r):",
  "    if not r: return ''",
  "    lines = ['    resources:']",
  "    lines.append(f'        cores={r.cores or 2},')",
  "    if r.mem_mb:  lines.append(f'        mem_mb={r.mem_mb},')",
  "    if r.disk_mb: lines.append(f'        disk_mb={r.disk_mb},')",
  "    if r.runtime: lines.append(f'        runtime={r.runtime},')",
  "    if r.gpu:     lines.append(f'        nvidia_gpu={r.gpu},')",
  "    return '\\n'.join(lines) + '\\n'",
  "",
  "def _generate(bench):",
  "    out = io.StringIO()",
  "    backend = bench.software_backend.value",
  "    out.write(f'# OmniBenchmark: {bench.id} v{bench.version}\\n')",
  "    out.write(f'# Benchmarker: {bench.benchmarker}\\n')",
  "    out.write(f'# Backend: {backend}\\n')",
  "    out.write('# Entrypoints are read from each module\\'s omnibenchmark.yaml by run.sh.\\n')",
  "    out.write('\\n')",
  "    out.write('configfile: \"module_config.json\"\\n\\n')",
  "    # Build the DAG with omnibenchmark's own graph code, then emit one rule per",
  "    # root->node lineage path. Reusing upstream's lineage join (each node's",
  "    # 'after' = the latest producing stage of its inputs) keeps the rule count",
  "    # identical to `ob run`, instead of re-deriving topology from YAML order.",
  "    from pathlib import Path as _Path",
  "    from omnibenchmark.core._graph import build_benchmark_dag, find_initial_and_terminal_nodes",
  "    from omnibenchmark.dag import all_simple_paths",
  "    _g = build_benchmark_dag(bench, _Path('.'))",
  "    _initial, _terminal = find_initial_and_terminal_nodes(_g)",
  "    _initial_set = set(_initial)",
  "    def _ob_node(prev, cn):",
  "        stage = cn.stage",
  "        module = cn.module",
  "        params = cn.parameters",
  "        input_ids = [e for ic in (stage.inputs or []) for e in ic.entries]",
  "        param_id = f'.{params.hash_short()}' if params else '.default'",
  "        inputs = {}",
  "        imap = {}",
  "        base_path = None",
  "        if prev:",
  "            for iid in input_ids:",
  "                skey = _ikey(iid)",
  "                ipath = prev['cmap'].get(iid)",
  "                if ipath:",
  "                    inputs[skey] = ipath",
  "                    imap[skey] = iid",
  "            if inputs:",
  "                deepest = max(inputs.values(), key=lambda p: p.count('/'))",
  "                base_path = '/'.join(deepest.split('/')[:-1])",
  "        dataset = prev['dataset'] if prev else module.id",
  "        outputs = []",
  "        omap = {}",
  "        for outf in (stage.outputs or []):",
  "            fname = outf.path.replace('{dataset}', dataset)",
  "            if base_path:",
  "                opath = f'{base_path}/{stage.id}/{module.id}/{param_id}/{fname}'",
  "            else:",
  "                opath = f'{stage.id}/{module.id}/{param_id}/{fname}'",
  "            outputs.append(opath)",
  "            omap[outf.id] = opath",
  "        cmap = dict(prev['cmap']) if prev else {}",
  "        cmap.update(omap)",
  "        if prev:",
  "            node_id = f\"{prev['node_id']}-{stage.id}-{module.id}{param_id}\"",
  "        else:",
  "            node_id = f'{stage.id}-{module.id}{param_id}'",
  "        return {'node_id': node_id, 'stage': stage, 'module': module, 'param_id': param_id, 'params': params, 'inputs': inputs, 'imap': imap, 'outputs': outputs, 'omap': omap, 'cmap': cmap, 'dataset': dataset}",
  "    # One rule per distinct lineage-node (root->node prefix) that lies on a",
  "    # non-excluded root->terminal path -- exactly what omnibenchmark's Snakefile",
  "    # emits. Excluded module combinations (module `exclude:`) are pruned, and",
  "    # dead-end prefixes (no reachable terminal) never appear. Dedup by node_id so",
  "    # a shared prefix is one rule. Caps bound the work on pathological benchmarks.",
  "    try:",
  "        from omnibenchmark.core._paths import collect_path_exclusions as _cpe, is_lineage_excluded as _ile",
  "        _excl = _cpe(bench)",
  "    except Exception:",
  "        _excl = {}",
  "        _ile = lambda mods, ex: False",
  "    _terminal_set = set(_terminal)",
  "    _NODE_BUDGET = 50000        # cap on emitted rules (keeps Monaco responsive)",
  "    _PATH_BUDGET = 300000       # cap on paths walked (bounds work on huge DAGs)",
  "    _seen = {}",
  "    _truncated = False",
  "    _npaths = 0",
  "    for _s in _initial:",
  "        if _truncated: break",
  "        for _t in _terminal:",
  "            if _truncated: break",
  "            for _path in all_simple_paths(_g, _s, _t):",
  "                _npaths += 1",
  "                if _npaths > _PATH_BUDGET: _truncated = True; break",
  "                if _ile({_cn.module_id for _cn in _path}, _excl): continue",
  "                _prev = None",
  "                for _cn in _path:",
  "                    _pid = f'.{_cn.parameters.hash_short()}' if _cn.parameters else '.default'",
  "                    _nid = (_prev['node_id'] + '-' if _prev else '') + f'{_cn.stage.id}-{_cn.module.id}{_pid}'",
  "                    if _nid in _seen:",
  "                        _prev = _seen[_nid]",
  "                    else:",
  "                        _prev = _ob_node(_prev, _cn)",
  "                        _seen[_nid] = _prev",
  "                if len(_seen) > _NODE_BUDGET: _truncated = True; break",
  "    # Standalone nodes that are both initial and terminal (no simple path walks",
  "    # through them) -- rare, but keep them as their own rule.",
  "    if not _truncated:",
  "        for _n in _g.nodes:",
  "            if _n in _initial_set and _n in _terminal_set:",
  "                _sn = _ob_node(None, _n)",
  "                _seen.setdefault(_sn['node_id'], _sn)",
  "    _stage_order = {s.id: i for i, s in enumerate(bench.stages)}",
  "    all_nodes = sorted(_seen.values(), key=lambda nd: (_stage_order.get(nd['stage'].id, 999), nd['node_id']))",
  "    if _truncated:",
  "        out.write(f'# WARNING: benchmark too large to fully expand; Snakefile truncated at {len(all_nodes)} rules.\\n')",
  "        out.write('# The Check tab shows the true total rule count. Filter the benchmark to preview the full Snakefile.\\n\\n')",
  "    # rule all",
  "    out.write('rule all:\\n    input:\\n')",
  "    for node in all_nodes:",
  "        for opath in node['outputs']:",
  "            out.write(f'        \"{opath}\",\\n')",
  "    out.write('    default_target: True\\n\\n\\n')",
  "    # Individual rules",
  "    prev_stage_id = None",
  "    for node in all_nodes:",
  "        stage   = node['stage']",
  "        module  = node['module']",
  "        params  = node['params']",
  "        inputs  = node['inputs']",
  "        imap    = node['imap']",
  "        outputs = node['outputs']",
  "        dataset = node['dataset']",
  "        if stage.id != prev_stage_id:",
  "            out.write(f'# {\"=\" * 60}\\n# Stage: {stage.id}\\n# {\"=\" * 60}\\n\\n')",
  "            prev_stage_id = stage.id",
  "        res       = module.resources or stage.resources",
  "        repo_slug = _repo_name(module.repository.url)",
  "        commit    = module.repository.commit[:7]",
  "        ekey      = module.repository.entrypoint or 'default'",
  "        rule      = _rulename(node['node_id'])",
  "        cli_str   = ' '.join(params.to_cli_args()) if params else ''",
  "        output_dir = '/'.join(outputs[0].split('/')[:-1]) if outputs else '.'",
  "        out.write(f'rule {rule}:\\n')",
  "        if inputs:",
  "            out.write('    input:\\n')",
  "            for skey, ipath in inputs.items():",
  "                out.write(f'        {skey}=\"{ipath}\",\\n')",
  "        out.write('    output:\\n')",
  "        for opath in outputs:",
  "            out.write(f'        \"{opath}\",\\n')",
  "        out.write('    params:\\n')",
  "        out.write(f'        module_dir=\".modules/{repo_slug}/{commit}/\",\\n')",
  "        out.write(f'        entrypoint=config[\"entrypoints\"][\"{module.id}\"][\"{ekey}\"],\\n')",
  "        out.write(f'        output_dir=\"{output_dir}\",\\n')",
  "        out.write(f'        cli_args=\"{cli_str}\",\\n')",
  "        out.write(f'    benchmark:\\n        \"{output_dir}/{dataset}_performance.txt\"\\n')",
  "        out.write(f'    log:\\n        \".logs/{rule}.log\"\\n')",
  "        ed = _env_dir(bench, module)",
  "        if ed: out.write(ed)",
  "        rd = _res_dir(res)",
  "        if rd: out.write(rd)",
  "        out.write('    shell:\\n        \"\"\"\\n')",
  "        out.write('        mkdir -p {params.output_dir} $(dirname {log})\\n')",
  "        out.write('        OUTPUT_DIR=$(cd {params.output_dir} && pwd)\\n')",
  "        for skey in inputs:",
  "            out.write(f'        INPUT_{skey}=$(cd $(dirname {{input.{skey}}}) && pwd)/$(basename {{input.{skey}}})\\n')",
  "        out.write('        LOG_FILE=$(pwd)/{log}\\n')",
  "        out.write('        exec > >(tee \"$LOG_FILE\") 2>&1\\n')",
  "        out.write(\"        echo '=== Rule: {rule} ==='\\n\")",
  "        out.write(\"        echo 'Started:' $(date -Iseconds)\\n\")",
  "        out.write(\"        echo '---'\\n\")",
  "        out.write('        cd {params.module_dir}\\n')",
  "        out.write('        python3 {params.entrypoint} \\\\\\n')",
  "        out.write('        --output_dir $OUTPUT_DIR \\\\\\n')",
  "        out.write(f'        --name {dataset} \\\\\\n')",
  "        for skey, iid in imap.items():",
  "            out.write(f'        --{iid} $INPUT_{skey} \\\\\\n')",
  "        out.write('        {params.cli_args}\\n')",
  "        out.write('        \"\"\"\\n\\n')",
  "    stats = {'total': len(all_nodes), 'stages': len(bench.stages), 'by_stage': {}}",
  "    for _n in all_nodes:",
  "        sid = _n['stage'].id",
  "        stats['by_stage'][sid] = stats['by_stage'].get(sid, 0) + 1",
  "    return out.getvalue(), stats",
  "",
  "def _module_url(module):",
  "    try:",
  "        url = (module.repository.url or '').strip()",
  "        commit = (module.repository.commit or '').strip()",
  "    except Exception:",
  "        return None",
  "    if not url:",
  "        return None",
  "    if url.endswith('.git'):",
  "        url = url[:-4]",
  "    url = url.rstrip('/')",
  "    if 'github.com' in url and commit:",
  "        return url + '/tree/' + commit",
  "    return url",
  "",
  "def _generate_mermaid(bench):",
  "    _sg = lambda sid: 'sg_' + re.sub(r'[^0-9A-Za-z_]', '_', sid)  # mermaid ids reject hyphens (e.g. EMBED-M)",
  "    stage_outs = {}",
  "    links = {}",
  "    for stage in bench.stages:",
  "        stage_outs[stage.id] = {o.id for o in (stage.outputs or [])}",
  "    lines = ['flowchart LR']",
  "    clicks = []",
  "    nondefault = []",
  "    defaultep = []",
  "    for stage in bench.stages:",
  "        lines.append(f'    subgraph {_sg(stage.id)}[{stage.id}]')",
  "        for module in stage.modules:",
  "            nid = re.sub(r'[^0-9A-Za-z_]', '_', 'n_' + stage.id + '_' + module.id)",
  "            ekey = getattr(module.repository, 'entrypoint', None)",
  "            if ekey and ekey != 'default':",
  "                lines.append(f'        {nid}[\"{module.id}<br/><small>{ekey}</small>\"]')",
  "                nondefault.append(nid)",
  "            else:",
  "                lines.append(f'        {nid}[\"{module.id}\"]')",
  "                defaultep.append(nid)",
  "            murl = _module_url(module)",
  "            if murl:",
  "                links[nid] = murl",
  "                clicks.append(f'    click {nid} obModuleClick \"Open module repository\"')",
  "        lines.append('    end')",
  "    # Topology edges. Prefer omnibenchmark's shared stage_adjacency() so this",
  "    # diagram stays in lock-step with `ob describe topology` / the dot export.",
  "    # It exists only in omnibenchmark >= 0.5.4; older wheels (selectable in the",
  "    # version dropdown) fall back to the equivalent inline computation below.",
  "    try:",
  "        from omnibenchmark.core._graph import stage_adjacency as _stage_adjacency",
  "        _edges = _stage_adjacency(bench)",
  "    except Exception:",
  "        _seen = set()",
  "        _edges = []",
  "        for stage in bench.stages:",
  "            input_ids = {e for ic in (stage.inputs or []) for e in ic.entries}",
  "            for prev in bench.stages:",
  "                if prev.id == stage.id: continue",
  "                shared = sorted(input_ids & stage_outs.get(prev.id, set()))",
  "                if shared and (prev.id, stage.id) not in _seen:",
  "                    _seen.add((prev.id, stage.id))",
  "                    _edges.append((prev.id, stage.id, shared))",
  "    for prev_id, stage_id, shared in _edges:",
  "        label = ', '.join(shared)",
  "        lines.append(f'    {_sg(prev_id)} -->|\"{label}\"| {_sg(stage_id)}')",
  "    lines.append('    classDef ep_default fill:#1f5132,stroke:#3fb950,color:#e6ffed')",
  "    lines.append('    classDef ep_custom fill:#3b2a5a,stroke:#a371f7,color:#f3eaff')",
  "    if defaultep:",
  "        lines.append('    class ' + ','.join(defaultep) + ' ep_default')",
  "    if nondefault:",
  "        lines.append('    class ' + ','.join(nondefault) + ' ep_custom')",
  "    lines.extend(clicks)",
  "    return '\\n'.join(lines), json.dumps(links)",
  "",
  "def _generate_io_summary(bench):",
  "    stage_outs = {}",
  "    for stage in bench.stages:",
  "        stage_outs[stage.id] = [o.id for o in (stage.outputs or [])]",
  "    rows = []",
  "    for stage in bench.stages:",
  "        input_ids = sorted({e for ic in (stage.inputs or []) for e in ic.entries})",
  "        rows.append({'stage': stage.id, 'inputs': input_ids, 'outputs': sorted(stage_outs[stage.id])})",
  "    return json.dumps(rows)",
  "",
  "def _generate_runner(bench, yaml_content, canonical_url=None):",
  "    out = io.StringIO()",
  "    out.write('#!/usr/bin/env bash\\n')",
  "    out.write(f'# obrun wrapper: {bench.id} v{bench.version}\\n')",
  "    out.write('# Generated by OBEditor\\n')",
  "    out.write('set -euo pipefail\\n\\n')",
  "    out.write('if ! command -v obrun &>/dev/null; then\\n')",
  "    out.write('  echo \"obrun not found in PATH.\" >&2\\n')",
  "    out.write('  echo \"Get it: https://github.com/btraven00/obflow/releases/tag/nightly\" >&2\\n')",
  "    out.write('  exit 1\\n')",
  "    out.write('fi\\n\\n')",
  "    if canonical_url:",
  "        out.write(f\"curl -fsSL '{canonical_url}' -o bench.yaml\\n\")",
  "    else:",
  "        out.write(\"cat > bench.yaml << 'OBEOF'\\n\")",
  "        out.write(yaml_content)",
  "        if not yaml_content.endswith('\\n'):",
  "            out.write('\\n')",
  "        out.write('OBEOF\\n')",
  "    out.write('\\n')",
  "    out.write('obrun use bench.yaml\\n')",
  "    out.write('obrun\\n')",
  "    return out.getvalue()",
  "",
  "def _normalize_yaml(text):",
  "    norm = '\\n'.join(line.rstrip() for line in text.splitlines())",
  "    if not norm.endswith('\\n'): norm += '\\n'",
  "    return norm",
  "",
  "def _parent_hash(text):",
  "    return hashlib.sha256(_normalize_yaml(text).encode('utf-8')).hexdigest()",
  "",
  "def _canonical_url(raw):",
  "    # The benchmark's published address lives under the provenance block.",
  "    if not isinstance(raw, dict): return None",
  "    prov = raw.get('provenance')",
  "    if not isinstance(prov, dict): return None",
  "    return prov.get('canonical_url') or prov.get('canonical-url')",
  "",
  "def _wizard(bench):",
  "    from omnibenchmark.model.params import Params",
  "    out = []",
  "    for stage in bench.stages:",
  "        mods = []",
  "        for m in stage.modules:",
  "            entry = {'id': m.id, 'name': (getattr(m, 'name', None) or m.id), 'params': None}",
  "            if m.parameters:",
  "                seen = set(); combos = []",
  "                for pset in m.parameters:",
  "                    try:",
  "                        for p in Params.expand_from_parameter(pset):",
  "                            h = p.hash_short()",
  "                            if h in seen: continue",
  "                            seen.add(h)",
  "                            combos.append({'hash': h,",
  "                                           'items': {k: v for k, v in p.items()},",
  "                                           'cli': ' '.join(p.to_cli_args())})",
  "                    except Exception: pass",
  "                entry['params'] = combos",
  "            mods.append(entry)",
  "        out.append({'id': stage.id, 'modules': mods})",
  "    return out",
  "",
  "def _rule_counts(bench):",
  "    # Exact Snakefile rule count, computed cheaply. A rule is one lineage-node =",
  "    # one distinct root->node prefix that lies on some non-excluded root->terminal",
  "    # path (exactly what omnibenchmark's Snakefile emits). We count these with a",
  "    # subset-state DP over the DAG rather than enumerating the (exponential) paths:",
  "    #   fwd(n): #distinct non-excluded prefixes reaching n, keyed by the set of",
  "    #           exclusion-relevant modules seen so far;",
  "    #   reach(n): the relevant-module sets of suffixes n->terminal.",
  "    # A prefix counts iff some suffix keeps the combined module set non-excluded.",
  "    from collections import defaultdict as _dd",
  "    from pathlib import Path as _Path",
  "    from omnibenchmark.core._graph import build_benchmark_dag, find_initial_and_terminal_nodes",
  "    try:",
  "        from omnibenchmark.core._paths import collect_path_exclusions as _cpe, is_lineage_excluded as _ile",
  "        excl = _cpe(bench)",
  "    except Exception:",
  "        excl = {}",
  "        _ile = lambda mods, ex: False",
  "    g = build_benchmark_dag(bench, _Path('.'))",
  "    initial, terminal = find_initial_and_terminal_nodes(g)",
  "    initial_set, terminal_set = set(initial), set(terminal)",
  "    relevant = set(excl) | {e for v in excl.values() for e in v}",
  "    def rmod(n): return frozenset({n.module_id} & relevant)",
  "    preds, succ = _dd(list), _dd(list)",
  "    for u, v in g.edges:",
  "        preds[v].append(u); succ[u].append(v)",
  "    fmemo = {}",
  "    def fwd(n):",
  "        if n in fmemo: return fmemo[n]",
  "        res = _dd(int); a = rmod(n)",
  "        if n in initial_set:",
  "            res[a] += 1",
  "        else:",
  "            for p in preds[n]:",
  "                for s, c in fwd(p).items():",
  "                    ns = frozenset(s | a)",
  "                    if not _ile(set(ns), excl): res[ns] += c",
  "        fmemo[n] = dict(res); return fmemo[n]",
  "    bmemo = {}",
  "    def reach(n):",
  "        if n in bmemo: return bmemo[n]",
  "        a = rmod(n); out = set()",
  "        if n in terminal_set: out.add(a)",
  "        for c in succ[n]:",
  "            for r in reach(c): out.add(frozenset(a | r))",
  "        bmemo[n] = out; return out",
  "    total = 0; by_stage = {}",
  "    for n in g.nodes:",
  "        rs = reach(n)",
  "        cnt = 0",
  "        for s, c in fwd(n).items():",
  "            if any(not _ile(set(s | r), excl) for r in rs): cnt += c",
  "        total += cnt",
  "        by_stage[n.stage_id] = by_stage.get(n.stage_id, 0) + cnt",
  "    return total, by_stage",
  "",
  "def _error_payload(e):",
  "    import traceback",
  "    line = None",
  "    if hasattr(e, 'problem_mark') and e.problem_mark:",
  "        line = e.problem_mark.line + 1",
  "    return {'ok': False, 'error': f'{type(e).__name__}: {e}',",
  "            'traceback': traceback.format_exc(), 'line': line}",
  "",
  "def yaml_to_views(yaml_content):",
  "    # Cheap views (graph / check / io / wizard). Derived straight from the parsed",
  "    # stages -- never expands the lineage DAG, so it can't hang on huge benchmarks.",
  "    try:",
  "        from omnibenchmark.model.benchmark import Benchmark",
  "        bench = Benchmark.from_yaml(yaml_content)",
  "        mermaid, mermaid_links = _generate_mermaid(bench)",
  "        io_summary = _generate_io_summary(bench)",
  "        raw = _yaml.safe_load(yaml_content)",
  "        canonical_url = _canonical_url(raw)",
  "        stats = {'stages': len(bench.stages),",
  "                 'modules': sum(len(s.modules) for s in bench.stages)}",
  "        try:",
  "            stats['total'], stats['by_stage'] = _rule_counts(bench)",
  "        except Exception:",
  "            # DAG unavailable (older wheel / parse quirk): fall back to module counts.",
  "            stats['by_stage'] = {s.id: len(s.modules) for s in bench.stages}",
  "        return {'ok': True, 'mermaid': mermaid, 'mermaid_links': mermaid_links,",
  "                'io_summary': io_summary, 'stats': json.dumps(stats),",
  "                'wizard': json.dumps(_wizard(bench)),",
  "                'parent_hash': _parent_hash(yaml_content),",
  "                'parent_url': canonical_url}",
  "    except Exception as e:",
  "        return _error_payload(e)",
  "",
  "def yaml_to_snakefile(yaml_content):",
  "    # Expensive: expands every lineage into rules. Generated lazily, only when the",
  "    # Snakefile / run.sh tab is opened.",
  "    try:",
  "        from omnibenchmark.model.benchmark import Benchmark",
  "        bench = Benchmark.from_yaml(yaml_content)",
  "        snakefile, _stats = _generate(bench)",
  "        raw = _yaml.safe_load(yaml_content)",
  "        canonical_url = _canonical_url(raw)",
  "        return {'ok': True, 'snakefile': snakefile,",
  "                'runner': _generate_runner(bench, yaml_content, canonical_url)}",
  "    except Exception as e:",
  "        return _error_payload(e)",
  "",
  "def filter_benchmark(yaml_content, picks_json, derived_from=None):",
  "    try:",
  "        from omnibenchmark.model.benchmark import Benchmark",
  "        from omnibenchmark.model.params import Params",
  "        picks = json.loads(picks_json)",
  "        raw = _yaml.safe_load(yaml_content)",
  "        if not isinstance(raw, dict):",
  "            return {'ok': False, 'error': 'YAML root must be a mapping'}",
  "        parent_hash = _parent_hash(yaml_content)",
  "        parent_url = _canonical_url(raw)",
  "        # Re-parse via pydantic so we can re-expand parameter sets when needed.",
  "        bench = Benchmark.from_yaml(yaml_content)",
  "        param_obj_map = {}",
  "        for _st in bench.stages:",
  "            for _m in _st.modules:",
  "                param_obj_map[(_st.id, _m.id)] = (_m.parameters or [])",
  "        def _apply_spec(new_mod, sid, mid, spec):",
  "            psets = param_obj_map.get((sid, mid)) or []",
  "            if not new_mod.get('parameters') or not psets: return",
  "            if spec == 'all': return",
  "            if spec == 'first':",
  "                try:",
  "                    expanded = Params.expand_from_parameter(psets[0])",
  "                    if expanded:",
  "                        new_mod['parameters'] = [{k: [v] for k, v in expanded[0].items()}]",
  "                except Exception: pass",
  "                return",
  "            if isinstance(spec, list):",
  "                wanted = set(spec); seen = set(); kept = []",
  "                for ps in psets:",
  "                    try:",
  "                        for p in Params.expand_from_parameter(ps):",
  "                            h = p.hash_short()",
  "                            if h in wanted and h not in seen:",
  "                                seen.add(h)",
  "                                kept.append({k: [v] for k, v in p.items()})",
  "                    except Exception: pass",
  "                if kept: new_mod['parameters'] = kept",
  "        new_stages = []",
  "        for st in (raw.get('stages') or []):",
  "            sid = st.get('id')",
  "            stage_picks = picks.get(sid)",
  "            if not stage_picks: continue",
  "            modules = []",
  "            for m in (st.get('modules') or []):",
  "                mid = m.get('id')",
  "                if mid not in stage_picks: continue",
  "                spec = stage_picks[mid]",
  "                new_mod = dict(m)",
  "                _apply_spec(new_mod, sid, mid, spec)",
  "                modules.append(new_mod)",
  "            if not modules: continue",
  "            new_st = dict(st)",
  "            new_st['modules'] = modules",
  "            new_stages.append(new_st)",
  "        out = dict(raw)",
  "        out.pop('derived-from', None); out.pop('derived_from', None)",
  "        # The filtered child isn't published at the parent's address; drop the",
  "        # parent's canonical_url (keep any other provenance keys).",
  "        if isinstance(out.get('provenance'), dict):",
  "            prov = dict(out['provenance'])",
  "            prov.pop('canonical_url', None); prov.pop('canonical-url', None)",
  "            if prov: out['provenance'] = prov",
  "            else: out.pop('provenance', None)",
  "        subset = {'sha256': parent_hash}",
  "        if parent_url: subset['url'] = parent_url",
  "        out['subset-of'] = subset",
  "        if derived_from: out['derived-from'] = derived_from",
  "        out['stages'] = new_stages",
  "        # Reorder header keys for readability.",
  "        head_keys = ['id', 'name', 'description', 'benchmarker', 'version', 'subset-of', 'derived-from']",
  "        ordered = {}",
  "        for k in head_keys:",
  "            if k in out: ordered[k] = out.pop(k)",
  "        ordered.update(out)",
  "        filtered_yaml = _yaml.safe_dump(ordered, sort_keys=False, default_flow_style=False, width=120)",
  "        blob = {'v': 2, 'parent': {'sha256': parent_hash}, 'picks': picks}",
  "        if parent_url: blob['parent']['url'] = parent_url",
  "        if derived_from: blob['parent']['derived_from'] = derived_from",
  "        blob_json = json.dumps(blob, separators=(',', ':'), sort_keys=True)",
  "        packed = base64.urlsafe_b64encode(",
  "            gzip.compress(blob_json.encode('utf-8'), compresslevel=9, mtime=0)",
  "        ).decode('ascii').rstrip('=')",
  "        return {'ok': True, 'filtered_yaml': filtered_yaml, 'blob_json': blob_json,",
  "                'blob_packed': packed, 'parent_hash': parent_hash,",
  "                'parent_url': parent_url}",
  "    except Exception as e:",
  "        import traceback",
  "        return {'ok': False, 'error': f'{type(e).__name__}: {e}',",
  "                'traceback': traceback.format_exc()}",
].join("\n");

async function init(wheelUrl) {
  try {
    self.postMessage({ type: "status", message: "Loading Pyodide runtime…" });
    pyodide = await loadPyodide();

    self.postMessage({ type: "status", message: "Loading pydantic / PyYAML…" });
    await pyodide.loadPackage(["pydantic", "pyyaml"]);

    self.postMessage({ type: "status", message: "Fetching omnibenchmark wheel…" });
    // Fetch the wheel in JS and write it to Pyodide's virtual FS.
    // Wheels are zip files — adding the path to sys.path lets zipimport handle it.
    // This completely avoids micropip and its download/resolution issues.
    const resp = await fetch(wheelUrl);
    if (!resp.ok) throw new Error(`Failed to fetch wheel: ${resp.status} ${resp.statusText}`);
    const wheelBytes = new Uint8Array(await resp.arrayBuffer());
    pyodide.FS.writeFile("/omnibenchmark.whl", wheelBytes);

    self.postMessage({ type: "status", message: "Initialising omnibenchmark…" });
    await pyodide.runPythonAsync(`
import sys, types

# Wheel is a zip — zipimport picks it up automatically from sys.path
sys.path.insert(0, "/omnibenchmark.whl")

def _mock(name, **attrs):
    m = types.ModuleType(name)
    for k, v in attrs.items():
        setattr(m, k, v)
    sys.modules[name] = m
    return m

# Mock every heavy / C-extension dep that omnibenchmark imports but we don't need.
# (We only use omnibenchmark.model.benchmark.Benchmark — pure pydantic + PyYAML.)
for _n in [
    "spdx_license_list",
    "matplotlib", "matplotlib.pyplot", "matplotlib.patches",
    "matplotlib.lines", "matplotlib.colors", "matplotlib.cm",
    "pydot",
    "humanfriendly",
    "filelock",
    "dulwich", "dulwich.repo", "dulwich.errors", "dulwich.porcelain",
    "snakemake",
    "copier",
    "dotenv", "python_dotenv",
    "tqdm", "tqdm.auto",
    "click",
    "rich", "rich.console", "rich.table", "rich.panel", "rich.progress",
]:
    _mock(_n)

sys.modules["spdx_license_list"].LICENSES = {}

# Stub the heavy siblings that omnibenchmark.core.__init__ pulls in, so importing
# the pure graph code (core._graph / _node / _paths + the dag toolkit) doesn't
# drag in execution/git/snakemake/mermaid. _node/_paths/_graph stay real.
_mock("omnibenchmark.core.execution", BenchmarkExecution=object)
_mock("omnibenchmark.core.validator", Validator=object)
_mock("omnibenchmark.core.prefetch", populate_git_cache=lambda *a, **k: None)

# Smoke-test: this is all we actually call at runtime
from omnibenchmark.model.benchmark import Benchmark
from omnibenchmark.core._graph import build_benchmark_dag
`);

    self.postMessage({ type: "status", message: "Initialising preview engine…" });
    await pyodide.runPythonAsync(PREVIEW_PY);

    const version = await pyodide.runPythonAsync(`
try:
    import importlib.metadata
    _ver = importlib.metadata.version('omnibenchmark')
except Exception:
    _ver = 'unknown'
_ver
`);
    self.postMessage({ type: "ready", version: String(version) });
  } catch (e) {
    self.postMessage({ type: "result", ok: false, error: String(e) });
  }
}

async function convert(yaml) {
  if (!pyodide) return;
  try {
    // Pass yaml via globals to avoid any quoting/injection issues
    pyodide.globals.set("_yaml_input", yaml);
    const result = await pyodide.runPythonAsync("yaml_to_views(_yaml_input)");
    const obj = result.toJs({ dict_converter: Object.fromEntries });
    result.destroy();
    self.postMessage({ type: "result", ok: obj.ok, stats: obj.stats, mermaid: obj.mermaid, mermaid_links: obj.mermaid_links, io_summary: obj.io_summary, error: obj.error, line: obj.line, wizard: obj.wizard, parent_hash: obj.parent_hash, parent_url: obj.parent_url });
  } catch (e) {
    self.postMessage({ type: "result", ok: false, error: e.message });
  }
}

async function genSnakefile(yaml) {
  if (!pyodide) return;
  try {
    pyodide.globals.set("_yaml_input", yaml);
    const result = await pyodide.runPythonAsync("yaml_to_snakefile(_yaml_input)");
    const obj = result.toJs({ dict_converter: Object.fromEntries });
    result.destroy();
    self.postMessage({ type: "snakefile", ok: obj.ok, snakefile: obj.snakefile, runner: obj.runner, error: obj.error, line: obj.line });
  } catch (e) {
    self.postMessage({ type: "snakefile", ok: false, error: e.message });
  }
}

async function filterCmd({ yaml, picks, derivedFrom }) {
  if (!pyodide) return;
  try {
    pyodide.globals.set("_yaml_input", yaml);
    pyodide.globals.set("_picks_json", JSON.stringify(picks));
    pyodide.globals.set("_derived_from", derivedFrom || null);
    const result = await pyodide.runPythonAsync("filter_benchmark(_yaml_input, _picks_json, _derived_from)");
    const obj = result.toJs({ dict_converter: Object.fromEntries });
    result.destroy();
    self.postMessage({ type: "filtered", ok: obj.ok, filtered_yaml: obj.filtered_yaml, blob_json: obj.blob_json, blob_packed: obj.blob_packed, parent_hash: obj.parent_hash, parent_url: obj.parent_url, error: obj.error });
  } catch (e) {
    self.postMessage({ type: "filtered", ok: false, error: e.message });
  }
}

self.onmessage = async ({ data }) => {
  if (data.type === "init")    await init(data.wheelUrl);
  if (data.type === "convert") await convert(data.yaml);
  if (data.type === "snakefile") await genSnakefile(data.yaml);
  if (data.type === "filter")  await filterCmd(data);
};