src/engine/pyrel-worker.js

/* The PyRel front-end, in a Web Worker.
 *
 * Everything is inside Pyodide now:
 *
 *   Pyodide         CPython 3.14 on wasm
 *   duckdb (wheel)  a real DuckDB 1.5.4 built for Pyodide — swan's own, because
 *                   Pyodide's stock package is 1.5.1 and the extension is built
 *                   against 1.5.4
 *   swan extension  loaded by DuckDB itself, found on sys.path
 *   pyrel_duckdb    one Nuitka-compiled wasm32-emscripten module, no source
 *
 * It is a worker for the ordinary reason: this is a hundred megabytes of wasm
 * and tens of seconds of work, and the main thread should not be holding it.
 * DuckDB lives inside Pyodide, so there is no JS underneath: no bridge to keep
 * synchronous, no shim, and no way for sql() and
 * pyrel() to end up looking at different databases. What remains is that this is
 * ~100 MB of wasm and multi-second queries, which should not run on the thread
 * that paints. The main thread talks to it in messages and never blocks.
 *
 * Verified as an arrangement by tools/verify_pyrel_wasm.mjs, which runs these
 * same pieces under Node against the same wasm binaries. This file is the
 * browser wiring of that; it has not itself been run in a browser.
 */

const BASE = new URL('../../web/swan-wasm/', import.meta.url);

let db = null;
let conn = null;
let pyodide = null;
let dbReady = false;
let engineManifest = null;
let pyReady = false;

/* The three artifacts swan publishes for Pyodide, staged next to each other.
 * Named here rather than hardcoded at each use so the manifest can carry them
 * and a version bump is one place. */
let PY = {
  duckdbWheel: 'duckdb-1.5.4-cp314-cp314-pyemscripten_2026_0_wasm32.whl',
  module: 'pyrel_duckdb.cpython-314-wasm32-emscripten.so',
  extension: 'swan.duckdb_extension.wasm',
};

/* Every page that connects gets a port, and they all stay open.
 *
 * This is a SharedWorker, so one engine outlives any single page. A reply goes
 * back to the port that asked; progress and the rules result go to everyone,
 * because a page that attached midway through a load has just as much reason to
 * see them as the page that started it. */
const ports = new Set();

const broadcast = (msg) => {
  for (const p of ports) {
    // A page that navigated away leaves a port that throws rather than one that
    // reports itself closed, so drop it here rather than accumulating dead ones.
    /* Dropping the port here is safe; dropping its *session* is not. A throw
     * from postMessage is not a reliable "this page is gone" signal — a live
     * page can miss one — and tearing the session down on that basis deleted
     * the namespace of a tab that was mid-run, which then met "no template
     * session s1 (have: s2)". Sessions are released on the page's own say-so
     * (templateClose) and bounded below, not guessed at from a failed send. */
    try { p.postMessage(msg); } catch { ports.delete(p); }
  }
};
const progress = (text) => broadcast({ type: 'progress', text });

/* The synchronous bridge Python calls into. Every statement pyrel_duckdb issues
 * arrives here.
 *
 * Parameters go through a prepared statement: `conn.query(sql, params)` silently
 * ignores its second argument, which is not a small bug — Swan's compiler is
 * invoked as a parameterized query, so unbound parameters disable compilation
 * while everything still looks like it is working. */
function execSql(sql) {
  if (!pyReady) {
    throw new Error('SQL needs the PyRel front-end started — the database lives '
      + 'inside Pyodide now, so call startPyrel() first');
  }
  pyodide.globals.set('_eh_sql', sql);
  return pyodide.runPython(`
_cur = _eh_con().execute(_eh_sql)
_cols = [d[0] for d in (_cur.description or [])]
json.dumps({"columns": _cols, "rows": [list(r) for r in _cur.fetchall()]}, default=str)
`);
}

/* No database on this side.
 *
 * DuckDB runs inside Pyodide, from a wheel, so this side owns nothing. One
 * database means sql() and pyrel() cannot end up looking at different data.
 *
 * ready() still resolves before Pyodide, because the ~50 MB behind startPyrel()
 * should stay opt-in. It just answers from the manifest now instead of from a
 * live connection.
 */
async function initDb() {
  const manifest = await (await fetch(new URL('manifest.json', BASE))).json();
  if (manifest.python && typeof manifest.python === 'object') {
    // A staged manifest names the three artifacts; the defaults above are only
    // a fallback for a stage written before they were recorded.
    PY = { ...PY, ...manifest.python };
  }
  dbReady = true;
  engineManifest = manifest;
  return { duckdbLoaded: false, swan: true, manifest };
}

/* ---- the database, cached in IndexedDB ---------------------------------
 *
 * The engine dies when the last page holding it goes away (see wasm_engine.js
 * on why a SharedWorker does not survive navigating to a page with no engine).
 * What survives is this: the DuckDB file itself, kept in IndexedDB and written
 * back into Pyodide's filesystem before the tables would otherwise be built.
 *
 * What that saves is the parquet fetch and 24 CREATE TABLEs -- about 8 s of a
 * 25 s load, measured. What it does NOT save is the rules, and that is worth
 * stating because it is the expensive part and the obvious thing to expect: a
 * fresh Python process re-declares every rule whether or not its output tables
 * are already there, so restoring a materialised database changes nothing
 * (22.0 s warm against 23.1 s cold, measured in CPython). Pyodide's own boot is
 * likewise untouched. This is a third off, not a shortcut to instant.
 */
const IDB = { name: 'event-horizon', store: 'duckdb', version: 1 };
const DB_FILE = '/eh/eh.duckdb';

function idb() {
  return new Promise((resolve, reject) => {
    const req = indexedDB.open(IDB.name, IDB.version);
    req.onupgradeneeded = () => {
      if (!req.result.objectStoreNames.contains(IDB.store)) req.result.createObjectStore(IDB.store);
    };
    req.onsuccess = () => resolve(req.result);
    req.onerror = () => reject(req.error);
  });
}

function idbGet(db, key) {
  return new Promise((resolve, reject) => {
    const req = db.transaction(IDB.store, 'readonly').objectStore(IDB.store).get(key);
    req.onsuccess = () => resolve(req.result);
    req.onerror = () => reject(req.error);
  });
}

function idbPut(db, key, value) {
  return new Promise((resolve, reject) => {
    const tx = db.transaction(IDB.store, 'readwrite');
    tx.objectStore(IDB.store).put(value, key);
    tx.oncomplete = () => resolve();
    tx.onerror = () => reject(tx.error);
  });
}

/* Everything, not the one key for the dataset in hand.
 *
 * Reload data means "forget what you cached and read the source again", and the
 * store holds nothing but this app's database snapshots -- so clearing it also
 * collects the entries left by earlier swan releases and other datasets, which
 * are dead weight no other code path ever removes. It also avoids reconstructing
 * a key from state that may not be loaded yet. */
function idbClear(db) {
  return new Promise((resolve, reject) => {
    const tx = db.transaction(IDB.store, 'readwrite');
    tx.objectStore(IDB.store).clear();
    tx.oncomplete = () => resolve();
    tx.onerror = () => reject(tx.error);
  });
}

/* The cache key carries everything a stored file could go stale against: the
 * Swan release that wrote it, the DuckDB it was written by, and the dataset
 * manifest. A snapshot from a different dataset is not merely unhelpful, it is
 * wrong -- the model module would import against someone else's tables -- so
 * this errs toward missing rather than toward reuse. */
const snapshotKey = (manifest, dataset) =>
  `db:${manifest.swan}:${manifest.duckdb}:${dataset.name || 'dataset'}:${
    (dataset.tables || []).map((t) => `${t.table}/${t.rows}`).join(',')}`;

/* What a snapshot records beyond its bytes.
 *
 * `rules` is the load-bearing field: a database cached before the rules were
 * evaluated has the tables but none of the derived columns, and the planned
 * fast path (declare the 69 rule declarations, skip the 48 defines -- see
 * docs/rules-fast-path.md) would read those columns and answer nothing. So a
 * snapshot has to say whether it is complete rather than leaving the reader to
 * assume. `v` guards the shape itself: an older entry is bare bytes with no
 * fields at all, and must not be read as a record with everything false. */
const SNAPSHOT_V = 1;

/** Write a cached database into place. Null when there was none. */
async function restoreDb(key) {
  try {
    const rec = await idbGet(await idb(), key);
    if (!rec) return null;
    if (rec.v !== SNAPSHOT_V || !rec.bytes) {
      // A snapshot from an older build of this file. Cheaper to drop than to
      // reason about, and it will be rewritten on this load anyway.
      progress('ignoring a cached database from an older build');
      return null;
    }
    pyodide.FS.mkdirTree('/eh');
    pyodide.FS.writeFile(DB_FILE, new Uint8Array(rec.bytes));
    progress('restoring the cached database'
      + `${rec.rules ? ` (${rec.rules.ok}/${rec.rules.total} rules)` : ''}…`);
    return rec;
  } catch (err) {
    /* A cache miss must never be a failure: private browsing, a storage quota,
     * or an eviction mid-read all land here, and every one of them is fine --
     * the load simply proceeds the slow way. Silence would be wrong too, since
     * a cache that never hits looks identical to one that is working. */
    progress(`could not read the cached database (${err && err.message || err}) — loading fresh`);
    return null;
  }
}

/* A content fingerprint for the rules file, so a cached database can be told
 * apart from the rules it was computed with.
 *
 * Not crypto.subtle, which is absent outside a secure context: this is a static
 * bundle someone may well serve over plain HTTP on a LAN, and a rules cache
 * that silently stopped working only there would be a miserable thing to debug.
 * Two FNV-1a accumulators plus the length, which answers the only question ever
 * asked of it -- is this the same file the cached values came from? */
const fingerprint = (text) => {
  let a = 0x811c9dc5, b = 0x01000193;
  for (let i = 0; i < text.length; i += 1) {
    const c = text.charCodeAt(i);
    a = Math.imul(a ^ c, 0x01000193);
    b = Math.imul(b + c, 0x85ebca6b) ^ (b >>> 13);
  }
  return `${(a >>> 0).toString(16)}-${(b >>> 0).toString(16)}-${text.length}`;
};

