diff --git a/static/cx-core/README.md b/static/cx-core/README.md
index a66576f87..baa76be4d 100644
--- a/static/cx-core/README.md
+++ b/static/cx-core/README.md
@@ -6,7 +6,9 @@ Calibre. It is client-agnostic: dash-node (K2Go) or the IIAB Console reach it ov
 REST API. See `controller/docs/ADR-content-exchange.md` for the frozen Pillar 0
 contracts (identity, manifest, adapter port, core REST API).
 
-This directory is the first slice: the READ side of Pillar 1, with the Kiwix adapter.
+The read, export, import and delete sides are built for all three adapters. Export runs
+as a liveness-paced long operation (prepare / poll / fetch), so a multi-gigabyte item does
+not time out on the client (K2GO-507).
 
 ## What is here
 
@@ -33,16 +35,39 @@ Environment (defaults match the box layout):
 - `CX_HOST` (default `127.0.0.1`), `CX_PORT` (default `9500`).
 - `CX_ZIMS_DIR` (default `/library/zims/content/`): where the ZIM files live.
 
-## Endpoints (read side)
+## Endpoints
 
-- `GET /v1/apps`: installed content apps that have a loaded adapter.
+Read:
+
+- `GET /v1/version`: the core build (mirrors dash-node `/system/version`).
+- `GET /v1/apps`: installed content apps that have a loaded adapter, plus `version`.
 - `GET /v1/content[?app=<app>]`: every item this box has, with identity + size. Filterable.
 - `GET /v1/content/{app}/{localRef}`: one item detail.
+- `GET /v1/storage`: content-volume disk summary + per-app content totals.
 
 Each item carries two identity handles: `localRef` (the app's own handle for
 export / delete) and `dedup` (the cross-box "already have this?" key). For Kiwix both
 are read, not computed: `dedup.build` is the ZIM header UUID.
 
+Export (a long operation, K2GO-507): a large item (a Kolibri channel) must stage before
+its bytes can stream, which takes minutes, so export is prepare / poll / fetch instead of
+one blocking request. The front owns the sequence; the core owns the op and the temp.
+
+- `POST /v1/export/prepare {app, localRef}` -> `{opId, state, totalBytes}`. `state` is
+  `staging` (a background thread is staging) or `ready` (nothing to stage). `507` when the
+  staging disk cannot hold the item.
+- `GET /v1/export/status/{opId}` -> `{state, stagedBytes, totalBytes, error?}`. `stagedBytes`
+  is the staging heartbeat (it grows as the export is assembled). `404` when the op is
+  unknown (swept), so the client re-prepares.
+- `GET /v1/export/fetch/{opId}` -> the chunked octet stream. `409` until the op is ready;
+  the stream's close drops the op and removes its temp.
+- `DELETE /v1/export/{opId}` -> `{ok: true}`. Client abort; idempotent.
+
+Import / delete:
+
+- `POST /v1/export/plan`, `POST /v1/dedup/check`, `POST /v1/import/validate`,
+  `POST /v1/import/item`, `DELETE /v1/content/{app}/{localRef}`.
+
 ## Test
 
 ```
@@ -51,17 +76,15 @@ python -m unittest discover -s tests -t .
 ```
 
 The tests cover the pure logic with no device: ZIM header UUID parsing, filename
-decomposition, soft-key normalization, the manifest JSON shape, and route dispatch.
-
-## Not in this slice
+decomposition, soft-key normalization, the manifest JSON shape, route dispatch, the export
+long-operation registry (prepare / status / fetch / delete, the TTL and startup sweeps),
+and the streaming over a real socket.
 
-Export, import, delete, the dedup-check endpoint, the Kolibri and Calibre adapters, the
-offline-disk and network carriers, strict-mode hashing, and the manager UI. The adapter
-boundary is kept clean so Kolibri and Calibre slot in next.
+## Not built yet
 
-Deployment (D1d: an Ansible-role baseline plus a dash-node-style live self-update) is
-not built here; this slice only has to be runnable. Live on-device verification needs a
-box and is a follow-up.
+The network and QR carriers, strict-mode content hashing, the cx-core live self-update
+(D1d; the Ansible role baseline is in `tools/upstream-patches/overlays/roles/cx_core`), and
+the dash-node <-> core listing consolidation (D1b).
 
 The registry resolves installed apps once at startup (a snapshot). Reacting to a module
 being installed or removed at runtime is a deferred design item: it must consume the
diff --git a/static/cx-core/cx_core/__init__.py b/static/cx-core/cx_core/__init__.py
index 38741f1fa..fe8a0fa7e 100644
--- a/static/cx-core/cx_core/__init__.py
+++ b/static/cx-core/cx_core/__init__.py
@@ -1,7 +1,11 @@
 """Content-exchange core: an in-box REST service for granular content exchange.
 
-See controller/docs/ADR-content-exchange.md for the frozen Pillar 0 contracts.
-This package holds the read side of Pillar 1 (list + identity + size).
+See controller/docs/ADR-content-exchange.md for the frozen Pillar 0 contracts and the
+K2GO-507 export long-operation design.
 """
 
-__version__ = "0.1.0"
+# The core build. Bumped per core change so a box can report which core it runs (GET
+# /v1/version), mirroring dash-node's package.json version and /system/version. The
+# cx_core Ansible role writes this into the [cx_core] section of iiab.ini.
+VERSION = "0.2.0"
+__version__ = VERSION
diff --git a/static/cx-core/cx_core/adapters/calibre.py b/static/cx-core/cx_core/adapters/calibre.py
index c8cf2c5e8..18afe9bd5 100644
--- a/static/cx-core/cx_core/adapters/calibre.py
+++ b/static/cx-core/cx_core/adapters/calibre.py
@@ -14,7 +14,6 @@
 
 from __future__ import annotations
 
-import os
 import os
 import shutil
 import sqlite3
@@ -25,7 +24,7 @@
 from ..export import resolve_within, tar_stream
 from ..importing import safe_extract_tar, single_child_dir
 from ..model import ABSENT, SAME_VERSION, SOFT, Dedup, Identity, ManifestItem, normalize_text
-from ..ports import ContentAdapter, ExportStream
+from ..ports import ContentAdapter, ExportPlan, ExportStream
 from .calibreweb import CalibreWebClient, resolve_calibre_credentials
 
 # One row per book, with its authors, formats, total byte-size and one external id if any.
@@ -218,15 +217,25 @@ def _book_path(self, uuid: str) -> Optional[str]:
             con.close()
         return row[0] if row and row[0] else None
 
-    def open_export(self, local_ref: str) -> Optional[ExportStream]:
-        # Export the book's own directory (its format files, cover and opf), the layout
-        # Calibre itself reads back. books.path comes from the DB, so a containment guard
-        # keeps a crafted path from escaping the library root.
-        rel = self._book_path(local_ref)
-        if rel is None:
+    def prepare_export(self, local_ref: str) -> Optional[ExportPlan]:
+        # A book's directory streams directly as a tar, so nothing is staged. total_bytes is
+        # the item's summed data size (the tar also carries the cover / opf sidecars, so the
+        # stream is a little larger; total_bytes is the progress denominator, not exact).
+        book_dir = self._export_dir(local_ref)
+        if book_dir is None:
             return None
-        book_dir = resolve_within(self._lib_root, rel)
-        if book_dir is None or not os.path.isdir(book_dir):
+        item = self.get(local_ref)
+        return ExportPlan(
+            stages=False,
+            total_bytes=item.bytes if item else 0,
+            filename=f"{local_ref}.tar",
+            content_type="application/x-tar",
+        )
+
+    def open_fetch(self, local_ref: str, staged_dir: Optional[str] = None) -> Optional[ExportStream]:
+        # Direct app: staged_dir is unused. Re-resolve at fetch time.
+        book_dir = self._export_dir(local_ref)
+        if book_dir is None:
             return None
         return ExportStream(
             filename=f"{local_ref}.tar",
@@ -234,6 +243,17 @@ def open_export(self, local_ref: str) -> Optional[ExportStream]:
             chunks=tar_stream(book_dir, arcname=local_ref),
         )
 
+    def _export_dir(self, local_ref: str) -> Optional[str]:
+        # books.path comes from the DB, so a containment guard keeps a crafted path from
+        # escaping the library root.
+        rel = self._book_path(local_ref)
+        if rel is None:
+            return None
+        book_dir = resolve_within(self._lib_root, rel)
+        if book_dir is None or not os.path.isdir(book_dir):
+            return None
+        return book_dir
+
     def _query(self, only_uuid: Optional[str]) -> List[ManifestItem]:
         if not os.path.exists(self._db):
             return []
diff --git a/static/cx-core/cx_core/adapters/kiwix.py b/static/cx-core/cx_core/adapters/kiwix.py
index 1586fe6a0..450b9ce49 100644
--- a/static/cx-core/cx_core/adapters/kiwix.py
+++ b/static/cx-core/cx_core/adapters/kiwix.py
@@ -32,7 +32,7 @@
     Identity,
     ManifestItem,
 )
-from ..ports import ContentAdapter, ExportStream
+from ..ports import ContentAdapter, ExportPlan, ExportStream
 
 # openzim header: magic number is uint32 little-endian 0x044D495A at offset 0, and the
 # 16-byte UUID is at offset 8. We only need the first 24 bytes.
@@ -160,13 +160,24 @@ def get(self, local_ref: str) -> Optional[ManifestItem]:
             return None
         return build_item(local_ref, size, read_zim_uuid_from_file(path))
 
