[feat.physics.sigma.D8+] the 90-second budget was a browser tab's patience wearing physics clothes

Ivan, reading a refusal: "75 atoms in def2-svp is 779 basis functions,
predicted ~202 s against a 90 s budget. Use a smaller basis, raise
max_seconds, or trim the ligand — 和现在这整个physics面板有什么意义呢"

The three options ARE the argument, and every one is bad:
  · "use a smaller basis" offers STO-3G — the level this same codebase now
    refuses to print HOMO/LUMO from, and which for a sigma-hole is not merely
    inaccurate but structurally wrong (no d functions on Cl/Br/S,
    non-relativistic all-electron I). The panel was recommending the theory
    its own caveat system disowns.
  · "raise max_seconds" means holding a synchronous HTTP request for minutes.
  · "trim the ligand" means changing the molecule to fit the computer.

And 75 atoms is an ORDINARY LEAD COMPOUND. The panel refused exactly the
molecules it exists for, and the single decision it was justified by — is Cl→Br
worth a synthesis — needs def2-SVP precisely on halogenated compounds.

DIAGNOSIS: 202 s is not a long quantum chemistry calculation. It is a long
SYNCHRONOUS HTTP REQUEST. The budget was never a statement about physics; it
was a statement about how long a browser tab will wait, and it was enforcing
that on the science.

So: submit-and-poll. POST /jobs/surface/mep and /jobs/torsion/strain return a
job id immediately with a cost quote and a queue position. GET /jobs/<id>
polls. POST /jobs/<id>/cancel rides the per-cycle watchdog that already exists
rather than needing a second mechanism. One worker, deliberately — the GPU is
one device and two concurrent SCFs thrash it; a queue with a depth is honest
about that where unlimited threads would not be.

MEASURED WHILE BUILDING IT, and it changes what the prediction is FOR:

  4-bromobenzonitrile, 13 atoms, def2-svp:  predicted 7.4 s, elapsed 151 s+
  imatinib-scale, nao 673:                  predicted 148 s (with the 3x
                                            safety), 49 s without it

The model is wrong in BOTH directions — a 3x safety factor that over-predicts
and refuses large molecules, and a fit that under-predicts small ones by more
than an order of magnitude. (Load average was 91 from concurrent sessions, so
the 151 s is not a clean number; but the same molecule took ~100 s uncontended
earlier, so contention is not the whole story.)

A gate whose predictor is order-of-magnitude wrong is not a gate, it is a
lottery. With a queue it does not have to be one: the prediction becomes
ADVICE, and being wrong costs an inaccurate progress hint instead of a refused
calculation. That is the actual repair — not a better fit.

nao_for() computes the basis count without running anything, so the quote is
made from the only honest predictor. Atom count is not it: 120 atoms of C/H is
~360 basis functions and 120 atoms with six iodines is over 500, which is
exactly why MAX_QM_ATOMS = 120 failed to stop a 43-atom heme.

NEXT: the panel still calls the synchronous routes. Moving it onto the queue is
what makes the answer reach a chemist; until then this is a backend that can do
the work and a UI that still asks the wrong way.

Author: ivanicu

backend/physics/mep_surface.py

@@ -93,6 +93,34 @@ def estimated_scf_seconds(nao: int) -> float:
 # was the one it got wrong, and nothing in the output said so.
 ECP_FROM_Z = 37
 