/** Store the current database. Best-effort: a failure here costs only speed. */
async function snapshotDb(key, rules = null) {
  try {
    /* CHECKPOINT first, or the file on disk is missing whatever is still in the
     * write-ahead log -- which for a database built entirely in this session is
     * potentially all of it. */
    execSql('CHECKPOINT');
    const bytes = pyodide.FS.readFile(DB_FILE);
    await idbPut(await idb(), key, { v: SNAPSHOT_V, rules, bytes });
    progress(`cached the database (${(bytes.length / 1e6).toFixed(1)} MB`
      + `${rules ? `, ${rules.ok}/${rules.total} rules` : ''})`);
  } catch (err) {
    progress(`could not cache the database (${err && err.message || err})`);
  }
}

/* Pyodide and pyrel_duckdb, on top of the same connection. Separate because it
 * is roughly 100 MB and tens of seconds — a page that paid that on open would
 * look broken rather than slow. */
async function initPython() {
  progress('booting Pyodide…');
  const { loadPyodide } = await import(new URL('pyodide/pyodide.mjs', BASE).href);
  pyodide = await loadPyodide({ indexURL: new URL('pyodide/', BASE).href });

  progress('loading numpy and pandas…');
  await pyodide.loadPackage(['numpy', 'pandas']);

  /* DuckDB runs inside Pyodide, from a wheel.
   *
   * The wheel must be swan's own, not Pyodide's stock `duckdb`: the stock
   * package is 1.5.1 and the extension is built against 1.5.4. */
  progress('installing duckdb…');
  await pyodide.loadPackage(new URL(`python/${PY.duckdbWheel}`, BASE).href);

  /* pyrel_duckdb as one Nuitka-compiled wasm32-emscripten module — no Python
   * source in the browser at all, which is what rai-swan 0.0.7 made possible
   * and, by dropping source from its wheel, also made necessary. */
  progress('installing pyrel_duckdb…');
  const site = pyodide.runPython('import site; site.getsitepackages()[0]');
  for (const name of [PY.module, PY.extension]) {
    const bytes = new Uint8Array(
      await (await fetch(new URL(`python/${name}`, BASE))).arrayBuffer(),
    );
    /* Both land in site-packages, and the extension's placement is load-bearing
     * rather than tidiness: swan finds it by scanning sys.path entries, because
     * Nuitka gives every compiled module a synthetic __file__ (`//pyrel_duckdb/
     * …`) with no relation to where the .so actually is, so package-relative
     * lookup cannot work. */
    pyodide.FS.writeFile(`${site}/${name}`, bytes);
  }

  progress('starting Swan…');
  await pyodide.runPythonAsync(`
import sys, json, contextlib
import duckdb            # must precede any Model(): swan promotes duckdb's own
                         # compiled module to a global symbol so that LOAD swan's
                         # internal dlopen can resolve against it
from pyrel_duckdb.nlq import execute_pyrel_code, serialize_model_schema
import pyrel_duckdb.nlq.sandbox as _sandbox

# Swan's sandbox caps execution with signal.alarm(), which Emscripten does not
# have -- Pyodide's signal module has no alarm at all, so every call to
# execute_pyrel_code dies with AttributeError before running anything.
#
# Neutralise *only* the timeout, by replacing the context manager rather than
# reimplementing execute_pyrel_code around it. That function does more than
# execute: it validates the AST, builds the restricted namespace, and afterwards
# diffs the model to catch properties the code invented and concepts it renamed
# (the latter being a real exploit -- reassigning Concept.name retargets which
# table a query resolves to). Rewriting it here would risk dropping one of those
# quietly; swapping one contextmanager cannot.
#
# What is genuinely lost is the wall-clock cap. The mitigation is structural:
# this is a Worker, so a runaway query blocks only itself, and the main thread
# can terminate it -- which a signal-based alarm could not have done any better.
@contextlib.contextmanager
def _no_timeout(*_args, **_kwargs):
    yield

# Refuse to patch a name that is not there. Plain assignment would create a new
# attribute nothing reads, leave the real _AlarmTimeout in place, and report
# nothing -- the engine would boot clean and every query would then die on
# signal.alarm, which Emscripten does not have. rai-swan is obfuscated from
# 0.0.6, so this name disappearing is a question of when.
if not hasattr(_sandbox, "_AlarmTimeout"):
    raise RuntimeError(
        "pyrel_duckdb.nlq.sandbox has no _AlarmTimeout. The browser engine "
        "replaces it because Pyodide's signal module has no alarm() (swan#152), "
        "and it cannot run without doing so. If the name was renamed or "
        "obfuscated, find its replacement; if the timeout was removed upstream, "
        "delete this patch."
    )

_sandbox._AlarmTimeout = _no_timeout

# One database on disk in Pyodide's filesystem, and SWAN_DB_PATH points every
# Model at it. Tables are loaded before the model module is imported (declaring a
# Concept creates its table if absent, so a model built first leaves empty tables
# in its own column order and the data then has nowhere correct to go) -- which
# means the loader and the model must agree on the database. SWAN_DB_PATH is how
# swan 0.0.7 supports that; it is the fix for the gap that previously forced a
# Model.__init__ monkeypatch (GitHub issue #155).
import os
os.makedirs("/eh/data", exist_ok=True)

# Set around the model import, never left standing.
#
# SWAN_DB_PATH is read by *every* Model(), not just this dataset's. Setting it
# here, once, at startup meant a template's Model("diet") -- which its notebook
# writes expecting the ':memory:' default — silently opened /eh/eh.duckdb
# instead. Every template in the browser then shared one database with TPC-DS
# and with each other, which is two bugs wearing one hat:
#
#   ConnectionException: Can't open a connection to same database file with a
#   different configuration than existing connections
#
# when _eh_con() already held that file on different terms than swan's Model
# opens it with, and
#
#   CatalogException: Column with name is_structuring already exists!
#
# when a second template session declared a property the first had already
# added to what was, unknown to either, the same table. Both looked like a
# broken template and neither pointed anywhere near this line.
def _eh_own_db(on):
    if on:
        os.environ["SWAN_DB_PATH"] = "/eh/eh.duckdb"
    else:
        os.environ.pop("SWAN_DB_PATH", None)

_eh_own_db(False)

_EH = {"model": None, "con": None}

def _eh_con():
    """The live connection: the model's once it exists, otherwise a transient one
    onto the same file. DuckDB will not open a file twice at once, so the loader
    closes its own before the model module is imported."""
    if _EH["con"] is None:
        _EH["con"] = duckdb.connect("/eh/eh.duckdb")
    return _EH["con"]

def _eh_release():
    if _EH["con"] is not None:
        _EH["con"].close()
        _EH["con"] = None
`);

  pyReady = true;
  /* Both versions, because the pairing is the thing that breaks: the extension
   * is built against one DuckDB and refuses to load against another, and there
   * is a stock Pyodide `duckdb` at a different version that pip would happily
   * have resolved instead. Reporting them makes a mismatch visible in the UI
   * rather than only as a LOAD failure. */
  return {
    duckdb: pyodide.runPython('duckdb.__version__'),
    swan: pyodide.runPython('import pyrel_duckdb; getattr(pyrel_duckdb, "__version__", "?")'),
  };
}

/* Load the staged demo dataset, then the model that describes it.
 *
 * Order matters and is not a preference: declaring a Concept creates its table
 * when the table is not already there, so a model loaded first leaves empty
 * tables behind in its own column order and the data then has nowhere correct
 * to go. Tables first, and the concepts bind to real ones.
 *
 * Parquet is fetched in JS and handed to Pyodide's filesystem — see the note
 * inside loadDataset on why DuckDB does no HTTP of its own here. */