-    def open_export(self, local_ref: str) -> Optional[ExportStream]:
-        # The item IS the .zim file: stream it as-is (no archive wrapper). Same containment
+    def prepare_export(self, local_ref: str) -> Optional[ExportPlan]:
+        # A ZIM streams directly (it IS the file), so nothing is staged. Same containment
         # guard as get(), so a traversal ref never reaches the filesystem.
-        if not is_safe_local_ref(local_ref) or not local_ref.lower().endswith(".zim"):
+        path = self._export_path(local_ref)
+        if path is None:
             return None
-        path = os.path.join(self._dir, local_ref)
-        if not os.path.isfile(path):
+        return ExportPlan(
+            stages=False,
+            total_bytes=os.path.getsize(path),
+            filename=local_ref,
+            content_type="application/octet-stream",
+        )
+
+    def open_fetch(self, local_ref: str, staged_dir: Optional[str] = None) -> Optional[ExportStream]:
+        # Direct app: staged_dir is unused. Re-validate at fetch time (the file may have
+        # vanished between prepare and fetch).
+        path = self._export_path(local_ref)
+        if path is None:
             return None
         return ExportStream(
             filename=local_ref,
@@ -174,6 +185,13 @@ def open_export(self, local_ref: str) -> Optional[ExportStream]:
             chunks=file_stream(path),
         )
 
+    def _export_path(self, local_ref: str) -> Optional[str]:
+        """The on-disk path of an exportable ZIM, or None. Containment-guarded, like get()."""
+        if not is_safe_local_ref(local_ref) or not local_ref.lower().endswith(".zim"):
+            return None
+        path = os.path.join(self._dir, local_ref)
+        return path if os.path.isfile(path) else None
+
     def check(self, dedup: dict) -> str:
         # Same (creator, lang, flavour) line is the comparable set; the UUID is the exact
         # build and the date is the version axis. Authoritative, so the verdict is a hard fact.
diff --git a/static/cx-core/cx_core/adapters/kolibri.py b/static/cx-core/cx_core/adapters/kolibri.py
index 2b806758e..43a6f7e2f 100644
--- a/static/cx-core/cx_core/adapters/kolibri.py
+++ b/static/cx-core/cx_core/adapters/kolibri.py
@@ -21,7 +21,7 @@
 import tempfile
 from typing import List, Optional
 
-from ..export import free_bytes, has_room, tar_stream
+from ..export import free_bytes, tar_stream
 from ..importing import safe_extract_tar, single_child_dir
 from ..model import (
     ABSENT,
@@ -33,7 +33,7 @@
     Identity,
     ManifestItem,
 )
-from ..ports import ContentAdapter, ExportStream
+from ..ports import ContentAdapter, ExportPlan, ExportStream
 
 _CHANNEL_ID_RE = re.compile(r"^[0-9a-fA-F]{32}$")
 
@@ -75,6 +75,27 @@ def delete_command(channel_id: str, kolibri_bin: str) -> List[str]:
     return [kolibri_bin, "manage", "deletechannel", channel_id]
 
 