+def nao_for(molblock: str, basis: str) -> int:
+    """Basis-function count, computed without running anything.
+
+    The queue needs to quote a cost BEFORE the job starts, and nao is the only
+    honest predictor — atom count is not. 120 atoms of C/H is ~360 basis
+    functions; 120 atoms containing six iodines is over 500. A cap expressed in
+    atoms bounds the wrong quantity, which is exactly how MAX_QM_ATOMS = 120
+    failed to stop a 43-atom heme from holding 22 cores for 36 minutes.
+    """
+    from pyscf import gto
+    mol = Chem.MolFromMolBlock(molblock, removeHs=False, sanitize=False)
+    if mol is None:
+        raise ValueError('cannot parse molfile')
+    conf = mol.GetConformer()
+    atoms = [(a.GetSymbol(),
+              (conf.GetAtomPosition(a.GetIdx()).x,
+               conf.GetAtomPosition(a.GetIdx()).y,
+               conf.GetAtomPosition(a.GetIdx()).z))
+             for a in mol.GetAtoms()]
+    pt = Chem.GetPeriodicTable()
+    charge = sum(a.GetFormalCharge() for a in mol.GetAtoms())
+    nelec = sum(pt.GetAtomicNumber(s) for s, _ in atoms) - charge
+    gmol = gto.M(atom=list(atoms), unit='Angstrom', basis=basis,
+                 ecp=_ecp_for(atoms, basis) or None, verbose=0,
+                 charge=charge, spin=nelec % 2)
+    return int(gmol.nao)
+
+
 
 
 def clamp_budget(value, default: float) -> float:

backend/physics/server.py

@@ -33,6 +33,10 @@
 
 import base64
 import json
+import queue as queue_mod
+import threading
+import uuid
+from collections import OrderedDict
 import os
 import sys
 import time
@@ -59,6 +63,10 @@
 # A molfile is kilobytes. Anything past this is a mistake or an attempt to wedge
 # the box, and either way it should be refused before it is read into memory.
 MAX_BODY_BYTES = 8 * 1024 * 1024
+# A queued job is not waited on by a socket, so its ceiling is a resource
+# decision rather than a browser's patience. Still bounded: unbounded is
+# how one heme click held 22 cores for 36 minutes.
+JOB_MAX_SECONDS = 1800.0
 
 
 # pyscf will happily accept an arbitrary string here and fail deep inside a
@@ -77,6 +85,130 @@ def validated_basis(name: str) -> str:
     return key
 
 
