[feat.physics.Omega.D8+] unify physics under durable job execution

WHY: surface and torsion descriptors claimed job semantics while a separate process-local queue owned their execution. The shared kernel now registers their real source identity, accepts durable queued jobs, persists artifacts, and exposes wait/cancel through one v2 control plane.

Author: ivanicu

backend/field_server.py

@@ -129,6 +129,7 @@ def _handle_admin(path: str):
 BOHR = 0.529177210859  # Å per Bohr
 MAX_QM_ATOMS = 120     # with hydrogens; STO-3G HF beyond this is not interactive
 DEFAULT_BASIS = 'sto-3g'
+IODINE_BASIS = 'def2-svp'
 
 # ── the wall-clock bound ────────────────────────────────────────────────────
 #
@@ -332,12 +333,30 @@ def handle_v2(method: str, path: str, body: dict | None) -> tuple[int, dict] | N
                          'meta': {'envelope': 2,
                                   'durability': getattr(svc, 'job_durability', None)}}
 
+        if method == 'POST' and path == '/v2/jobs':
+            body = body or {}
+            mid = body.get('method_id')
+            if not mid or 'input' not in body:
+                return _v2_error(
+                    'PARSE', 'job submission requires method_id and input',
+                    got_keys=sorted(body))
+            kw = {k: body[k] for k in
+                  ('inline_max', 'budget_seconds', 'request_id')
+                  if body.get(k) is not None}
+            return 202, _kernel().submit(mid, body['input'], **kw)
+
         if path.startswith('/v2/jobs/'):
             rest = urllib.parse.unquote(path[len('/v2/jobs/'):]).strip('/')
             if method == 'POST' and rest.endswith('/cancel'):
                 job_id = rest[:-len('/cancel')].strip('/')
                 return 200, {'ok': True, 'data': _kernel().cancel_job(job_id),
                              'meta': {'envelope': 2}}
+            if method == 'GET' and rest.endswith('/wait'):
+                job_id = rest[:-len('/wait')].strip('/')
+                timeout = float((query.get('timeout') or ['300'])[0])
+                return 200, {'ok': True,
+                             'data': _kernel().wait_job(job_id, timeout=timeout),
+                             'meta': {'envelope': 2}}
             if method == 'GET' and rest and '/' not in rest:
                 return 200, {'ok': True, 'data': _kernel().get_job(rest),
                              'meta': {'envelope': 2}}
@@ -507,12 +526,11 @@ def db_init() -> bool:
         # that decides whether a cached SCF is still valid.
         try:
             import method_registry
-            _method_ids = method_registry.register_all(_db, sys.modules[__name__],
-                                                      _toolkit_ids.get('pyscf'))
+            _method_ids = method_registry.register_all(
+                _db, sys.modules[__name__], _toolkit_ids)
             _method_versions = {
-                mid: method_registry.unit_version(sys.modules[__name__],
-                                                  u['fns'], u['consts'])[0]
-                for mid, u in method_registry.UNITS.items()}
+                row['method_id']: row['version']
+                for row in method_registry.plan(sys.modules[__name__])}
             print(f'[db] {len(_method_ids)} compute units registered '
                   f'({len(_method_versions)} source digests)', file=sys.stderr, flush=True)
             global _jobs
