[feat.jobs.Omega.D8+] the kernel gets one durable JobStore contract, canonical request identity, honest cancellation, and replaceable execution

The v2 kernel had a ledger parameter but never received the database ledger, so every SDK and CLI result reported job_id null while physics kept an unrelated process-local queue. These files are one action because the schema, the two JobStore implementations, the executor seam, kernel wiring, v2 control routes, and lifecycle tests must change together or one transport can claim durability or cancellation the runtime does not have. request_digest replaces split input-plus-md5 identity; queued work is cancellable, while running work records intent without pretending Python interrupted native compute.

Author: ivanicu

backend/db/migrations/014_job_store_contract.sql

@@ -0,0 +1,62 @@
+-- 014 · app.job becomes the durable implementation of the public JobStore contract.
+--
+-- input_sha256 describes input bytes. It is not the identity of an invocation: basis,
+-- spin and other parameters can change the science while the molecule bytes remain the
+-- same. The old unique index compensated with md5(params::text), which is both split
+-- across two columns and weaker than every other content identity in this system.
+-- request_digest is the one canonical SHA-256 used for open, dedup and wait.
+--
+-- Cancellation is also separated from interruption. cancel_requested_at records what a
+-- caller asked; state='cancelled' records what the executor actually achieved. A running
+-- SCF that cannot be interrupted must never be relabelled cancelled merely because the
+-- browser stopped waiting.
+
+BEGIN;
+
+ALTER TABLE app.job
+    ADD COLUMN request_digest bytea,
+    ADD COLUMN durability text NOT NULL DEFAULT 'durable'
+        CHECK (durability IN ('durable', 'process')),
+    ADD COLUMN cancel_requested_at timestamptz,
+    ADD COLUMN result_summary jsonb;
+
+UPDATE app.job
+   SET request_digest = digest(
+       method_row_id::text || ':' || encode(input_sha256, 'hex') || ':' || params::text,
+       'sha256');
+
+ALTER TABLE app.job
+    ALTER COLUMN request_digest SET NOT NULL,
+    ADD CONSTRAINT job_request_digest_sha256
+        CHECK (octet_length(request_digest) = 32);
+
+DROP INDEX app.job_one_inflight;
+CREATE UNIQUE INDEX job_one_inflight
+    ON app.job (method_row_id, request_digest)
+ WHERE state IN ('queued', 'running');
+
+CREATE OR REPLACE VIEW app.v_job_live AS
+SELECT j.id, m.method_id, m.version AS method_version, j.state,
+       j.compound_id, j.budget_seconds, j.est_seconds,
+       round(extract(epoch FROM now() - coalesce(j.started_at, j.created_at))::numeric, 1)
+           AS age_seconds,
+       j.worker, j.created_at, j.started_at,
+       j.request_digest, j.durability, j.cancel_requested_at
+  FROM app.job j JOIN meta.method m ON m.id = j.method_row_id
+ WHERE j.state IN ('queued', 'running')
+ ORDER BY j.created_at;
+
+COMMENT ON COLUMN app.job.request_digest IS
+    'Canonical SHA-256 of the complete scientific invocation identity; the sole in-flight dedup key beside method_row_id.';
+COMMENT ON COLUMN app.job.cancel_requested_at IS
+    'A caller asked to cancel. This does not claim a running executor was interrupted.';
+COMMENT ON COLUMN app.job.result_summary IS
+    'Small JSON summary only; result bytes remain first-class artifacts.';
+
+INSERT INTO meta.migration (filename, sha256, content_sha256, hash_source)
+VALUES ('014_job_store_contract.sql',
+        '\xcf7999ac21a5392f6ddcc12addeb654768133083ea8b8db96661ccebda0400d9'::bytea,
+        '\xcf7999ac21a5392f6ddcc12addeb654768133083ea8b8db96661ccebda0400d9'::bytea, 'content')
+ON CONFLICT (filename) DO NOTHING;
+
+COMMIT;

