chore(sync): mirror backend .py + Dockerfile to Space (hf-sync-backend)

Automated backend sync from szl-holdings/a11oy main via hf-sync-backend.
Updated (differed from the Space): a11oy_live_feeds.py, serve.py
Deleted (gone from the repo + Dockerfile COPY set): (none)

Keeps the Space-built backend (serve.py + the Dockerfile-COPY'd .py
modules) identical to GitHub main so the Space never rebuilds from a
stale backend, new endpoints don't 404 there, and orphaned modules
removed from the repo don't linger in the Space tree.

a11oy_live_feeds.py

@@ -7,12 +7,13 @@ same-origin proxy, keeping the Space sovereign — 0 runtime CDN from the client
 
 Every response carries an HONEST label:
     {"source": <human source>, "source_url": <upstream URL>,
-     "mode": "live" | "cached" | "self",   # never fabricated
+     "mode": "live" | "cached" | "unavailable" | "self",   # never fabricated
      "fetched_at": <iso8601>, "ttl_s": <int>, ...payload}
 
   - "live"   = freshly fetched from upstream this request (or within TTL).
   - "cached" = upstream was unreachable; serving the last good in-memory value
                or the bundled on-disk snapshot (stage resilience).
+  - "unavailable" = upstream was unreachable and no cached or bundled data exists.
   - "self"   = our own internal real data (not third-party) — used by callers
                that pass through this layer's helpers; the feed endpoints here
                are all third-party live/cached.
@@ -27,22 +28,22 @@ Feeds + TTLs:
   fhir        (hapi.fhir.org/baseR4 Observation/Immunization)     TTL 10m
 
 No auth required for any of these feeds. NEVER fabricates: a down feed returns