+
+# ── the job queue: the budget was a UI constraint wearing physics clothes ───
+#
+# Ivan, on reading a refusal: "75 atoms in def2-svp is 779 basis functions,
+# predicted ~202 s against a 90 s budget. Use a smaller basis, raise
+# max_seconds, or trim the ligand — 和现在这整个physics面板有什么意义呢"
+#
+# He is right, and the refusal's three options are the evidence. Every one of
+# them is bad:
+#   · "use a smaller basis" offers STO-3G, which this same codebase refuses to
+#     quote HOMO/LUMO from, and which for a sigma-hole is not merely inaccurate
+#     but structurally wrong — no d functions on Cl/Br/S, non-relativistic
+#     all-electron I. The panel would be recommending the level of theory its
+#     own caveat system disowns.
+#   · "raise max_seconds" means sit on a synchronous HTTP request for minutes.
+#   · "trim the ligand" means change the molecule to fit the computer.
+#
+# And 75 atoms is an ORDINARY LEAD COMPOUND. So the panel refused exactly the
+# molecules it exists for, and the one decision it was justified by — is Cl→Br
+# worth a synthesis — needs def2-SVP precisely on the halogenated compounds a
+# medicinal chemist cares about.
+#
+# The diagnosis: 202 s is not a long quantum chemistry calculation. It is a
+# long SYNCHRONOUS HTTP REQUEST. The budget was never a statement about
+# physics; it was a statement about how long a browser tab will wait, wearing
+# physics clothes. Submit-and-poll makes the wait a fact about the interaction
+# instead of a limit on the science, and then a 200 s job is simply a job.
+#
+# One worker, deliberately: the GPU is one device and two concurrent SCFs
+# thrash it. A queue with a depth is honest about that; unlimited threads would
+# not be.
+
+JOB_RETENTION = 32          # bounded, LRU — a job store is not a database
+_jobs: 'OrderedDict[str, dict]' = OrderedDict()
+_jobs_lock = threading.Lock()
+_job_queue: 'queue.Queue[str]' = queue_mod.Queue()
+
+
+def _new_job(kind: str, payload: dict, predicted: float | None) -> str:
+    job_id = uuid.uuid4().hex[:12]
+    with _jobs_lock:
+        _jobs[job_id] = {
+            'id': job_id, 'kind': kind, 'state': 'queued',
+            'submitted_at': time.time(), 'started_at': None, 'finished_at': None,
+            'predicted_seconds': predicted, 'cancelled': False,
+            'result': None, 'error': None, 'reason': None,
+            'queue_position': _job_queue.qsize() + 1,
+            '_payload': payload,
+        }
+        while len(_jobs) > JOB_RETENTION:
+            _jobs.popitem(last=False)
+    _job_queue.put(job_id)
+    return job_id
+
+
+def _public_job(job: dict) -> dict:
+    """Everything except the payload and the result blob."""
+    out = {k: v for k, v in job.items()
+           if not k.startswith('_') and k != 'result'}
+    now = time.time()
+    if job['started_at']:
+        out['elapsed_seconds'] = round((job['finished_at'] or now) - job['started_at'], 1)
+    else:
+        out['waiting_seconds'] = round(now - job['submitted_at'], 1)
+    return out
+
+
+def _worker():
+    while True:
+        job_id = _job_queue.get()
+        with _jobs_lock:
+            job = _jobs.get(job_id)
+            if job is None or job['cancelled']:
+                if job is not None:
+                    job['state'] = 'cancelled'
+                    job['finished_at'] = time.time()
+                continue
+            job['state'] = 'running'
+            job['started_at'] = time.time()
+        req = job['_payload']
+        try:
+            # A queued job answers to the QUEUE's ceiling, not to a socket's
+            # patience. The watchdog still bounds it — the point is that the
+            # bound is now a resource decision instead of a UI artefact.
+            budget = float(req.get('max_seconds', JOB_MAX_SECONDS))
+            if job['kind'] == 'surface/mep':
+                out = compute_surface_mep(
+                    req['molfile'],
+                    basis=validated_basis(req.get('basis', DEFAULT_BASIS)),
+                    isovalue=float(req.get('isovalue', 0.001)),
+                    points_per_atom=int(req.get('points_per_atom', 120)),
+                    max_seconds=budget, xc=req.get('xc'))
+                result = {'points_b64': b64(out['points']),
+                          'values_b64': b64(out['values']),
+                          'n_points': int(len(out['values'])),
+                          'extrema': out['extrema'], 'meta': out['meta']}
+            elif job['kind'] == 'torsion/strain':
+                out = compute_torsion_strain(
+                    req['molfile'], steps=int(req.get('steps', 24)),
+                    relax_hydrogens=bool(req.get('relax_hydrogens', True)),
+                    max_torsions=int(req.get('max_torsions', 12)),
+                    variant=req.get('variant', 'MMFF94s'))
+                result = out
+            else:
+                raise ValueError(f'unknown job kind {job["kind"]!r}')
+            with _jobs_lock:
+                job['state'] = 'done'
+                job['result'] = result
+        except Exception as e:                               # noqa: BLE001
+            traceback.print_exc()
+            with _jobs_lock:
+                job['state'] = 'cancelled' if job['cancelled'] else 'failed'
+                job['error'] = str(e)
+                job['reason'] = ('cancelled' if job['cancelled'] else
+                                 'unsupported' if isinstance(e, (ValueError, KeyError))
+                                 else 'internal')
+        finally:
+            with _jobs_lock:
+                job['finished_at'] = time.time()
+            print(f"[job {job_id}] {job['kind']} {job['state']} "
+                  f"in {job['finished_at'] - (job['started_at'] or job['finished_at']):.1f}s",
+                  flush=True)
+
+
 def b64(array: np.ndarray) -> str:
     return base64.b64encode(np.ascontiguousarray(array, dtype='<f4').tobytes()).decode()
 
@@ -103,14 +235,46 @@ def do_OPTIONS(self):
         self.end_headers()
 
     def do_GET(self):