async function loadDataset() {
  const base = new URL('dataset/', BASE);
  const manifest = await (await fetch(new URL('manifest.json', base))).json();

  /* Fetch each file here and hand DuckDB a buffer, rather than giving it a URL.
   *
   * `read_parquet('http://…')` needs the httpfs extension, and this database's
   * `custom_extension_repository` points at the staged directory that holds only
   * Swan — so an implicit INSTALL looks in the wrong place. What it actually did
   * was hang: the first table never returned and the load sat there forever
   * with no error. Fetching in JS avoids DuckDB doing any HTTP at all. */
  /* A cached database stands in for the whole parquet step.
   *
   * It has to happen before anything opens the file -- restoring underneath a
   * live connection would leave DuckDB holding a handle to bytes that are no
   * longer what it read. Nothing has called _eh_con() yet at this point, which
   * is why this sits at the top rather than beside the loop it replaces. */
  const key = snapshotKey(engineManifest || {}, manifest);
  const cached = await restoreDb(key);
  const restored = !!cached;

  pyodide.FS.mkdirTree('/eh/data');
  for (const { table, file, rows } of (restored ? [] : manifest.tables)) {
    progress(`loading ${table} (${rows.toLocaleString()} rows)…`);
    const bytes = new Uint8Array(await (await fetch(new URL(file, base))).arrayBuffer());
    /* Into Pyodide's filesystem, because that is where DuckDB is. The file
     * simply exists and read_parquet opens it like any other path. */
    pyodide.FS.writeFile(`/eh/data/${file}`, bytes);
    // The statement is built here and handed over as a value. Composing it
    // inside runPython means an f-string nested in a JS template literal, three
    // levels of quoting for no gain.
    execSql(`CREATE OR REPLACE TABLE "${table}" AS `
      + `SELECT * FROM read_parquet('/eh/data/${file}')`);
    pyodide.FS.unlink(`/eh/data/${file}`);   // the rows are in the table now
  }

  /* The model may be several modules that import each other — TPC-DS is five,
   * with model.py importing the sales-channel modules at the bottom and each of
   * those doing `from model import …` back. They resolve only when all of them
   * sit in one directory on sys.path, so write the files and import the entry
   * rather than concatenating anything. */
  /* Hand the database over. DuckDB will not open one file from two
   * connections at once, and the model module is about to open it via
   * SWAN_DB_PATH, so the loader's connection has to go first. */
  pyodide.runPython('_eh_release()');

  progress('building the model…');
  const spec = manifest.model;
  const files = {};
  for (const name of spec.files) {
    files[name] = await (await fetch(new URL(name, base))).text();
  }
  const info = JSON.parse(await loadModel({ entry: spec.entry, files }));
  await loadPromptModule(manifest, base);

  /* Rules are not awaited. Evaluating the 117 statements costs about 40s here,
   * and measurement says none of it is recoverable: exec'ing them is 19.3s of a
   * 19.9s total in CPython, so the expense is Swan's own define() work rather
   * than the flush, and restoring an already-materialised database changes
   * nothing because a fresh process re-declares everything anyway (22.0s warm
   * against 23.1s cold).
   *
   * So the fix is not to make it faster but to stop charging the whole of it
   * before the first question. The concept tier is complete the moment the model
   * is imported; the rules tier arrives about 35 seconds later, and a `rules`
   * message says when. */
  if (manifest.rules) {
    fetch(new URL(manifest.rules, base))
      .then((res) => res.text())
      .then(async (source) => {
        /* Take the fast path only if the cached values were computed from *this*
         * rules file. The cache key covers the swan release, the DuckDB version
         * and the dataset's tables — none of which move when someone edits a
         * rule, so without this an edited rules.py would be declared over stale
         * values and answer confidently wrong numbers. Hashing the source turns
         * that into a recompute.
         *
         * A snapshot written before this check existed carries no fingerprint,
         * so it mismatches and earns one full run before going fast again —
         * which is the right direction to fail in, and cheaper than throwing
         * away 21 MB that is still perfectly good. */
        const digest = fingerprint(source);
        const usable = restored && cached.rules && cached.rules.ok > 0;
        const fresh = usable && cached.rules.source === digest;
        if (usable && !fresh) progress('the rules changed since this database was cached — recomputing…');

        const rules = await loadRules(source, fresh ? cached.rules : null);
        progress(rules.fast
          ? `rules: ${rules.ok} of ${rules.total} in force `
            + `(${rules.fast.declared} declared, ${rules.fast.skipped} reused from cache)`
          : `rules: ${rules.ok} of ${rules.total} evaluated`);
        rulesInfo = rules;
        broadcast({ type: 'rules', rules });

        /* Snapshot here rather than beside the parquet load, because only now is
         * there anything worth caching for the fast path: the derived columns
         * exist. Cached before this point, the database has the tables and none
         * of the rule outputs, and declaring rules over it would read empty
         * columns and answer nothing -- silently, which is the worst version.
         *
         * After the broadcast, deliberately. Nothing waits on the cache, so a
         * page should hear that its rules are ready before this spends a second
         * writing tens of megabytes to IndexedDB.
         *
         * The tally goes in the record rather than being required to be
         * perfect. 116 of 117 is TPC-DS's *normal* state -- one statement is
         * non-stratifiable and loadRules rolls it back every time -- so
         * demanding ok === total would have cached nothing, forever, silently.
         * What matters is not that every rule evaluated but that a later load
         * can tell which ones did, so store the count and let the reader
         * decide. The failure records ride along too, so a page on the fast path
         * can still name the rule that did not evaluate rather than quietly
         * presenting 116 rules as if they were 117.
         *
         * Written on a cold load, and again when a restored database had to
         * recompute because the rules changed -- otherwise the next load would
         * pay for the same edit a second time. The wart in re-snapshotting: a
         * table left behind by a rule that was *deleted* is carried forward too,
         * since this stores the database as it stands. Reload data is the way
         * out of that, and it is the reason that button exists. */
        if (!restored || !fresh) {
          await snapshotDb(key, {
            ok: rules.ok, total: rules.total, failures: rules.failures, source: digest,
          });
        }
      })
      .catch((err) => {
        rulesInfo = { error: String(err && err.message || err) };
        broadcast({ type: 'rules', error: rulesInfo.error });
      });
  }

  /* Report the cache outcome rather than leaving it to be inferred.
   *
   * A restore that silently did not happen looks exactly like one that did --
   * the tables get built from parquet instead and every query answers the same
   * numbers. Scraping the progress line cannot tell them apart either, since
   * `restoring the cached database…` is overwritten within milliseconds by
   * `building the model…`. So the fact has to travel in the reply, where a test
   * can assert it. */
  return {
    ...manifest,
    model: info,
    rules: null,
    rulesPending: !!manifest.rules,
    cache: cached ? { restored: true, rules: cached.rules || null } : { restored: false },
  };
}

/* Re-read the model and recompute every rule against the data as it stands.
 *
 * The escape hatch for the fast path. A warm load declares the rules and reuses
 * values a previous session computed, guarded by a fingerprint of rules.py --
 * which catches an edited rules file but cannot catch a define whose *inputs*
 * were changed by something outside the cache key. This is what to press when
 * the answers are not trusted, and it is deliberately a button rather than a
 * heuristic: the alternative is guessing, and a wrong guess here serves stale
 * numbers with no signal at all.
 *
 * The model module is re-imported rather than reused. Re-declaring a property
 * onto a model that already has it is not obviously safe, a fresh import is
 * simpler to reason about, and it is what the button says -- the model source is
 * read again too, not only the rules. */
async function reloadModel() {
  if (!datasetInfo) throw new Error('there is no model loaded to reload');
  const base = new URL('dataset/', BASE);
  const manifest = await (await fetch(new URL('manifest.json', base))).json();

  /* Close the old connection before the new import opens its own. DuckDB shares
   * one instance per file within a process, so the two would most likely
   * coexist -- but only while their configuration matches, and "most likely" is
   * a poor foundation for the button someone reaches for when they already
   * suspect something is wrong. */
  pyodide.runPython(`
_m = _EH.get("model")
if _m is not None:
    try:
        _m.con.close()
    except Exception:
        pass
_EH["model"] = None
# loadModel adopted this connection from the model, so it is the one just closed.
# Clearing it matters if the re-import below fails: _eh_con() then opens a fresh
# one rather than handing every later query a dead handle.
_EH["con"] = None
`);

  progress('rebuilding the model…');
  const spec = manifest.model;
  const files = {};
  for (const name of spec.files) {
    files[name] = await (await fetch(new URL(name, base))).text();
  }
  const info = JSON.parse(await loadModel({ entry: spec.entry, files }));
  await loadPromptModule(manifest, base);
  /* `cache` is left as it was on purpose: it records how the *data* got here,
   * which a model rebuild does not change. */
  datasetInfo = { ...datasetInfo, model: info };
  rulesInfo = null;
  if (!manifest.rules) return { model: info, rules: null };

  const source = await (await fetch(new URL(manifest.rules, base))).text();
  const rules = await loadRules(source, null);   // full: defines and all
  progress(`rules: ${rules.ok} of ${rules.total} evaluated`);
  rulesInfo = rules;
  broadcast({ type: 'rules', rules });

  /* Re-cache, so the next visit starts from the recomputed values rather than
   * repeating this. The fingerprint goes in with them: whatever prompted the
   * rebuild, what is stored now was computed from the rules file as it is. */
  await snapshotDb(snapshotKey(engineManifest || {}, manifest), {
    ok: rules.ok, total: rules.total, failures: rules.failures, source: fingerprint(source),
  });
  return { model: info, rules };
}

/* Evaluate a rules file one top-level statement at a time, rolling back what
 * fails.
 *
 * A plain import is all-or-nothing, and rule evaluation is deferred: every later
 * execute() calls flush_and_evaluate(), so one rule that generates SQL DuckDB
 * rejects re-raises on every subsequent query and takes the engine down with it.
 * Dark Matter's TPC-DS rules have five such statements out of 117 — without
 * this, loading them costs you the other 112 and the browser engine with them.
 *
 * The same approach as tools/check_rules.py, in Python here because the rules
 * are Python and the model they attach to lives in Pyodide. */
