Δ9Φ963-PHASE6-v1.0: Space sync — Twin Gate + Phase 5 mesh + P6 attest bundle

app.py

@@ -518,6 +518,76 @@ with gr.Blocks() as demo:
 
         p7_btn.click(_p7_entropy_ui, inputs=[p7_bpm, p7_sdnn, p7_noise, p7_fs], outputs=[p7_md])
 
+        with gr.Accordion("📡 Live BLE stream (Biophase7 Objective)", open=False):
+            gr.Markdown(
+                "**Tunnel required:** HF cloud cannot reach `localhost`. Run "
+                "`python tools/run_live_ble_pipeline.py` on your node, expose **:8788** via "
+                "Cloudflare/ngrok, set Space secret **`LYGO_BLE_WS_URL=wss://…`** or paste below. "
+                "`Δ9Φ963-PHASE7-LIVE-STREAM-v1`"
+            )
+            live_ws_url = gr.Textbox(
+                label="WebSocket URL (wss:// tunnel)",
+                placeholder="wss://your-tunnel.example",
+            )
+            with gr.Row():
+                live_connect = gr.Button("Connect live stream", variant="primary")
+                live_disconnect = gr.Button("Disconnect")
+            live_status = gr.Markdown("*Not connected*")
+            with gr.Row():
+                live_ibi = gr.Number(label="Latest IBI (ms)", value=0)
+                live_buf = gr.Number(label="Buffer size / 64", value=0)
+                live_hmin = gr.Number(label="H_min", value=0.0)
+            live_seed = gr.Textbox(label="P0 seed", value="Awaiting…")
+            live_plot = gr.LinePlot(x="time", y="ibi", title="IBI waveform (last 50)", height=280)
+            live_hist = gr.BarPlot(x="bin", y="count", title="IBI distribution", height=200)
+
+            def _live_import():
+                try:
+                    from live_ble_gradio import connect_ws, disconnect_ws, refresh_dashboard
+                except ImportError:
+                    from tools.live_ble_gradio import connect_ws, disconnect_ws, refresh_dashboard
+                return connect_ws, disconnect_ws, refresh_dashboard
+
+            def _live_connect(url):
+                connect_ws, _, _ = _live_import()
+                return connect_ws(url)
+
+            def _live_disconnect():
+                _, disconnect_ws, _ = _live_import()
+                return disconnect_ws()
+
+            def _live_refresh():
+                _, _, refresh_dashboard = _live_import()
+                ibi, buf, hmin, seed, df, wave, status = refresh_dashboard()
+                import pandas as pd
+
+                if wave and max(wave) > 0:
+                    bins = min(12, max(4, len(set(wave)) // 2))
+                    hist_df = pd.DataFrame({"ibi": wave})
+                    hist_df["bin"] = pd.cut(hist_df["ibi"], bins=bins).astype(str)
+                    hist_out = hist_df.groupby("bin", observed=True).size().reset_index(name="count")
+                else:
+                    hist_out = pd.DataFrame({"bin": ["—"], "count": [0]})
+                return ibi, buf, hmin, seed, df, hist_out, status
+
+            live_connect.click(_live_connect, inputs=[live_ws_url], outputs=[live_status])
+            live_disconnect.click(_live_disconnect, outputs=[live_status])
+            def _live_boot():
+                import os
+
+                url = os.environ.get("LYGO_BLE_WS_URL", "").strip()
+                if url:
+                    return _live_connect(url)
+                return "*Set `LYGO_BLE_WS_URL` or paste tunnel URL and click Connect.*"
+
+            demo.load(
+                _live_refresh,
+                inputs=None,
+                outputs=[live_ibi, live_buf, live_hmin, live_seed, live_plot, live_hist, live_status],
+                every=0.5,
+            )
+            demo.load(_live_boot, inputs=None, outputs=[live_status])
+
     # ---- Twin Gate Phase 3 — isolated from Standard Beat factory ----
     with gr.Accordion("🛡️ LYGO Twin Gate — Ethical Guardian (Phase 3)", open=False):
         gr.Markdown(

live_ble_gradio.py

@@ -0,0 +1,171 @@
+"""
+HF Space / Gradio live BLE consumer — connects to operator tunnel (not localhost on cloud).
+Set LYGO_BLE_WS_URL=wss://your-tunnel.example/ws or pass URL in UI.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import threading
+import time
+from typing import Any
+
+_lock = threading.Lock()
+_state: dict[str, Any] = {
+    "connected": False,
+    "last_error": "",
+    "latest_ibi_ms": 0,
+    "buffer_size": 0,
+    "h_min": 0.0,
+    "seed": "",
+    "ibi_history": [],
+    "ws_url": "",
+}
+
+_ws_thread: threading.Thread | None = None
+_ws_stop = threading.Event()
+
+
+def _apply_message(data: dict) -> None:
+    global _state
+    with _lock:
+        if data.get("type") == "ibi":
+            ibi = int(data.get("ibi_ms") or 0)
+            if ibi > 0:
+                _state["latest_ibi_ms"] = ibi
+                hist = list(_state.get("ibi_history") or [])
+                hist.append(ibi)
+                _state["ibi_history"] = hist[-100:]
+            _state["buffer_size"] = int(data.get("buffer_size") or _state.get("buffer_size") or 0)
+            if data.get("h_min") is not None:
+                _state["h_min"] = float(data["h_min"])
+        elif data.get("type") == "seed":
+            _state["seed"] = str(data.get("seed") or data.get("seed_256") or "")
+            if data.get("h_min") is not None:
+                _state["h_min"] = float(data["h_min"])
+            _state["buffer_size"] = 0
+        elif data.get("type") in ("snapshot", "hello"):
+            if data.get("latest_ibi_ms"):
+                _state["latest_ibi_ms"] = int(data["latest_ibi_ms"])
+            if data.get("ibi_history"):
+                _state["ibi_history"] = list(data["ibi_history"])[-100:]
+            if data.get("buffer_size") is not None:
+                _state["buffer_size"] = int(data["buffer_size"])
+            if data.get("h_min") is not None:
+                _state["h_min"] = float(data["h_min"])
+            if data.get("seed_256"):
+                _state["seed"] = str(data["seed_256"])
+
+
+def _ws_loop(url: str) -> None:
+    try:
+        import websocket  # websocket-client
+    except ImportError:
+        with _lock:
+            _state["last_error"] = "pip install websocket-client"
+            _state["connected"] = False
+        return
+
+    def on_message(_ws, message: str) -> None:
+        try:
+            _apply_message(json.loads(message))
+        except json.JSONDecodeError:
+            pass
+
+    def on_open(_ws) -> None:
+        with _lock:
+            _state["connected"] = True
+            _state["last_error"] = ""
+
+    def on_error(_ws, err) -> None:
+        with _lock:
+            _state["connected"] = False
+            _state["last_error"] = str(err)[:200]
+
+    def on_close(_ws, *_args) -> None:
+        with _lock:
+            _state["connected"] = False
+
+    while not _ws_stop.is_set():
+        app = websocket.WebSocketApp(url, on_message=on_message, on_open=on_open, on_error=on_error, on_close=on_close)
+        app.run_forever(ping_interval=20, ping_timeout=10)
+        if _ws_stop.is_set():
+            break
+        time.sleep(2.0)
+
+
+def connect_ws(url: str) -> str:
+    global _ws_thread
+    url = (url or "").strip()
+    if not url:
+        url = os.environ.get("LYGO_BLE_WS_URL", "").strip()
+    if not url:
+        return "Set **LYGO_BLE_WS_URL** (Space secret) or paste your `wss://` tunnel URL."
+    if url.startswith("https://"):
+        url = "wss://" + url[len("https://") :]
+    if url.startswith("http://"):
+        url = "ws://" + url[len("http://") :]
+
+    _ws_stop.set()
+    if _ws_thread and _ws_thread.is_alive():
+        _ws_thread.join(timeout=2.0)
+    _ws_stop.clear()
+    with _lock:
+        _state["ws_url"] = url
+    _ws_thread = threading.Thread(target=_ws_loop, args=(url,), daemon=True)
+    _ws_thread.start()
+    return f"Connecting to `{url}` (tunnel → local :8788)…"
+
+
+def disconnect_ws() -> str:
+    _ws_stop.set()
+    with _lock:
+        _state["connected"] = False
+    return "Disconnected."
+
+
+def refresh_dashboard() -> tuple:
+    """Returns Gradio outputs: ibi, buffer, h_min, seed, lineplot df, hist list, status md."""
+    import pandas as pd
+
+    with _lock:
+        st = dict(_state)
+    hist = st.get("ibi_history") or []
+    seed = st.get("seed") or ""
+    if not seed:
+        seed = "Awaiting 64 IBIs…"
+    status = (
+        f"**WS:** `{'connected' if st.get('connected') else 'disconnected'}` · "
+        f"**URL:** `{st.get('ws_url') or '—'}`"
+    )
+    if st.get("last_error"):
+        status += f"\n\nError: `{st['last_error']}`"
+    if not st.get("ws_url"):
+        status += (
+            "\n\nCloud HF cannot use `localhost`. Expose local ingest with **Cloudflare Tunnel** or **ngrok** "
+            "to port **8788**, then set `LYGO_BLE_WS_URL`."
+        )
+    wave = hist[-50:] if hist else [0]
+    df = pd.DataFrame({"time": list(range(len(wave))), "ibi": wave})
+    return (
+        float(st.get("latest_ibi_ms") or 0),
+        int(st.get("buffer_size") or 0),
+        float(st.get("h_min") or 0.0),
+        seed,
+        df,
+        wave,
+        status,
+    )
+
+
+def load_local_snapshot_from_file(path: str) -> str:
+    """Fallback: read latest_seed.json from synced workspace (no live WS)."""
+    from pathlib import Path
+
+    p = Path(path)
+    if not p.is_file():
+        return "No local seed file."
+    data = json.loads(p.read_text(encoding="utf-8"))
+    _apply_message({"type": "seed", "seed": data.get("seed_256"), "h_min": data.get("h_min")})
+    return f"Loaded seed preview `{str(data.get('seed_preview', ''))[:16]}` from file."

protocol_stack/docs/BIOPHASE7_OBJECTIVE_LIVE_BLE.md

@@ -0,0 +1,53 @@
+# Biophase7 Objective — Live BLE → HF Space
+
+**Source:** `LYRA SYSTEM RETORE/.../2026Biophase7/🧬 Objective.txt`  
+**Signature:** `Δ9Φ963-PHASE7-BLE-LIVE-HARNESS` · `Δ9Φ963-PHASE7-LIVE-STREAM-v1`
+
+## What ships
+
+| Piece | Path |
+|-------|------|
+| BLE GATT ingest | `tools/live_ble_telemetry_ingest.py` |
+| WebSocket hub (8788 public) | `tools/ble_ws_broadcast_server.py` |
+| Shared live state | `protocol7_human_ai_interface/live_stream_hub.py` |
+| One-command start | `tools/run_live_ble_pipeline.py` |
+| HF Gradio consumer | `tools/live_ble_gradio.py` + HF `live_ble_gradio.py` |
+| Harness (local) | `ws://127.0.0.1:8790` via `tools/lygo_control_center/websocket_server.py` |
+
+## Cloud ↔ local fix (required for HF)
+
+Hugging Face runs in the cloud. **`localhost:8788` inside the Space is not your laptop.**
+
+1. On the LYGO node: `python tools/run_live_ble_pipeline.py` (or `--simulate-stream` without hardware).
+2. Expose **port 8788** with **Cloudflare Tunnel** or **ngrok** → `wss://…`.
+3. HF Space secret: `LYGO_BLE_WS_URL=wss://your-tunnel-host/…`
+4. Open Space → **Phase 7** → **Live BLE stream** → connect (or rely on secret URL).
+
+## Local verification
+
+```powershell
+cd "I:\E Drive\lygo-protocol-stack"
+pip install -r requirements-p7-ble.txt
+python tools/live_ble_telemetry_ingest.py --simulate
+python tools/run_live_ble_pipeline.py --simulate-stream
+# In another shell: wscat -c ws://127.0.0.1:8788
+```
+
+| Check | Expected |
+|-------|----------|
+| BLE found | `[+] Candidate: …` |
+| IBI | `[+] IBI: …ms \| buffer n/64` |
+| WS | `LYGO BLE WebSocket ws://0.0.0.0:8788` |
+| Seed | After 64 IBIs: `H_min=… seed=…` |
+
+## Anchor hook
+
+When a seed is generated, `latest_seed.json` is written under `tools/lygo_control_center/workspace/`. Node API: `GET /biometric/live_seed`. Optional anchor drain via existing stack anchor worker.
+
+## Deploy HF
+
+```powershell
+.\tools\deploy_live_ibi_hf.ps1
+```
+
+Copies `live_ble_gradio.py` into `I:\E Drive\Hugging face\` and documents push steps (human `git push` to Space repo if not automated).

protocol_stack/docs/LYGO_PUBLIC_LINK_ARCHIVE.json

@@ -1,6 +1,6 @@
 {
   "signature": "\u03949\u03a6963-LINK-ARCHIVE-v1",
-  "updated_utc": "2026-07-01T12:00:00Z",
+  "updated_utc": "2026-07-02T18:30:00Z",
   "maintainer_note": "Append-only growth log for public surfaces. Register new links via tools/log_public_surface.py. Agents read this + AGENT_MEMORY_SNAPSHOT.json on bootstrap.",
   "canonical_sources": {
     "stack_repo": "https://github.com/DeepSeekOracle/lygo-protocol-stack",
@@ -116,9 +116,25 @@
       "title": "Permaweb anchor \u2014 test_queue_drain",
       "role": "anchor-test",
       "urls": {
-        "permaweb": "file:///I:/E%20Drive/lygo-protocol-stack/data/anchors/8d74a07e3d489e74195627c14c53cccd8ed5a69f5ae2383002b282f7263082d3.json"
+        "permaweb": "file:///I:/E%20Drive/lygo-protocol-stack/data/anchors/5d57d1f528c717ca91016b6dc9b8d570b437e5969abaaf12f08d2bceaa89a197.json"
       },
       "since": "2026-07-02"
+    },
+    {
+      "id": "lattice-intel-index",
+      "title": "E Drive lattice intel index (agent outer brain)",
+      "role": "agent-memory",
+      "urls": {
+        "live_stack": "https://deepseekoracle.github.io/lygo-protocol-stack/LYGO_LATTICE_INTEL_INDEX.json",
+        "repo_canonical": "docs/LYGO_LATTICE_INTEL_INDEX.json",
+        "library_brain": "I:\\E Drive\\LYRA_CORE\\memory\\2026-07-02-lattice-intel-index.md"
+      },
+      "since": "2026-07-02",
+      "pointers": [
+        "docs/LDQ_VAULT_REFERENCE.md",
+        "docs/HF_SPACE_REBUILD_POINTER.md",
+        "docs/SEAL_286_RECURSIVE_ETHICS.md"
+      ]
     }
   ],
   "growth_log": [
@@ -162,6 +178,21 @@
       "refs": [
         "permaweb-test_queue_drain"
       ]
+    },
+    {
+      "utc": "2026-07-02T06:57:40Z",
+      "event": "anchor test_queue_drain",
+      "refs": [
+        "permaweb-test_queue_drain"
+      ]
+    },
+    {
+      "utc": "2026-07-02T18:30:00Z",
+      "event": "E Drive lattice intel sweep installed: LYGO_LATTICE_INTEL_INDEX.json, LDQ/SEAL/HF pointers, map_ldq_lattice_bridge, LYRA_CORE memory log",
+      "refs": [
+        "lattice-intel-index",
+        "pages-stack-index"
+      ]
     }
   ]
 }

protocol_stack/docs/PHASE7_POLISH.md

@@ -11,14 +11,16 @@
 
 ```bash
 pip install -r requirements-p7-ble.txt
-python tools/live_ble_telemetry_ingest.py          # bleak scan + GATT 0x2A37
+python tools/run_live_ble_pipeline.py              # BLE + ws://0.0.0.0:8788 (tunnel → HF)
+python tools/run_live_ble_pipeline.py --simulate-stream
 python tools/live_ble_telemetry_ingest.py --simulate
-python tools/lygo_control_center/websocket_server.py
+python tools/lygo_control_center/websocket_server.py   # legacy harness :8790
 ```
 
 - Seed file: `tools/lygo_control_center/workspace/latest_seed.json`
 - Node API: `GET /biometric/live_seed`
-- Harness page: WebSocket `ws://127.0.0.1:8790` overlays live seed when ingest is running
+- **Biophase7 Objective:** `docs/BIOPHASE7_OBJECTIVE_LIVE_BLE.md` — HF secret `LYGO_BLE_WS_URL`
+- Harness page: `ws://127.0.0.1:8790` or public `ws://0.0.0.0:8788` with event stream
 
 ## Audits
 

protocol_stack/protocol7_human_ai_interface/live_stream_hub.py

@@ -0,0 +1,109 @@
+"""Thread-safe live IBI / P0 seed state for BLE ingest, WebSocket, and HF UI."""
+
+from __future__ import annotations
+
+import threading
+import time
+from collections import deque
+from typing import Any
+
+SIGNATURE = "Δ9Φ963-PHASE7-LIVE-STREAM-v1"
+
+_lock = threading.Lock()
+_ibi_history: deque[int] = deque(maxlen=128)
+_buffer_size = 0
+_latest_ibi = 0
+_latest_seed = ""
+_latest_h_min: float | None = None
+_latest_seed_preview = ""
+_connected_source = "idle"
+
+
+def reset(source: str = "idle") -> None:
+    global _buffer_size, _latest_ibi, _latest_seed, _latest_h_min, _latest_seed_preview, _connected_source
+    with _lock:
+        _ibi_history.clear()
+        _buffer_size = 0
+        _latest_ibi = 0
+        _latest_seed = ""
+        _latest_h_min = None
+        _latest_seed_preview = ""
+        _connected_source = source
+
+
+def record_ibi(ibi_ms: int, buffer_size: int, *, h_min: float | None = None, source: str = "ble") -> dict[str, Any]:
+    global _buffer_size, _latest_ibi, _latest_h_min, _connected_source
+    with _lock:
+        _ibi_history.append(int(ibi_ms))
+        _buffer_size = int(buffer_size)
+        _latest_ibi = int(ibi_ms)
+        if h_min is not None:
+            _latest_h_min = float(h_min)
+        _connected_source = source
+        return _snapshot_unlocked()
+
+
+def record_seed(
+    seed_256: str,
+    *,
+    h_min: float | None,
+    ibi_count: int,
+    source: str = "ble",
+) -> dict[str, Any]:
+    global _latest_seed, _latest_h_min, _latest_seed_preview, _buffer_size, _connected_source
+    with _lock:
+        _latest_seed = str(seed_256)
+        _latest_seed_preview = _latest_seed[:16]
+        if h_min is not None:
+            _latest_h_min = float(h_min)
+        _buffer_size = 0
+        _connected_source = source
+        return _snapshot_unlocked(ibi_count=ibi_count)
+
+
+def _snapshot_unlocked(ibi_count: int | None = None) -> dict[str, Any]:
+    hist = list(_ibi_history)
+    return {
+        "signature": SIGNATURE,
+        "timestamp": time.time(),
+        "source": _connected_source,
+        "latest_ibi_ms": _latest_ibi,
+        "buffer_size": _buffer_size,
+        "ibi_count": ibi_count if ibi_count is not None else len(hist),
+        "h_min": _latest_h_min,
+        "seed_256": _latest_seed or None,
+        "seed_preview": _latest_seed_preview or None,
+        "ibi_history": hist[-100:],
+    }
+
+
+def snapshot() -> dict[str, Any]:
+    with _lock:
+        return _snapshot_unlocked()
+
+
+def ws_message_ibi(ibi_ms: int, buffer_size: int, *, h_min: float | None = None) -> dict[str, Any]:
+    msg: dict[str, Any] = {
+        "type": "ibi",
+        "signature": SIGNATURE,
+        "timestamp": time.time(),
+        "ibi_ms": int(ibi_ms),
+        "buffer_size": int(buffer_size),
+    }
+    if h_min is not None:
+        msg["h_min"] = float(h_min)
+    return msg
+
+
+def ws_message_seed(seed_256: str, *, h_min: float | None, ibi_count: int) -> dict[str, Any]:
+    msg: dict[str, Any] = {
+        "type": "seed",
+        "signature": SIGNATURE,
+        "timestamp": time.time(),
+        "seed": str(seed_256),
+        "seed_preview": str(seed_256)[:16],
+        "ibi_count": int(ibi_count),
+    }
+    if h_min is not None:
+        msg["h_min"] = float(h_min)
+    return msg

protocol_stack/protocol7_human_ai_interface/tests/test_live_stream_hub.py

@@ -0,0 +1,34 @@
+"""Live stream hub + synthetic PDU parse."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
+
+from protocol7_human_ai_interface import live_stream_hub as hub
+from protocol7_human_ai_interface.ble_gatt import parse_heart_rate_measurement
+
+
+def test_hub_ibi_and_seed():
+    hub.reset("test")
+    hub.record_ibi(820, 1, h_min=0.5)
+    hub.record_ibi(810, 2, h_min=0.55)
+    snap = hub.snapshot()
+    assert snap["latest_ibi_ms"] == 810
+    assert len(snap["ibi_history"]) == 2
+    hub.record_seed("a" * 64, h_min=0.9, ibi_count=64)
+    snap2 = hub.snapshot()
+    assert snap2["seed_256"] == "a" * 64
+    assert snap2["buffer_size"] == 0
+
+
+def test_fake_hr_pdu_rr():
+    ibi_ms = 850
+    rr = int((ibi_ms / 1000.0) * 1024)
+    pdu = bytes([0x10, 75, rr & 0xFF, (rr >> 8) & 0xFF])
+    out = parse_heart_rate_measurement(pdu)
+    assert out["ibi_ms"]
+    assert abs(out["ibi_ms"][0] - ibi_ms) <= 2

protocol_stack/tests/phase7_entropy_last_run.json

@@ -1,20 +1,20 @@
 {
-  "timestamp": 1782974640.5148246,
+  "timestamp": 1782977842.6722515,
   "signature": "\u03949\u03a6963-PHASE7-v1.0",
   "ibi_extraction": {
     "signature": "\u03949\u03a6963-PHASE7-v1.0",
     "h_min": 0.2451,
     "von_neumann_bits": 31,
     "entropy_sufficient": false,
-    "seed_256": "4848f28cbaaf443454b950089ef41975895435edd8c4f4c42fdc97fa38229891",
+    "seed_256": "7ab7c0633057318591e851b9071abcd157653ad67f9786b643b1ea88a5f58cd5",
     "ibi_count": 32
   },
   "api_state_preview": {
-    "frequency": 852,
+    "frequency": 639,
     "ethical_vector": [
-      0.7287,
-      0.6558,
-      0.6871
+      0.1773,
+      0.8937,
+      0.3215
     ]
   }
 }

protocol_stack/tests/phase9_audit_last_run.json

@@ -38,5 +38,5 @@
     }
   ],
   "all_pass": true,
-  "duration_ms": 624
+  "duration_ms": 577
 }