@@ -1352,10 +1370,10 @@ def run_scf(mol: Chem.Mol, basis: str, max_seconds: float = DEFAULT_MAX_SECONDS,
         # the v1 golden does not move.
         raise failures.DiracUnsupported(
             f'basis {basis} does not cover {"/".join(uncovered)} — '
-            'def2-svp is the only whitelisted basis with iodine',
+            f'{IODINE_BASIS} is the only whitelisted basis with iodine',
             details={'basis': basis, 'unsupported_elements': sorted(uncovered),
-                     'supported_alternatives': ['def2-svp']},
-            hint={'parameters': {'basis': 'def2-svp'}})
+                     'supported_alternatives': [IODINE_BASIS]},
+            hint={'parameters': {'basis': IODINE_BASIS}})
     if len(syms) > MAX_QM_ATOMS:
         raise failures.DiracTooLarge(
             f'{len(syms)} atoms (with H) exceeds the interactive QM cap of {MAX_QM_ATOMS}',

backend/handlers.py

@@ -27,6 +27,7 @@
 """
 from __future__ import annotations
 
+import json
 from typing import Any
 
 import failures
@@ -551,3 +552,113 @@ def region_handler(payload: dict, ctx: InvocationContext) -> HandlerResult:
         warnings=warnings,
         parameters_used={'dielectric': dielectric, 'spacing': spacing},
         cache='computed')
+
+
+def surface_handler(payload: dict, ctx: InvocationContext) -> HandlerResult:
+    """Both QM surface methods through the canonical invocation kernel."""
+    import numpy as np
+    from physics import mep_surface as MS
+
+    molecule = payload['molecule']['content']
+    params = dict(payload.get('parameters') or {})
+    basis = params.get('basis', MS.DEFAULT_BASIS)
+    budget = ctx.budget_seconds or params.get('max_seconds') or MS.DEFAULT_MAX_SECONDS
+    try:
+        if ctx.method_id == 'surface.mep':
+            out = MS.compute_surface_mep(
+                molecule, basis=basis,
+                isovalue=float(params.get('isovalue', MS.DEFAULT_ISOVALUE)),
+                points_per_atom=int(params.get('points_per_atom', 120)),
+                max_seconds=float(budget), xc=params.get('xc'),
+                use_gpu=params.get('use_gpu', 'auto'))
+            points = np.ascontiguousarray(out['points'], dtype='<f4')
+            values = np.ascontiguousarray(out['values'], dtype='<f4')
+            meta = out['meta']
+            result = {'summary': {'n_points': int(len(values)),
+                                  'extrema': out['extrema'], 'meta': meta}}
+            artifacts = [('surface.points', points.tobytes()),
+                         ('surface.values', values.tobytes())]
+        elif ctx.method_id == 'surface.mep_at':
+            points = np.asarray(payload['points'], dtype=float)
+            values, meta = MS.mep_at_points(
+                molecule, points, basis=basis, max_seconds=float(budget))
+            values = np.ascontiguousarray(values, dtype='<f4')
+            result = {'summary': {'n_points': int(len(values)),
+                                  'min': float(values.min()),
+                                  'max': float(values.max()), 'meta': meta}}
+            artifacts = [('surface.values', values.tobytes())]
+        else:                                                       # pragma: no cover
+            raise failures.DiracInternal(f'unknown surface method {ctx.method_id}')
+    except Exception as exc:                                       # noqa: BLE001
+        _raise_physics_failure(exc)
+    return HandlerResult(
+        result=result, artifacts=artifacts,
+        provenance={'n_atoms': meta.get('n_atoms'),
+                    'charge': meta.get('charge'), 'spin': meta.get('spin'),
+                    'scf_seconds': meta.get('scf_seconds')},
+        parameters_used={'basis': basis, 'max_seconds': float(budget)},
+        cache='computed')
+
+
+def surface_estimate(payload: dict) -> dict:
+    from physics import mep_surface as MS
+    try:
+        basis = (payload.get('parameters') or {}).get('basis', MS.DEFAULT_BASIS)
+        nao = MS.nao_for(payload['molecule']['content'], basis)
+        seconds = MS.estimated_scf_seconds(nao)
+        return {'available': True, 'n_basis_functions_estimated': nao,
+                'seconds': round(seconds, 2),
+                'confidence': 'same fitted SCF cost model used by the runtime preflight'}
+    except Exception as exc:                                       # noqa: BLE001
+        return {'available': False, 'reason': f'{type(exc).__name__}: {exc}'}
+
+
+def torsion_handler(payload: dict, ctx: InvocationContext) -> HandlerResult:
+    from physics.torsion import compute_torsion_strain
+    params = dict(payload.get('parameters') or {})
+    try:
+        out = compute_torsion_strain(
+            payload['molecule']['content'], steps=int(params.get('steps', 24)),
+            relax_hydrogens=bool(params.get('relax_hydrogens', True)),
+            max_torsions=int(params.get('max_torsions', 12)),
+            variant=params.get('variant', 'MMFF94s'))
+    except Exception as exc:                                       # noqa: BLE001
+        _raise_physics_failure(exc)
+    profile = json.dumps(out, sort_keys=True, separators=(',', ':')).encode()
+    meta = out.get('meta') or {}
+    return HandlerResult(
+        result={'summary': {'total_strain_kcal': out['total_strain_kcal'],
+                            'total_verdict': out['total_verdict'],
+                            'n_scanned': meta.get('n_scanned'), 'meta': meta}},
+        artifacts=[('torsion.profile', profile)],
+        provenance={'n_atoms': None},
+        parameters_used={'steps': int(params.get('steps', 24)),
+                         'relax_hydrogens': bool(params.get('relax_hydrogens', True)),
+                         'max_torsions': int(params.get('max_torsions', 12)),
+                         'variant': params.get('variant', 'MMFF94s')},
+        cache='computed')
+
+
+def torsion_estimate(payload: dict) -> dict:
+    return {'available': False,
+            'reason': 'cost depends on the number of rotatable bonds and MMFF convergence'}
+
+
+def _raise_physics_failure(exc: Exception) -> None:
+    """Translate producer-native refusals once, at the science boundary."""
+    from physics.mep_surface import PhysicsBudgetExceeded
+    if isinstance(exc, failures.DiracFailure):
+        raise exc
+    message = str(exc)
+    lower = message.lower()
+    if isinstance(exc, PhysicsBudgetExceeded) or 'budget' in lower or 'predicted' in lower:
+        raise failures.DiracBudgetExceeded(message) from exc
+    if 'did not converge' in lower:
+        raise failures.DiracUnconverged(message) from exc
+    if 'parse' in lower or 'molfile' in lower and 'no 3d' not in lower:
+        raise failures.DiracParseFailure(message) from exc
+    if 'exceeds' in lower and 'atoms' in lower:
+        raise failures.DiracTooLarge(message) from exc
+    if 'type' in lower or 'force field' in lower:
+        raise failures.DiracUnparameterized(message) from exc
+    raise failures.DiracUnsupported(message) from exc

backend/invocation.py

@@ -41,6 +41,7 @@
 from __future__ import annotations
 
 import sys
+import threading
 import time
 from dataclasses import dataclass, field
 from typing import Any, Callable, Protocol
@@ -140,6 +141,8 @@ def __init__(self, catalog: C.MethodCatalog, *,
         self.cache = cache
         self.executor = executor or E.InlineExecutor()
         self.toolkit_versions = toolkit_versions or {}
+        self._futures: dict[str, Any] = {}
+        self._futures_lock = threading.Lock()
         self.counters = {'invoked': 0, 'cache_hit': 0, 'refused': 0, 'failed': 0,
                          'artifacts_registered': 0}
 
@@ -211,18 +214,96 @@ def get_job(self, job_id: str) -> dict:
         return row
 
     def cancel_job(self, job_id: str) -> dict:
+        with self._futures_lock:
+            future = self._futures.get(job_id)
+        if future is not None and not future.running() and not future.done():
+            future.cancel()
         row = self.ledger.request_cancel(job_id) if self.ledger is not None and \
             hasattr(self.ledger, 'request_cancel') else None
         if row is None:
             raise failures.DiracNotFound(
                 f'no job {job_id!r}', details={'job_id': job_id})
         return row
 
+    def wait_job(self, job_id: str, *, timeout: float = 300.0,
+                 poll: float = 0.1) -> dict:
+        """Wait for a handle without coupling the caller to executor internals."""
+        deadline = time.time() + max(0.0, float(timeout))
+        while True:
+            row = self.get_job(job_id)
+            if row.get('state') in ('done', 'failed', 'cancelled'):
+                return row
+            if time.time() >= deadline:
+                row['wait'] = {'timed_out': True, 'timeout_seconds': float(timeout)}
+                return row
+            time.sleep(max(0.01, float(poll)))
+
+    def submit(self, method_id: str, payload: dict, *,
+               inline_max: int | None = None,
+               budget_seconds: float | None = None,
+               request_id: str | None = None) -> dict:
+        """Create a reconnectable queued Job and execute it on the injected executor."""
+        spec = self.catalog.get(method_id)
+        self.catalog.validate(method_id, payload)
+        spec.handler()  # fail before minting a handle that can never run
+        if 'job' not in spec.execution.get('supported_modes', []):
+            raise failures.DiracUnsupported(
+                f'{method_id} does not declare job execution',
+                details={'method_id': method_id,
+                         'supported_modes': spec.execution.get('supported_modes', [])})
+        if self.ledger is None:
+            raise failures.DiracInternal(
+                'job submission requires a JobStore; this kernel has none')
+        if not getattr(self.executor, 'supports_submission', False):
+            raise failures.DiracInternal(
+                'job submission requires an executor with submit(); '
+                f'got {getattr(self.executor, "kind", type(self.executor).__name__)}')
+        budget = budget_seconds
+        if budget is None:
+            budget = float((payload.get('parameters') or {}).get('max_seconds')
+                           or spec.execution.get('default_budget_seconds') or 0.0) or None
+        job_id, conflicted = self._open_job(spec, payload, budget, queued=True)
+        if job_id is None:
+            raise failures.DiracInternal(
+                'the JobStore could not create or resolve a durable job handle')
+        if not conflicted:
+            future = self.executor.submit(
+                self._run_submitted, job_id, method_id, payload,
+                inline_max, budget, request_id)
+            with self._futures_lock:
+                self._futures[job_id] = future
+            future.add_done_callback(lambda _f, jid=job_id: self._forget_future(jid))
+        return {
+            'ok': True,
+            'data': {'job': self.get_job(job_id)},
+            'artifacts': [], 'warnings': [],
+            'meta': {'envelope': 2, 'method_id': method_id,
+                     'version': spec.version, 'job_id': job_id,
+                     'execution_mode': 'job', 'deduplicated': conflicted,
+                     'request_id': request_id},
+        }
+
+    def _run_submitted(self, job_id: str, method_id: str, payload: dict,
+                       inline_max: int | None, budget_seconds: float | None,
+                       request_id: str | None) -> None:
+        row = self.ledger.get(job_id)
+        if row is None or row.get('state') == 'cancelled':
+            return
+        self.ledger.start(job_id)
+        self.invoke(method_id, payload, inline_max=inline_max,
+                    budget_seconds=budget_seconds, request_id=request_id,
+                    _preopened_job_id=job_id)
+
+    def _forget_future(self, job_id: str) -> None:
+        with self._futures_lock:
+            self._futures.pop(job_id, None)
+
     # ── the write side ────────────────────────────────────────────────────────
     def invoke(self, method_id: str, payload: dict, *,
                inline_max: int | None = None,
                budget_seconds: float | None = None,
-               request_id: str | None = None) -> dict:
+               request_id: str | None = None,
+               _preopened_job_id: str | None = None) -> dict:
         """Run a method and return a v2 envelope. Never raises for a REFUSAL.
 
         A refusal is a RESULT — the molecule is too large, the basis does not cover
@@ -237,7 +318,7 @@ def invoke(self, method_id: str, payload: dict, *,
         """
         t0 = time.time()
         self.counters['invoked'] += 1
-        job_id = None
+        job_id = _preopened_job_id
         spec = None
         try:
             spec = self.catalog.get(method_id)
@@ -265,12 +346,16 @@ def invoke(self, method_id: str, payload: dict, *,
                 # the COMPUTED path while cache hits are the majority of responses. A rule
                 # that applies to the minority of answers is not a rule.
                 self.catalog.validate_output(method_id, hit.result)
-                return self._envelope(spec, hit, t0, cache='db',
-                                      inline_max=inline_max, request_id=request_id,
-                                      job_id=None)
+                env = self._envelope(spec, hit, t0, cache='db',
+                                     inline_max=inline_max, request_id=request_id,
+                                     job_id=job_id)
+                if job_id is not None and self.ledger is not None:
+                    self.ledger.done(job_id, seconds=round(time.time() - t0, 3),
+                                     result_summary={'ok': True, 'cache': 'db'})
+                return env
 
-            if self.ledger is not None:
-                job_id = self._open_job(spec, payload, budget)
+            if self.ledger is not None and job_id is None:
+                job_id, _ = self._open_job(spec, payload, budget)
 
             ctx = InvocationContext(
                 method_id=method_id, version=spec.version, budget_seconds=budget,
@@ -346,7 +431,7 @@ def invoke(self, method_id: str, payload: dict, *,
 
     # ── internals ─────────────────────────────────────────────────────────────
     def _open_job(self, spec: C.MethodSpec, payload: dict,
-                  budget: float | None) -> str | None:
+                  budget: float | None, *, queued: bool = False) -> tuple[str | None, bool]:
         """Open a ledger row, and never let the ledger break the invocation.
 
         A job row is OBSERVABILITY. If the database is down, the science must still
@@ -357,18 +442,18 @@ def _open_job(self, spec: C.MethodSpec, payload: dict,
             import hashlib
             import json as _json
             canonical = _json.dumps(payload, sort_keys=True, separators=(',', ':'))
-            job_id, _conflict = self.ledger.open(
+            job_id, conflict = self.ledger.open(
                 method_row_id=getattr(self.ledger, 'method_row_for',
                                       lambda _m: None)(spec.method_id),
                 input_sha256=hashlib.sha256(canonical.encode()).digest(),
                 params=dict(payload.get('parameters') or {}),
-                budget_seconds=budget, queued=False)
-            return job_id
+                budget_seconds=budget, queued=queued)
+            return job_id, conflict
         except Exception as e:                                     # noqa: BLE001
             print(f'[invoke] ledger unavailable ({type(e).__name__}: {e}) — running '
                   f'anyway, and provenance will say job_id: null', file=sys.stderr,
                   flush=True)
-            return None
+            return None, False
 
     def _require_declared_artifacts(self, spec: C.MethodSpec,
                                     out: HandlerResult) -> None:

backend/jobs.py

@@ -240,7 +240,7 @@ def done(self, job_id: str | None, *, seconds: float,
             _bump('write_failed', f'{type(e).__name__}: {e}')
 
     def failed(self, job_id: str | None, *, code: str, detail: str,
-               seconds: float | None = None) -> None:
+               seconds: float | None = None, retryable: bool | None = None) -> None:
         """Terminal-failed. `code` is mapped onto app.job_error, never raising.
 
         A CANCELLED code is stored as state='cancelled', because the schema's
@@ -533,7 +533,7 @@ def done(self, job_id: str | None, *, seconds: float,
                            finished_at=_now(), result_summary=result_summary)
 
     def failed(self, job_id: str | None, *, code: str, detail: str,
-               seconds: float | None = None) -> None:
+               seconds: float | None = None, retryable: bool | None = None) -> None:
         with self._lock:
             row = self._rows.get(job_id or '')
             if row and row['state'] in ('queued', 'running'):