async function loadRules(source, cachedTally = null) {
  /* Passing a tally puts this on the fast path: declare the rules, do not
   * recompute them.
   *
   * rules.py is two kinds of statement. `Customer.store_spend = m.Property(…)`
   * declares a name — it is what makes the property resolve on the model object,
   * and skipping it fails every query touching it with "referenced property not
   * declared in the schema". `m.define(Customer.store_spend(…))` computes the
   * values, which is where all the time goes. Against a database where a
   * previous session already materialised those values, running only the
   * declarations answers every rule kind correctly — derivations, aggregates and
   * classifications alike. Measured in CPython: 20.7 s to 0.6 s, with all five
   * expectations from verify_tpcds_browser unchanged. docs/rules-fast-path.md
   * has the table.
   *
   * The filter is structural rather than textual. Declarations are assignments;
   * defines are bare expressions wrapping a define() call. Matching the node
   * shape means a define inside a string, a comment or a name like `redefine`
   * cannot be mistaken for one, which a regex over source lines would do.
   *
   * Note what this does NOT change: the stepper below, its rollback, its
   * failure records. The fast path is a shorter list of nodes handed to exactly
   * the same loop, which is deliberate -- a second loading path that skipped the
   * rollback would trade 20 s for the entire engine, since one non-stratifiable
   * statement poisons every subsequent query, including plain column reads. */
  const declareOnly = !!cachedTally;
  pyodide.globals.set('_eh_rules_src', source);
  pyodide.globals.set('_eh_declare_only', declareOnly);
  /* The loop is driven from JS, one statement per call, so it can yield.
   *
   * As a single runPythonAsync it was 40 seconds of uninterrupted synchronous
   * Python: this worker has one thread, so nothing else ran — not a query, not
   * even the delivery of the message asking for one. Evaluating in the
   * background is only meaningful if the background actually gives the thread
   * back, which is what the await below does. */
  await pyodide.runPythonAsync(`
import ast as _ast

_model = _EH["model"]
_ns = _EH["ns"]
_tree = _ast.parse(_eh_rules_src, filename="rules.py")
_lines = _eh_rules_src.splitlines()

def _eh_is_define(_n):
    # A statement that computes values, rather than one that declares a name.
    if not isinstance(_n, _ast.Expr):
        return False
    for _sub in _ast.walk(_n):
        if isinstance(_sub, _ast.Call):
            _f = _sub.func
            if getattr(_f, "attr", None) == "define" or getattr(_f, "id", None) == "define":
                return True
    return False

_EH["rule_statements"] = len(_tree.body)
_EH["rule_nodes"] = ([_n for _n in _tree.body if not _eh_is_define(_n)]
                     if _eh_declare_only else _tree.body)
_EH["rule_failures"] = []

def _marks():
    return (len(_model.datalog_rules),
            len(getattr(_model, "_pending_rules", []) or []),
            getattr(_model, "datalog_dirty", False))

def _rollback(m):
    n_datalog, n_pending, dirty = m
    del _model.datalog_rules[n_datalog:]
    if getattr(_model, "_pending_rules", None) is not None:
        del _model._pending_rules[n_pending:]
    _model.datalog_dirty = dirty

def _eh_has_bodyless_new(_tree):
    """Any statement creating entities with .new() and no .where() body?

    Such a rule is not a Datalog rule: it sits in model._pending_rules and is
    consumed by the first flush_pending_rules() that sees it, running exactly
    once and never again. With evaluation suppressed that flush would consume it
    against still-empty data -- permanently. It is the one shape the fast path
    cannot defer, so a file containing one is loaded the slow way.

    Conservative on purpose: it asks whether the statement mentions .where at
    all, not whether that .where binds to this .new(). Anything it cannot prove
    safe is merely slower.
    """
    for _n in _tree.body:
        _attrs = [_a for _a in _ast.walk(_n) if isinstance(_a, _ast.Attribute)]
        if any(_a.attr == "new" for _a in _attrs) and not any(_a.attr == "where" for _a in _attrs):
            return True
    return False


_EH["rules_bodyless"] = _eh_has_bodyless_new(_tree)


def _eh_rules_fast():
    """Declare the whole file with evaluation suppressed, then evaluate once.

    This is tools/swan_sidecar.py's load_rules_file, which measured 26.3s to
    1.3s on this same file. The saving is not in skipping our own
    flush_and_evaluate -- dropping that does nothing, because
    ConnectionProxy.execute() calls it on every statement it runs, including
    the DDL each new m.Property() issues, so the flush just moves onto the next
    property. auto_evaluate_rules is the flag evaluate_datalog_rules checks
    first and returns on, so it is the one that actually defers the work.

    Returns "" on success, or the error that made it give up.
    """
    _m = _marks()
    _prior = getattr(_model, "auto_evaluate_rules", True)
    try:
        _model.auto_evaluate_rules = False
        try:
            for _node in _EH["rule_nodes"]:
                exec(compile(_ast.Module(body=[_node], type_ignores=[]), "rules.py", "exec"), _ns)
        finally:
            _model.auto_evaluate_rules = _prior
        # define() sets this itself; set it again because the fast path's whole
        # premise is that no evaluation has happened yet, and this is the flag
        # that says one is owed.
        _model.datalog_dirty = True
        _model.flush_and_evaluate()
        return ""
    except Exception as _e:
        _model.auto_evaluate_rules = _prior
        _rollback(_m)
        return str(_e).strip().splitlines()[0][:200] or "evaluation failed"


def _eh_rule_step(i):
    _node = _EH["rule_nodes"][i]
    _m = _marks()
    try:
        exec(compile(_ast.Module(body=[_node], type_ignores=[]), "rules.py", "exec"), _ns)
        _model.flush_and_evaluate()
    except Exception as _e:
        _rollback(_m)
        # Same record the sidecar reports, so one banner renders either engine.
        _EH["rule_failures"].append({"line": _node.lineno,
                                     "statement": _lines[_node.lineno - 1].strip()[:120],
                                     "error": str(_e).strip().splitlines()[0][:200]})
`);

  const total = pyodide.runPython('len(_EH["rule_nodes"])');
  const bodyless = pyodide.runPython('bool(_EH["rules_bodyless"])');

  /* Declare the whole file with evaluation suppressed, then evaluate once.
   *
   * Every evaluation pass recomputes *every* registered rule from scratch, so
   * statement N re-derives the N-1 before it. Measured in this browser on
   * TPC-DS the curve is unmistakable — the first rules landed in ~0.4s each and
   * the last in ~4s, the file taking over 200s. The sidecar hit the same wall
   * and solved it (tools/swan_sidecar.py:load_rules_file, 26.3s → 1.3s); this
   * is that fix on the browser side, so the two engines load rules the same way
   * as well as answering the same.
   *
   * Note what does NOT work, because it is the obvious thing to try and I tried
   * it: dropping our own flush_and_evaluate() from the loop changes nothing.
   * ConnectionProxy.execute() calls it on every statement it runs, including
   * the DDL each new m.Property() issues, and rules.py alternates declaration
   * with definition — so the flush does not go away, it moves onto the next
   * property. `auto_evaluate_rules` is the flag evaluate_datalog_rules checks
   * first, which is why it is the one that defers anything.
   *
   * The per-statement stepper stays, for the two cases that need it. */
  let usedFastPath = false;
  if (!bodyless) {
    progress(`${declareOnly ? 'declaring' : 'evaluating'} ${total} rules…`);
    // Yield first: the fast path is one synchronous span, and this worker has a
    // single thread. Better to hand it back once here than to look hung.
    await new Promise((resolve) => setTimeout(resolve, 0));
    const fastError = pyodide.runPython('_eh_rules_fast()');
    usedFastPath = !fastError;
    if (fastError) progress('a rule did not evaluate — finding which…');
  }

  /* Statement by statement: either because the file has a bodyless .new() rule
   * the fast path cannot defer, or because the single evaluation raised and the
   * only way to name the culprit is to reintroduce them one at a time. Rolling
   * back a failure matters more than the speed — one non-stratifiable statement
   * left in the model poisons every later query, including plain column reads
   * that touch no rule. The quadratic cost is now paid only by a file that is
   * already failing, or one shaped so it cannot be avoided. */
  if (!usedFastPath) {
    for (let i = 0; i < total; i += 1) {
      pyodide.runPython(`_eh_rule_step(${i})`);
      await new Promise((resolve) => setTimeout(resolve, 0));
      if (i % 4 === 3 || i === total - 1) {
        progress(`${declareOnly ? 'declaring' : 'evaluating'} rules… ${i + 1} of ${total}`);
      }
    }
  }

  const ran = JSON.parse(pyodide.runPython(`json.dumps({
    "statements": _EH["rule_statements"],
    "ran": len(_EH["rule_nodes"]),
    "failures": _EH["rule_failures"],
    "concepts": len(_EH["model"]._concepts),
})`));

  if (!declareOnly) {
    return {
      total: ran.statements,
      ok: ran.ran - ran.failures.length,
      failures: ran.failures,
      concepts: ran.concepts,
    };
  }

  /* On the fast path `ok` and `total` describe the rules in force, not the
   * statements this call ran -- otherwise a page that reused cached values would
   * report "69 of 69 rules" and read as though the file had shrunk.
   *
   * What is in force is what the caching load evaluated, less anything that
   * failed to declare now. A failed declaration is not cosmetic: the property
   * stops resolving, so its cached values become unreachable and it is correctly
   * no longer counted. Its record joins the caching load's own failures, which
   * ride along in the snapshot so a warm page can still say which rule did not
   * evaluate rather than quietly presenting 116 rules as 117. */
  return {
    total: cachedTally.total,
    ok: Math.max(0, (cachedTally.ok || 0) - ran.failures.length),
    failures: [...(cachedTally.failures || []), ...ran.failures],
    concepts: ran.concepts,
    fast: { declared: ran.ran, skipped: ran.statements - ran.ran },
  };
}

/* Stage the fine-tune's prompt shape, when the dataset carries one.
 *
 * `render_schema` renders a live Swan model in the nested foreign-key form the
 * adapter was trained on, which is not what pyrel_duckdb.nlq emits. That is not
 * a cosmetic difference: black_swan's own measurements name the flat form as the
 * largest single failure class, 11 of 52 evaluation failures naming a property
 * that belongs to the parent as if it were local. So the browser renders the
 * shape the weights expect.
 *
 * It runs the exported copy of tools/local_generator.py rather than a JS
 * translation of it, because the model is right here in Pyodide and a second
 * implementation would drift from the training data with nothing to catch it —
 * tools/verify_local_generator.py asserts that file against the training split,
 * and a reimplementation would be outside what it checks.
 *
 * Absent is fine and quiet-ish: a dataset staged before this existed has no
 * promptModule, and nlq falls back to its own schema renderer, which is right
 * for a general model and merely off-distribution for the fine-tune. */
async function loadPromptModule(manifest, base) {
  if (!manifest.promptModule) return;
  try {
    const source = await (await fetch(new URL(manifest.promptModule, base))).text();
    pyodide.FS.writeFile('/eh/model/nlq_prompt.py', source);
    pyodide.runPython(`
import importlib, sys
sys.modules.pop("nlq_prompt", None)
_EH["prompt_mod"] = importlib.import_module("nlq_prompt")
`);
  } catch (err) {
    _EHpromptFailed = String(err && err.message || err);
    progress(`the fine-tune prompt shape did not load (${_EHpromptFailed}) — falling back to swan's schema`);
  }
}
let _EHpromptFailed = null;

/* Build the model inside Pyodide from source the caller supplies. The model is
 * Python, so it is defined the same way it is everywhere else — by running the
 * module — rather than by inventing a serialization format for it. */