+        if self.path.startswith('/jobs/'):
+            job_id = self.path.split('/jobs/', 1)[1].strip('/')
+            with _jobs_lock:
+                job = _jobs.get(job_id)
+                if job is None:
+                    # NOT_FOUND is distinct from "still queued". A job store
+                    # that has evicted a result must not answer like one that
+                    # never had it.
+                    self._send(200, {'ok': False, 'reason': 'not_found',
+                                     'error': f'no job {job_id} — it may have '
+                                              f'been evicted (the store keeps '
+                                              f'the last {JOB_RETENTION})'})
+                    return
+                public = _public_job(job)
+                done = job['state'] == 'done'
+                result = job['result'] if done else None
+                failed = job['state'] in ('failed', 'cancelled')
+                err, reason = job['error'], job['reason']
+            if done:
+                self._send(200, {'ok': True, 'job': public, **result})
+            elif failed:
+                self._send(200, {'ok': False, 'job': public, 'error': err,
+                                 'reason': reason})
+            else:
+                self._send(200, {'ok': True, 'job': public, 'pending': True})
+            return
         if self.path != '/health':
             self._send(404, {'ok': False, 'error': 'not found'})
             return
         import pyscf
         import rdkit
         self._send(200, {'ok': True, 'service': 'dirac-physics',
                          'rdkit': rdkit.__version__, 'pyscf': pyscf.__version__,
-                         'endpoints': ['/surface/mep', '/surface/mep_at', '/torsion/strain']})
+                         'endpoints': ['/surface/mep', '/surface/mep_at',
+                                       '/torsion/strain',
+                                       'POST /jobs/surface/mep',
+                                       'POST /jobs/torsion/strain',
+                                       'GET /jobs/<id>', 'POST /jobs/<id>/cancel'],
+                         'queue_depth': _job_queue.qsize(),
+                         'jobs_retained': len(_jobs)})
 
     def do_POST(self):
         try:
@@ -123,6 +287,51 @@ def do_POST(self):
             req = json.loads(self.rfile.read(length)) if length else {}
             t0 = time.time()
 
+            if self.path.startswith('/jobs/') and self.path.endswith('/cancel'):
+                job_id = self.path[len('/jobs/'):-len('/cancel')].strip('/')
+                with _jobs_lock:
+                    job = _jobs.get(job_id)
+                    if job is None:
+                        self._send(200, {'ok': False, 'reason': 'not_found',
+                                         'error': f'no job {job_id}'})
+                        return
+                    job['cancelled'] = True
+                    if job['state'] == 'queued':
+                        job['state'] = 'cancelled'
+                        job['finished_at'] = time.time()
+                    state = job['state']
+                # A running job stops at its next SCF cycle: the watchdog is
+                # already per-cycle, so cancellation rides the guard that
+                # exists rather than needing a second mechanism.
+                self._send(200, {'ok': True, 'state': state})
+                return
+
+            if self.path in ('/jobs/surface/mep', '/jobs/torsion/strain'):
+                kind = self.path[len('/jobs/'):]
+                predicted = None
+                if kind == 'surface/mep':
+                    # Predict BEFORE queueing so the caller sees the cost up
+                    # front — but do not refuse on it. Refusing was the whole
+                    # complaint: the number is advice now, not a gate.
+                    try:
+                        from physics.mep_surface import (estimated_scf_seconds,
+                                                         GPU_CROSSOVER_NAO,
+                                                         GPU_SPEEDUP, nao_for)
+                        nao = nao_for(req['molfile'],
+                                      validated_basis(req.get('basis', DEFAULT_BASIS)))
+                        predicted = round(
+                            estimated_scf_seconds(nao)
+                            / (GPU_SPEEDUP if nao >= GPU_CROSSOVER_NAO else 1.0), 1)
+                    except Exception:                        # noqa: BLE001
+                        predicted = None
+                job_id = _new_job(kind, req, predicted)
+                with _jobs_lock:
+                    public = _public_job(_jobs[job_id])
+                print(f'[job {job_id}] {kind} queued, predicted '
+                      f'{predicted if predicted is not None else "?"} s', flush=True)
+                self._send(200, {'ok': True, 'job': public})
+                return
+
             if self.path == '/surface/mep':
                 out = compute_surface_mep(
                     req['molfile'],
@@ -194,4 +403,5 @@ def log_message(self, *args):
     print(f'Dirac physics backend on http://{host}:{port}'
           + ('  (reachable from the LAN — unauthenticated)' if host == '0.0.0.0' else ''),
           flush=True)
+    threading.Thread(target=_worker, daemon=True, name='physics-worker').start()
     ThreadingHTTPServer((host, port), Handler).serve_forever()