backend/execution.py

@@ -0,0 +1,46 @@
+"""Execution strategies for the invocation kernel.
+
+An executor runs a callable; it never decides method semantics, cache identity, response
+shape, or error vocabulary. Keeping that boundary explicit is what lets an inline call
+become a worker/cluster submission without changing SDK, CLI, HTTP, or MCP adapters.
+"""
+from __future__ import annotations
+
+from concurrent.futures import Future, ThreadPoolExecutor as _ThreadPool
+from typing import Any, Callable
+
+
+class InlineExecutor:
+    kind = 'inline'
+    supports_submission = False
+
+    def execute(self, fn: Callable[..., Any], *args, **kwargs) -> Any:
+        return fn(*args, **kwargs)
+
+
+class ThreadExecutor:
+    """A bounded in-process executor for reconnectable job-mode work.
+
+    Cancellation is honest: a queued future may be cancelled; Python cannot interrupt a
+    callable that is already running, so ``cancel`` returns false in that state.
+    """
+
+    kind = 'thread'
+    supports_submission = True
+
+    def __init__(self, max_workers: int = 1) -> None:
+        self._pool = _ThreadPool(max_workers=max_workers,
+                                 thread_name_prefix='dirac-job')
+
+    def execute(self, fn: Callable[..., Any], *args, **kwargs) -> Any:
+        return fn(*args, **kwargs)
+
+    def submit(self, fn: Callable[..., Any], *args, **kwargs) -> Future:
+        return self._pool.submit(fn, *args, **kwargs)
+
+    @staticmethod
+    def cancel(future: Future) -> bool:
+        return future.cancel()
+
+    def shutdown(self, *, wait: bool = True) -> None:
+        self._pool.shutdown(wait=wait)

backend/field_server.py