async function loadModel(spec) {
  /* Accepts a bare source string, or {entry, files} for a model split across
   * modules. The multi-file form writes them to a directory and imports the
   * entry the ordinary way, because these modules import *each other* — TPC-DS's
   * model.py imports its four sales-channel modules at the bottom and each of
   * those does `from model import …` back. exec'ing a concatenation cannot
   * satisfy that; a directory on sys.path can. */
  const files = typeof spec === 'string'
    ? { 'model.py': spec }
    : spec.files;
  const entry = typeof spec === 'string' ? 'model' : spec.entry;

  pyodide.FS.mkdirTree('/eh/model');
  for (const [name, text] of Object.entries(files)) {
    pyodide.FS.writeFile(`/eh/model/${name}`, text);
  }
  pyodide.globals.set('_eh_entry', entry);

  await pyodide.runPythonAsync(`
import importlib, sys

if "/eh/model" not in sys.path:
    sys.path.insert(0, "/eh/model")
# Drop any previous load, so a second call re-imports rather than returning the
# module object from the first one and silently ignoring new source.
for _name in [n for n in sys.modules if n == _eh_entry or n.startswith("model")]:
    sys.modules.pop(_name, None)

# Only this import gets to claim /eh/eh.duckdb — see _eh_own_db. The loader and
# the model must agree on one database (swan issue #155); nothing else should.
_eh_own_db(True)
try:
    _module = importlib.import_module(_eh_entry)
finally:
    _eh_own_db(False)
_EH["model"] = next(
    (v for v in vars(_module).values()
     if hasattr(v, "_concepts") and hasattr(v, "con")), None
)
if _EH["model"] is None:
    raise RuntimeError("no pyrel_duckdb.Model instance found in " + _eh_entry)
# Kept for the rules loader: rules.py expects the model module's own namespace
# (m, Customer, Integer, …) in scope, exactly as importing it normally would.
_EH["ns"] = vars(_module)
# Adopt the model's own connection: from here it is the one holding the file, so
# sql() and pyrel() cannot drift onto different databases.
_EH["con"] = _EH["model"].con
`);
  return pyodide.runPython(`json.dumps({
      "name": getattr(_EH["model"], "name", "model"),
      "concepts": len(_EH["model"]._concepts),
  })`);
}

/* Natural language, with generation handed out to whatever the page points at.
 *
 * Everything except producing text already runs here: `pyrel_duckdb.nlq` is in
 * the compiled module, and the worker has been importing it all along for
 * execute_pyrel_code and serialize_model_schema. `answer_question` takes the
 * model, the question and a `generate_fn` — so the only thing missing from a
 * browser-only natural-language path was a way to turn a prompt into a string.
 *
 * That makes the backend pluggable rather than chosen: a self-hosted fine-tune,
 * an OpenAI-compatible server, eventually a model running in this same browser.
 * None of them change anything above this function.
 *
 * XMLHttpRequest rather than fetch, and synchronously, because answer_question
 * is synchronous Python: it calls generate_fn(prompt), inspects what comes back,
 * and re-prompts around its own sandbox. There is no point in the call stack
 * where a promise could be awaited. Sync XHR is barred on the main thread and
 * allowed in a worker, which is one more thing this worker is for. The cost is
 * that the thread blocks for the length of a generation — the same bargain rule
 * evaluation made before it learned to yield, and an explicit question is a
 * reasonable place to pay it.
 */
/* Gradio's REST API, for a Space that cannot expose an OpenAI-compatible route.
 *
 * ZeroGPU -- the free GPU tier -- only works with the Gradio SDK, and the
 * platform already runs a server on the Space's ports, so mounting FastAPI
 * beside it fights for 7860/7861 and its GPU supervisor never finds a
 * @spaces.GPU function because it is watching its own app. A Gradio endpoint is
 * what such a Space can offer, so this speaks it. What it buys is f16 instead of
 * q4, which is worth about 3 questions in 22 on a wide schema.
 *
 * Two steps rather than one: POST returns an event id, and a second GET drains
 * the result as server-sent events. Both are synchronous XHRs for the same
 * reason the OpenAI path is -- answer_question is synchronous Python and there
 * is nowhere to await a promise. The SSE body arrives whole because the stream
 * closes when the run completes.
 */
/** Host of a URL, for saying which generator answered without printing a path. */
function hostOf(url) {
  try {
    return new URL(url).host;
  } catch {
    return String(url);
  }
}

/* Gradio's REST API, for a Space that cannot expose an OpenAI-compatible route.
 *
 * ZeroGPU -- the free GPU tier -- only works with the Gradio SDK, and the
 * platform already runs a server on the Space's ports, so mounting FastAPI
 * beside it fights for 7860/7861 and its GPU supervisor never finds a
 * @spaces.GPU function because it is watching its own app.
 *
 * Two steps rather than one: POST returns an event id, and a second GET drains
 * the result as server-sent events. Both are synchronous XHRs for the same
 * reason the OpenAI path is -- answer_question is synchronous Python and there
 * is nowhere to await a promise. The SSE body arrives whole because the stream
 * closes when the run completes.
 */
function callGradio(endpoint, _model, messagesJson) {
  const xhr = new XMLHttpRequest();
  xhr.open('POST', endpoint, false);
  xhr.setRequestHeader('Content-Type', 'application/json');
  xhr.send(JSON.stringify({ data: [messagesJson] }));
  if (xhr.status < 200 || xhr.status >= 300) {
    throw new Error(`answered ${xhr.status}: ${String(xhr.responseText).slice(0, 160)}`);
  }
  const started = JSON.parse(xhr.responseText);
  const id = started && (started.event_id || started.eventId);
  if (!id) throw new Error(`no event id: ${JSON.stringify(started).slice(0, 120)}`);

  const poll = new XMLHttpRequest();
  poll.open('GET', `${endpoint.replace(/\/$/, '')}/${id}`, false);
  poll.send();
  if (poll.status < 200 || poll.status >= 300) throw new Error(`stream answered ${poll.status}`);

  /* The LAST data frame, not the first: the stream carries progress events
   * before the result, and an earlier one is a queue position rather than a
   * program. A frame of `null` is how this endpoint reports a refused GPU
   * allocation -- no message, no traceback -- so it has to count as a failure
   * rather than as an empty answer, or the fallback never fires. */
  let out = null;
  for (const line of String(poll.responseText).split('\n')) {
    if (!line.startsWith('data: ')) continue;
    try {
      const parsed = JSON.parse(line.slice(6));
      if (Array.isArray(parsed) && parsed.length && typeof parsed[0] === 'string') out = parsed[0];
    } catch { /* keepalives and non-JSON frames */ }
  }
  if (out === null) {
    throw new Error(`no result: ${String(poll.responseText).slice(0, 120).replace(/\n/g, ' ')}`);
  }
  return out;
}

/* The OpenAI-compatible chat shape, which vLLM, llama.cpp's server, Ollama and
 * TGI all speak. `model` is passed through untouched: hosts serving a single
 * fine-tune usually ignore it, and ones serving several need it. */
function callOpenAI(endpoint, model, messagesJson) {
  const xhr = new XMLHttpRequest();
  xhr.open('POST', endpoint, false);
  xhr.setRequestHeader('Content-Type', 'application/json');
  const body = {
    messages: JSON.parse(messagesJson),
    // Deterministic. A schema-to-code task has a right answer, and sampling
    // makes a wrong one irreproducible.
    temperature: 0,
    stream: false,
  };
  if (model) body.model = model;
  xhr.send(JSON.stringify(body));
  if (xhr.status < 200 || xhr.status >= 300) {
    throw new Error(`answered ${xhr.status}: ${String(xhr.responseText).slice(0, 160)}`);
  }
  let data;
  try { data = JSON.parse(xhr.responseText); } catch {
    throw new Error(`did not answer JSON: ${String(xhr.responseText).slice(0, 120)}`);
  }
  const text = data?.choices?.[0]?.message?.content
    ?? data?.choices?.[0]?.text
    ?? data?.content?.[0]?.text          // Anthropic messages
    ?? data?.message?.content            // Ollama /api/chat
    ?? data?.response;                   // Ollama /api/generate
  if (typeof text !== 'string') {
    throw new Error(`no text in the reply: ${JSON.stringify(data).slice(0, 120)}`);
  }
  return text;
}

/* Try each generator in turn, and remember which one answered.
 *
 * The default chain is a token-holding proxy onto a GPU Space, followed by a CPU
 * one. The GPU is ~15x faster and runs the weights unquantized, which is worth
 * about three questions in twenty-two on a wide schema -- but its daily quota is
 * finite and shared by everyone reading the page, and when it runs out the
 * allocation is refused in ~0.4s with no status code, no message and no
 * traceback. Falling through to the slower host turns that from an outage into a
 * slow answer, which is why the chain outlives the problem that created it.
 *
 * The refusal being *fast* is what makes the chain cheap: a doomed first hop
 * costs less than half a second before the real one starts.
 *
 * Chosen by URL shape rather than a flag, so a chain can mix protocols: the
 * GPU Space speaks Gradio because ZeroGPU forces it, the CPU one speaks OpenAI.
 */
function generateVia(endpoints, model) {
  const chain = (Array.isArray(endpoints) ? endpoints : [endpoints]).filter(Boolean);
  const state = { host: null, fellBack: false, tried: [] };

  const generate = (messagesJson) => {
    for (let i = 0; i < chain.length; i += 1) {
      const url = chain[i];
      const call = /\/gradio_api\/call\//.test(url) ? callGradio : callOpenAI;
      try {
        const out = call(url, model, messagesJson);
        state.host = hostOf(url);
        state.fellBack = i > 0;
        return out;
      } catch (err) {
        const why = String(err && err.message || err);
        state.tried.push(`${hostOf(url)}: ${why}`);
        if (i === chain.length - 1) {
          throw new Error(`no generator answered — ${state.tried.join('; ')}`);
        }
        progress(`${hostOf(url)} did not answer, trying ${hostOf(chain[i + 1])}…`);
      }
    }
    throw new Error('no generation endpoint configured');
  };

  generate.state = state;
  return generate;
}