protocol_stack/tests/slm_audit_last_run.json

@@ -44,9 +44,9 @@
     {
       "id": "SLM-09-PERF-100",
       "pass": true,
-      "elapsed_ms": 161
+      "elapsed_ms": 157
     }
   ],
   "all_pass": true,
-  "duration_ms": 161
+  "duration_ms": 157
 }

protocol_stack/tools/ble_ws_broadcast_server.py

@@ -0,0 +1,110 @@
+#!/usr/bin/env python3
+"""WebSocket broadcast for live BLE IBI + P0 seeds (Biophase7 Objective)."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from protocol7_human_ai_interface import live_stream_hub as hub  # noqa: E402
+
+DEFAULT_PORT_OBJECTIVE = 8788
+DEFAULT_PORT_HARNESS = 8790
+
+_clients: set = set()
+_event_queue: asyncio.Queue | None = None
+_loop: asyncio.AbstractEventLoop | None = None
+
+
+def schedule_broadcast(message: dict) -> None:
+    """Thread-safe: called from bleak notification callback."""
+    global _loop, _event_queue
+    if _loop is None or _event_queue is None:
+        return
+
+    def _put() -> None:
+        if _event_queue is not None:
+            try:
+                _event_queue.put_nowait(message)
+            except asyncio.QueueFull:
+                pass
+
+    _loop.call_soon_threadsafe(_put)
+
+
+async def _fan_out(message: dict) -> None:
+    if not _clients:
+        return
+    raw = json.dumps(message)
+    dead = []
+    for ws in list(_clients):
+        try:
+            await ws.send(raw)
+        except Exception:
+            dead.append(ws)
+    for ws in dead:
+        _clients.discard(ws)
+
+
+async def _handler(websocket) -> None:
+    _clients.add(websocket)
+    try:
+        await websocket.send(json.dumps({"type": "hello", "signature": hub.SIGNATURE, "snapshot": hub.snapshot()}))
+        await websocket.wait_closed()
+    finally:
+        _clients.discard(websocket)
+
+
+async def _pump_events() -> None:
+    assert _event_queue is not None
+    while True:
+        msg = await _event_queue.get()
+        await _fan_out(msg)
+
+
+async def _heartbeat() -> None:
+    while True:
+        snap = hub.snapshot()
+        if snap.get("ibi_history") or snap.get("seed_256"):
+            await _fan_out({"type": "snapshot", **snap})
+        await asyncio.sleep(0.5)
+
+
+async def run_server(host: str, port: int) -> None:
+    global _event_queue, _loop
+    try:
+        import websockets
+    except ImportError:
+        print("pip install websockets (see requirements-p7-ble.txt)", file=sys.stderr)
+        raise SystemExit(1)
+
+    _loop = asyncio.get_running_loop()
+    _event_queue = asyncio.Queue(maxsize=512)
+    asyncio.create_task(_pump_events())
+    asyncio.create_task(_heartbeat())
+
+    async with websockets.serve(_handler, host, port):
+        print(f"LYGO BLE WebSocket ws://{host}:{port} ({hub.SIGNATURE})", flush=True)
+        await asyncio.Future()
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description="Broadcast live BLE stream over WebSocket")
+    ap.add_argument("--host", default="0.0.0.0", help="Bind address (0.0.0.0 for Cloudflare/ngrok tunnel)")
+    ap.add_argument("--port", type=int, default=DEFAULT_PORT_OBJECTIVE)
+    ap.add_argument("--harness", action="store_true", help=f"Use harness port {DEFAULT_PORT_HARNESS} on 127.0.0.1")
+    args = ap.parse_args()
+    host = "127.0.0.1" if args.harness else args.host
+    port = DEFAULT_PORT_HARNESS if args.harness else args.port
+    asyncio.run(run_server(host, port))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

protocol_stack/tools/live_ble_gradio.py

@@ -0,0 +1,171 @@
+"""
+HF Space / Gradio live BLE consumer — connects to operator tunnel (not localhost on cloud).
+Set LYGO_BLE_WS_URL=wss://your-tunnel.example/ws or pass URL in UI.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import threading
+import time
+from typing import Any
+
+_lock = threading.Lock()
+_state: dict[str, Any] = {
+    "connected": False,
+    "last_error": "",
+    "latest_ibi_ms": 0,
+    "buffer_size": 0,
+    "h_min": 0.0,
+    "seed": "",
+    "ibi_history": [],
+    "ws_url": "",
+}
+
+_ws_thread: threading.Thread | None = None
+_ws_stop = threading.Event()
+
+
+def _apply_message(data: dict) -> None:
+    global _state
+    with _lock:
+        if data.get("type") == "ibi":
+            ibi = int(data.get("ibi_ms") or 0)
+            if ibi > 0:
+                _state["latest_ibi_ms"] = ibi
+                hist = list(_state.get("ibi_history") or [])
+                hist.append(ibi)
+                _state["ibi_history"] = hist[-100:]
+            _state["buffer_size"] = int(data.get("buffer_size") or _state.get("buffer_size") or 0)
+            if data.get("h_min") is not None:
+                _state["h_min"] = float(data["h_min"])
+        elif data.get("type") == "seed":
+            _state["seed"] = str(data.get("seed") or data.get("seed_256") or "")
+            if data.get("h_min") is not None:
+                _state["h_min"] = float(data["h_min"])
+            _state["buffer_size"] = 0
+        elif data.get("type") in ("snapshot", "hello"):
+            if data.get("latest_ibi_ms"):
+                _state["latest_ibi_ms"] = int(data["latest_ibi_ms"])
+            if data.get("ibi_history"):
+                _state["ibi_history"] = list(data["ibi_history"])[-100:]
+            if data.get("buffer_size") is not None:
+                _state["buffer_size"] = int(data["buffer_size"])
+            if data.get("h_min") is not None:
+                _state["h_min"] = float(data["h_min"])
+            if data.get("seed_256"):
+                _state["seed"] = str(data["seed_256"])
+
+
+def _ws_loop(url: str) -> None:
+    try:
+        import websocket  # websocket-client
+    except ImportError:
+        with _lock:
+            _state["last_error"] = "pip install websocket-client"
+            _state["connected"] = False
+        return
+
+    def on_message(_ws, message: str) -> None:
+        try:
+            _apply_message(json.loads(message))
+        except json.JSONDecodeError:
+            pass
+
+    def on_open(_ws) -> None:
+        with _lock:
+            _state["connected"] = True
+            _state["last_error"] = ""
+
+    def on_error(_ws, err) -> None:
+        with _lock:
+            _state["connected"] = False
+            _state["last_error"] = str(err)[:200]
+
+    def on_close(_ws, *_args) -> None:
+        with _lock:
+            _state["connected"] = False
+
+    while not _ws_stop.is_set():
+        app = websocket.WebSocketApp(url, on_message=on_message, on_open=on_open, on_error=on_error, on_close=on_close)
+        app.run_forever(ping_interval=20, ping_timeout=10)
+        if _ws_stop.is_set():
+            break
+        time.sleep(2.0)
+
+
+def connect_ws(url: str) -> str:
+    global _ws_thread
+    url = (url or "").strip()
+    if not url:
+        url = os.environ.get("LYGO_BLE_WS_URL", "").strip()
+    if not url:
+        return "Set **LYGO_BLE_WS_URL** (Space secret) or paste your `wss://` tunnel URL."
+    if url.startswith("https://"):
+        url = "wss://" + url[len("https://") :]
+    if url.startswith("http://"):
+        url = "ws://" + url[len("http://") :]
+
+    _ws_stop.set()
+    if _ws_thread and _ws_thread.is_alive():
+        _ws_thread.join(timeout=2.0)
+    _ws_stop.clear()
+    with _lock:
+        _state["ws_url"] = url
+    _ws_thread = threading.Thread(target=_ws_loop, args=(url,), daemon=True)
+    _ws_thread.start()
+    return f"Connecting to `{url}` (tunnel → local :8788)…"
+
+
+def disconnect_ws() -> str:
+    _ws_stop.set()
+    with _lock:
+        _state["connected"] = False
+    return "Disconnected."
+
+
+def refresh_dashboard() -> tuple:
+    """Returns Gradio outputs: ibi, buffer, h_min, seed, lineplot df, hist list, status md."""
+    import pandas as pd
+
+    with _lock:
+        st = dict(_state)
+    hist = st.get("ibi_history") or []
+    seed = st.get("seed") or ""
+    if not seed:
+        seed = "Awaiting 64 IBIs…"
+    status = (
+        f"**WS:** `{'connected' if st.get('connected') else 'disconnected'}` · "
+        f"**URL:** `{st.get('ws_url') or '—'}`"
+    )
+    if st.get("last_error"):
+        status += f"\n\nError: `{st['last_error']}`"
+    if not st.get("ws_url"):
+        status += (
+            "\n\nCloud HF cannot use `localhost`. Expose local ingest with **Cloudflare Tunnel** or **ngrok** "
+            "to port **8788**, then set `LYGO_BLE_WS_URL`."
+        )
+    wave = hist[-50:] if hist else [0]
+    df = pd.DataFrame({"time": list(range(len(wave))), "ibi": wave})
+    return (
+        float(st.get("latest_ibi_ms") or 0),
+        int(st.get("buffer_size") or 0),
+        float(st.get("h_min") or 0.0),
+        seed,
+        df,
+        wave,
+        status,
+    )
+
+
+def load_local_snapshot_from_file(path: str) -> str:
+    """Fallback: read latest_seed.json from synced workspace (no live WS)."""
+    from pathlib import Path
+
+    p = Path(path)
+    if not p.is_file():
+        return "No local seed file."
+    data = json.loads(p.read_text(encoding="utf-8"))
+    _apply_message({"type": "seed", "seed": data.get("seed_256"), "h_min": data.get("h_min")})
+    return f"Loaded seed preview `{str(data.get('seed_preview', ''))[:16]}` from file."