@@ -310,10 +310,38 @@ def handle_v2(method: str, path: str, body: dict | None) -> tuple[int, dict] | N
     """
     if not path.startswith('/v2/'):
         return None
-    q = path.split('?', 1)
-    path = q[0]
+    split = urllib.parse.urlsplit(path)
+    path = split.path
+    query = urllib.parse.parse_qs(split.query)
 
     try:
+        if method == 'GET' and path == '/v2/meta':
+            svc = _kernel()
+            return 200, {'ok': True,
+                         'data': {'api_version': '2.0.0',
+                                  'supported_envelopes': [2],
+                                  'capabilities': svc.capabilities()},
+                         'meta': {'envelope': 2}}
+
+        if method == 'GET' and path == '/v2/jobs':
+            svc = _kernel()
+            state = (query.get('state') or [None])[0]
+            limit = int((query.get('limit') or ['100'])[0])
+            return 200, {'ok': True,
+                         'data': {'jobs': svc.list_jobs(state=state, limit=limit)},
+                         'meta': {'envelope': 2,
+                                  'durability': getattr(svc, 'job_durability', None)}}
+
+        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 and '/' not in rest:
+                return 200, {'ok': True, 'data': _kernel().get_job(rest),
+                             'meta': {'envelope': 2}}
+
         if method == 'GET' and path == '/v2/methods':
             svc = _kernel()
             return 200, {'ok': True, 'data': {'methods': svc.list_methods()},
@@ -489,7 +517,8 @@ def db_init() -> bool:
                   f'({len(_method_versions)} source digests)', file=sys.stderr, flush=True)
             global _jobs
             _jobs = jobs.JobLedger(
-                _db, jobs.make_worker_name(os.getpid(), _producer_version()))
+                _db, jobs.make_worker_name(os.getpid(), _producer_version()),
+                method_rows=_method_ids)
             # A restart leaves rows 'running' that no process is doing. Without
             # the reap, v_job_live's age_seconds grows without bound and the
             # ledger reports work nobody is performing — worse than no ledger,

backend/invocation.py

@@ -47,6 +47,7 @@
 
 import artifacts as A
 import catalog as C
+import execution as E
 import failures
 
 
@@ -131,11 +132,13 @@ def __init__(self, catalog: C.MethodCatalog, *,
                  store: Any | None = None,
                  ledger: Any | None = None,
                  cache: Any | None = None,
+                 executor: Any | None = None,
                  toolkit_versions: dict[str, str] | None = None) -> None:
         self.catalog = catalog
         self.store = store or A.MemoryArtifactStore()
         self.ledger = ledger
         self.cache = cache
+        self.executor = executor or E.InlineExecutor()
         self.toolkit_versions = toolkit_versions or {}
         self.counters = {'invoked': 0, 'cache_hit': 0, 'refused': 0, 'failed': 0,
                          'artifacts_registered': 0}
@@ -175,6 +178,46 @@ def estimate(self, method_id: str, payload: dict) -> dict:
         out['version'] = spec.version
         return out
 
+    # ── job control surface ─────────────────────────────────────────────────
+    def capabilities(self) -> dict:
+        return {
+            'job_store': {
+                'kind': getattr(self.ledger, 'kind', 'none') if self.ledger else 'none',
+                'durability': (getattr(self.ledger, 'durability', 'none')
+                               if self.ledger else 'none'),
+            },
+            'artifact_store': {
+                'kind': getattr(self, 'store_kind', type(self.store).__name__),
+                'durability': ('durable' if getattr(self, 'store_kind', '') == 'postgres'
+                               else 'process'),
+            },
+            'executor': {'kind': getattr(self.executor, 'kind', 'unknown')},
+            # Inline Python cannot interrupt C/Fortran work. Queued cancellation becomes
+            # available when ThreadExecutor owns a submitted future.
+            'cancellation': 'queued-only',
+        }
+
+    def list_jobs(self, *, state: str | None = None, limit: int = 100) -> list[dict]:
+        if self.ledger is None or not hasattr(self.ledger, 'list'):
+            return []
+        return self.ledger.list(state=state, limit=limit)
+
+    def get_job(self, job_id: str) -> dict:
+        row = self.ledger.get(job_id) if self.ledger is not None and \
+            hasattr(self.ledger, 'get') else None
+        if row is None:
+            raise failures.DiracNotFound(
+                f'no job {job_id!r}', details={'job_id': job_id})
+        return row
+
+    def cancel_job(self, job_id: str) -> dict:
+        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
+
     # ── the write side ────────────────────────────────────────────────────────
     def invoke(self, method_id: str, payload: dict, *,
                inline_max: int | None = None,
@@ -233,7 +276,7 @@ def invoke(self, method_id: str, payload: dict, *,
                 method_id=method_id, version=spec.version, budget_seconds=budget,
                 job_id=job_id, spec=spec,
                 deadline=(t0 + budget) if budget else None)
-            out = handler(payload, ctx)
+            out = self.executor.execute(handler, payload, ctx)
             if not isinstance(out, HandlerResult):
                 raise failures.DiracInternal(
                     f'{method_id}: its handler returned {type(out).__name__}, not a '
@@ -259,7 +302,11 @@ def invoke(self, method_id: str, payload: dict, *,
                                  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))
+                self.ledger.done(
+                    job_id, seconds=round(time.time() - t0, 3),
+                    result_summary={'ok': True,
+                                    'result_keys': sorted(out.result),
+                                    'artifact_roles': [role for role, _ in out.artifacts]})
             return env
 
         except failures.DiracFailure as f:
@@ -314,7 +361,7 @@ def _open_job(self, spec: C.MethodSpec, payload: dict,
                 method_row_id=getattr(self.ledger, 'method_row_for',
                                       lambda _m: None)(spec.method_id),
                 input_sha256=hashlib.sha256(canonical.encode()).digest(),
-                params={'method_id': spec.method_id},
+                params=dict(payload.get('parameters') or {}),
                 budget_seconds=budget, queued=False)
             return job_id
         except Exception as e:                                     # noqa: BLE001