async function runAsk({ question, history, endpoint, model, prune = true }) {
  if (!endpoint) throw new Error('no generation endpoint — load the page with ?nlq=<url>');
  if (!pyodide.runPython('_EH["model"] is not None')) {
    throw new Error('no model loaded — load the dataset first');
  }
  const bridge = generateVia(endpoint, model);
  pyodide.globals.set('_eh_generate_js', bridge);
  pyodide.globals.set('_eh_question', question);
  pyodide.globals.set('_eh_prune', prune !== false);
  pyodide.globals.set('_eh_history', history && history.length ? JSON.stringify(history) : null);

  const out = await pyodide.runPythonAsync(`
from pyrel_duckdb.nlq import NLQGenerationError, answer_question

_kw = {}
_pm = _EH.get("prompt_mod")
_eh_state = {"code": "", "escalated": False, "pruned": None, "full": None}

if _eh_history:
    _kw["history"] = json.loads(_eh_history)

if _pm is not None:
    # Rebuild the adapter's own turn, rather than passing swan's prompt through.
    #
    # The shape has two halves and only one of them was right before: the system
    # prompt was the trained one, but the user turn was still swan's wrapper with
    # our schema inside it. LocalGenerator ignores swan's prompt for exactly this
    # reason -- it reads only the *error* back out of a retry prompt and builds
    # the Schema/Question turn itself. This is the same thing on the JS side of
    # the bridge, calling that same build_user_turn and extract_code, so there is
    # one definition of the shape rather than a copy to drift from.
    #
    # Pruning, and the full schema behind it: the first attempt sees only the
    # concepts the question appears to need, and a retry escalates to everything.
    # TPC-DS renders to 8,203 tokens whole against the 317-640 the adapter was
    # trained on, and pruning brings that to ~1,850.
    _eh_state["full"] = _pm.render_schema(_EH["model"])
    try:
        _eh_state["pruned"] = (_pm.render_schema(_EH["model"], _eh_question)
                               if _eh_prune else _eh_state["full"])
    except TypeError:
        # The staged module is a *copy* of tools/local_generator.py taken at
        # export time, so it can be older than this worker -- a dataset staged
        # before pruning existed has a render_schema that takes only the model.
        # Falling back to the whole schema costs speed; raising would cost the
        # feature, and re-running the export is not something a page can do.
        _eh_state["pruned"] = _eh_state["full"]
    _eh_sys = getattr(_pm, "SYSTEM", None)

    def _eh_generate(_swan_prompt):
        _schema = _eh_state["pruned"]
        _retry = None
        _marker = "# That attempt failed"
        if _marker in _swan_prompt and _eh_state["code"]:
            _err = _swan_prompt.split(_marker, 1)[1].strip()
            _err = _err.split("Write a corrected version", 1)[0].strip()[:800]
            _retry = {"code": _eh_state["code"], "error": _err}
            _schema = _eh_state["full"]
            _eh_state["escalated"] = True
        _msgs = []
        if _eh_sys:
            _msgs.append({"role": "system", "content": _eh_sys})
        _msgs.append({"role": "user",
                      "content": _pm.build_user_turn(_schema, _eh_question)})
        if _retry:
            # Same minimal correction turn LocalGenerator sends: the failed
            # program, the error, and the instruction. The adapter never saw a
            # repair turn in training, so anything more elaborate is invention.
            _msgs.append({"role": "assistant", "content": _retry["code"]})
            _msgs.append({"role": "user", "content":
                          "That program failed: " + _retry["error"]
                          + "\\nReply with only the corrected PyRel program."})
        _code = _pm.extract_code(_eh_generate_js(json.dumps(_msgs)))
        _eh_state["code"] = _code
        return _code
else:
    # No staged prompt module: swan's own prompt, unmodified. Right for a general
    # model, merely off-distribution for the fine-tune.
    def _eh_generate(_swan_prompt):
        return _eh_generate_js(json.dumps([{"role": "user", "content": _swan_prompt}]))

try:
    _r = answer_question(_EH["model"], _eh_question, _eh_generate, **_kw)
except NLQGenerationError as _e:
    # The pipeline has already retried and translated each failure. Surface the
    # last attempt rather than "generation failed", so the page can show what was
    # tried and why it did not run.
    raise RuntimeError(json.dumps({
        "error": str(_e),
        "code": getattr(_e, "last_code", None),
        "detail": getattr(_e, "last_error", None),
    }))

json.dumps({
    "question": _r.question,
    "code": _r.code,
    "attempts": _r.attempts,
    "rows": _r.result if isinstance(_r.result, list) else str(_r.result),
    "examples": [_ex.get("question", "") for _ex in (_r.examples_used or [])],
    # Reported rather than inferred: an answer that needed the whole schema is a
    # question the pruner got wrong, and that is worth being able to count.
    "schema": {
        "pruned": bool(_eh_prune and _pm is not None),
        "escalated": _eh_state["escalated"],
        "chars": len(_eh_state["pruned"] or "") if _eh_state["pruned"] else None,
        "fullChars": len(_eh_state["full"] or "") if _eh_state["full"] else None,
    },
}, default=str)
`);
  const answer = JSON.parse(out);
  /* Which generator actually answered, and whether it was the first choice.
   * The chain exists because the fast host refuses anonymous callers, so "it
   * worked" and "it worked on the host we wanted" are different facts, and the
   * gap between them is ~110 seconds of waiting. The page shows both. */
  answer.via = { host: bridge.state.host, fellBack: bridge.state.fellBack, tried: bridge.state.tried };
  return answer;
}


/* Installed once per template. Kept out of the function above so the Python is
 * one readable block rather than a string with holes in it. */
const TEMPLATE_SETUP = String.raw`
import io as _io, json as _json, sys as _sys, ast as _ast, traceback as _tb
import pandas as _pd, duckdb as _duckdb

if "_tpl" not in globals():
    # One session per page, not one per worker.
    #
    # This is a SharedWorker: every template page in the browser is served by
    # this one Pyodide. _tpl used to be a single dict, so opening a second
    # template reset the namespace, the connection and the table map underneath
    # the first — the second page's prepare wiped the first page's variables
    # mid-run. Two tabs produced "NameError: name 'graph' is not defined" in
    # *both*, and a DuckDB "different configuration" error when the timing put
    # two connections on one file. Neither looked like a tab collision; both
    # looked like the template was broken.
    #
    # Sessions are keyed by page. _tpl still names the active one, so every
    # helper below reads it unchanged; _tpl_activate swaps which session that
    # is, and does it inside the same call that runs a cell so no message can
    # land between the two.
    _tpl_sessions = {}
    _tpl_active = None
    _tpl = {"con": None, "tables": {}, "ns": {}, "written": []}
    _tpl_real_read_csv = _pd.read_csv

    def _tpl_read_csv(path, *a, **kw):
        key = str(path).replace(chr(92), "/")
        base = key.rsplit("/", 1)[-1]
        table = _tpl["tables"].get(key) or _tpl["tables"].get(base)
        if table is None or _tpl["con"] is None:
            # Not one of ours. Let pandas fail in its own words rather than
            # ours -- a template reading something we never extracted should
            # say "no such file", not raise from inside a shim.
            return _tpl_real_read_csv(path, *a, **kw)
        return _tpl_frame(table, *a, **kw)

    def _tpl_frame(table, *a, **kw):
        """Hand back what read_csv would have produced, not what DuckDB knows.

        DuckDB's sniffer is better, and that is the problem: it parses
        "1985-03-12" into datetime64 and a nullable integer into masked Int64,
        where read_csv gives a string and float64. The templates were written
        against read_csv's answers. So the database carries bytes and pandas
        does the typing -- verified column-for-column against the original CSVs.
        Do not "improve" this by returning .df() directly.
        """
        cols = [r[0] for r in _tpl["con"].execute('DESCRIBE "%s"' % table).fetchall()]
        sel = ", ".join('CAST("%s" AS VARCHAR) AS "%s"' % (c, c) for c in cols)
        text = _tpl["con"].execute('SELECT %s FROM "%s"' % (sel, table)).df()
        return _tpl_real_read_csv(_io.StringIO(text.to_csv(index=False)), *a, **kw)

    _pd.read_csv = _tpl_read_csv

    def _tpl_render_frame(frame):
        head = frame.head(50)
        return {"kind": "frame", "columns": [str(c) for c in head.columns],
                "rows": _json.loads(head.to_json(orient="values", default_handler=str)),
                "total": int(len(frame))}

    def _tpl_render(value):
        """A cell's trailing expression, the way a notebook shows it."""
        if value is None:
            return None
        if isinstance(value, _pd.DataFrame):
            return _tpl_render_frame(value)
        # A PyRel Query carries no __repr__ of its own -- it inherits object's,
        # so a cell ending in one reads "<pyrel_duckdb.query.Query object at
        # 0x...>". That is not a browser artifact; the same cell shows the same
        # string in Jupyter, which is why so many of these notebooks end on it.
        # The rows are one call away, and someone who ended a cell with a query
        # wanted the answer, not the address of the object that would give it.
        if hasattr(value, "to_df"):
            try:
                out = value.to_df()
                if isinstance(out, _pd.DataFrame):
                    return _tpl_render_frame(out)
            except Exception as exc:
                # Evaluating is the risk this takes; say so rather than falling
                # back to the same opaque repr and looking like nothing tried.
                return {"kind": "text",
                        "text": "%s could not be evaluated: %s" % (type(value).__name__, exc)}
        return {"kind": "text", "text": repr(value)[:4000]}

    def _tpl_run(src):
        """Run one cell: stdout, the trailing expression, or the error.

        Split into "everything but the last statement" and "the last one" so a
        cell ending in an expression shows its value, which is most of what
        makes a notebook readable. A cell ending in an assignment shows nothing,
        exactly as it would in Jupyter.
        """
        buf = _io.StringIO()
        old = _sys.stdout
        _sys.stdout = buf
        result, error = None, None
        try:
            tree = _ast.parse(src)
            body, tail = tree.body[:-1], tree.body[-1:] if tree.body else []
            if body:
                exec(compile(_ast.Module(body=body, type_ignores=[]), "<cell>", "exec"), _tpl["ns"])
            if tail:
                node = tail[0]
                if isinstance(node, _ast.Expr):
                    value = eval(compile(_ast.Expression(node.value), "<cell>", "eval"), _tpl["ns"])
                    result = _tpl_render(value)
                else:
                    exec(compile(_ast.Module(body=tail, type_ignores=[]), "<cell>", "exec"), _tpl["ns"])
        except Exception:
            # The last frames only. A template traceback runs through pandas and
            # pyrel internals, and the useful line is the last one.
            lines = _tb.format_exc().strip().splitlines()
            error = chr(10).join(lines[-3:])[:1200]
        finally:
            _sys.stdout = old
        return _json.dumps({"stdout": buf.getvalue()[:8000], "result": result, "error": error},
                           default=str)

def _tpl_materialise():
    """Write each table back out as the CSV file the notebook expects to find.

    The read_csv shim above only intercepts pandas. Templates do not all read
    their data through pandas: transaction-screening-local hands the path
    straight to DuckDB --

        model.con.execute("CREATE TABLE txn AS SELECT * FROM read_csv_auto('data/transactions.csv')")

    -- which looks in the filesystem, finds nothing, and fails with "No files
    found that match the pattern". That was our gap, not the template's bug, and
    it would have been reported on the page as the template being broken.

    So put the bytes where any reader will find them. This is the general fix
    and the shim is now the belt to its braces: whatever a template uses to read
    a CSV -- pandas, DuckDB, csv.reader -- there is a real file there, holding
    the same text the original did.

    Written to "data/<name>" as well as "<name>": every notebook in the corpus
    resolves its data through DATA_DIR = Path("data"), and the extractor records
    only the basename.
    """
    import os as _os

    # Clear what a previous template left behind. The browser gives each page a
    # fresh worker, but run_templates.mjs drives all 49 through one Pyodide, and
    # several of them read a file called "transactions.csv". Without this, one
    # template can quietly be served another's data -- and pass, or fail, for a
    # reason that has nothing to do with it.
    for path in _tpl.get("written", []):
        try:
            _os.remove(path)
        except OSError:
            pass
    _tpl["written"] = []

    if _tpl["con"] is None:
        return
    for table in set(_tpl["tables"].values()):
        cols = [r[0] for r in _tpl["con"].execute('DESCRIBE "%s"' % table).fetchall()]
        sel = ", ".join('CAST("%s" AS VARCHAR) AS "%s"' % (c, c) for c in cols)
        text = _tpl["con"].execute('SELECT %s FROM "%s"' % (sel, table)).df().to_csv(index=False)
        for name in [n for n, t in _tpl["tables"].items() if t == table]:
            for path in (name, "data/" + name.rsplit("/", 1)[-1]):
                _os.makedirs(_os.path.dirname(path) or ".", exist_ok=True)
                with open(path, "w", newline="") as fh:
                    fh.write(text)
                _tpl["written"].append(path)

def _tpl_activate(sid):
    """Make this page's session the one every helper above reads.

    Switching also rewrites that session's CSVs. Two templates can name a file
    the same thing — several in the corpus read "transactions.csv" — and the
    emscripten filesystem has one path for it, so whichever session prepared
    last owns the bytes. Re-materialising on the way in means a cell always
    reads its own template's data rather than whatever the other tab left.
    """
    global _tpl, _tpl_active
    sess = _tpl_sessions.get(sid)
    if sess is None:
        raise RuntimeError("no template session %r (have: %s) — reload the page"
                           % (sid, sorted(_tpl_sessions)))
    switched = _tpl_active != sid
    _tpl = sess
    _tpl_active = sid
    if switched:
        _tpl_materialise()


def _tpl_open(sid):
    """Start (or restart) a session for one page."""
    global _tpl, _tpl_active
    old = _tpl_sessions.get(sid)
    if old is not None and old.get("con") is not None:
        # Close before the caller writes the database file again: a connection
        # left open over a rewritten file is how the "different configuration"
        # error appears, and it depends on refcount timing, hence "sometimes".
        try:
            old["con"].close()
        except Exception:
            pass
    _tpl = {"con": _duckdb.connect(_tpl_db, read_only=True) if _tpl_db else None,
            "tables": dict(_tpl_map), "ns": {"__name__": "__template__"}, "written": []}
    _tpl_sessions[sid] = _tpl
    _tpl_active = sid
    _tpl_materialise()


_tpl_open(_tpl_sid)
`;