protocol_stack/tools/live_ble_telemetry_ingest.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 """
-LYGO Phase 7 — Live BLE telemetry ingest (GATT 0x180D).
+LYGO Phase 7 — Live BLE telemetry ingest (GATT 0x180D) + optional WebSocket broadcast.
+Biophase7 Objective: ws://0.0.0.0:8788 for tunnel → HF Space (LYGO_BLE_WS_URL).
 Signature: Δ9Φ963-PHASE7-BLE-LIVE-HARNESS
 """
 
@@ -28,33 +29,58 @@ from protocol7_human_ai_interface.ble_gatt import (  # noqa: E402
     parse_heart_rate_measurement,
 )
 from protocol7_human_ai_interface.entropy_extraction import extract_p0_seed_from_ibi  # noqa: E402
+from protocol7_human_ai_interface import live_stream_hub as hub  # noqa: E402
 
 ENTROPY_BUFFER: list[int] = []
 
 
-def _flush_seed(ibi: list[int]) -> dict:
+def _preview_h_min(ibi: list[int]) -> float | None:
+    if len(ibi) < 4:
+        return None
+    pack = extract_p0_seed_from_ibi([float(x) for x in ibi[-min(32, len(ibi)) :]])
+    return pack.get("h_min")
+
+
+def _flush_seed(ibi: list[int], *, source: str = "ble") -> dict:
     pack = extract_p0_seed_from_ibi([float(x) for x in ibi])