-the cached snapshot labelled "cached".
+real cached data labelled "cached", or "unavailable" when no cached data exists.
 """
 import json
 import os
 import time
 import threading
-import urllib.request
-import urllib.error
 from datetime import datetime, timezone
 from pathlib import Path
 
+import httpx
 from starlette.routing import Route
 from starlette.responses import JSONResponse
 
 _SNAP_DIR = Path(os.environ.get("A11OY_LIVE_SNAPSHOTS", "/app/live_snapshots"))
 _UA = "a11oy-live-proxy/1.0 (+https://szlholdings-a11oy.hf.space)"
+_MAX_RESPONSE_BYTES = 32 * 1024 * 1024
 
 # in-memory cache: feed -> {"data":..., "ts":..., "mode":...}
 _CACHE = {}
@@ -75,13 +76,59 @@ def _now_iso():
     return datetime.now(timezone.utc).isoformat()
 
 
-def _http_get(url, timeout=20, headers=None, data=None, method=None):
+def _http_get(url, timeout=20, headers=None, data=None, method=None, deadline=None):
+    """Fetch one bounded JSON response.
+
+    Socket timeouts alone are insufficient because a peer can trickle bytes and
+    reset the per-read timer forever.  ``iter_bytes`` exposes each received
+    network chunk so the absolute monotonic deadline is checked throughout the
+    body read.  The response context is closed before a deadline error escapes,
+    which lets the calling worker finish instead of being abandoned.
+    """
     h = {"User-Agent": _UA, "Accept": "application/json"}
     if headers:
         h.update(headers)
-    req = urllib.request.Request(url, data=data, headers=h, method=method)
-    with urllib.request.urlopen(req, timeout=timeout) as r:
-        return json.loads(r.read())
+    request_timeout = _remaining_timeout(deadline, timeout)
+    request_deadline = (
+        deadline
+        if deadline is not None
+        else time.monotonic() + request_timeout
+    )
+    request_method = method or ("POST" if data is not None else "GET")
+    chunks = []
+    received = 0
+    try:
+        with httpx.stream(
+            request_method,
+            url,
+            content=data,
+            headers=h,
+            timeout=httpx.Timeout(request_timeout),
+            follow_redirects=True,
+        ) as response:
+            response.raise_for_status()
+            content_length = response.headers.get("content-length")
+            if content_length is not None:
+                try:
+                    declared_bytes = int(content_length)
+                except (TypeError, ValueError):
+                    declared_bytes = None
+                if declared_bytes is not None and declared_bytes > _MAX_RESPONSE_BYTES:
+                    raise ValueError("live-feed response exceeds size limit")
+            for chunk in response.iter_bytes():
+                if time.monotonic() >= request_deadline:
+                    raise TimeoutError("live-feed response deadline exhausted")
+                if not chunk:
+                    continue
+                received += len(chunk)
+                if received > _MAX_RESPONSE_BYTES:
+                    raise ValueError("live-feed response exceeds size limit")
+                chunks.append(chunk)
+            if time.monotonic() >= request_deadline:
+                raise TimeoutError("live-feed response deadline exhausted")
+    except httpx.TimeoutException as exc:
+        raise TimeoutError("live-feed response deadline exhausted") from exc
+    return json.loads(b"".join(chunks))
 
 
 def _load_snapshot(feed):
@@ -92,7 +139,43 @@ def _load_snapshot(feed):
         return None
 
 
-def _fetch(feed):
+def get_cached_feed(feed, error):
+    """Return only real cached evidence, without attempting an upstream read."""
+    ttl = _TTL.get(feed, 60)
+    src, url = _SOURCE.get(feed, ("unknown", ""))
+    with _LOCK:
+        ent = _CACHE.get(feed)
+    if ent:
+        return {"source": src, "source_url": url, "mode": "cached",
+                "fetched_at": ent["iso"], "ttl_s": ttl,
+                "cache_note": "upstream unreachable (%s) — serving last good value"
+                              % type(error).__name__,
+                "data": ent["data"]}
+    snap = _load_snapshot(feed)
+    if snap is not None:
+        return {"source": src, "source_url": url, "mode": "cached",
+                "fetched_at": "bundled-snapshot", "ttl_s": ttl,
+                "cache_note": "upstream unreachable (%s) — serving bundled in-image snapshot"
+                              % type(error).__name__,
+                "data": snap}
+    return {"source": src, "source_url": url, "mode": "unavailable",
+            "fetched_at": None, "ttl_s": ttl,
+            "error": "upstream unreachable and no snapshot (%s): %s"
+                     % (type(error).__name__, error),
+            "data": None}
+
+
+def _remaining_timeout(deadline, default):
+    """Return a cooperative per-request socket timeout within ``deadline``."""
+    if deadline is None:
+        return float(default)
+    remaining = deadline - time.monotonic()
+    if remaining <= 0:
+        raise TimeoutError("live-feed network budget exhausted")
+    return min(float(default), remaining)
+
+
+def _fetch(feed, deadline=None):
     """Return raw upstream JSON for a feed (raises on failure)."""
     if feed == "prometheus":
         import urllib.parse
@@ -102,17 +185,25 @@ def _fetch(feed):
                      ("cpu", 'rate(node_cpu_seconds_total{mode="user"}[5m])'),
                      ("mem", "node_memory_MemAvailable_bytes"),
                      ("http_req", "rate(prometheus_http_requests_total[5m])")):
-            out[k] = _http_get(base + urllib.parse.quote(q), timeout=12)
+            out[k] = _http_get(
+                base + urllib.parse.quote(q),
+                timeout=_remaining_timeout(deadline, 12),
+                deadline=deadline,
+            )
         return out
     if feed == "kev":
-        return _http_get(_SOURCE["kev"][1], timeout=40)
+        return _http_get(
+            _SOURCE["kev"][1], timeout=_remaining_timeout(deadline, 40),
+            deadline=deadline)
     if feed == "osv":
         out = {}
         for pkg, eco in (("tensorflow", "PyPI"), ("torch", "PyPI"),
                          ("transformers", "PyPI"), ("numpy", "PyPI"), ("requests", "PyPI")):
             body = json.dumps({"package": {"name": pkg, "ecosystem": eco}}).encode()
-            r = _http_get("https://api.osv.dev/v1/query", timeout=20, data=body,
-                          headers={"Content-Type": "application/json"}, method="POST")
+            r = _http_get("https://api.osv.dev/v1/query",
+                          timeout=_remaining_timeout(deadline, 20), data=body,
+                          headers={"Content-Type": "application/json"}, method="POST",
+                          deadline=deadline)
             vulns = r.get("vulns", [])
             out[pkg] = {"ecosystem": eco, "count": len(vulns),
                         "vulns": [{"id": v.get("id"), "summary": v.get("summary"),
@@ -120,16 +211,24 @@ def _fetch(feed):
                                    "aliases": (v.get("aliases") or [])[:4]} for v in vulns[:25]]}
         return out
     if feed == "rekor":
-        return {"log": _http_get(_SOURCE["rekor"][1], timeout=15)}
+        return {"log": _http_get(
+            _SOURCE["rekor"][1], timeout=_remaining_timeout(deadline, 15),
+            deadline=deadline)}
     if feed == "celestrak":
-        return _http_get(_SOURCE["celestrak"][1], timeout=20)
+        return _http_get(
+            _SOURCE["celestrak"][1], timeout=_remaining_timeout(deadline, 20),
+            deadline=deadline)
     if feed == "iss":
-        return _http_get(_SOURCE["iss"][1], timeout=12)
+        return _http_get(
+            _SOURCE["iss"][1], timeout=_remaining_timeout(deadline, 12),
+            deadline=deadline)
     if feed == "fhir":
         out = {}
         for rt in ("Immunization", "Observation"):
-            b = _http_get("https://hapi.fhir.org/baseR4/%s?_count=10" % rt, timeout=25,
-                          headers={"Accept": "application/fhir+json"})
+            b = _http_get("https://hapi.fhir.org/baseR4/%s?_count=10" % rt,
+                          timeout=_remaining_timeout(deadline, 25),
+                          headers={"Accept": "application/fhir+json"},
+                          deadline=deadline)
             entries = b.get("entry", [])
             out[rt] = {"total": b.get("total"), "count": len(entries),
                        "entries": [e.get("resource", {}) for e in entries[:10]]}
@@ -137,8 +236,8 @@ def _fetch(feed):
     raise ValueError("unknown feed: %s" % feed)
 
 
-def get_feed(feed):
-    """Cached, snapshot-fallback, honestly-labelled feed accessor."""
+def get_feed(feed, timeout_s=None):
+    """Cached, honestly-labelled accessor with an optional cooperative network budget."""
     ttl = _TTL.get(feed, 60)
     src, url = _SOURCE.get(feed, ("unknown", ""))
     with _LOCK:
@@ -149,30 +248,18 @@ def get_feed(feed):
                 "fetched_at": ent["iso"], "ttl_s": ttl, "data": ent["data"]}
     # need refresh
     try:
-        data = _fetch(feed)
+        deadline = (
+            time.monotonic() + max(0.001, float(timeout_s))
+            if timeout_s is not None else None
+        )
+        data = _fetch(feed, deadline=deadline)
         iso = _now_iso()
         with _LOCK:
             _CACHE[feed] = {"data": data, "ts": now, "mode": "live", "iso": iso}
         return {"source": src, "source_url": url, "mode": "live",
                 "fetched_at": iso, "ttl_s": ttl, "data": data}
     except Exception as e:
-        # serve last good in-memory value if present
-        if ent:
-            return {"source": src, "source_url": url, "mode": "cached",
-                    "fetched_at": ent["iso"], "ttl_s": ttl,
-                    "cache_note": "upstream unreachable (%s) — serving last good value" % type(e).__name__,
-                    "data": ent["data"]}
-        # else bundled on-disk snapshot
-        snap = _load_snapshot(feed)
-        if snap is not None:
-            return {"source": src, "source_url": url, "mode": "cached",
-                    "fetched_at": "bundled-snapshot",
-                    "ttl_s": ttl,
-                    "cache_note": "upstream unreachable (%s) — serving bundled in-image snapshot" % type(e).__name__,
-                    "data": snap}
-        return {"source": src, "source_url": url, "mode": "cached",
-                "fetched_at": None, "ttl_s": ttl,
-                "error": "upstream unreachable and no snapshot: %s" % e, "data": None}
+        return get_cached_feed(feed, e)
 
 
 def register(app, ns="a11oy"):
@@ -201,8 +288,9 @@ def register(app, ns="a11oy"):
         return JSONResponse({
             "layer": "a11oy live-data proxy",
             "honest": "Every feed is server-side fetched + cached, CORS-safe via OUR same-origin "
-                      "proxy (0 client CDN). Mode is honestly labelled live/cached; a down feed "
-                      "serves the bundled in-image snapshot labelled 'cached', never fabricated.",
+                      "proxy (0 client CDN). Mode is honestly labelled live/cached/unavailable; "
+                      "a down feed serves real cached data when present and reports unavailable "
+                      "when absent, never fabricated.",
             "count": len(feeds), "feeds": feeds,
         })
 

serve.py

@@ -10708,6 +10708,13 @@ try:
     # real mode (live/cached), age, and round-trip latency. This is the data-
     # provenance heartbeat for the governed-AI mission: a governed system must
     # know whether its evidence feeds are actually live. 0 fabricated status.
+    try:
+        _KL_FEED_PULSE_TIMEOUT_S = max(
+            0.25, min(10.0, float(os.environ.get("A11OY_FEED_PULSE_TIMEOUT_SEC", "4.0")))
+        )
+    except (TypeError, ValueError):
+        _KL_FEED_PULSE_TIMEOUT_S = 4.0
+
     @app.get("/api/a11oy/v1/feeds/pulse")
     async def _feeds_pulse():
         import anyio, time as _t
@@ -10715,27 +10722,42 @@ try:
         async def _one(f):
             t0 = _t.time()
             try:
-                p = await anyio.to_thread.run_sync(_kl_live.get_feed, f)
-                dt = round((_t.time()-t0)*1000)
-                d = p.get("data")
-                # honest payload-size signal
-                try:
-                    import json as _j; size = len(_j.dumps(d)) if d is not None else 0
-                except Exception:
-                    size = 0
-                return {"feed": f, "source": p.get("source"),
-                        "source_url": p.get("source_url"),
-                        "mode": p.get("mode"), "fetched_at": p.get("fetched_at"),
-                        "ttl_s": p.get("ttl_s"), "latency_ms": dt,
-                        "payload_bytes": size,
-                        "error": p.get("error")}
+                # get_feed enforces the absolute deadline inside every streamed
+                # network read. Keep the worker joined: it must exit and return
+                # its honest cache fallback instead of being orphaned.
+                p = await anyio.to_thread.run_sync(
+                    _kl_live.get_feed,
+                    f,
+                    _KL_FEED_PULSE_TIMEOUT_S,
+                )
+            except TimeoutError as e:
+                p = _kl_live.get_cached_feed(f, e)
+                if p.get("mode") == "unavailable":
+                    p["error"] = "probe timeout after %.2fs" % _KL_FEED_PULSE_TIMEOUT_S
             except Exception as e:
-                return {"feed": f, "mode": "unavailable",
-                        "latency_ms": round((_t.time()-t0)*1000),
-                        "error": "%s: %s" % (type(e).__name__, e)}
-        items = []
-        for f in feeds:
-            items.append(await _one(f))
+                p = _kl_live.get_cached_feed(f, e)
+                if p.get("mode") == "unavailable":
+                    p["error"] = "%s: %s" % (type(e).__name__, e)
+            dt = round((_t.time()-t0)*1000)
+            d = p.get("data")
+            # honest payload-size signal
+            try:
+                import json as _j; size = len(_j.dumps(d)) if d is not None else 0
+            except Exception:
+                size = 0
+            return {"feed": f, "source": p.get("source"),
+                    "source_url": p.get("source_url"),
+                    "mode": p.get("mode"), "fetched_at": p.get("fetched_at"),
+                    "ttl_s": p.get("ttl_s"), "latency_ms": dt,
+                    "payload_bytes": size,
+                    "error": p.get("error"),
+                    "cache_note": p.get("cache_note")}
+        items = [None] * len(feeds)
+        async def _collect(index, feed):
+            items[index] = await _one(feed)
+        async with anyio.create_task_group() as task_group:
+            for index, feed in enumerate(feeds):
+                task_group.start_soon(_collect, index, feed)
         live = sum(1 for i in items if i.get("mode")=="live")
         return JSONResponse({
             "probed_at": _kl_live._now_iso(),
@@ -10744,8 +10766,10 @@ try:
             "cached_count": sum(1 for i in items if i.get("mode")=="cached"),
             "down_count": sum(1 for i in items if i.get("mode")=="unavailable"),
             "note": ("Real-time provenance heartbeat: each row is a live server-side "
-                     "probe of an upstream evidence feed. mode/latency are measured, "
-                     "never fabricated. A governed-AI system must know its feeds are live."),
+                     "bounded probe of an upstream evidence feed. mode/latency are measured, "
+                     "internally handled timeouts remain visible through cache_note when real "
+                     "cached data exists and otherwise count as unavailable. A governed-AI "
+                     "system must know its feeds are live."),
             "items": items,
         })