/* ---- templates ---------------------------------------------------------
 *
 * Swan's ~52 templates run here, one notebook cell at a time. This is the one
 * place the worker executes code it did not shape: `pyrel` compiles a PyRel
 * expression in a sandbox, while a template cell is ordinary Python that builds
 * its own Model, its own DuckDB connection and its own solver.
 *
 * That is a real capability, so it is worth being plain about the boundary. The
 * code comes from data/templates/, extracted from a swan checkout at build time
 * and committed — it is not user input and there is no path from the page to
 * arbitrary source. What a template can reach is what any script in this tab
 * can reach, which is the tab.
 *
 * Cells share one namespace, in order, because that is what a notebook means:
 * cell 3 uses the model cell 1 built. A fresh namespace per template keeps one
 * from leaking into the next — and, since this worker is shared, per *page*
 * rather than per worker: see the session comment in TEMPLATE_SETUP.
 */
const templateStates = new Map();     // port -> {sid, slug, loaded, cells}
let nextSessionId = 0;

/* One template operation at a time, across every page.
 *
 * Arguments reach Python through `pyodide.globals.set`, and runPythonAsync does
 * not run the body at call time — it returns a promise and the code executes
 * later. So four pages preparing at once each set `_tpl_sid` first and every
 * body then read the *last* value written: all four tabs claimed one session
 * id, and three of them addressed a session that had never been opened. The
 * cell source travels the same way, so the same race would have run one page's
 * code in another page's namespace.
 *
 * Interpolating the id into the source would fix the id and not the source,
 * which is a template's own code and must never be spliced into a string. A
 * queue fixes both: set globals and run inside a section nothing else can
 * enter. Python is single-threaded here regardless, so this costs no
 * throughput — it only stops the arguments from being overwritten in flight. */
let templateQueue = Promise.resolve();

function serializeTemplateOp(fn) {
  const done = templateQueue.then(fn, fn);
  // The chain must survive a rejection, or one failed cell wedges every page.
  templateQueue = done.then(() => {}, () => {});
  return done;
}

/** This page's session, created on its first prepare. */
function sessionFor(port) {
  let state = templateStates.get(port);
  if (!state) {
    /* Bound the worker's memory without guessing which pages are alive: a
     * session holds a model and an open connection, and a browser left open
     * across a browsing session can visit a lot of templates. Map iterates in
     * insertion order, so the oldest goes — and it is the oldest *page*, which
     * is the one least likely to still be watched. */
    while (templateStates.size >= MAX_SESSIONS) {
      dropTemplateSession(templateStates.keys().next().value);
    }
    state = { sid: `s${nextSessionId += 1}`, slug: null, loaded: new Set(), cells: 0 };
    templateStates.set(port, state);
  }
  return state;
}

/* Release a page's session. Without this every visited template leaves a
 * namespace holding a model and an open DuckDB connection, in a worker that
 * outlives the pages by design.
 *
 * Called when a page says it is going (templateClose, sent on pagehide) and
 * when the number of live sessions passes MAX_SESSIONS. Never inferred from a
 * failed postMessage — see the comment in broadcast(). */
const MAX_SESSIONS = 8;

function dropTemplateSession(port) {
  const state = templateStates.get(port);
  if (!state) return;
  templateStates.delete(port);
  if (!pyReady) return;
  try {
    pyodide.globals.set('_tpl_sid', state.sid);
    pyodide.runPython(`
_gone = _tpl_sessions.pop(_tpl_sid, None)
if _gone is not None and _gone.get("con") is not None:
    try:
        _gone["con"].close()
    except Exception:
        pass
if _tpl_active == _tpl_sid:
    _tpl_active = None
del _gone
`);
  } catch { /* the worker outlives this; a failed cleanup is not worth raising */ }
}

async function templatePrepare(port, { slug, database, tables, packages }) {
  needsPython();
  const state = sessionFor(port);

  /* Extra wheels, only when a template asks. Four use networkx for reachability
   * and centrality; loading it for the other 45 is 2 MB nobody reads.
   * Tracked per page, but loadPackage itself is global and idempotent. */
  for (const name of packages || []) {
    if (state.loaded.has(name)) continue;
    progress(`loading ${name}…`);
    await pyodide.loadPackage(name, { messageCallback: () => {}, errorCallback: () => {} });
  }

  let bytes = null;
  if (database) {
    progress('loading the template data…');
    /* Resolved against the site root, not the worker. A bare relative path in a
     * Worker resolves against the *worker's* URL — src/engine/ — so this asked
     * for src/engine/data/templates/diet.duckdb and got a 404 that read as
     * missing data rather than a wrong base. BASE points into web/swan-wasm/,
     * hence the two levels up. */
    const url = new URL(`../../data/templates/${database}`, BASE);
    const res = await fetch(url);
    if (!res.ok) throw new Error(`could not fetch ${url.pathname}: HTTP ${res.status}`);
    bytes = new Uint8Array(await res.arrayBuffer());
    pyodide.FS.writeFile(`/${database}`, bytes);
  }

  /* Set and run with no await in between. Another page's message can only be
   * delivered at an await, so globals cannot be overwritten under this call —
   * which is the whole reason activation and execution are fused below too. */
  pyodide.globals.set('_tpl_db', database ? `/${database}` : null);
  pyodide.globals.set('_tpl_map', pyodide.toPy(tables || {}));
  pyodide.globals.set('_tpl_sid', state.sid);
  await pyodide.runPythonAsync(TEMPLATE_SETUP);

  for (const name of packages || []) state.loaded.add(name);
  state.slug = slug;
  state.cells = 0;
  return { slug, database: database || null, bytes: bytes ? bytes.length : 0 };
}

async function templateCell(port, { source }) {
  needsPython();
  const state = templateStates.get(port);
  if (!state || !state.slug) throw new Error('no template prepared — send templatePrepare first');
  pyodide.globals.set('_tpl_src', source);
  pyodide.globals.set('_tpl_sid', state.sid);
  // One call, so no other page can activate its own session between the two.
  const out = await pyodide.runPythonAsync('_tpl_activate(_tpl_sid)\n_tpl_run(_tpl_src)');
  state.cells += 1;
  return JSON.parse(out);
}