+def _run_cancellable(argv: List[str], env: dict, should_cancel=None) -> bool:
+    """Run one child process to completion, polling ``should_cancel`` about once a second.
+    When it returns True the process is terminated (then killed if it does not exit) and this
+    returns False. Otherwise returns True on exit code 0. This is how an aborted export stops
+    the long-running Kolibri CLI instead of leaving it writing into a removed dir."""
+    proc = subprocess.Popen(argv, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    while True:
+        try:
+            proc.wait(timeout=1.0)
+            return proc.returncode == 0
+        except subprocess.TimeoutExpired:
+            if should_cancel and should_cancel():
+                proc.terminate()
+                try:
+                    proc.wait(timeout=5)
+                except subprocess.TimeoutExpired:
+                    proc.kill()
+                    proc.wait()
+                return False
+
+
 def _to_int(*values) -> Optional[int]:
     """First value that parses as an int, else None. The version may arrive as an int or a
     string over JSON, and the build mirrors it."""
@@ -169,48 +190,52 @@ def get(self, local_ref: str) -> Optional[ManifestItem]:
         items = self._query(local_ref)
         return items[0] if items else None
 
-    def open_export(self, local_ref: str) -> Optional[ExportStream]:
-        # Delegate to Kolibri's own export: it stages the channel DB + the channel's
-        # content-addressed files into a staging tree, which we then tar-stream. Read-only on
-        # the source (export copies out). The tree is removed after streaming.
+    def prepare_export(self, local_ref: str) -> Optional[ExportPlan]:
+        # A channel must be staged (the Kolibri export CLI writes the DB + content tree)
+        # before it can tar-stream, so stages=True. The op layer preflights the staging disk,
+        # creates the temp and runs the staging thread; this only validates and reports.
         if not _CHANNEL_ID_RE.match(local_ref or ""):
             return None
         channel = self._query(local_ref)
         if not channel:  # not an installed channel
             return None
-        # Preflight the staging disk against the channel size, so we fail cleanly here
-        # instead of hitting ENOSPC part way through the export (the box owns its own disk
-        # check, mirroring the front's phone-disk preflight).
-        try:
-            os.makedirs(self._staging_dir, exist_ok=True)
-        except OSError:
-            return None
-        if not has_room(self._staging_dir, channel[0].bytes):
-            return None
-        dest = tempfile.mkdtemp(prefix="cxkolibri-", dir=self._staging_dir)
+        return ExportPlan(
+            stages=True,
+            total_bytes=channel[0].bytes,
+            filename=f"{local_ref}.tar",
+            content_type="application/x-tar",
+        )
+
+    def stage_export(self, local_ref: str, dest: str, should_cancel=None) -> bool:
+        # Stage the channel into dest (the op layer's temp). Delegate to Kolibri's own export:
+        # it writes the channel DB + the content-addressed files that import dedups for free.
+        # Read-only on the source (export copies out). ``should_cancel`` lets the op layer stop
+        # the export CLI on abort instead of leaving it running into a deleted dir.
+        if not _CHANNEL_ID_RE.match(local_ref or ""):
+            return False
         try:
-            staged = self._stage_export(local_ref, dest)
+            return self._stage_export(local_ref, dest, should_cancel)
         except OSError:
-            staged = False
-        if not staged:
-            shutil.rmtree(dest, ignore_errors=True)
+            return False
+
+    def open_fetch(self, local_ref: str, staged_dir: Optional[str] = None) -> Optional[ExportStream]:
+        # Tar-stream the staged tree. The op layer sets on_close (drop the op + remove temp).
+        if not staged_dir or not os.path.isdir(staged_dir):
             return None
         return ExportStream(
             filename=f"{local_ref}.tar",
             content_type="application/x-tar",
-            chunks=tar_stream(dest, arcname=local_ref),
-            on_close=lambda: shutil.rmtree(dest, ignore_errors=True),
+            chunks=tar_stream(staged_dir, arcname=local_ref),
         )
 
-    def _stage_export(self, channel_id: str, dest: str) -> bool:
+    def _stage_export(self, channel_id: str, dest: str, should_cancel=None) -> bool:
         """Run the Kolibri export CLI into dest. A seam, so a test can stage a fake tree
-        without a live Kolibri. Returns True on success."""
+        without a live Kolibri. Each command runs as a child process polled against
+        ``should_cancel``, so an abort terminates the export instead of letting it finish into a
+        deleted dir. Returns True on success, False on CLI failure or cancellation."""
         env = dict(os.environ, KOLIBRI_HOME=self._home)
         for argv in export_commands(channel_id, dest, self._bin):
-            result = subprocess.run(
-                argv, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False
-            )
-            if result.returncode != 0:
+            if not _run_cancellable(argv, env, should_cancel):
                 return False
         return True
 
diff --git a/static/cx-core/cx_core/export_ops.py b/static/cx-core/cx_core/export_ops.py
new file mode 100644
index 000000000..802cb95ea
--- /dev/null
+++ b/static/cx-core/cx_core/export_ops.py
@@ -0,0 +1,274 @@
+"""Export long-operation registry (K2GO-507).
+
+Exporting a large item (a multi-gigabyte Kolibri channel) must stage the whole channel
+before the first byte can stream, which takes minutes. A single blocking request times out
+on the client. So export becomes a long operation: prepare starts the work and returns an
+opId at once, the client polls status for liveness and progress, then fetches the bytes.
+The front owns the sequence (ADR D1c); this registry owns the in-flight set and the temp
+lifecycle, uniform for every app. The adapters only report a plan, stage into a given
+directory and open the stream: the orchestration is here, not in an adapter.
+
+Single source of truth: the op state and the staged bytes are owned by the core. The op
+registry is in-memory, so a cx-core restart loses the ops; this is acceptable because the
+front recovers by RE-PREPARE and the startup sweep reclaims any orphan temp left by a dead
+process. The filesystem is the source for staged bytes; the registry is the source for the
+in-flight set. No durable op state, no second owner.
+
+Every op temp has exactly one cleanup owner across the four end states:
+  1. normal completion : the fetch stream's on_close drops the op and removes the temp.
+  2. client abort       : DELETE drops the op and removes the temp.
+  3. abandonment        : a lazy TTL sweep (run at the head of each op request) removes ops
+                          whose last_seen is older than the timeout, so an abandoned prepare
+                          never orphans a multi-gigabyte temp while the service keeps running.
+  4. process death      : a startup sweep removes every op temp left on disk, since a fresh
+                          process holds no ops and anything matching the temp prefix is an
+                          orphan.
+The TTL sweep is lazy (no background timer to own): it needs no thread lifecycle, and the
+startup sweep is the backstop for a temp left with no further requests coming.
+"""
+
+from __future__ import annotations
+
+import os
+import secrets
+import shutil
+import tempfile
+import threading
+import time
+from dataclasses import dataclass, field
+from typing import Any, Dict, Optional, Tuple
+
+from .export import has_room
+
+Response = Tuple[int, Any]
+
+# The op temp prefix. Op-owned and app-agnostic (the op layer creates the temp, not the
+# adapter), and distinct from the import flow's "cximport-" temps, so the startup sweep
+# reclaims only export op temps and never touches an in-flight import.
+TEMP_PREFIX = "cxexport-"
+
+# Prefixes the startup sweep also reclaims for one release: before this change the Kolibri
+# adapter staged into "cxkolibri-" temps, so an orphan from an older build (or a device test)
+# must still be swept. Not used for the import flow's "cximport-" temps, which self-clean.
+_LEGACY_SWEEP_PREFIXES = ("cxkolibri-",)
+
+# How long an op may go without a status or fetch before the lazy sweep reclaims it. The
+# client polls status throughout a staging export, so a live op keeps bumping last_seen;
+# the timeout only catches an abandoned op.
+DEFAULT_TTL_SECONDS = 600
+
+# Op states.
+STAGING = "staging"
+READY = "ready"
+ERROR = "error"
+
+
+@dataclass
+class _Op:
+    app: str
+    local_ref: str
+    temp: Optional[str]  # the staging dir for a staging op, None for a direct app
+    state: str
+    total_bytes: int
+    error: Optional[str] = None
+    last_seen: float = field(default_factory=time.monotonic)
+    # True while a fetch stream is open on this op. The TTL sweep never reclaims an in-use op,
+    # so a multi-gigabyte fetch that streams longer than the TTL (and bumps nothing meanwhile)
+    # is not swept out from under tar_stream. Set by fetch() when the stream opens, cleared by
+    # dropping the op in the stream's on_close. Lost on process death, but then the op itself
+    # is gone and the startup sweep reclaims the orphan temp.
+    in_use: bool = False
+
+
+def _dir_size(path: str) -> int:
+    """Sum of the file sizes under path (a staged tree), or 0 when it is gone. This is the
+    staging heartbeat: it grows as the export CLI writes, so the client sees progress."""
+    total = 0
+    for dirpath, _dirnames, filenames in os.walk(path):
+        for name in filenames:
+            try:
+                total += os.path.getsize(os.path.join(dirpath, name))
+            except OSError:
+                pass
+    return total
+
+
+class ExportOps:
+    """Thread-safe registry of in-flight export operations. One instance lives for the
+    server's lifetime; the handler threads call into it concurrently."""
+
+    def __init__(self, registry, staging_dir: str, ttl_seconds: int = DEFAULT_TTL_SECONDS):
+        self._registry = registry
+        self._staging_dir = staging_dir
+        self._ttl = ttl_seconds
+        self._lock = threading.Lock()
+        self._ops: Dict[str, _Op] = {}
+
+    # ---- REST-facing operations (return (status, payload)) ----
+
+    def prepare(self, app: str, local_ref: str) -> Response:
+        """Resolve the item, preflight the staging disk, and either start a background
+        staging thread (apps that stage) or mark the op ready at once (direct apps)."""
+        self._sweep_expired()
+        if not self._registry.has(app):
+            return 404, {"error": f"unknown app: {app}"}
+        plan = self._registry.get(app).prepare_export(local_ref)
+        if plan is None:
+            return 404, {"error": "item not found or not exportable"}
+        if plan.stages and not has_room(self._staging_dir, plan.total_bytes):
+            return 507, {"error": "insufficient space to stage export"}
+        op_id = secrets.token_hex(16)
+        if plan.stages:
+            try:
+                os.makedirs(self._staging_dir, exist_ok=True)
+                temp = tempfile.mkdtemp(prefix=TEMP_PREFIX, dir=self._staging_dir)
+            except OSError:
+                return 507, {"error": "cannot create staging dir"}
+            op = _Op(app, local_ref, temp, STAGING, plan.total_bytes)
+            with self._lock:
+                self._ops[op_id] = op
+            worker = threading.Thread(
+                target=self._run_stage, args=(op_id, app, local_ref, temp), daemon=True)
+            worker.start()
+            return 200, {"opId": op_id, "state": STAGING, "totalBytes": plan.total_bytes}
+        op = _Op(app, local_ref, None, READY, plan.total_bytes)
+        with self._lock:
+            self._ops[op_id] = op
+        return 200, {"opId": op_id, "state": READY, "totalBytes": plan.total_bytes}
+
+    def status(self, op_id: str) -> Response:
+        """Report the op state plus the staged-bytes heartbeat. A 404 means the op is
+        unknown (swept or never existed), so the client re-prepares."""
+        self._sweep_expired()
+        with self._lock:
+            op = self._ops.get(op_id)
+            if op is None:
+                return 404, {"error": "unknown op"}
+            op.last_seen = time.monotonic()
+            state, total, temp, error = op.state, op.total_bytes, op.temp, op.error
+        if state == READY:
+            staged = total
+        elif temp:
+            staged = _dir_size(temp)  # du outside the lock: the walk may be slow
+        else:
+            staged = 0
+        payload = {"state": state, "stagedBytes": staged, "totalBytes": total}
+        if error:
+            payload["error"] = error
+        return 200, payload
+
+    def fetch(self, op_id: str):
+        """Open the byte stream for a ready op. Returns (200, ExportStream) for the server
+        to stream; the stream's on_close drops the op and removes the temp. 409 when the op
+        is not ready, 404 when it is unknown."""
+        self._sweep_expired()
+        with self._lock:
+            op = self._ops.get(op_id)
+            if op is None:
+                return 404, {"error": "unknown op"}
+            op.last_seen = time.monotonic()
+            if op.state != READY:
+                return 409, {"error": f"op not ready: {op.state}"}
+            app, local_ref, temp = op.app, op.local_ref, op.temp
+        # Open the stream OUTSIDE the lock. The op is not pinned yet: on_open pins it once the
+        # server actually begins streaming, so a failure here (or before streaming starts)
+        # leaves the op merely fresh and reclaimable by the normal sweep, never stuck in_use.
+        try:
+            stream = self._registry.get(app).open_fetch(local_ref, temp)
+        except Exception:
+            self._finish(op_id)
+            return 500, {"error": "failed to open export"}
+        if stream is None:
+            # The item vanished between prepare and fetch: clean up and report absent.
+            self._finish(op_id)
+            return 404, {"error": "item not found or not exportable"}
+        stream.on_open = lambda: self._pin(op_id)
+        stream.on_close = lambda: self._finish(op_id)
+        return 200, stream
+
+    def delete(self, op_id: str) -> Response:
+        """Client abort: drop the op and remove its temp. Idempotent, so a repeat or an
+        unknown op is still 200 {ok: True}."""
+        self._sweep_expired()
+        self._finish(op_id)
+        return 200, {"ok": True}
+
+    # ---- lifecycle ----
+
+    def sweep_startup(self) -> None:
+        """Remove every export op temp on disk. The server calls this once on start: a fresh
+        process holds no ops, so anything under the staging dir with the op prefix is an
+        orphan left by a dead process."""
+        try:
+            names = os.listdir(self._staging_dir)
+        except OSError:
+            return
+        prefixes = (TEMP_PREFIX,) + _LEGACY_SWEEP_PREFIXES
+        for name in names:
+            if name.startswith(prefixes):
+                shutil.rmtree(os.path.join(self._staging_dir, name), ignore_errors=True)
+
+    def _run_stage(self, op_id: str, app: str, local_ref: str, temp: str) -> None:
+        """Stage the item into temp off the request thread, then flip the op to ready or
+        error. If the op was dropped meanwhile (DELETE or sweep), the flip is a no-op and the
+        temp was already removed by whoever dropped it. The adapter polls ``should_cancel``
+        and stops its staging work (terminates the subprocess) once the op is gone, so an
+        abort actually cancels the server-side work instead of letting it run on into a
+        deleted dir."""
+        def should_cancel() -> bool:
+            with self._lock:
+                return op_id not in self._ops
+
+        try:
+            ok = self._registry.get(app).stage_export(local_ref, temp, should_cancel)
+        except Exception as exc:  # any adapter / CLI failure surfaces as an op error
+            self._set_error(op_id, f"stage failed: {exc}")
+            return
+        if ok:
+            self._set_state(op_id, STAGING, READY)
+        else:
+            self._set_error(op_id, "staging failed")
+
+    def _set_state(self, op_id: str, expect: str, new: str) -> None:
+        with self._lock:
+            op = self._ops.get(op_id)
+            if op is not None and op.state == expect:
+                op.state = new
+                op.last_seen = time.monotonic()
+
+    def _set_error(self, op_id: str, message: str) -> None:
+        with self._lock:
+            op = self._ops.get(op_id)
+            if op is not None and op.state == STAGING:
+                op.state = ERROR
+                op.error = message
+                op.last_seen = time.monotonic()
+
+    def _pin(self, op_id: str) -> None:
+        """Mark the op in use for the life of an open fetch stream. The server calls this at the
+        top of streaming; _finish (the stream's on_close) drops the op and clears the pin with
+        it. Keeping the pin here, not in fetch(), means a pre-stream failure never leaves it
+        set."""
+        with self._lock:
+            op = self._ops.get(op_id)
+            if op is not None:
+                op.in_use = True
+
+    def _finish(self, op_id: str) -> None:
+        """Drop the op and remove its temp. The one cleanup primitive, idempotent."""
+        with self._lock:
+            op = self._ops.pop(op_id, None)
+        if op is not None and op.temp:
+            shutil.rmtree(op.temp, ignore_errors=True)
+
+    def _sweep_expired(self) -> None:
+        now = time.monotonic()
+        stale = []
+        with self._lock:
+            for op_id, op in list(self._ops.items()):
+                # Never reclaim an op with an open fetch stream: its temp is in use.
+                if not op.in_use and now - op.last_seen > self._ttl:
+                    stale.append(self._ops.pop(op_id))
+        for op in stale:
+            if op.temp:
+                shutil.rmtree(op.temp, ignore_errors=True)
diff --git a/static/cx-core/cx_core/ports.py b/static/cx-core/cx_core/ports.py
index e1c664530..25a49a9f6 100644
--- a/static/cx-core/cx_core/ports.py
+++ b/static/cx-core/cx_core/ports.py
@@ -19,13 +19,34 @@
 
 @dataclass
 class ExportStream:
-    """One item staged for export: a byte stream the core serves and the front moves
-    (ADR D1c). ``on_close`` releases any staging (a temp export tree) after streaming."""
+    """One item's byte stream the core serves and the front moves (ADR D1c). ``on_open`` runs
+    when the server begins streaming and ``on_close`` after the last byte (or on any streaming
+    error); the EXPORT OP LAYER sets both: on_open pins the op for the stream's life, on_close
+    unpins and drops it. The server fires on_open at the top of the same try whose finally runs
+    on_close, so the pin and its release are a matched pair even when the client disconnects.
+    The adapter only opens the stream; the op layer owns the lifecycle (K2GO-507)."""
 
     filename: str
     content_type: str
     chunks: Iterable[bytes]
     on_close: Optional[Callable[[], None]] = None
+    on_open: Optional[Callable[[], None]] = None
+
+
+@dataclass
+class ExportPlan:
+    """How to export one item (K2GO-507). ``stages`` is True when the item must be staged
+    into a temp before its bytes can stream (a Kolibri channel), False when it streams
+    directly (a Kiwix .zim, a Calibre book dir). ``total_bytes`` is the size to move: the
+    disk preflight and the progress denominator. ``filename`` / ``content_type`` describe
+    the stream. Returned by ``prepare_export``; None when the item is absent or not
+    exportable. The op layer owns the temp and the staging thread; the adapter only reports
+    this plan and, when asked, stages and opens the stream."""
+
+    stages: bool
+    total_bytes: int
+    filename: str
+    content_type: str
 
 
 class ContentAdapter(ABC):
@@ -53,9 +74,25 @@ def get(self, local_ref: str) -> Optional[ManifestItem]:
                 return item
         return None
 
-    def open_export(self, local_ref: str) -> Optional[ExportStream]:
-        """Stage one item and return a byte stream, or None when the item is absent or the
-        adapter cannot export it. Read-only on the source. Default: export not supported."""
+    def prepare_export(self, local_ref: str) -> "Optional[ExportPlan]":
+        """Validate that this item can be exported and report its export plan (K2GO-507).
+        Read-only on the source: no staging, no temp. Default: export not supported."""
+        return None
+
+    def stage_export(self, local_ref: str, dest: str,
+                     should_cancel: Optional[Callable[[], bool]] = None) -> bool:
+        """Stage the item into ``dest`` (the op layer's temp dir), for a plan with
+        stages=True. Read-only on the source (copies out). Runs off the request thread, so it
+        may be slow. ``should_cancel``, when supplied, is polled during long work: once it
+        returns True (the op was aborted or swept) the adapter must stop and return False, so a
+        cancel actually stops the server-side work. Returns True on success. Default: nothing
+        to stage."""
+        return False
+
+    def open_fetch(self, local_ref: str, staged_dir: Optional[str] = None) -> Optional[ExportStream]:
+        """Open the byte stream for a prepared export: the staged tar (``staged_dir`` set) or
+        the direct file / tar (``staged_dir`` None). The op layer sets ``on_close``. Returns
+        None when the item is no longer exportable. Default: not supported."""
         return None
 
     def check(self, dedup: dict) -> str:
diff --git a/static/cx-core/cx_core/router.py b/static/cx-core/cx_core/router.py
index d7592cf70..23a289031 100644
--- a/static/cx-core/cx_core/router.py
+++ b/static/cx-core/cx_core/router.py
@@ -10,7 +10,9 @@
 from typing import Any, List, Optional, Tuple
 from urllib.parse import unquote
 
+from . import VERSION
 from .export import disk_total_free, has_room
+from .export_ops import ExportOps
 from .importing import is_within_dir
 from .registry import Registry
 
@@ -149,13 +151,31 @@ def _storage(registry: Registry) -> Response:
     }
 
 