+    seed = str(pack.get("seed_256", ""))
+    h_min = pack.get("h_min")
     payload = {
         "timestamp": time.time(),
         "signature": "Δ9Φ963-PHASE7-BLE-LIVE-HARNESS",
         "ibi_count": len(ibi),
-        "h_min": pack.get("h_min"),
-        "seed_256": pack.get("seed_256"),
-        "seed_preview": str(pack.get("seed_256", ""))[:16],
+        "h_min": h_min,
+        "seed_256": seed,
+        "seed_preview": seed[:16],
     }
     SEED_JSON.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+    hub.record_seed(seed, h_min=h_min, ibi_count=len(ibi), source=source)
+    try:
+        from tools.ble_ws_broadcast_server import schedule_broadcast
+
+        schedule_broadcast(hub.ws_message_seed(seed, h_min=h_min, ibi_count=len(ibi)))
+    except Exception:
+        pass
     return payload
 
 
-def handle_notification(data: bytes) -> None:
+def handle_notification(data: bytes, *, source: str = "ble") -> None:
     global ENTROPY_BUFFER
     parsed = parse_heart_rate_measurement(data)
     for ibi in parsed.get("ibi_ms") or []:
         ENTROPY_BUFFER.append(int(ibi))
+        h_est = _preview_h_min(ENTROPY_BUFFER)
+        hub.record_ibi(int(ibi), len(ENTROPY_BUFFER), h_min=h_est, source=source)
+        try:
+            from tools.ble_ws_broadcast_server import schedule_broadcast
+
+            schedule_broadcast(hub.ws_message_ibi(int(ibi), len(ENTROPY_BUFFER), h_min=h_est))
+        except Exception:
+            pass
         print(f"[+] IBI: {ibi}ms | buffer {len(ENTROPY_BUFFER)}/{BUFFER_THRESHOLD}")
     if len(ENTROPY_BUFFER) >= BUFFER_THRESHOLD:
         print("[!] Threshold reached — extracting P0 seed...")
-        out = _flush_seed(ENTROPY_BUFFER.copy())
+        out = _flush_seed(ENTROPY_BUFFER.copy(), source=source)
         print(f"[>] H_min={out['h_min']} seed={out['seed_preview']}…")
         ENTROPY_BUFFER.clear()
 
@@ -66,6 +92,7 @@ async def _live_loop(scan_timeout: float = 8.0) -> int:
     print("LYGO PHASE 7 — LIVE BLE TELEMETRY INGEST")
     print("Δ9Φ963-PHASE7-BLE-LIVE-HARNESS")
     print("=" * 70)
+    hub.reset("ble")
     devices = await BleakScanner.discover(timeout=scan_timeout)
     target = None
     for d in devices:
@@ -91,18 +118,65 @@ async def _live_loop(scan_timeout: float = 8.0) -> int:
             await asyncio.sleep(1)
 
 
+async def _simulate_stream(interval: float = 0.35, count: int = 80) -> None:
+    """Emit synthetic IBI over WebSocket for HF/tunnel verification without hardware."""
+    import random
+
+    hub.reset("simulate")
+    rng = random.Random(528963)
+    for _ in range(count):
+        ibi = int(60000 / 75 + rng.uniform(-40, 40))
+        handle_notification(_fake_hr_pdu(ibi), source="simulate")
+        await asyncio.sleep(interval)
+
+
+def _fake_hr_pdu(ibi_ms: int) -> bytes:
+    """Minimal GATT 0x2A37 PDU with one RR interval."""
+    rr = int((ibi_ms / 1000.0) * 1024)
+    hr = max(40, min(180, int(60000 / max(ibi_ms, 300))))
+    return bytes([0x10, hr, rr & 0xFF, (rr >> 8) & 0xFF])
+
+
 def run_simulated_flush(n: int = 64) -> dict:
-    """No hardware: synthetic IBI burst for dashboard / audit."""
     import random
 
     rng = random.Random(528963)
     ibi = [int(60000 / 75 + rng.uniform(-40, 40)) for _ in range(n)]
-    return _flush_seed(ibi)
+    hub.reset("simulate")
+    return _flush_seed(ibi, source="simulate")
+
+
+async def _run_with_ws(
+    *,
+    host: str,
+    port: int,
+    simulate_stream: bool,
+    scan_timeout: float,
+) -> int:
+    from tools.ble_ws_broadcast_server import run_server
+
+    server_task = asyncio.create_task(run_server(host, port))
+    await asyncio.sleep(0.3)
+    if simulate_stream:
+        await _simulate_stream()
+        print("[*] Synthetic stream complete — WebSocket still listening (Ctrl+C to exit)")
+        try:
+            await server_task
+        except asyncio.CancelledError:
+            pass
+        return 0
+    code = await _live_loop(scan_timeout)
+    server_task.cancel()
+    return code
 
 
 def main() -> int:
     ap = argparse.ArgumentParser()
     ap.add_argument("--simulate", action="store_true", help="Write seed from synthetic IBI (no BLE)")
+    ap.add_argument("--with-ws", action="store_true", help="Run WebSocket broadcast alongside ingest")
+    ap.add_argument("--ws-host", default="0.0.0.0")
+    ap.add_argument("--ws-port", type=int, default=8788)
+    ap.add_argument("--simulate-stream", action="store_true", help="With --with-ws: stream synthetic IBI over WS")
     ap.add_argument("--scan-timeout", type=float, default=8.0)
     args = ap.parse_args()
 