/* ---- the model a template just built, as a graph -----------------------
 *
 * The Derivation Ladder reads a {nodes, edges} document with three tiers. For
 * TPC-DS that document is `data/tpcds.json`, produced offline by three scripts
 * over a checked-out model directory. A template page has no checkout — but it
 * has the model itself, standing in this worker's namespace, built by the cells
 * the reader just ran and carrying whatever they edited before running them.
 *
 * tools/live_graph.py collapses those three stages into one call against a live
 * Model. It is fetched rather than inlined: it is ~100 KB of Python across four
 * modules, it is the same code the offline path runs (so the two cannot drift),
 * and no template page pays for it unless someone opens the model view.
 */
const GRAPH_MODULES = ['extract_duckdb.py', 'extract_swan.py', 'build_graph.py', 'live_graph.py'];
let graphModulesReady = null;

async function ensureGraphModules() {
  if (graphModulesReady) return graphModulesReady;
  graphModulesReady = (async () => {
    progress('loading the model reader…');
    pyodide.FS.mkdirTree('/eh/graph');
    for (const name of GRAPH_MODULES) {
      // Same two-levels-up as the template database: a bare relative path in a
      // worker resolves against src/engine/, not the site root.
      const url = new URL(`../../tools/${name}`, BASE);
      const res = await fetch(url);
      if (!res.ok) throw new Error(`could not fetch ${url.pathname}: HTTP ${res.status}`);
      pyodide.FS.writeFile(`/eh/graph/${name}`, await res.text());
    }
    await pyodide.runPythonAsync(`
import sys
if "/eh/graph" not in sys.path:
    sys.path.insert(0, "/eh/graph")
`);
  })();
  return graphModulesReady;
}

async function templateGraph(port) {
  needsPython();
  const state = templateStates.get(port);
  if (!state || !state.slug) throw new Error('no template prepared — send templatePrepare first');
  await ensureGraphModules();
  progress('reading the model…');
  pyodide.globals.set('_tpl_sid', state.sid);
  const out = await pyodide.runPythonAsync(`
_tpl_activate(_tpl_sid)
import json as _json
from pyrel_duckdb import Model as _Model
from live_graph import graph_for_model as _graph_for_model

# A notebook usually names it \`model\`; prefer that over scanning, so a template
# that builds a throwaway Model first still reports the one it went on to use.
_cands = [v for v in _tpl["ns"].values() if isinstance(v, _Model)]
_m = _tpl["ns"].get("model") if isinstance(_tpl["ns"].get("model"), _Model) else (_cands[-1] if _cands else None)

_json.dumps({"graph": _graph_for_model(_m) if _m is not None else None,
             "models": len(_cands)}, default=str)
`);
  return JSON.parse(out);
}

async function runPyrel(code) {
  // The same sandbox the sidecar uses. Hand-written and generated code are
  // indistinguishable by the time they reach here, so neither gets more trust.
  pyodide.globals.set('_eh_code', code);
  const out = await pyodide.runPythonAsync(`
if _EH["model"] is None:
    raise RuntimeError("no model loaded — send a 'model' message first")
_res = execute_pyrel_code(_EH["model"], _eh_code)
json.dumps(_res if isinstance(_res, list) else str(_res), default=str)
`);
  return JSON.parse(out);
}

const needsDb = () => { if (!dbReady) throw new Error('the worker has no database yet — send init first'); };
/* What a newly-connected page needs to know about work already done. Kept
 * beside the caches it reads so the two cannot drift. */
let pythonInfo = null;
let datasetInfo = null;
let rulesInfo = null;

const liveState = () => ({
  pyrelStarted: pyReady,
  python: pythonInfo,
  dataset: datasetInfo,
  rules: rulesInfo,
});

const needsPython = () => {
  needsDb();
  if (!pyReady) throw new Error('the Python front-end is not started — send python first');
};

async function handle(port, e) {
  const { id, type, payload } = e.data || {};
  try {
    let result;
    switch (type) {
      case 'init':
        /* Report what is already up, not just what the manifest says. A page
         * attaching to a SharedWorker that a previous page already loaded must
         * be able to tell -- otherwise it offers "Load it" for a model that is
         * sitting right there, and clicking it rebuilds what already exists. */
        result = { ...await initDb(), ...liveState() };
        break;
      case 'python':
        needsDb();
        /* Idempotent, because "has someone else already done this?" is now a
         * real question. initPython() is ~100 MB and rebinds _EH; running it a
         * second time for a second page would discard a loaded model. */
        result = pyReady ? pythonInfo : (pythonInfo = await initPython());
        break;
      case 'sql':
        needsDb();
        result = JSON.parse(execSql(payload, null));
        break;
      case 'dataset':
        needsPython();
        // Same reasoning as `python`: a second page must reattach, not reload.
        result = datasetInfo || (datasetInfo = await loadDataset());
        break;
      case 'model':
        needsPython();
        result = JSON.parse(await loadModel(payload));
        break;
      case 'templatePrepare':
        result = await serializeTemplateOp(() => templatePrepare(port, payload));
        break;
      case 'templateCell':
        result = await serializeTemplateOp(() => templateCell(port, payload));
        break;
      case 'templateGraph':
        result = await serializeTemplateOp(() => templateGraph(port));
        break;
      case 'templateClose':
        // The page is going. Nothing to report back, and no reason to fail if
        // it never prepared anything.
        await serializeTemplateOp(() => dropTemplateSession(port));
        result = { closed: true };
        break;
      case 'reloadModel':
        needsPython();
        result = await reloadModel();
        break;
      case 'reloadData':
        /* Forget the cached database and stand down, so the page that asked can
         * reload into a worker with nothing behind it.
         *
         * Not a rebuild in place. Pulling the database out from under a live
         * model means closing its connections, deleting files inside Pyodide and
         * re-running the load with half the engine still warm -- a lot of moving
         * parts for the one action someone takes when they already think
         * something is wrong. Standing down cannot leave a half-state, and
         * Pyodide comes back from the HTTP cache, so what this really costs is
         * the boot rather than the ~100 MB.
         *
         * If the clear throws -- private browsing, a locked store -- it throws
         * before the close is scheduled, so the worker stays up and the page
         * hears about it. Reloading into the same cached database while
         * reporting success is the one outcome worth avoiding here.
         *
         * The close is deferred because the reply still has to be posted below.
         * It has to happen at all because a SharedWorker outlives one page's
         * reload whenever another tab is holding it. */
        await idbClear(await idb());
        setTimeout(() => self.close(), 250);
        result = { ok: true };
        break;
      case 'pyrel':
        needsPython();
        result = { code: payload, rows: await runPyrel(payload) };
        break;
      case 'ask':
        needsPython();
        result = await runAsk(payload || {});
        break;
      case 'schema':
        needsDb();
        // With a model loaded, the model's schema is the useful answer; without
        // one, the tables are all there is to describe.
        /* `model` carries the same stats the sidecar's /schema does, not a
         * boolean. The page renders `${s.model.concepts} concepts`, and against
         * `true` that read "undefined concepts" — the panel reporting on itself
         * in a way no engine check would catch, since the call succeeded. */
        result = pyReady && pyodide.runPython('_EH["model"] is not None')
          ? {
            schema: pyodide.runPython('serialize_model_schema(_EH["model"])'),
            model: JSON.parse(pyodide.runPython(`json.dumps({
                "name": getattr(_EH["model"], "name", "model"),
                "concepts": len(_EH["model"]._concepts),
            })`)),
          }
          /* No Python yet means no database yet — it lives inside Pyodide now,
           * so there are no tables to describe rather than an error to report.
           * Querying information_schema here threw, and horizon.js calls
           * schema() straight after ready(), so every visit opened on a red
           * engine-failure banner before touching anything. It survived the
           * headless checks because those call startPyrel() first. */
          : !pyReady
            ? { schema: [], model: false }
            : { schema: JSON.parse(execSql(
                'SELECT table_name, column_name, data_type FROM information_schema.columns '
                + 'ORDER BY table_name, ordinal_position')).rows, model: false };
        break;
      default:
        throw new Error(`unknown message: ${type}`);
    }
    port.postMessage({ id, ok: true, result });
  } catch (err) {
    port.postMessage({ id, ok: false, ...describeFailure(err) });
  }
}

/* Turn a Pyodide failure into something a person can read.
 *
 * Pyodide sets a PythonError's `message` to the *entire formatted traceback*, so
 * posting `String(err.message)` put frames from `_pyodide/_base.py` in front of
 * the user. A question that failed for the ordinary reason -- the model wrote a
 * query that would not run -- was reported as what looks like a crash in the
 * engine, and the actual explanation was on line 30 of it.
 *
 * Two recoveries, in order:
 *
 *   the JSON payload runAsk raises, which carries the failed program and the
 *   translated database error -- the whole point of building it
 *
 *   failing that, the last non-empty line, which in a Python traceback is the
 *   exception itself. Never better than the payload, always better than frames.
 */
function describeFailure(err) {
  const raw = String(err && err.message || err || 'unknown failure');

  // JSON.stringify escapes newlines, so the payload is always one line. Find it
  // by its shape rather than by position: `raise ... from` and chained handlers
  // both append frames after it.
  for (const line of raw.split('\n').reverse()) {
    const at = line.indexOf('{"error"');
    if (at < 0) continue;
    try {
      const payload = JSON.parse(line.slice(at));
      if (payload && typeof payload.error === 'string') {
        return { error: payload.error, code: payload.code || null, detail: payload.detail || null };
      }
    } catch { /* a line that merely looks like the payload */ }
  }

  const lines = raw.split('\n').map((l) => l.trim()).filter(Boolean);
  const last = lines[lines.length - 1];
  // Only collapse to the last line when this really is a traceback; an ordinary
  // Error's message is already the thing to show, and may be several lines.
  return { error: /Traceback \(most recent call last\)/.test(raw) && last ? last : raw };
}

/* One connection per page. Request ids are per-port counters, so two pages both
 * starting at 1 is fine as long as a reply goes back to the port it came from --
 * which is why handle() takes the port rather than reading a global. */
self.onconnect = (e) => {
  const port = e.ports[0];
  ports.add(port);
  port.onmessage = (ev) => handle(port, ev);
  port.start();
};