-def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any = None) -> Response:
+def _export_prepare(ops: Optional[ExportOps], body: Any) -> Response:
+    """Start (or mark ready) one export long operation and return its opId (K2GO-507)."""
+    if ops is None:
+        return 500, {"error": "export ops unavailable"}
+    if not isinstance(body, dict):
+        return 400, {"error": "body must be {app, localRef}"}
+    app = body.get("app")
+    local_ref = body.get("localRef")
+    if not isinstance(app, str) or not isinstance(local_ref, str) or not local_ref:
+        return 400, {"error": "app and localRef are required"}
+    return ops.prepare(app, local_ref)
+
+
+def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any = None,
+             ops: Optional[ExportOps] = None) -> Response:
     """Route one request. ``query`` is a dict of first values; ``body`` is the parsed JSON
-    for a POST."""
+    for a POST; ``ops`` is the export long-operation registry (the server supplies it, the
+    export-op routes need it)."""
     trimmed = path.rstrip("/") or "/"
     segments = [unquote(s) for s in trimmed.split("/") if s != ""]
 
     if method == "POST":
+        # POST /v1/export/prepare : start an export long operation, return its opId.
+        if segments == ["v1", "export", "prepare"]:
+            return _export_prepare(ops, body)
         # POST /v1/export/plan : plan an export (sizes only, no payload moved).
         if segments == ["v1", "export", "plan"]:
             return _export_plan(registry, body)
@@ -171,6 +191,11 @@ def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any
         return 404, {"error": "not found"}
 
     if method == "DELETE":
+        # DELETE /v1/export/{opId} : abort an export long operation.
+        if len(segments) == 3 and segments[0] == "v1" and segments[1] == "export":
+            if ops is None:
+                return 500, {"error": "export ops unavailable"}
+            return ops.delete(segments[2])
         # DELETE /v1/content/{app}/{localRef} : reclaim space by removing one item.
         if len(segments) == 4 and segments[0] == "v1" and segments[1] == "content":
             return _delete_item(registry, segments[2], segments[3])
@@ -179,9 +204,26 @@ def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any
     if method != "GET":
         return 405, {"error": "method not allowed"}
 
-    # GET /v1/apps : installed apps that have a loaded adapter.
+    # GET /v1/version : which core build this box runs (mirrors dash-node /system/version).
+    if segments == ["v1", "version"]:
+        return 200, {"version": VERSION}
+
+    # GET /v1/apps : installed apps that have a loaded adapter, plus the core version.
     if segments == ["v1", "apps"]:
-        return 200, {"apps": registry.apps()}
+        return 200, {"apps": registry.apps(), "version": VERSION}
+
+    # GET /v1/export/status/{opId} : the op state + the staged-bytes heartbeat.
+    if len(segments) == 4 and segments[0] == "v1" and segments[1] == "export" and segments[2] == "status":
+        if ops is None:
+            return 500, {"error": "export ops unavailable"}
+        return ops.status(segments[3])
+
+    # GET /v1/export/fetch/{opId} : stream the prepared item (payload is an ExportStream the
+    # server streams). 409 until the op is ready; the stream's on_close drops the op.
+    if len(segments) == 4 and segments[0] == "v1" and segments[1] == "export" and segments[2] == "fetch":
+        if ops is None:
+            return 500, {"error": "export ops unavailable"}
+        return ops.fetch(segments[3])
 
     # GET /v1/storage : box content-volume disk summary + per-app content totals.
     if segments == ["v1", "storage"]:
@@ -201,15 +243,4 @@ def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any
             return 404, {"error": "item not found"}
         return 200, item.to_json()
 
-    # GET /v1/export/item/{app}/{localRef} : stream that one item's archive (payload is an
-    # ExportStream; the server streams it). The front moves the bytes and owns the sequence.
-    if len(segments) == 5 and segments[0] == "v1" and segments[1] == "export" and segments[2] == "item":
-        app, local_ref = segments[3], segments[4]
-        if not registry.has(app):
-            return 404, {"error": f"unknown app: {app}"}
-        stream = registry.get(app).open_export(local_ref)
-        if stream is None:
-            return 404, {"error": "item not found or not exportable"}
-        return 200, stream
-
     return 404, {"error": "not found"}
diff --git a/static/cx-core/cx_core/server.py b/static/cx-core/cx_core/server.py
index 9c9dc7b82..704db9891 100644
--- a/static/cx-core/cx_core/server.py
+++ b/static/cx-core/cx_core/server.py
@@ -9,7 +9,9 @@
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from urllib.parse import parse_qs, urlsplit
 
+from . import VERSION
 from .config import Config
+from .export_ops import ExportOps
 from .ports import ExportStream
 from .registry import Registry
 from .router import dispatch
@@ -19,9 +21,9 @@
 _TOO_LARGE = object()  # sentinel: declared body exceeds _MAX_BODY
 
 
-def make_handler(registry: Registry):
+def make_handler(registry: Registry, ops: ExportOps):
     class Handler(BaseHTTPRequestHandler):
-        server_version = "cx-core/0.1"
+        server_version = f"cx-core/{VERSION}"
         # HTTP/1.1 so the chunked export stream is valid (HTTP/1.0 would hand the client the
         # chunk-size lines as body). The JSON path sends Content-Length, so keep-alive holds.
         protocol_version = "HTTP/1.1"
@@ -29,7 +31,7 @@ class Handler(BaseHTTPRequestHandler):
         def do_GET(self) -> None:  # noqa: N802 (http.server naming)
             parts = urlsplit(self.path)
             query = {k: v[0] for k, v in parse_qs(parts.query).items()}
-            status, payload = dispatch(registry, "GET", parts.path, query)
+            status, payload = dispatch(registry, "GET", parts.path, query, ops=ops)
             if isinstance(payload, ExportStream):
                 self._stream(status, payload)
             else:
@@ -46,13 +48,13 @@ def do_POST(self) -> None:  # noqa: N802 (http.server naming)
             if body is _BAD_JSON:
                 self._send(400, {"error": "invalid JSON body"})
                 return
-            status, payload = dispatch(registry, "POST", parts.path, {}, body)
+            status, payload = dispatch(registry, "POST", parts.path, {}, body, ops=ops)
             self._send(status, payload)
 
         def do_DELETE(self) -> None:  # noqa: N802 (http.server naming)
             parts = urlsplit(self.path)
             query = {k: v[0] for k, v in parse_qs(parts.query).items()}
-            status, payload = dispatch(registry, "DELETE", parts.path, query)
+            status, payload = dispatch(registry, "DELETE", parts.path, query, ops=ops)
             self._send(status, payload)
 
         def _read_json_body(self):
@@ -91,6 +93,10 @@ def _stream(self, status: int, stream: ExportStream) -> None:
             self.send_header("Connection", "close")
             self.end_headers()
             try:
+                # Pin the op for the stream's life inside the same try whose finally runs
+                # on_close, so the pin is always released (even on a client disconnect).
+                if stream.on_open:
+                    stream.on_open()
                 for block in stream.chunks:
                     if not block:
                         continue
@@ -111,7 +117,11 @@ def log_message(self, *_args) -> None:
 
 def serve(config: Config) -> None:
     registry = Registry.load(config)
-    httpd = ThreadingHTTPServer((config.host, config.port), make_handler(registry))
-    print(f"cx-core listening on http://{config.host}:{config.port} "
+    ops = ExportOps(registry, config.staging_dir)
+    # Reclaim export op temps left by a dead process before serving (K2GO-507 process-death
+    # cleanup owner): a fresh process holds no ops, so any op temp on disk is an orphan.
+    ops.sweep_startup()
+    httpd = ThreadingHTTPServer((config.host, config.port), make_handler(registry, ops))
+    print(f"cx-core {VERSION} listening on http://{config.host}:{config.port} "
           f"(apps: {', '.join(registry.apps()) or 'none'})")
     httpd.serve_forever()
diff --git a/static/cx-core/tests/test_calibre.py b/static/cx-core/tests/test_calibre.py
index 561fcd03b..02b10dbe5 100644
--- a/static/cx-core/tests/test_calibre.py
+++ b/static/cx-core/tests/test_calibre.py
@@ -90,7 +90,18 @@ def test_absent_db_not_installed(self):
 
 
 class CalibreExportTest(unittest.TestCase):
-    def test_open_export_tars_the_book_dir(self):
+    def test_prepare_export_is_direct_with_total_bytes(self):
+        with tempfile.TemporaryDirectory() as lib:
+            db = os.path.join(lib, "metadata.db")
+            _make_db(db)
+            os.makedirs(os.path.join(lib, "Tolkien", "The Hobbit (1)"))
+            plan = CalibreAdapter(db).prepare_export("u-1")
+            self.assertIsNotNone(plan)
+            self.assertFalse(plan.stages)  # a book dir streams directly
+            self.assertEqual(plan.total_bytes, 3000)  # 1000 EPUB + 2000 PDF
+            self.assertEqual(plan.filename, "u-1.tar")
+
+    def test_open_fetch_tars_the_book_dir(self):
         import io
         import tarfile
 
@@ -104,7 +115,7 @@ def test_open_export_tars_the_book_dir(self):
             with open(os.path.join(book_dir, "cover.jpg"), "wb") as f:
                 f.write(b"JPG")
 
-            stream = CalibreAdapter(db).open_export("u-1")
+            stream = CalibreAdapter(db).open_fetch("u-1")
             self.assertIsNotNone(stream)
             self.assertEqual(stream.filename, "u-1.tar")
             blob = b"".join(stream.chunks)
@@ -112,20 +123,22 @@ def test_open_export_tars_the_book_dir(self):
                 names = sorted(m.name for m in tar.getmembers() if m.isfile())
                 self.assertEqual(names, ["u-1/cover.jpg", "u-1/hobbit.epub"])
 
-    def test_open_export_missing_book_or_dir_is_none(self):
+    def test_export_missing_book_or_dir_is_none(self):
         with tempfile.TemporaryDirectory() as lib:
             db = os.path.join(lib, "metadata.db")
             _make_db(db)  # row exists but the book dir was never created
             adapter = CalibreAdapter(db)
-            self.assertIsNone(adapter.open_export("u-1"))
-            self.assertIsNone(adapter.open_export("no-such-uuid"))
+            self.assertIsNone(adapter.prepare_export("u-1"))
+            self.assertIsNone(adapter.prepare_export("no-such-uuid"))
+            self.assertIsNone(adapter.open_fetch("u-1"))
 
-    def test_open_export_rejects_traversal_path(self):
+    def test_export_rejects_traversal_path(self):
         with tempfile.TemporaryDirectory() as lib:
             db = os.path.join(lib, "metadata.db")
             # a malicious books.path must not escape the library root.
             _make_db(db, book_path="../../etc")
-            self.assertIsNone(CalibreAdapter(db).open_export("u-1"))
+            self.assertIsNone(CalibreAdapter(db).prepare_export("u-1"))
+            self.assertIsNone(CalibreAdapter(db).open_fetch("u-1"))
 
 
 class CalibreCheckTest(unittest.TestCase):
diff --git a/static/cx-core/tests/test_export_ops.py b/static/cx-core/tests/test_export_ops.py
new file mode 100644
index 000000000..5a16bef8f
--- /dev/null
+++ b/static/cx-core/tests/test_export_ops.py
@@ -0,0 +1,328 @@
+import os
+import tempfile
+import threading
+import time
+import unittest
+
+import cx_core.export_ops as export_ops
+from cx_core.export_ops import ExportOps
+from cx_core.ports import ContentAdapter, ExportPlan, ExportStream
+from cx_core.registry import Registry
+
+
+class _DirectFake(ContentAdapter):
+    """A direct app (Kiwix / Calibre): nothing to stage, ready at once."""
+
+    app = "kiwix"
+
+    def __init__(self, total=42, present=True):
+        self._total = total
+        self._present = present
+
+    def is_installed(self):
+        return True
+
+    def list(self):
+        return []
+
+    def prepare_export(self, local_ref):
+        if not self._present:
+            return None
+        return ExportPlan(False, self._total, local_ref, "application/octet-stream")
+
+    def open_fetch(self, local_ref, staged_dir=None):
+        if not self._present:
+            return None
+        return ExportStream(local_ref, "application/octet-stream", [b"x" * self._total])
+
+
+class _StagingFake(ContentAdapter):
+    """A staging app (Kolibri): stage_export writes into dest, gated by an event so a test
+    can observe the staging state before the op flips to ready."""
+
+    app = "kolibri"
+
+    def __init__(self, total=300, gate=None, fail=False):
+        self._total = total
+        self._gate = gate
+        self._fail = fail
+
+    def is_installed(self):
+        return True
+
+    def list(self):
+        return []
+
+    def prepare_export(self, local_ref):
+        if local_ref == "missing":
+            return None
+        return ExportPlan(True, self._total, f"{local_ref}.tar", "application/x-tar")
+
+    def stage_export(self, local_ref, dest, should_cancel=None):
+        if self._gate is not None:
+            self._gate.wait(5)
+        if self._fail:
+            return False
+        with open(os.path.join(dest, "part.bin"), "wb") as f:
+            f.write(b"x" * self._total)
+        return True
+
+    def open_fetch(self, local_ref, staged_dir=None):
+        path = os.path.join(staged_dir or "", "part.bin")
+        data = b""
+        if os.path.isfile(path):
+            with open(path, "rb") as f:
+                data = f.read()
+        return ExportStream(f"{local_ref}.tar", "application/x-tar", [data])
+
+
+def _ops(adapter, staging, ttl=60):
+    reg = Registry([adapter])
+    return ExportOps(reg, staging_dir=staging, ttl_seconds=ttl)
+
+
+def _wait_state(ops, op_id, want, timeout=5):
+    end = time.time() + timeout
+    body = {}
+    while time.time() < end:
+        _, body = ops.status(op_id)
+        if body.get("state") == want:
+            return body
+        time.sleep(0.01)
+    return body
+
+
+class PrepareTest(unittest.TestCase):
+    def test_direct_app_is_ready_at_once(self):
+        with tempfile.TemporaryDirectory() as d:
+            ops = _ops(_DirectFake(total=42), d)
+            status, body = ops.prepare("kiwix", "a.zim")
+            self.assertEqual(status, 200)
+            self.assertEqual(body["state"], "ready")
+            self.assertEqual(body["totalBytes"], 42)
+            self.assertIn("opId", body)
+
+    def test_staging_app_starts_staging(self):
+        with tempfile.TemporaryDirectory() as d:
+            gate = threading.Event()
+            gate.set()  # let staging run to completion
+            ops = _ops(_StagingFake(total=300, gate=gate), d)
+            status, body = ops.prepare("kolibri", "chan")
+            self.assertEqual(status, 200)
+            self.assertEqual(body["state"], "staging")
+            self.assertEqual(body["totalBytes"], 300)
+            # the op flips to ready once the staging thread finishes.
+            ready = _wait_state(ops, body["opId"], "ready")
+            self.assertEqual(ready["state"], "ready")
+
+    def test_unknown_app_404(self):
+        with tempfile.TemporaryDirectory() as d:
+            ops = _ops(_DirectFake(), d)
+            status, _ = ops.prepare("nope", "x")
+            self.assertEqual(status, 404)
+
+    def test_not_exportable_404(self):
+        with tempfile.TemporaryDirectory() as d:
+            ops = _ops(_StagingFake(), d)
+            status, _ = ops.prepare("kolibri", "missing")
+            self.assertEqual(status, 404)
+
+    def test_no_room_507(self):
+        with tempfile.TemporaryDirectory() as d:
+            ops = _ops(_StagingFake(total=10 ** 9), d)
+            orig = export_ops.has_room
+            export_ops.has_room = lambda path, n: False
+            try:
+                status, _ = ops.prepare("kolibri", "chan")
+            finally:
+                export_ops.has_room = orig
+            self.assertEqual(status, 507)
+
+
+class StatusTest(unittest.TestCase):
+    def test_staged_bytes_grows_then_flips_to_ready(self):
+        with tempfile.TemporaryDirectory() as d:
+            gate = threading.Event()  # hold staging open
+            ops = _ops(_StagingFake(total=300, gate=gate), d)
+            _, prep = ops.prepare("kolibri", "chan")
+            op_id = prep["opId"]
+            # Before the gate opens the temp is empty: staging, 0 staged.
+            _, mid = ops.status(op_id)
+            self.assertEqual(mid["state"], "staging")
+            self.assertEqual(mid["stagedBytes"], 0)
+            gate.set()
+            ready = _wait_state(ops, op_id, "ready")
+            self.assertEqual(ready["state"], "ready")
+            self.assertEqual(ready["stagedBytes"], 300)
+            self.assertEqual(ready["totalBytes"], 300)
+
+    def test_unknown_op_404(self):
+        with tempfile.TemporaryDirectory() as d:
+            ops = _ops(_DirectFake(), d)
+            status, _ = ops.status("deadbeef")
+            self.assertEqual(status, 404)
+
+    def test_staging_failure_becomes_error(self):
+        with tempfile.TemporaryDirectory() as d:
+            gate = threading.Event()
+            gate.set()
+            ops = _ops(_StagingFake(gate=gate, fail=True), d)
+            _, prep = ops.prepare("kolibri", "chan")
+            body = _wait_state(ops, prep["opId"], "error")
+            self.assertEqual(body["state"], "error")
+            self.assertIn("error", body)
+
+
+class FetchTest(unittest.TestCase):
+    def test_fetch_streams_then_on_close_drops_op_and_temp(self):
+        with tempfile.TemporaryDirectory() as d:
+            gate = threading.Event()
+            gate.set()
+            ops = _ops(_StagingFake(total=120, gate=gate), d)
+            _, prep = ops.prepare("kolibri", "chan")
+            op_id = prep["opId"]
+            _wait_state(ops, op_id, "ready")
+            temp = ops._ops[op_id].temp
+            self.assertTrue(os.path.isdir(temp))
+            status, stream = ops.fetch(op_id)
+            self.assertEqual(status, 200)
+            data = b"".join(stream.chunks)
+            self.assertEqual(len(data), 120)
+            # the server calls on_close after streaming: it drops the op and removes the temp.
+            stream.on_close()
+            self.assertFalse(os.path.exists(temp))
+            self.assertEqual(ops.status(op_id)[0], 404)
+
+    def test_fetch_before_ready_409(self):
+        with tempfile.TemporaryDirectory() as d:
+            gate = threading.Event()  # keep it staging
+            ops = _ops(_StagingFake(gate=gate), d)
+            _, prep = ops.prepare("kolibri", "chan")
+            status, _ = ops.fetch(prep["opId"])
+            self.assertEqual(status, 409)
+            # Release the staging thread and let it finish before the temp dir is cleaned up,
+            # so the thread is not writing into a dir being removed under it.
+            gate.set()
+            _wait_state(ops, prep["opId"], "ready")
+
+    def test_fetch_unknown_op_404(self):
+        with tempfile.TemporaryDirectory() as d:
+            ops = _ops(_DirectFake(), d)
+            status, _ = ops.fetch("nope")
+            self.assertEqual(status, 404)
+
+    def test_fetch_direct_app_has_no_temp(self):
+        with tempfile.TemporaryDirectory() as d:
+            ops = _ops(_DirectFake(total=7), d)
+            _, prep = ops.prepare("kiwix", "a.zim")
+            status, stream = ops.fetch(prep["opId"])
+            self.assertEqual(status, 200)
+            self.assertEqual(len(b"".join(stream.chunks)), 7)
+            stream.on_close()  # no temp, just drops the op
+            self.assertEqual(ops.status(prep["opId"])[0], 404)
+
+
+class DeleteTest(unittest.TestCase):
+    def test_delete_drops_op_and_temp_idempotent(self):
+        with tempfile.TemporaryDirectory() as d:
+            gate = threading.Event()
+            gate.set()
+            ops = _ops(_StagingFake(gate=gate), d)
+            _, prep = ops.prepare("kolibri", "chan")
+            op_id = prep["opId"]
+            _wait_state(ops, op_id, "ready")
+            temp = ops._ops[op_id].temp
+            self.assertEqual(ops.delete(op_id), (200, {"ok": True}))
+            self.assertFalse(os.path.exists(temp))
+            self.assertEqual(ops.status(op_id)[0], 404)
+            # idempotent: a repeat and an unknown op are both 200 ok.
+            self.assertEqual(ops.delete(op_id), (200, {"ok": True}))
+            self.assertEqual(ops.delete("never-existed"), (200, {"ok": True}))
+
+
+class SweepTest(unittest.TestCase):
+    def test_ttl_sweep_reclaims_an_abandoned_op(self):
+        with tempfile.TemporaryDirectory() as d:
+            gate = threading.Event()
+            gate.set()
+            ops = _ops(_StagingFake(gate=gate), d, ttl=60)
+            _, prep = ops.prepare("kolibri", "chan")
+            op_id = prep["opId"]
+            _wait_state(ops, op_id, "ready")
+            temp = ops._ops[op_id].temp
+            # Backdate last_seen past the TTL, then run the lazy sweep.
+            ops._ops[op_id].last_seen -= 120
+            ops._sweep_expired()
+            self.assertNotIn(op_id, ops._ops)
+            self.assertFalse(os.path.exists(temp))
+
+    def test_ttl_sweep_never_reclaims_an_op_being_fetched(self):
+        # A multi-gigabyte fetch can stream longer than the TTL without bumping last_seen; the
+        # sweep must not rmtree its temp out from under the open stream.
+        with tempfile.TemporaryDirectory() as d:
+            gate = threading.Event()
+            gate.set()
+            ops = _ops(_StagingFake(gate=gate), d, ttl=60)
+            _, prep = ops.prepare("kolibri", "chan")
+            op_id = prep["opId"]
+            _wait_state(ops, op_id, "ready")
+            status, stream = ops.fetch(op_id)  # returns the stream, not yet pinned
+            self.assertEqual(status, 200)
+            self.assertFalse(ops._ops[op_id].in_use)  # the pin waits for streaming to start
+            temp = ops._ops[op_id].temp
+            stream.on_open()                    # the server pins at the top of streaming
+            self.assertTrue(ops._ops[op_id].in_use)
+            ops._ops[op_id].last_seen -= 120    # older than the TTL
+            ops._sweep_expired()
+            self.assertIn(op_id, ops._ops)      # pinned: survives the sweep
+            self.assertTrue(os.path.exists(temp))
+            b"".join(stream.chunks)             # drain, then close as the server would
+            stream.on_close()
+            self.assertNotIn(op_id, ops._ops)
+            self.assertFalse(os.path.exists(temp))
+
+    def test_fetch_failure_before_streaming_does_not_leave_op_pinned(self):
+        # If open_fetch raises (or the stream never starts), the op must not be stuck in_use:
+        # it is cleaned up so it cannot survive the sweep forever.
+        class _RaisingFetch(_StagingFake):
+            def open_fetch(self, local_ref, staged_dir=None):
+                raise RuntimeError("boom")
+
+        with tempfile.TemporaryDirectory() as d:
+            gate = threading.Event()
+            gate.set()
+            ops = _ops(_RaisingFetch(gate=gate), d, ttl=60)
+            _, prep = ops.prepare("kolibri", "chan")
+            op_id = prep["opId"]
+            _wait_state(ops, op_id, "ready")
+            temp = ops._ops[op_id].temp
+            status, _ = ops.fetch(op_id)
+            self.assertEqual(status, 500)
+            # The op (and its temp) were dropped, so nothing is left pinned or orphaned.
+            self.assertNotIn(op_id, ops._ops)
+            self.assertFalse(os.path.exists(temp))
+
+    def test_startup_sweep_removes_export_and_legacy_temps_only(self):
+        with tempfile.TemporaryDirectory() as d:
+            orphan = os.path.join(d, "cxexport-abc123")
+            legacy = os.path.join(d, "cxkolibri-old99")  # a pre-change temp (e.g. on the A16)
+            an_import = os.path.join(d, "cximport-xyz")   # an in-flight import must survive
+            unrelated = os.path.join(d, "kolibri")
+            for p in (orphan, legacy, an_import, unrelated):
+                os.makedirs(p)
+                with open(os.path.join(p, "f"), "wb") as f:
+                    f.write(b"x")
+            ops = _ops(_DirectFake(), d)
+            ops.sweep_startup()
+            self.assertFalse(os.path.exists(orphan))
+            self.assertFalse(os.path.exists(legacy))
+            self.assertTrue(os.path.exists(an_import))
+            self.assertTrue(os.path.exists(unrelated))
+
+    def test_startup_sweep_tolerates_missing_dir(self):
+        ops = _ops(_DirectFake(), "/no/such/staging/dir")
+        ops.sweep_startup()  # must not raise
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/static/cx-core/tests/test_kiwix.py b/static/cx-core/tests/test_kiwix.py
index 9907d6fbf..3decd3917 100644
--- a/static/cx-core/tests/test_kiwix.py
+++ b/static/cx-core/tests/test_kiwix.py
@@ -143,23 +143,38 @@ def test_absent_dir_is_not_installed(self):
 
 
 class KiwixExportTest(unittest.TestCase):
-    def test_open_export_streams_the_zim_as_is(self):
+    def test_prepare_export_is_direct_with_total_bytes(self):
         with tempfile.TemporaryDirectory() as d:
             name = "devdocs_en_angular_2026-05.zim"
             payload = _zim_header(uuid.uuid4()) + os.urandom(200)
             with open(os.path.join(d, name), "wb") as f:
                 f.write(payload)
-            stream = KiwixAdapter(d).open_export(name)
+            plan = KiwixAdapter(d).prepare_export(name)
+            self.assertIsNotNone(plan)
+            self.assertFalse(plan.stages)  # a ZIM streams directly, nothing to stage
+            self.assertEqual(plan.total_bytes, len(payload))
+            self.assertEqual(plan.filename, name)
+            self.assertEqual(plan.content_type, "application/octet-stream")
+
+    def test_open_fetch_streams_the_zim_as_is(self):
+        with tempfile.TemporaryDirectory() as d:
+            name = "devdocs_en_angular_2026-05.zim"
+            payload = _zim_header(uuid.uuid4()) + os.urandom(200)
+            with open(os.path.join(d, name), "wb") as f:
+                f.write(payload)
+            stream = KiwixAdapter(d).open_fetch(name)
             self.assertIsNotNone(stream)
             self.assertEqual(stream.filename, name)
             self.assertEqual(stream.content_type, "application/octet-stream")
             self.assertEqual(b"".join(stream.chunks), payload)
 
-    def test_open_export_rejects_traversal_and_missing(self):
+    def test_export_rejects_traversal_and_missing(self):
         with tempfile.TemporaryDirectory() as d:
             adapter = KiwixAdapter(d)
-            self.assertIsNone(adapter.open_export("../x.zim"))
-            self.assertIsNone(adapter.open_export("absent.zim"))
+            self.assertIsNone(adapter.prepare_export("../x.zim"))
+            self.assertIsNone(adapter.prepare_export("absent.zim"))
+            self.assertIsNone(adapter.open_fetch("../x.zim"))
+            self.assertIsNone(adapter.open_fetch("absent.zim"))
 
 
 class KiwixCheckTest(unittest.TestCase):
diff --git a/static/cx-core/tests/test_kolibri.py b/static/cx-core/tests/test_kolibri.py
index 924ec0aee..2241cedd3 100644
--- a/static/cx-core/tests/test_kolibri.py
+++ b/static/cx-core/tests/test_kolibri.py
@@ -1,6 +1,8 @@
 import os
 import sqlite3
+import sys
 import tempfile
+import time
 import unittest
 
 import io
@@ -9,6 +11,7 @@
 import cx_core.adapters.kolibri as kolibri_mod
 from cx_core.adapters.kolibri import (
     KolibriAdapter,
+    _run_cancellable,
     build_channel_item,
     delete_command,
     export_commands,
@@ -178,7 +181,7 @@ class _StubbedKolibri(KolibriAdapter):
     """Overrides the CLI seam so the export assembly + streaming is testable without a live
     Kolibri: it writes a tiny Kolibri-shaped export tree into dest."""
 
-    def _stage_export(self, channel_id, dest):
+    def _stage_export(self, channel_id, dest, should_cancel=None):
         self.staged_dest = dest
         dbs = os.path.join(dest, "content", "databases")
         storage = os.path.join(dest, "content", "storage", "a", "b")
@@ -192,7 +195,18 @@ def _stage_export(self, channel_id, dest):
 
 
 class KolibriExportTest(unittest.TestCase):
-    def test_open_export_streams_tar_of_the_bundle_and_cleans_up(self):
+    def test_prepare_export_is_staging_with_total_bytes(self):
+        with tempfile.TemporaryDirectory() as d:
+            db = os.path.join(d, "db.sqlite3")
+            _make_db(db, cid=VALID_ID)  # channel available bytes = 100 in the fixture
+            adapter = _StubbedKolibri(db, staging_dir=os.path.join(d, "staging"))
+            plan = adapter.prepare_export(VALID_ID)
+            self.assertIsNotNone(plan)
+            self.assertTrue(plan.stages)  # a channel must be staged before it streams
+            self.assertEqual(plan.total_bytes, 100)
+            self.assertEqual(plan.filename, f"{VALID_ID}.tar")
+
+    def test_stage_export_then_open_fetch_streams_tar_of_the_bundle(self):
         import io
         import tarfile
 
@@ -200,41 +214,55 @@ def test_open_export_streams_tar_of_the_bundle_and_cleans_up(self):
             db = os.path.join(d, "db.sqlite3")
             _make_db(db, cid=VALID_ID)
             adapter = _StubbedKolibri(db, staging_dir=os.path.join(d, "staging"))
-            stream = adapter.open_export(VALID_ID)
+            dest = os.path.join(d, "op-temp")
+            os.makedirs(dest)
+            self.assertTrue(adapter.stage_export(VALID_ID, dest))
+            stream = adapter.open_fetch(VALID_ID, dest)
             self.assertIsNotNone(stream)
             self.assertEqual(stream.filename, f"{VALID_ID}.tar")
+            # The adapter does NOT set on_close: the op layer owns the temp lifecycle.
+            self.assertIsNone(stream.on_close)
             blob = b"".join(stream.chunks)
             with tarfile.open(fileobj=io.BytesIO(blob), mode="r") as tar:
                 names = sorted(m.name for m in tar.getmembers() if m.isfile())
             self.assertIn(f"{VALID_ID}/content/databases/{VALID_ID}.sqlite3", names)
             self.assertIn(f"{VALID_ID}/content/storage/a/b/abc.mp4", names)
-            # the staging tree is removed after streaming.
-            self.assertTrue(os.path.isdir(adapter.staged_dest))
-            stream.on_close()
-            self.assertFalse(os.path.exists(adapter.staged_dest))
 
-    def test_open_export_rejects_non_channel_id_and_uninstalled(self):
+    def test_prepare_export_rejects_non_channel_id_and_uninstalled(self):
         with tempfile.TemporaryDirectory() as d:
             db = os.path.join(d, "db.sqlite3")
             _make_db(db, cid=VALID_ID)
             adapter = _StubbedKolibri(db, staging_dir=os.path.join(d, "staging"))
-            self.assertIsNone(adapter.open_export("not-a-channel-id"))
-            self.assertIsNone(adapter.open_export("a" * 32))  # valid shape, not installed
+            self.assertIsNone(adapter.prepare_export("not-a-channel-id"))
+            self.assertIsNone(adapter.prepare_export("a" * 32))  # valid shape, not installed
 
-    def test_open_export_fails_when_staging_disk_is_short(self):
+    def test_open_fetch_without_staged_dir_is_none(self):
         with tempfile.TemporaryDirectory() as d:
             db = os.path.join(d, "db.sqlite3")
-            _make_db(db, cid=VALID_ID)  # channel available bytes = 100 in the fixture
+            _make_db(db, cid=VALID_ID)
             adapter = _StubbedKolibri(db, staging_dir=os.path.join(d, "staging"))
-            # Force the preflight to see almost no free space.
-            adapter._stage_export = lambda *a: (_ for _ in ()).throw(AssertionError("must not stage"))
-            import cx_core.adapters.kolibri as k
-            orig = k.shutil.disk_usage
-            k.shutil.disk_usage = lambda p: type("U", (), {"total": 0, "used": 0, "free": 1})()
-            try:
-                self.assertIsNone(adapter.open_export(VALID_ID))
-            finally:
-                k.shutil.disk_usage = orig
+            self.assertIsNone(adapter.open_fetch(VALID_ID, None))
+            self.assertIsNone(adapter.open_fetch(VALID_ID, os.path.join(d, "missing")))
+
+
+class RunCancellableTest(unittest.TestCase):
+    def test_completed_command_returns_true(self):
+        self.assertTrue(_run_cancellable([sys.executable, "-c", "pass"], dict(os.environ)))
+
+    def test_failed_command_returns_false(self):
+        self.assertFalse(
+            _run_cancellable([sys.executable, "-c", "import sys; sys.exit(3)"], dict(os.environ)))
+
+    def test_cancel_terminates_a_long_command(self):
+        start = time.time()
+        # A 30 s child that is cancelled at once: it must be terminated promptly, not waited out.
+        ok = _run_cancellable(
+            [sys.executable, "-c", "import time; time.sleep(30)"],
+            dict(os.environ),
+            should_cancel=lambda: True,
+        )
+        self.assertFalse(ok)
+        self.assertLess(time.time() - start, 10)
 
 
 class DeleteCommandTest(unittest.TestCase):
diff --git a/static/cx-core/tests/test_router.py b/static/cx-core/tests/test_router.py
index e67c21fa0..eb727c66a 100644
--- a/static/cx-core/tests/test_router.py
+++ b/static/cx-core/tests/test_router.py
@@ -2,6 +2,7 @@
 import tempfile
 import unittest
 
+from cx_core import VERSION
 from cx_core.model import AUTHORITATIVE, Dedup, Identity, ManifestItem
 from cx_core.ports import ContentAdapter, ExportStream
 from cx_core.registry import Registry
@@ -38,9 +39,6 @@ def is_installed(self):
     def list(self):
         return self._items
 
-    def open_export(self, local_ref):
-        return self._export
-
     def check(self, dedup):
         return self._verdict
 
@@ -67,7 +65,12 @@ def setUp(self):
     def test_apps(self):
         status, body = dispatch(self.reg, "GET", "/v1/apps", {})
         self.assertEqual(status, 200)
-        self.assertEqual(body, {"apps": ["calibre", "kiwix"]})
+        self.assertEqual(body, {"apps": ["calibre", "kiwix"], "version": VERSION})
+
+    def test_version(self):
+        status, body = dispatch(self.reg, "GET", "/v1/version", {})
+        self.assertEqual(status, 200)
+        self.assertEqual(body, {"version": VERSION})
 
     def test_content_all(self):
         status, body = dispatch(self.reg, "GET", "/v1/content", {})
@@ -131,20 +134,31 @@ def test_export_plan_bad_body(self):
         status, _ = dispatch(self.reg, "POST", "/v1/export/plan", {}, {"nope": 1})
         self.assertEqual(status, 400)
 
-    def test_export_item_streams(self):
-        stream = ExportStream("a.zim", "application/octet-stream", [b"abc"])
-        reg = Registry([FakeAdapter("kiwix", [_item("kiwix", "a.zim")], export=stream)])
-        status, payload = dispatch(reg, "GET", "/v1/export/item/kiwix/a.zim", {})
-        self.assertEqual(status, 200)
-        self.assertIs(payload, stream)
 
-    def test_export_item_not_exportable_404(self):
-        status, _ = dispatch(self.reg, "GET", "/v1/export/item/kiwix/a.zim", {})
-        self.assertEqual(status, 404)  # FakeAdapter has no export stream
+class ExportOpRouteTest(unittest.TestCase):
+    """The export long-op routes forward to the ops registry; the op mechanics live in
+    test_export_ops. Here we check the routing and the missing-ops guard."""
 
-    def test_export_item_unknown_app_404(self):
-        status, _ = dispatch(self.reg, "GET", "/v1/export/item/nope/x", {})
-        self.assertEqual(status, 404)
+    def setUp(self):
+        self.reg = _registry()
+
+    def test_prepare_bad_body_400(self):
+        from cx_core.export_ops import ExportOps
+        ops = ExportOps(self.reg, staging_dir="/tmp", ttl_seconds=60)
+        status, _ = dispatch(self.reg, "POST", "/v1/export/prepare", {}, {"nope": 1}, ops=ops)
+        self.assertEqual(status, 400)
+
+    def test_prepare_without_ops_500(self):
+        status, _ = dispatch(self.reg, "POST", "/v1/export/prepare", {},
+                             {"app": "kiwix", "localRef": "a.zim"}, ops=None)
+        self.assertEqual(status, 500)
+
+    def test_status_fetch_delete_without_ops_500(self):
+        for method, path in [("GET", "/v1/export/status/abc"),
+                             ("GET", "/v1/export/fetch/abc"),
+                             ("DELETE", "/v1/export/abc")]:
+            status, _ = dispatch(self.reg, method, path, {}, ops=None)
+            self.assertEqual(status, 500)
 
 
 class ImportRouteTest(unittest.TestCase):
diff --git a/static/cx-core/tests/test_server.py b/static/cx-core/tests/test_server.py
index fb2f32ba0..fc9c8ecc8 100644
--- a/static/cx-core/tests/test_server.py
+++ b/static/cx-core/tests/test_server.py
@@ -7,7 +7,9 @@
 import uuid
 from http.server import ThreadingHTTPServer
 
+from cx_core import VERSION
 from cx_core.adapters.kiwix import KiwixAdapter
+from cx_core.export_ops import ExportOps
 from cx_core.registry import Registry
 from cx_core.server import make_handler
 
@@ -30,7 +32,8 @@ def setUp(self):
         self.zim_name = "wikipedia_en_all_2024-01.zim"
         self.zim_bytes = _zim(os.path.join(self._dir.name, self.zim_name), body_len=200000)
         reg = Registry([KiwixAdapter(self._dir.name)])
-        self.httpd = ThreadingHTTPServer(("127.0.0.1", 0), make_handler(reg))
+        ops = ExportOps(reg, staging_dir=self._dir.name, ttl_seconds=60)
+        self.httpd = ThreadingHTTPServer(("127.0.0.1", 0), make_handler(reg, ops))
         self.port = self.httpd.server_address[1]
         self._t = threading.Thread(target=self.httpd.serve_forever, daemon=True)
         self._t.start()
@@ -43,15 +46,47 @@ def tearDown(self):
     def _conn(self):
         return http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
 
-    def test_export_item_stream_matches_source(self):
+    def test_export_longop_stream_matches_source(self):
+        # Prepare -> status (ready, direct app) -> fetch over a real socket, so the chunked
+        # stream is decoded by a real client (K2GO-507 flow, Kiwix = direct).
         c = self._conn()
-        c.request("GET", f"/v1/export/item/kiwix/{self.zim_name}")
+        body = json.dumps({"app": "kiwix", "localRef": self.zim_name})
+        c.request("POST", "/v1/export/prepare", body=body,
+                  headers={"Content-Type": "application/json"})
+        prep = json.loads(c.getresponse().read())
+        c.close()
+        self.assertEqual(prep["state"], "ready")
+        self.assertEqual(prep["totalBytes"], len(self.zim_bytes))
+        op_id = prep["opId"]
+
+        c = self._conn()
+        c.request("GET", f"/v1/export/status/{op_id}")
+        st = json.loads(c.getresponse().read())
+        c.close()
+        self.assertEqual(st["state"], "ready")
+        self.assertEqual(st["stagedBytes"], len(self.zim_bytes))
+
+        c = self._conn()
+        c.request("GET", f"/v1/export/fetch/{op_id}")
         resp = c.getresponse()
         self.assertEqual(resp.status, 200)
         got = resp.read()
         c.close()
         self.assertEqual(got, self.zim_bytes)
 
+        # The op is dropped after fetch (on_close), so a second status is 404.
+        c = self._conn()
+        c.request("GET", f"/v1/export/status/{op_id}")
+        self.assertEqual(c.getresponse().status, 404)
+        c.close()
+
+    def test_version_over_http(self):
+        c = self._conn()
+        c.request("GET", "/v1/version")
+        out = json.loads(c.getresponse().read())
+        c.close()
+        self.assertEqual(out, {"version": VERSION})
+
     def test_export_plan_over_http(self):
         c = self._conn()
         payload = json.dumps({"items": [{"app": "kiwix", "localRef": self.zim_name}]})
@@ -68,7 +103,7 @@ def test_apps_json_over_http(self):
         resp = c.getresponse()
         out = json.loads(resp.read())
         c.close()
-        self.assertEqual(out, {"apps": ["kiwix"]})
+        self.assertEqual(out, {"apps": ["kiwix"], "version": VERSION})
 
     def test_delete_item_over_http(self):
         c = self._conn()
diff --git a/tools/upstream-patches/overlays/roles/cx_core/tasks/main.yml b/tools/upstream-patches/overlays/roles/cx_core/tasks/main.yml
index 33b3e9d89..43e07aa79 100644
--- a/tools/upstream-patches/overlays/roles/cx_core/tasks/main.yml
+++ b/tools/upstream-patches/overlays/roles/cx_core/tasks/main.yml
@@ -20,6 +20,18 @@
     - name: Enable/Disable the content-exchange core service
       include_tasks: enable-or-disable.yml
 
+    # Read the core build from the installed source (the single source: cx_core/__init__.py
+    # VERSION), so the box can report which core it runs. Mirrors dash-node's package.json
+    # version in /system/version (K2GO-507).
+    - name: Read the installed cx_core source version
+      slurp:
+        src: "{{ cx_core_serve_path }}/cx_core/__init__.py"
+      register: cx_core_init_py
+
+    - name: Set cx_core_version from the source
+      set_fact:
+        cx_core_version: "{{ cx_core_init_py.content | b64decode | regex_search('VERSION\\s*=\\s*\"([^\"]+)\"', '\\1') | first }}"
+
     - name: Add 'cx_core' variable values to {{ iiab_ini_file }}
       ini_file:
         path: "{{ iiab_ini_file }}"    # /etc/iiab/iiab.ini
@@ -37,6 +49,8 @@
           value: "{{ cx_core_enabled }}"
         - option: cx_core_port
           value: "{{ cx_core_port }}"
+        - option: version
+          value: "{{ cx_core_version }}"
 
   rescue: