K2GO-376 feat(cx-core): /v1/share gate (two-lifetime, token-scoped)

static/cx-core/CHANGELOG.md

@@ -0,0 +1,10 @@
+# cx-core (content-exchange core) - changelog
+
+One line per version, newest first. Every REST-facing change bumps `cx_core.VERSION` (reported by
+`GET /v1/version`, and written into the `[cx_core]` section of iiab.ini by the cx_core Ansible
+role), so this file is the human record of what each bump enables. Keep entries short:
+`version - change (TICKET)`.
+
+- **0.3.0** - /v1/share gate: a token-scoped, two-lifetime share session (ephemeral connection vs durable session) over a scoped item set, reusing the export op per item, for granular wireless Share. (K2GO-376)
+- **0.2.0** - export as a prepare/poll/fetch long operation, so a large Kolibri channel stages without timing out the client. (K2GO-507)
+- **0.1.0** - initial content-exchange core: read (list + identity), export, import and delete over /v1 across Kiwix, Kolibri and Calibre. (K2GO-427)

static/cx-core/README.md

@@ -8,14 +8,18 @@ contracts (identity, manifest, adapter port, core REST API).
 
 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).
+not time out on the client (K2GO-507). Share (K2GO-376) is the network twin: a token-gated
+session over a scoped item set that reuses the export op per item (the gate contract only;
+pairing and the network carrier are later app slices).
 
 ## What is here
 
 - `cx_core/model.py`: app-agnostic content model (identity, dedup, manifest item).
 - `cx_core/ports.py`: the inner contract, the app-adapter port (read methods).
 - `cx_core/adapters/kiwix.py`: the first adapter, read side (ZIM header UUID + listing).
 - `cx_core/registry.py`: loads an adapter only when its app is installed.
+- `cx_core/export_ops.py`: the export long-operation registry (op lifecycle + temp sweeps).
+- `cx_core/share_ops.py`: the share gate (token + scoped item set + two lifetimes), K2GO-376.
 - `cx_core/router.py`: pure `/v1` route dispatch (unit-testable without a socket).
 - `cx_core/server.py` + `__main__.py`: a thin stdlib `http.server` wrapper, localhost only.
 
@@ -68,6 +72,38 @@ Import / delete:
 - `POST /v1/export/plan`, `POST /v1/dedup/check`, `POST /v1/import/validate`,
   `POST /v1/import/item`, `DELETE /v1/content/{app}/{localRef}`.
 
+Share (a gate over the export op, K2GO-376): Share is the network twin of Export. The
+receiver pulls selected items from the sender's live cx-core, so the transfer can close
+itself (the sender machine is on, unlike Clone). The gate has two lifetimes: the CONNECTION
+(one open stream) reuses the export op's idle TTL + in_use pin per item, and the durable
+SESSION (a token + a scoped item set + per-item delivered state) does not expire on the short
+connection idle. The session closes on COMPLETE (the receiver has every item), CANCEL
+(DELETE, by host or receiver; the sender's foreground service calls it in onDestroy), a long
+SAFETY CAP (hours of no activity), or process death (in-memory + the export startup sweep).
+Each item is a lazily created export fetch, so staging, the in_use pin and the on_close
+cleanup are the export op's. An item counts as delivered only on a CLEAN close (the server
+marks a full body), so a mid-stream drop stays pending and a resume re-fetches it. This slice
+is loopback only: pairing (QR, hotspot, WLAN) is a later app slice.
+
+The session token travels in the `X-Share-Token` request header, never the URL (it is the
+only auth, so it must not reach access logs).
+
+- `POST /v1/share {items: [{app, localRef}]}` -> `{sessionId, token, manifestUrl}`. Opens a
+  session scoped to the item set.
+- `GET /v1/share/{sessionId}/manifest` -> the bundle manifest over the scope (per-item
+  identity, size, upstream hash when present, each with its `itemRef` and whether it is already
+  `delivered`), plus `missing` and `totalBytes`.
+- `GET /v1/share/{sessionId}/fetch/{itemRef}` -> the chunked octet stream for one item.
+  `409 {state: staging}` while a staging item (Kolibri) stages (the receiver retries),
+  `409 {state: busy}` when the item already has an open stream (single-flight), `200` streams
+  once ready.
+- `POST /v1/share/{sessionId}/complete` -> `{ok: true}`. The receiver's clean-success close.
+- `DELETE /v1/share/{sessionId}` -> `{ok: true}`. Cancel (abort) by host or receiver. Both
+  closers revoke the token and every child temp; `404` on an unknown or already-closed session.
+
+Token checks return `401` when no token is sent, `403` on a mismatch, `404` on an unknown or
+expired session.
+
 ## Test
 
 ```
@@ -77,14 +113,16 @@ 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, route dispatch, the export
-long-operation registry (prepare / status / fetch / delete, the TTL and startup sweeps),
-and the streaming over a real socket.
+long-operation registry (prepare / status / fetch / delete, the TTL and startup sweeps), the
+share gate (create / manifest / fetch / complete / delete, header-token enforcement,
+single-flight, clean-close delivery, the safety cap), and the streaming over a real socket.
 
 ## Not built yet
 
-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 Share pairing and network carrier (QR, hotspot, WLAN: the app slices that build on the
+`/v1/share` gate above), 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

static/cx-core/cx_core/__init__.py

@@ -7,5 +7,5 @@
 # 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 = "0.3.0"
 __version__ = VERSION

static/cx-core/cx_core/ports.py

@@ -24,13 +24,19 @@ class ExportStream:
     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)."""
+    The adapter only opens the stream; the op layer owns the lifecycle (K2GO-507).
+
+    ``completed`` is set True by the server after the FULL body has been written (the clean
+    terminator), and stays False on a mid-stream client drop. The export op ignores it (it drops
+    the temp either way), but the share gate reads it so a share item counts as delivered only on
+    a clean close, never on a drop (K2GO-376)."""
 
     filename: str
     content_type: str
     chunks: Iterable[bytes]
     on_close: Optional[Callable[[], None]] = None
     on_open: Optional[Callable[[], None]] = None
+    completed: bool = False
 
 
 @dataclass

static/cx-core/cx_core/router.py

@@ -15,6 +15,7 @@
 from .export_ops import ExportOps
 from .importing import is_within_dir
 from .registry import Registry
+from .share_ops import ShareOps
 
 Response = Tuple[int, Any]
 
@@ -165,10 +166,14 @@ def _export_prepare(ops: Optional[ExportOps], body: Any) -> Response:
 
 
 def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any = None,
-             ops: Optional[ExportOps] = None) -> Response:
+             ops: Optional[ExportOps] = None, share: Optional[ShareOps] = None,
+             token: Optional[str] = None) -> Response:
     """Route one request. ``query`` is a dict of first values; ``body`` is the parsed JSON
-    for a POST; ``ops`` is the export long-operation registry (the server supplies it, the
-    export-op routes need it)."""
+    for a POST; ``ops`` is the export long-operation registry and ``share`` is the share-gate
+    registry (the server supplies both; the routes that need them guard against None). ``token``
+    is the share session token: the server reads it from the ``X-Share-Token`` request header
+    (never the URL, so it stays out of access logs) and passes it here, so the router stays
+    header-agnostic and unit-testable without a socket."""
     trimmed = path.rstrip("/") or "/"
     segments = [unquote(s) for s in trimmed.split("/") if s != ""]
 
@@ -179,6 +184,17 @@ def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any
         # POST /v1/export/plan : plan an export (sizes only, no payload moved).
         if segments == ["v1", "export", "plan"]:
             return _export_plan(registry, body)
+        # POST /v1/share : open a share session, return its sessionId + token + manifest pointer.
+        if segments == ["v1", "share"]:
+            if share is None:
+                return 500, {"error": "share unavailable"}
+            items = body.get("items") if isinstance(body, dict) else None
+            return share.create(items)
+        # POST /v1/share/{sessionId}/complete : the receiver signals a clean, done transfer.
+        if len(segments) == 4 and segments[0] == "v1" and segments[1] == "share" and segments[3] == "complete":
+            if share is None:
+                return 500, {"error": "share unavailable"}
+            return share.complete(segments[2], token)
         # POST /v1/dedup/check : receiver dedup verdicts, nothing moved.
         if segments == ["v1", "dedup", "check"]:
             return _dedup_check(registry, body)
@@ -196,6 +212,11 @@ def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any
             if ops is None:
                 return 500, {"error": "export ops unavailable"}
             return ops.delete(segments[2])
+        # DELETE /v1/share/{sessionId} : cancel a share session (abort, by host or receiver).
+        if len(segments) == 3 and segments[0] == "v1" and segments[1] == "share":
+            if share is None:
+                return 500, {"error": "share unavailable"}
+            return share.delete(segments[2], token)
         # 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])
@@ -225,6 +246,19 @@ def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any
             return 500, {"error": "export ops unavailable"}
         return ops.fetch(segments[3])
 
+    # GET /v1/share/{sessionId}/manifest : the bundle manifest over the session's item set.
+    if len(segments) == 4 and segments[0] == "v1" and segments[1] == "share" and segments[3] == "manifest":
+        if share is None:
+            return 500, {"error": "share unavailable"}
+        return share.manifest(segments[2], token)
+
+    # GET /v1/share/{sessionId}/fetch/{itemRef} : stream one shared item (payload is an
+    # ExportStream the server streams). 409 while the item stages; the receiver retries.
+    if len(segments) == 5 and segments[0] == "v1" and segments[1] == "share" and segments[3] == "fetch":
+        if share is None:
+            return 500, {"error": "share unavailable"}
+        return share.fetch(segments[2], segments[4], token)
+
     # GET /v1/storage : box content-volume disk summary + per-app content totals.
     if segments == ["v1", "storage"]:
         return _storage(registry)

static/cx-core/cx_core/server.py

@@ -15,23 +15,30 @@
 from .ports import ExportStream
 from .registry import Registry
 from .router import dispatch
+from .share_ops import ShareOps
 
 _MAX_BODY = 1 * 1024 * 1024  # a plan request is small; cap the read defensively
 _BAD_JSON = object()  # sentinel: body present but not valid JSON
 _TOO_LARGE = object()  # sentinel: declared body exceeds _MAX_BODY
 
 
-def make_handler(registry: Registry, ops: ExportOps):
+def make_handler(registry: Registry, ops: ExportOps, share: ShareOps = None):
     class Handler(BaseHTTPRequestHandler):
         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"
 
+        def _share_token(self):
+            # The share session token travels in a header, never the URL query, so it does not
+            # land in access logs (it is the only auth for the gate).
+            return self.headers.get("X-Share-Token")
+
         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, ops=ops)
+            status, payload = dispatch(registry, "GET", parts.path, query, ops=ops, share=share,
+                                       token=self._share_token())
             if isinstance(payload, ExportStream):
                 self._stream(status, payload)
             else:
@@ -48,13 +55,15 @@ 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, ops=ops)
+            status, payload = dispatch(registry, "POST", parts.path, {}, body, ops=ops, share=share,
+                                       token=self._share_token())
             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, ops=ops)
+            status, payload = dispatch(registry, "DELETE", parts.path, query, ops=ops, share=share,
+                                       token=self._share_token())
             self._send(status, payload)
 
         def _read_json_body(self):
@@ -86,15 +95,18 @@ def _stream(self, status: int, stream: ExportStream) -> None:
             # Close after the stream: the body length is unknown up front, so a chunked
             # response plus an explicit close keeps the connection state simple.
             self.close_connection = True
-            self.send_response(status)
-            self.send_header("Content-Type", stream.content_type)
-            self.send_header("Content-Disposition", f'attachment; filename="{stream.filename}"')
-            self.send_header("Transfer-Encoding", "chunked")
-            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).
+                # Write the headers INSIDE the same try whose finally runs on_close, so on_close
+                # is guaranteed on every exit: a client that disconnects during the header write
+                # must still release the op pin and the share single-flight slot (K2GO-376),
+                # never strand them. on_open follows the headers, so a pre-header failure runs
+                # on_close without a matching on_open: the op cleanup (_finish) is safe either way.
+                self.send_response(status)
+                self.send_header("Content-Type", stream.content_type)
+                self.send_header("Content-Disposition", f'attachment; filename="{stream.filename}"')
+                self.send_header("Transfer-Encoding", "chunked")
+                self.send_header("Connection", "close")
+                self.end_headers()
                 if stream.on_open:
                     stream.on_open()
                 for block in stream.chunks:
@@ -104,6 +116,10 @@ def _stream(self, status: int, stream: ExportStream) -> None:
                     self.wfile.write(block)
                     self.wfile.write(b"\r\n")
                 self.wfile.write(b"0\r\n\r\n")
+                # The full body (and its terminator) is written: a clean close. A mid-stream
+                # client drop raises above and leaves this False, so the share gate counts the
+                # item delivered only on a clean close (K2GO-376).
+                stream.completed = True
             finally:
                 if stream.on_close:
                     stream.on_close()
@@ -118,10 +134,14 @@ def log_message(self, *_args) -> None:
 def serve(config: Config) -> None:
     registry = Registry.load(config)
     ops = ExportOps(registry, config.staging_dir)
+    share = ShareOps(registry, ops)
     # 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.
+    # cleanup owner): a fresh process holds no ops, so any op temp on disk is an orphan. The
+    # share sweep then drops any in-memory sessions (K2GO-376 closer 2); a share session's own
+    # on-disk footprint is its child export temps, already reclaimed by the export sweep above.
     ops.sweep_startup()
-    httpd = ThreadingHTTPServer((config.host, config.port), make_handler(registry, ops))
+    share.sweep_startup()
+    httpd = ThreadingHTTPServer((config.host, config.port), make_handler(registry, ops, share))
     print(f"cx-core {VERSION} listening on http://{config.host}:{config.port} "
           f"(apps: {', '.join(registry.apps()) or 'none'})")
     httpd.serve_forever()