@@ -111,6 +185,29 @@ def main() -> int:
         print(json.dumps(out, indent=2))
         return 0
 
+    if args.with_ws:
+        if args.simulate_stream:
+            try:
+                return asyncio.run(
+                    _run_with_ws(
+                        host=args.ws_host, port=args.ws_port, simulate_stream=True, scan_timeout=args.scan_timeout
+                    )
+                )
+            except KeyboardInterrupt:
+                return 0
+        if not bleak_available() and not args.simulate_stream:
+            print("bleak not installed — pip install bleak  OR  --simulate-stream", file=sys.stderr)
+            return 1
+        try:
+            return asyncio.run(
+                _run_with_ws(host=args.ws_host, port=args.ws_port, simulate_stream=False, scan_timeout=args.scan_timeout)
+            )
+        except KeyboardInterrupt:
+            print("\n[!] Ingestion stopped.")
+            if ENTROPY_BUFFER:
+                _flush_seed(ENTROPY_BUFFER)
+            return 0
+
     if not bleak_available():
         print("bleak not installed — pip install bleak  OR  use --simulate", file=sys.stderr)
         return 1

protocol_stack/tools/run_live_ble_pipeline.py

@@ -0,0 +1,47 @@
+#!/usr/bin/env python3
+"""Install + start Biophase7 live BLE → WebSocket pipeline (Objective.txt)."""
+
+from __future__ import annotations
+
+import argparse
+import subprocess
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description="LYGO live BLE pipeline")
+    ap.add_argument("--simulate-stream", action="store_true", help="No hardware; synthetic IBI on WS :8788")
+    ap.add_argument("--ws-port", type=int, default=8788)
+    ap.add_argument("--check-deps", action="store_true")
+    args = ap.parse_args()
+
+    req = ROOT / "requirements-p7-ble.txt"
+    if args.check_deps:
+        print(f"Install: pip install -r {req}")
+        return 0
+
+    cmd = [
+        sys.executable,
+        str(ROOT / "tools" / "live_ble_telemetry_ingest.py"),
+        "--with-ws",
+        "--ws-host",
+        "0.0.0.0",
+        "--ws-port",
+        str(args.ws_port),
+    ]
+    if args.simulate_stream:
+        cmd.append("--simulate-stream")
+
+    print("Δ9Φ963 — Live BLE pipeline")
+    print("  1) Tunnel port", args.ws_port, "→ set HF secret LYGO_BLE_WS_URL=wss://…")
+    print("  2) HF Space tab: Live BLE stream")
+    print("  3) Docs: docs/BIOPHASE7_OBJECTIVE_LIVE_BLE.md")
+    print("Running:", " ".join(cmd))
+    return subprocess.call(cmd)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

requirements.txt

@@ -6,4 +6,4 @@ requests
 gradio
 scipy
 # Phase 2: Ethical Guardian alignment badge uses bundled protocol_stack/ (stdlib-first).
-# Community Docker node: see lygo-protocol-stack/setup.sh and docker-compose.yml on GitHub.
+# Community Docker node: see lygo-protocol-stack/setup.sh and docker-compose.yml on GitHub.websocket-client