CI: push to ipcr-push-85-17-246-67.ip.nsl.sh (the box's cert no longer covers nip.io) [skip ci]

.gitattributes

@@ -0,0 +1,6 @@
+# FIX CRLF always when developer has not set
+# Linux/Mac: git config --global core.autocrlf input
+# Windows: git config --global core.autocrlf true
+# Auto detect text files and perform LF normalization
+* text=auto
+* eol=lf

.gitea/workflows/docker-publish.yml

@@ -0,0 +1,98 @@
+# Gitea twin of .github/workflows/docker-publish.yml. Gitea runs only .gitea/workflows/ once this
+# folder exists, and GitHub never reads it.
+#
+# Publishes to IPCR on watch.nsl.sh instead of ghcr.io:
+#   push  ipcr-push-85-17-246-67.ip.nsl.sh/metadec.eth/<repo>:<tag>   docker login as metadec.eth with the
+#         IPCR_PUSH_TOKEN secret (the metadec.eth namespace token from the IPCR dashboard)
+#   pull  ipcr-watch.nsl.sh/ipns/metadec.eth/metadec.eth/<repo>:<tag>   anonymous
+# The push goes to ipcr-push-85-17-246-67.ip.nsl.sh, the box's own address, not ipcr-push-watch.nsl.sh: Cloudflare fronts
+# *.watch.nsl.sh and refuses any request over 100 MB (413), which a large image layer is.
+# That host's certificate is signed by the box's private Mesh Router CA, given to buildkit from
+# the IPCR_CA_PEM repo variable. docker login would make the job's Docker daemon check the
+# certificate without that CA, so the login is written straight into ~/.docker/config.json.
+# The runner's job image (node:22-bookworm) has no docker CLI, so the static one is
+# installed first; buildx installs itself. No `type=gha` cache: the runner has none.
+
+name: Build and Push Docker Image
+
+on:
+  push:
+    branches:
+      - main
+    tags:
+      - 'v*'
+  workflow_dispatch:
+    inputs:
+      tag:
+        description: 'Tag for the image (optional, defaults to sha)'
+        required: false
+        type: string
+      meta_core_version:
+        description: 'meta-core version to use (default: latest)'
+        required: false
+        type: string
+        default: 'latest'
+
+env:
+  REGISTRY: ipcr-push-85-17-246-67.ip.nsl.sh
+  IPCR_PULL: ipcr-watch.nsl.sh/ipns/metadec.eth
+  IMAGE_NAME: metadec.eth/${{ github.event.repository.name }}
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Install Docker CLI
+        run: curl -fsSL https://download.docker.com/linux/static/stable/x86_64/docker-28.5.1.tgz | tar -xz -C /usr/local/bin --strip-components=1 docker/docker
+
+      - name: Trust the Mesh Router CA
+        run: printf '%s\n' "$IPCR_CA_PEM" > /tmp/ipcr-ca.pem
+        env:
+          IPCR_CA_PEM: ${{ vars.IPCR_CA_PEM }}
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+        with:
+          version: latest
+          buildkitd-config-inline: |
+            [registry."${{ env.REGISTRY }}"]
+              ca = ["/tmp/ipcr-ca.pem"]
+
+      - name: Log in to IPCR
+        env:
+          IPCR_PUSH_TOKEN: ${{ secrets.IPCR_PUSH_TOKEN }}
+        run: |
+          mkdir -p ~/.docker
+          printf '{"auths":{"%s":{"auth":"%s"}}}\n' "$REGISTRY" "$(printf 'metadec.eth:%s' "$IPCR_PUSH_TOKEN" | base64 -w0)" > ~/.docker/config.json
+
+      - name: Extract metadata (tags, labels)
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
+          tags: |
+            type=semver,pattern={{version}}
+            # A fix to release 1.2.3 is tagged v1.2.3.1: four numbers are not semver, so
+            # the pattern above makes no tag for it; this one does (image tag 1.2.3.1).
+            type=match,pattern=v(\d+\.\d+\.\d+\.\d+),group=1
+            type=semver,pattern={{major}}.{{minor}}
+            type=sha,prefix=
+            type=raw,value=latest,enable={{is_default_branch}}
+            type=raw,value=${{ inputs.tag }},enable=${{ inputs.tag != '' }}
+
+      - name: Build and push Docker image
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          push: true
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+          build-args: |
+            META_CORE_IMAGE=${{ env.IPCR_PULL }}/metadec.eth/meta-core:${{ inputs.meta_core_version || 'latest' }}

.github/workflows/docker-publish.yml

@@ -50,6 +50,9 @@ jobs:
           images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
           tags: |
             type=semver,pattern={{version}}
+            # A fix to release 1.2.3 is tagged v1.2.3.1: four numbers are not semver, so
+            # the pattern above makes no tag for it; this one does (image tag 1.2.3.1).
+            type=match,pattern=v(\d+\.\d+\.\d+\.\d+),group=1
             type=semver,pattern={{major}}.{{minor}}
             type=sha,prefix=
             type=raw,value=latest,enable={{is_default_branch}}

README.md

@@ -1,16 +1,11 @@
 # Meta-Stremio
 
-A standalone Stremio addon service for streaming personal media with on-the-fly HLS transcoding. Part of the MetaMesh decentralized media ecosystem.
+Stremio addon for the MetaMesh stack. Reads file metadata from meta-core's HTTP API (no direct Redis access), reads files via meta-core's WebDAV (or a direct mount), and serves HLS with on-the-fly FFmpeg transcoding plus the standard Stremio addon protocol.
+
+External port in the dev stack: **8182** (auth-gated, via Caddy + nginx-hash-lock). Internal container port is `7000`. Debug-direct port (bypasses the perimeter): **18182**. Container name: `metastremio-app` (backend) / `metastremio` (hash-lock proxy).
 
 ## Overview
 
-Meta-Stremio is a **process-read** service that:
-- Reads file metadata from the shared KV store (Redis)
-- Reads media files via WebDAV (from meta-sort) or direct volume access
-- Provides HLS streaming with adaptive real-time FFmpeg transcoding
-- Serves as a Stremio addon for seamless playback
-- Supports SMB/rclone mounts via meta-sort's WebDAV endpoint
-
 ```
 ┌─────────────────────────────────────────────────────────────────────┐
 │                        MetaMesh Ecosystem                           │
@@ -23,246 +18,228 @@ Meta-Stremio is a **process-read** service that:
 │         │                   │                   │                   │
 │         ▼                   ▼                   ▼                   │
 │  ┌─────────────────────────────────────────────────────┐           │
-│  │              Shared KV Store (Redis)                │           │
+│  │     meta-core (owns Redis; /urls + /meta APIs)      │           │
-│  │   Leader election via flock on shared filesystem    │           │
+│  │   located over UDP multicast (beacon v2)    │           │
-│  │   DB stored in: DATA/Apps/meta-core/DB              │           │
 │  └─────────────────────────────────────────────────────┘           │
 │                             │                                       │
 │                             ▼                                       │
 │  ┌─────────────────────────────────────────────────────┐           │
-│  │           Shared DATA Volume (media files)          │           │
+│  │      File access via WebDAV (or direct mount)       │           │
-│  │   meta-sort writes → meta-stremio/meta-fuse reads   │           │
 │  └─────────────────────────────────────────────────────┘           │
-│                                                                     │
 └─────────────────────────────────────────────────────────────────────┘
 ```
 
 ## Features
 
-### High Performance
+- **HLS transcoding** — adaptive preset+CRF targeting a 60–80% transcode ratio, prefetch of N segments ahead, persistent segment cache.
-- **Adaptive Quality** - Dynamic preset + CRF adjustment targeting 60-80% transcode ratio
+- **Stremio protocol** — manifest, catalog, meta, stream; one HLS stream per audio track; subtitles extracted to VTT.
-- **Intelligent Prefetch** - 4 segments ahead for smooth playback
+- **Direct file serving** — range-request capable, for clients that can play the source directly.
-- **Single-threaded Encode** - 100% CPU per segment for fastest encoding
+- **meta-core over HTTP** — `LeaderStorage` locates meta-core via **beacon v2** (UDP multicast `239.255.99.1:9099`; see [`beacon-v2.md`](../../docs/project-architecture/beacon-v2.md)) or a pinned `META_CORE_URL`, reads every record through meta-core's `/meta/*` API, and keeps its cache fresh from the SSE meta-stream. The WebDAV URL is taken from meta-core's announce/`/urls` — nothing is configured by hand and no `/meta-core` volume is mounted. (`RedisStorage` remains in `src/storage/` as a legacy, unused backend; meta-core no longer publishes a Redis URL.)
-- **Muxed Segments** - Video+audio in one FFmpeg call
+- **Neighbour nav** — this service announces itself on the discovery group and serves its own neighbour map at `/api/neighbors` for the dashboard's shared `<meta-service-menu>`.
-- **Segment Caching** - Persistent cache for repeated access
+- **Path-token gate for addon URLs** — when `HASH_API_SEED` is set, addon paths must be prefixed with `/s/<token>`. Dashboard/browser paths are protected separately by the upstream hash-lock + OIDC sidecar (Stremio clients can't carry cookies, so they get the path-token instead).
-
+- **Language-configurable manifest** — `displayLanguage` in the per-install config (base64-in-path) selects which `titles` translation meta-sort's TMDB plugin populated.
-### Stremio Integration
-- **Catalog Endpoint** - Browse media library by type (movies, series)
-- **Meta Endpoint** - Detailed video information with codec details
-- **Stream Endpoint** - Multiple stream options per video
-- **Multi-Audio** - Separate HLS streams per audio track
-- **Subtitles** - Extract and serve embedded subtitles as VTT
-- **Language Configuration** - Configure display language (uses metadata stored by meta-sort's TMDB plugin)
-
-### Storage Abstraction
-- **Leader Storage** - Auto-discovers meta-sort's Redis via leader election (recommended)
-- **Redis Storage** - Direct Redis connection via URL
-
-## Project Structure
-
-```
-meta-stremio/
-├── README.md
-├── CLAUDE.md                   # Development guide
-├── Dockerfile
-├── docker-compose.yml
-├── requirements.txt            # redis + requests
-├── src/
-│   ├── server.py               # HTTP server + routing
-│   ├── stremio.py              # Stremio handlers + storage integration
-│   ├── transcoder.py           # HLS transcoding engine
-│   ├── fileserver.py           # File serving (direct or WebDAV)
-│   ├── webdav_client.py        # WebDAV client for meta-sort file access
-│   └── storage/
-│       ├── __init__.py
-│       ├── provider.py         # Abstract StorageProvider
-│       ├── redis_storage.py    # Direct Redis connection
-│       ├── leader_storage.py   # Leader-aware Redis connection
-│       └── leader_discovery.py # Leader election discovery
-└── www/
-    └── index.html              # Dashboard
-```
 
 ## Installation
 
-### Docker (Recommended)
+### Docker (in the MetaMesh dev stack)
 
 ```bash
-# Build
-docker build -t meta-stremio .
-
-# Run with leader discovery (recommended - auto-discovers meta-sort's Redis)
-docker run -d \
-  -p 7000:7000 \
-  -v /path/to/media:/files:ro \
-  -v /path/to/meta-core:/meta-core \
-  meta-stremio
-
-# Run with direct Redis connection
-docker run -d \
-  -p 7000:7000 \
-  -v /path/to/media:/files:ro \
-  -e STORAGE_MODE=redis \
-  -e REDIS_URL=redis://your-redis:6379 \
-  meta-stremio
-```
-
-### Docker Compose
-
-```bash
-# Run as part of the MetaMesh stack (uses leader discovery)
 cd dev
-./start.sh
+./scripts/start.sh
 ```
 
-### Standalone (Development)
+This brings up `metastremio-app` on port 18182 (debug-direct) and the auth-gated entry on `https://metastremio-dev.localhost:8182`.
+
+### Docker (standalone)
 
 ```bash
-# Prerequisites: Python 3.9+, FFmpeg
+docker build -t meta-stremio packages/meta-stremio
 
-# Install dependencies
+# Against an existing meta-core (found over UDP on the same Docker network,
-pip install -r requirements.txt
+# or pinned with META_CORE_URL)
+docker run -d \
+  -p 8182:7000 \
+  -v /path/to/media:/files:ro \
+  -e STORAGE_MODE=direct \
+  -e META_CORE_URL=http://metacore-app:9000 \
+  meta-stremio
+```
 
-# Run
+The container always listens on port `7000` internally — map it to whatever you want externally. The image bundles a `meta-core` binary (copied from `ghcr.io/worph/meta-core`, overridable with `--build-arg META_CORE_IMAGE=…`): with the image default `STORAGE_MODE=leader`, `docker/start.sh` starts it as an in-container sidecar first; any other value (the dev stack and the CasaOS app use `direct`) skips it and uses an external meta-core. Published images: `ghcr.io/worph/meta-stremio` (semver tags from `v*` git tags; in this repo `:latest` follows `main` — see `.github/workflows/docker-publish.yml`); the CasaOS app is `MetaStremio` in [`MetaAppStore`](../MetaAppStore/Apps/MetaStremio/docker-compose.yml).
+
+### Standalone (development)
+
+```bash
+# Requires Python 3.9+ and FFmpeg.
+pip install -r requirements.txt   # redis, watchdog, Pillow, requests
 cd src
-python server.py
+python server.py   # finds meta-core over UDP, or set META_CORE_URL=<meta-core API URL>
 ```
 
 ## Configuration
 
-### Environment Variables
+### Environment variables
 
 | Variable | Default | Description |
-|----------|---------|-------------|
+|---|---|---|
-| `PORT` | `7000` | Server listen port |
+| `PORT` | `7000` | Internal HTTP listen port (the dev stack maps 18182 / 8182 to this). |
-| `MEDIA_DIR` | `/data/media` | Media files directory |
+| `BASE_URL` | — | External URL emitted in manifests / poster URLs (e.g. `https://metastremio-dev.localhost:8182`). |
-| `CACHE_DIR` | `/data/cache` | Transcoded segment cache |
+| `MEDIA_DIR` | `/files/watch` | Media directory (consumed by `transcoder.py`). |
-| `STORAGE_MODE` | `leader` | `leader` or `redis` |
+| `CACHE_DIR` | `/data/cache` | Transcoded segment cache. |
-| `META_CORE_PATH` | `/meta-core` | Path to meta-core shared volume (for leader discovery) |
+| `PUBLIC_URL` | — | Browser-facing URL announced to neighbours (nav menu); wins over `BASE_URL` for the announce. |
-| `FILES_PATH` | `/files` | Path to files volume |
+| `STORAGE_MODE` | `leader` (image) | Read only by `docker/start.sh`: `leader` starts the bundled meta-core sidecar in-container, anything else (e.g. `direct`) skips it. The Python server always uses `LeaderStorage`. |
-| `REDIS_URL` | `redis://localhost:6379` | Redis connection URL (for `redis` mode) |
+| `META_CORE_URL` | — | Pins meta-core's HTTP API (e.g. `http://metacore-app:9000`); when set, UDP discovery never overrides it. Unset → meta-core is located over UDP. |
-| `REDIS_PREFIX` | `meta-sort:` | Redis key prefix |
+| `ENABLE_UDP_DISCOVERY` | `true` | beacon v2 announce/listen. |
-| `SEGMENT_DURATION` | `4` | HLS segment length (seconds) |
+| `LEADER_WAIT_TIMEOUT` | `0` | Seconds to wait for meta-core at startup (`0` = forever); on timeout, keeps retrying in the background. |
-| `PREFETCH_SEGMENTS` | `4` | Segments to prefetch ahead |
+| `LEADER_RETRY_INTERVAL` | `5` | Seconds between meta-core lookup attempts. |
-| `SCHEME` | `auto` | URL scheme (`http`, `https`, `auto`) |
+| `FILES_PATH` | `/files` | Files volume (local fallback when WebDAV is not configured). |
-| `META_CORE_WEBDAV_URL` | - | WebDAV URL for file access (e.g., `http://meta-core/webdav`) |
+| `SCHEME` | `auto` | URL scheme for generated URLs: `http`, `https`, or `auto`. |
+| `SEGMENT_DURATION` | `4` | HLS segment length (s). |
+| `PREFETCH_SEGMENTS` | `4` | Segments to prefetch ahead. |
+| `HASH_API_SEED` | — | When set, derives a 16-char path-token; all addon routes must be prefixed with `/s/<token>`. Dashboard routes bypass. |
 
-**WebDAV File Access**: When `META_CORE_WEBDAV_URL` is set, meta-stremio reads files via HTTP from meta-core's WebDAV server instead of direct filesystem access. This enables access to SMB/rclone mounts that are dynamically mounted inside meta-core.
+`META_CORE_PATH`, `REDIS_URL` and `REDIS_PREFIX` are still set by the Dockerfile but are vestigial: meta-core is no longer found through the `/meta-core` volume, and reads go over meta-core's HTTP API rather than Redis.
 
-## API Endpoints
+## API endpoints
+
+All routes below are matched in `src/server.py`. When `HASH_API_SEED` is set, **every addon path is prefixed with `/s/<token>`**; the dashboard paths in the first table are *not* prefixed (they're protected by hash-lock/OIDC upstream).
+
+### Dashboard / browser
 
-### Dashboard
 | Endpoint | Method | Description |
-|----------|--------|-------------|
+|---|---|---|
-| `/` | GET | Setup dashboard |
+| `/` , `/index.html` | GET | Setup dashboard |
-| `/health` | GET | Health check with storage status |
+| `/configure` | GET | Per-install language configuration page |
-
+| `/health` | GET | Liveness + storage status |
-### Dashboard API
-| Endpoint | Method | Description |
-|----------|--------|-------------|
 | `/api/stats` | GET | Library statistics |
-| `/api/library` | GET | Full library list |
+| `/api/library` | GET | Full library list (videos + count) |
+| `/api/neighbors` | GET | This peer's UDP neighbour map (beacon v2) for the nav menu |
+| `/api/services` | GET | Alias of `/api/neighbors` (kept for older dashboards) |
+| `/api/languages` | GET | Languages selectable from the `/configure` page |
+| `/meta-service-menu.js` | GET | Shared `<meta-service-menu>` nav element loaded by the dashboard |
+| `/transcode/metrics` | GET | Transcoder metrics (also listed below; exempt from the path-token) |
+
+### Stremio addon protocol
+
+`/manifest.json` and `/stremio/manifest.json` are aliases; same for catalog/meta/stream. Stremio per-install configuration is encoded as URL-safe base64 JSON between the `/s/<token>` prefix and the addon path (e.g. `/s/<token>/<configB64>/manifest.json`).
 
-### Stremio Protocol
 | Endpoint | Method | Description |
-|----------|--------|-------------|
+|---|---|---|
-| `/manifest.json` | GET | Addon manifest |
+| `/manifest.json` | GET, HEAD | Addon manifest |
-| `/catalog/:type/:id.json` | GET | Video catalog |
+| `/stremio/manifest.json` | GET, HEAD | Alias of `/manifest.json` |
+| `/catalog/:type/:id.json` | GET | Catalog (with optional `…/:extra.json`) |
+| `/stremio/catalog/:type/:id.json` | GET | Alias |
 | `/meta/:type/:id.json` | GET | Video metadata |
+| `/stremio/meta/:type/:id.json` | GET | Alias |
 | `/stream/:type/:id.json` | GET | Stream URLs |
+| `/stremio/stream/:type/:id.json` | GET | Alias |
+
+### File serving
 
-### Transcoding
 | Endpoint | Method | Description |
-|----------|--------|-------------|
+|---|---|---|
-| `/transcode/:path/master.m3u8` | GET | ABR master playlist |
+| `/file/{cid}` | GET, HEAD | Serve file by CID |
-| `/transcode/:path/master_original_a0.m3u8` | GET | Original quality playlist |
+| `/file/{cid}/w{width}` | GET, HEAD | Serve resized image by CID (uses Pillow) |
-| `/transcode/:path/stream_a0_720p.m3u8` | GET | Quality stream playlist |
+| `/poster/{cid}` | GET, HEAD | Legacy alias of `/file/{cid}` |
-| `/transcode/:path/seg_a0_720p_0.ts` | GET | Video segment |
+| `/poster/{cid}/w{width}` | GET, HEAD | Legacy alias of `/file/{cid}/w{width}` |
-| `/transcode/:path/subtitle_0.vtt` | GET | Subtitle track |
+| `/direct/{path}` | GET, HEAD | Direct file serving with HTTP `Range` support |
-| `/direct/:path` | GET | Direct file serving |
-| `/transcode/metrics` | GET | Transcoding metrics |
 
-## KV Store Schema
+### Transcoder
 
-Meta-stremio reads from meta-sort's Redis using this key format:
+| Endpoint | Method | Description |
+|---|---|---|
+| `/transcode/{path}/master.m3u8` | GET, HEAD | ABR master playlist |
+| `/transcode/{path}/master_{resolution}.m3u8` | GET, HEAD | Master playlist for a single resolution |
+| `/transcode/{path}/master_a{audio}.m3u8` | GET, HEAD | Master playlist for a single audio track |
+| `/transcode/{path}/master_{resolution}_a{audio}.m3u8` | GET, HEAD | Combined variant |
+| `/transcode/{path}/stream_a{audio}_{resolution}.m3u8` | GET, HEAD | Per-variant stream playlist |
+| `/transcode/{path}/seg_a{audio}_{resolution}_{n}.ts` | GET, HEAD | Video segment |
+| `/transcode/{path}/subtitle_{idx}.m3u8` | GET, HEAD | Subtitle playlist |
+| `/transcode/{path}/subtitle_{idx}.vtt` | GET, HEAD | Subtitle VTT |
+| `/transcode/metrics` | GET | Transcoder metrics (includes `total_files`) |
+| `/transcode/reset-metrics` | POST | Reset transcoder metrics |
+
+## Project structure
 
 ```
-meta-sort:file:{hashId}              → Hash containing all metadata
+meta-stremio/
-  - filePath                         → /media/movies/Movie.mkv
+├── README.md
-  - title                            → Movie Title
+├── Dockerfile
-  - type                             → movie|series|anime
+├── docker-compose.yml
-  - year                             → 2024
+├── docker/                            # container entry scripts
-  - season                           → 1 (for series)
+├── requirements.txt                   # redis, watchdog, Pillow, requests
-  - episode                          → 1 (for series)
+├── src/
-  - duration                         → 7200 (seconds)
+│   ├── server.py                      # HTTP server + routing (entry point)
-  - width                            → 1920
+│   ├── stremio.py                     # Stremio handlers + library/language helpers
-  - height                           → 1080
+│   ├── transcoder.py                  # HLS transcoding engine + metrics + segment/subtitle managers
-  - videoCodec                       → h264
+│   ├── fileserver.py                  # serve_file(cid, width) — file/poster endpoints
-  - audioCodec                       → aac
+│   ├── poster.py                      # poster URL/CID helpers
-  - audioTracks                      → JSON string
+│   ├── webdav_client.py               # WebDAV client (file_exists, stream_file, stream_range, get_file_size)
-  - subtitles                        → JSON string
+│   └── storage/
-  - imdbId                           → tt1234567
+│       ├── __init__.py
-  - poster                           → URL
+│       ├── provider.py                # StorageProvider + VideoMetadata abstract base
+│       ├── leader_storage.py          # Storage backend in use: meta-core HTTP API + SSE cache invalidation
+│       ├── leader_client.py           # Locates meta-core (UDP or META_CORE_URL pin), exposes its /urls set
+│       ├── meshdisco.py               # beacon v2 (Python port of the Go reference in meta-core)
+│       ├── meta_consumer.py           # SSE consumer from meta-core /meta stream
+│       ├── meta_core_api_client.py    # HTTP client for meta-core REST API
+│       └── redis_storage.py           # Legacy direct-Redis backend (unused)
+└── www/
+    ├── index.html                     # Dashboard
+    └── meta-service-menu.js           # Shared nav element
 ```
 
-## Stream Types
+The `/configure` page is rendered inline by `server.py` (there is no `www/configure.html`).
 
-Each video provides multiple stream options:
+## KV store schema
 
-1. **Direct File** - Original file, no transcoding
+meta-stremio reads the per-file record `/file/{cid}` (via meta-core's API) populated by meta-sort and its plugins. Field semantics (`videoType`, `originalTitle`, `titles`, `season`, `episode`, `movieYear`, `cid_*`, etc.) are documented in the repo-root [`METADATA_KEYS.md`](../../METADATA_KEYS.md) — that is the single source of truth, and matches the `@metazla/meta-interface` types.
-   - Best quality, may not play on all devices
 
-2. **HLS Original** - Transcoded at source resolution
+## Stream types
-   - H.264/AAC in HLS container
-   - One stream per audio track
 
-3. **HLS ABR** - Adaptive bitrate with quality ladder
+For each video, the addon advertises multiple streams:
-   - Multiple resolutions (1080p, 720p, 480p, 360p)
-   - Automatic quality switching
 
-## Transcoding Details
+1. **Direct file** — original bytes, served via `/direct/...` with `Range` support. Best quality, may not play on every device.
+2. **HLS original** — transcoded at source resolution, H.264/AAC, one stream per audio track.
+3. **HLS ABR** — adaptive ladder (rungs above the source height are skipped). Presets are baked into `transcoder.py`: video is x264 **CRF-based** (the adaptive controller shifts preset and CRF offset at runtime), audio is AAC 128 kbps stereo. `BANDWIDTH` is only the hint written to the master playlist.
 
-### Adaptive Quality System
+| Rung | Base CRF | Master-playlist `BANDWIDTH` hint |
-
+|---|---|---|
-The transcoder automatically adjusts encoding settings to maintain a 60-80% transcode ratio (segment encode time / segment duration):
+| original | 23 | — |
-
+| 1080p | 23 | 4000 kbps |
-- **Too Fast (<60%)**: Increase quality (lower CRF, slower preset)
+| 720p  | 24 | 2500 kbps |
-- **Too Slow (>80%)**: Decrease quality (higher CRF, faster preset)
+| 480p  | 25 | 1200 kbps |
-
+| 360p  | 26 |  800 kbps |
-### Quality Ladder
-
-| Resolution | Video Bitrate | Audio Bitrate |
-|------------|---------------|---------------|
-| 1080p | 5000 kbps | 192 kbps |
-| 720p | 3000 kbps | 128 kbps |
-| 480p | 1500 kbps | 128 kbps |
-| 360p | 800 kbps | 96 kbps |
 
 ## Adding to Stremio
 
-1. Open the dashboard at `http://localhost:7000`
+1. Open the dashboard at `https://metastremio-dev.localhost:8182/` (or debug-direct `http://localhost:18182/`).
-2. Click "Install in Stremio" or copy the manifest URL
+2. Pick your display language on `/configure` if you want non-English titles.
-3. In Stremio: Settings → Addons → Enter URL → Install
+3. Use the **Install in Stremio** button (uses the `stremio://` protocol handler) or copy the manifest URL into Stremio: Settings → Addons → Add Addon.
 
-## Development
+## Debugging
 
 ```bash
-# Run in development mode (requires meta-sort running with Redis)
+# Container logs
-cd src
+docker compose -f dev/docker-compose.yml logs -f metastremio-app
-python server.py
 
-# Or with direct Redis URL
+# Live health
-STORAGE_MODE=redis REDIS_URL=redis://localhost:6379 python server.py
+curl -k https://metastremio-dev.localhost:8182/health
+curl    http://localhost:18182/health   # debug-direct (no auth, no TLS)
 
-# View logs
+# Transcoder metrics
-tail -f /data/cache/*.log
+curl    http://localhost:18182/transcode/metrics
+curl -X POST http://localhost:18182/transcode/reset-metrics
 ```
 
-## Related Projects
+Container lifecycle: do **not** `docker restart metastremio-app` to apply config — use the dev-stack reload script (`./scripts/reload-stremio.sh`) so supervisord restarts the right process inside the container.
 
-- **[meta-sort](../meta-sort)** - File indexer (writes metadata to Redis)
+## Related projects
-- **[meta-fuse](../meta-fuse)** - Virtual filesystem (reads metadata)
+
-- **[meta-orbit](../meta-orbit)** - P2P metadata sharing
+- **[meta-sort](../meta-sort)** — file indexer; writes the metadata meta-stremio reads.
+- **[meta-core](../meta-core)** — owns Redis; the HTTP API + WebDAV meta-stremio reads from.
+- **[meta-fuse](../meta-fuse)** — virtual filesystem; reads the same records.
+- **[meta-watch](../meta-watch)** — browser video client for the meta-share network (server-side zero media work, unlike this addon's transcoder).
 
 ## License
 

src/server.py

@@ -35,7 +35,6 @@ import threading
 
 import stremio
 import transcoder
-from storage import init_service_discovery, get_service_discovery
 import fileserver
 import webdav_client
 
@@ -58,19 +57,24 @@ API_PREFIX = f'/s/{API_KEY}' if API_KEY else ''
 # upstream by hash-lock/OIDC instead of the path-token.
 DASHBOARD_PATHS = frozenset({
     '/', '/index.html', '/configure', '/health',
-    '/api/stats', '/api/library', '/api/services', '/api/languages',
+    '/api/stats', '/api/library', '/api/services', '/api/neighbors', '/api/languages',
+    # The dashboard's shared nav element, loaded as a module from index.html.
+    '/meta-service-menu.js',
     '/transcode/metrics',
 })
 
 # Initialize storage
 storage = stremio.init_storage()
 
-# Initialize service discovery (only if META_CORE_PATH exists)
+# Service discovery is UDP now (beacon v2), so there is no
+# META_CORE_PATH gate: the leader client's node announces this service on the
+# same multicast group it listens on. Being discoverable and discovering are
+# one thing. See docs/project-architecture/service-discovery.md.
 service_discovery = None
-if os.path.exists(META_CORE_PATH):
+try:
-    try:
+    from storage.leader_client import get_leader_client
-        service_discovery = init_service_discovery(base_url=BASE_URL if BASE_URL else None)
+    get_leader_client().start_watching()
-    except Exception as e:
+except Exception as e:
     print(f"[Server] Service discovery init failed: {e}")
 
 # Global Stremio handler
@@ -252,6 +256,13 @@ class Handler(BaseHTTPRequestHandler):
         if path == '/' or path == '/index.html':
             return self.serve_setup_page()
 
+        # The one static asset under www/ that the dashboard loads as a module:
+        # the shared <meta-service-menu> element. There is no generic static
+        # route here (index.html is string-injected on every request), so this
+        # is an explicit allowlist of one rather than an open file server.
+        if path == '/meta-service-menu.js':
+            return self.serve_www_asset('meta-service-menu.js', 'text/javascript')
+
         # Configure page
         if path == '/configure':
             return self.serve_configure_page()
@@ -287,7 +298,7 @@ class Handler(BaseHTTPRequestHandler):
             })
 
         # Discovered services (for inter-service navigation)
-        if path == '/api/services':
+        if path == '/api/services' or path == '/api/neighbors':
             return self.handle_services_api()
 
         # === File API ===
@@ -368,6 +379,22 @@ class Handler(BaseHTTPRequestHandler):
 
         self.send_error(404)
 
+    def serve_www_asset(self, name, content_type):
+        """Serve one named file from www/ verbatim (no template injection)."""
+        for base in (os.path.dirname(__file__), os.path.dirname(os.path.dirname(__file__))):
+            candidate = os.path.join(base, 'www', name)
+            if os.path.exists(candidate):
+                with open(candidate, 'rb') as f:
+                    body = f.read()
+                self.send_response(200)
+                self.send_header('Content-Type', content_type)
+                self.send_header('Content-Length', len(body))
+                self.end_headers()
+                self.wfile.write(body)
+                return
+        self.send_response(404)
+        self.end_headers()
+
     def serve_setup_page(self):
         """Serve the setup/dashboard page."""
         html_path = os.path.join(os.path.dirname(__file__), 'www', 'index.html')
@@ -627,35 +654,42 @@ class Handler(BaseHTTPRequestHandler):
             self.wfile.write(content.encode('utf-8'))
 
     def handle_services_api(self):
-        """Return list of discovered services for dashboard navigation.
+        """Neighbours for the dashboard nav, from this peer's own beacon v2 view.
 
-        Fetches services from meta-core (single source of truth) instead of
+        No hop through meta-core any more: every service keeps its own
-        doing local filesystem-based discovery.
+        neighbour table, so the nav still renders when meta-core is down.
+        `?all=1` returns every instance, `?cap=<pattern>` filters by
+        capability (docs/project-architecture/beacon-v2.md). `services` is kept
+        as an alias of `neighbors` so the existing dashboards keep working.
         """
+        from urllib.parse import parse_qs, urlparse
+
         from storage.leader_client import get_leader_client
-        from urllib.request import urlopen
-        from urllib.error import URLError
 
         try:
-            leader_client = get_leader_client()
+            query = parse_qs(urlparse(self.path).query)
-            leader_info = leader_client.get_leader_info() if leader_client else None
+            client = get_leader_client()
-
+            neighbors = client.neighbors(
-            if leader_info and leader_info.api_url:
+                all=bool(query.get('all')),
-                # Fetch services from meta-core API
+                cap=(query.get('cap') or [None])[0],
-                url = f"{leader_info.api_url}/services?current=meta-stremio"
+            ) if client else []
-                with urlopen(url, timeout=5) as response:
+            return self.send_json({
-                    data = json.loads(response.read().decode())
+                'current': 'meta-stremio',
-                    return self.send_json(data)
+                'enabled': True,
-        except URLError as e:
+                'count': len(neighbors),
-            print(f"[Server] Error fetching services from meta-core: {e}")
+                'neighbors': neighbors,
+                'services': neighbors,
+                'self': client.self_announce() if client else None,
+            })
         except Exception as e:
             print(f"[Server] Error in handle_services_api: {e}")
 
-        # Fallback: return empty list if meta-core is unavailable
         return self.send_json({
-            'services': [],
             'current': 'meta-stremio',
+            'enabled': False,
             'count': 0,
+            'neighbors': [],
+            'services': [],
         })
 
     def handle_file(self, cid: str, width: int = None):
@@ -906,11 +940,14 @@ def shutdown_handler(signum, frame):
     """Handle graceful shutdown."""
     print("\n[Server] Shutting down...")
 
-    # Stop service discovery
+    # Stop service discovery. close() emits a final "stopping" announce so
+    # neighbours drop us immediately instead of waiting out the staleness
+    # window.
     try:
-        sd = get_service_discovery()
+        from storage.leader_client import get_leader_client
-        if sd:
+        client = get_leader_client()
-            sd.stop()
+        if client:
+            client.close()
     except Exception as e:
         print(f"[Server] Error stopping service discovery: {e}")
 

src/storage/__init__.py

@@ -13,15 +13,6 @@ from .provider import StorageProvider, VideoMetadata
 from .redis_storage import RedisStorage
 from .leader_storage import LeaderStorage
 from .leader_client import LeaderClient, LeaderLockInfo, get_leader_client
-from .service_registration import (
-    ServiceRegistration,
-    ServiceInfo,
-    get_service_discovery,  # Kept for backward compatibility
-    init_service_discovery  # Kept for backward compatibility
-)
-
-# Backward compatibility alias
-ServiceDiscovery = ServiceRegistration
 
 __all__ = [
     'StorageProvider',
@@ -31,9 +22,4 @@ __all__ = [
     'LeaderClient',
     'LeaderLockInfo',
     'get_leader_client',
-    'ServiceRegistration',
-    'ServiceDiscovery',  # Backward compatibility alias
-    'ServiceInfo',
-    'get_service_discovery',
-    'init_service_discovery',
 ]

src/storage/leader_client.py

@@ -1,40 +1,41 @@
+"""LeaderClient — locates meta-core and exposes its URLs.
+
+Since UDP discovery (meta-discovery v1, now beacon v2) this is a thin adapter
+over ``MetaCoreLocator`` (``meshdisco.py``): meta-core is found by UDP advertise
+instead of by reading ``/meta-core/locks/kv-leader.info``, so meta-stremio no
+longer needs the ``/meta-core`` volume mounted at all.
+
+The public surface is deliberately unchanged so ``LeaderStorage`` did not have
+to move: ``get_leader_info``, ``get_urls``, ``get_api_url``,
+``get_webdav_url*``, ``wait_for_leader``, ``start_watching`` / ``on_change``
+(the watchdog on the lock directory became "an announce arrived carrying a
+different apiUrl", feeding the same callback).
+
+Precedence is the pin: when ``meta_core_url`` (``META_CORE_URL``) is set it
+always wins and the wire is never consulted for core selection. See
+docs/project-architecture/service-discovery.md.
+
+The name "leader" is legacy: meta-core's flock election is vestigial, and
+``redis_url`` has been empty since the api-mediated-access lockdown — every
+read goes over HTTP.
 """
-LeaderClient - Client for reading leader info from meta-core
 
-This replaces the old LeaderDiscovery.py by delegating leader election
-to meta-core (Go sidecar) and just reading the results.
-
-Features:
-- Reads kv-leader.info file for leader URLs
-- Calls meta-core /urls API for current leader URLs
-- Watches for leader changes via file system watcher
-"""
-from __future__ import annotations
-
-import os
 import json
-import threading
+import os
+import time
+import urllib.request
 from dataclasses import dataclass
-from pathlib import Path
+from typing import Callable, List, Optional
-from typing import Optional, Callable
-from urllib.request import urlopen, Request
-from urllib.error import URLError
 
-# Try to import watchdog for file watching (optional)
+from .meshdisco import MetaCoreLocator
-try:
-    from watchdog.observers import Observer
-    from watchdog.events import FileSystemEventHandler
-    WATCHDOG_AVAILABLE = True
-except ImportError:
-    WATCHDOG_AVAILABLE = False
-    Observer = None
-    FileSystemEventHandler = object
 
 
 @dataclass
 class LeaderLockInfo:
-    """Leader lock file content format (kv-leader.info).
+    """meta-core's identity and URLs.
-    Written by meta-core leader, read by other services.
+
+    Historically the parsed kv-leader.info + /urls pair; now assembled from the
+    announce (or from /urls when meta-core is pinned).
     """
     hostname: str
     base_url: str
@@ -53,125 +54,133 @@ class URLsResponse:
     base_url: str
     api_url: str
     redis_url: str
-    webdav_url: str           # External WebDAV URL (via nginx/HTTPS)
+    webdav_url: str
-    webdav_url_internal: str  # Internal WebDAV URL (direct to port 9000)
+    webdav_url_internal: str
     is_leader: bool
 
 
-class LockFileHandler(FileSystemEventHandler):
-    """File system event handler for lock file changes."""
-
-    def __init__(self, callback: Callable[[], None], lock_filename: str):
-        self.callback = callback
-        self.lock_filename = lock_filename
-
-    def on_modified(self, event):
-        if event.src_path.endswith(self.lock_filename):
-            self.callback()
-
-    def on_created(self, event):
-        if event.src_path.endswith(self.lock_filename):
-            self.callback()
-
-    def on_deleted(self, event):
-        if event.src_path.endswith(self.lock_filename):
-            self.callback()
-
-
 class LeaderClient:
-    """
+    """Locates meta-core over UDP and resolves its URL set."""
-    LeaderClient for meta-stremio.
-
-    Reads leader info from kv-leader.info file written by meta-core.
-    """
 
     def __init__(
         self,
-        meta_core_path: str,
+        meta_core_path: Optional[str] = None,
-        meta_core_url: Optional[str] = None
+        meta_core_url: Optional[str] = None,
-    ):
+        service_name: str = "meta-stremio",
-        self.meta_core_path = meta_core_path
+        base_url: Optional[str] = None,
-        self.meta_core_url = meta_core_url
+        version: str = "",
+    ) -> None:
+        # meta_core_path is accepted and ignored — kept so existing callers
+        # construct unchanged. Nothing reads the volume any more.
+        self._meta_core_path = meta_core_path
+        self._meta_core_url = (meta_core_url or "").strip() or None
+        self._leader_info: Optional[LeaderLockInfo] = None
+        self._on_change_callbacks: List[Callable[[], None]] = []
+        self._started = False
 
-        self.info_file_path = os.path.join(meta_core_path, 'locks', 'kv-leader.info')
+        # Only used on the pinned path, where the URLs still come from an HTTP
-        self.leader_info: Optional[LeaderLockInfo] = None
+        # GET /urls. A discovered core carries them in the announce itself.
-
-        self._file_observer = None
-        self._on_change_callbacks: list[Callable[[], None]] = []
-        self._stop_event = threading.Event()
-
-        # URL caching
         self._cached_urls: Optional[URLsResponse] = None
-        self._urls_cache_time: float = 0
+        self._urls_cache_time: float = 0.0
-        self._urls_cache_ttl: float = 5.0  # 5 seconds
+        self._urls_cache_ttl: float = 5.0
 
-    def _get_api_url_from_file(self) -> Optional[str]:
+        self._locator = MetaCoreLocator(
-        """Read API URL from file (plain text format)."""
+            service_name=service_name,
-        try:
+            base_url=base_url,
-            with open(self.info_file_path, 'r') as f:
+            version=version,
-                content = f.read()
+            meta_core_url=self._meta_core_url,
-            return content.strip() or None
+        )
-        except FileNotFoundError:
+        self._locator.on_change(self._on_locator_change)
-            print(f"[LeaderClient] Leader info file not found: {self.info_file_path}")
+
-            return None
+    # -- lifecycle ---------------------------------------------------------
-        except Exception as e:
+
-            print(f"[LeaderClient] Error reading API URL from file: {e}")
+    def _ensure_started(self) -> None:
-            return None
+        if self._started:
+            return
+        self._started = True
+        self._locator.start()
+
+    def _on_locator_change(self) -> None:
+        self._cached_urls = None
+        self._urls_cache_time = 0.0
+        self._notify_change()
+
+    # -- url resolution ----------------------------------------------------
+
+    def _assert_no_redis_url(self, redis_url: Optional[str]) -> None:
+        """Guard against a rollback reintroducing direct Redis exposure.
+
+        api-mediated-access PR D removed redis_url; ALLOW_LEGACY_REDIS_URL=1
+        downgrades this to a warning during a deliberate temporary rollback.
+        """
+        if not redis_url:
+            return
+        msg = (
+            "meta-core still publishes redisUrl; direct Redis access was retired "
+            "by the api-mediated-access lockdown. Verify meta-core version."
+        )
+        if os.environ.get("ALLOW_LEGACY_REDIS_URL") == "1":
+            print(f"[LeaderClient] WARNING: {msg}")
+        else:
+            raise RuntimeError(msg)
 
     def _fetch_urls(self, api_url: str) -> Optional[URLsResponse]:
-        """Fetch URLs from meta-core /urls API with caching."""
-        import time
-
-        # Check cache
         now = time.time()
         if self._cached_urls and (now - self._urls_cache_time) < self._urls_cache_ttl:
             return self._cached_urls
-
         try:
-            url = f"{api_url}/urls"
+            req = urllib.request.Request(
-            req = Request(url, headers={'Accept': 'application/json'})
+                f"{api_url.rstrip('/')}/urls", headers={"Accept": "application/json"}
-            with urlopen(req, timeout=5) as response:
+            )
-                if response.status != 200:
+            with urllib.request.urlopen(req, timeout=5) as resp:
-                    print(f"[LeaderClient] Failed to fetch URLs: {response.status}")
+                data = json.loads(resp.read().decode("utf-8"))
-                    return None
+            self._assert_no_redis_url(data.get("redisUrl"))
-
-                data = json.loads(response.read().decode())
-
             self._cached_urls = URLsResponse(
-                    hostname=data.get('hostname', ''),
+                hostname=data.get("hostname", ""),
-                    base_url=data.get('baseUrl', ''),
+                base_url=data.get("baseUrl", ""),
-                    api_url=data.get('apiUrl', ''),
+                api_url=data.get("apiUrl", ""),
-                    redis_url=data.get('redisUrl', ''),
+                redis_url=data.get("redisUrl", "") or "",
-                    webdav_url=data.get('webdavUrl', ''),
+                webdav_url=data.get("webdavUrl", ""),
-                    webdav_url_internal=data.get('webdavUrlInternal', ''),
+                webdav_url_internal=data.get("webdavUrlInternal", ""),
-                    is_leader=data.get('isLeader', False)
+                is_leader=bool(data.get("isLeader", True)),
             )
             self._urls_cache_time = now
             return self._cached_urls
-
+        except Exception as e:
-        except URLError as e:
             print(f"[LeaderClient] Error calling /urls API: {e}")
             return None
-        except Exception as e:
-            print(f"[LeaderClient] Error fetching URLs: {e}")
-            return None
 
-    def get_leader_info(self) -> Optional[LeaderLockInfo]:
+    def get_urls(self) -> Optional[URLsResponse]:
-        """Read leader info from file and /urls API."""
+        """Resolve meta-core's URLs.
-        try:
+
-            # Read API URL from file (plain text)
+        Straight from the announce when discovered (no HTTP hop needed); over
-            api_url = self._get_api_url_from_file()
+        HTTP when meta-core is pinned.
+        """
+        self._ensure_started()
+
+        wire = self._locator.get_urls()
+        if wire and not self._meta_core_url:
+            return URLsResponse(
+                hostname=wire.get("hostname", ""),
+                base_url=wire.get("baseUrl", ""),
+                api_url=wire.get("apiUrl", ""),
+                redis_url="",
+                webdav_url=wire.get("webdavUrl", ""),
+                webdav_url_internal=wire.get("webdavUrlInternal", ""),
+                is_leader=True,
+            )
+
+        api_url = self._locator.get_api_url()
         if not api_url:
             return None
+        return self._fetch_urls(api_url)
 
-            # Fetch full info from /urls API
+    def get_leader_info(self) -> Optional[LeaderLockInfo]:
-            urls = self._fetch_urls(api_url)
+        try:
+            urls = self.get_urls()
             if not urls:
                 return None
-
+            self._leader_info = LeaderLockInfo(
-            # Convert URLsResponse to LeaderLockInfo
-            import time
-            self.leader_info = LeaderLockInfo(
                 hostname=urls.hostname,
                 base_url=urls.base_url,
                 api_url=urls.api_url,
@@ -179,145 +188,111 @@ class LeaderClient:
                 webdav_url=urls.webdav_url,
                 webdav_url_internal=urls.webdav_url_internal,
                 timestamp=int(time.time() * 1000),
-                pid=0  # Unknown for remote leader
+                pid=0,
             )
-
+            return self._leader_info
-            return self.leader_info
-
         except Exception as e:
-            print(f"[LeaderClient] Error reading leader info: {e}")
+            print(f"[LeaderClient] Failed to read leader info: {e}")
             return None
 
     def get_redis_url(self) -> Optional[str]:
-        """Get Redis URL from leader info."""
         info = self.get_leader_info()
         return info.redis_url if info else None
 
     def get_webdav_url(self) -> Optional[str]:
-        """Get WebDAV URL from leader info (external, via nginx/HTTPS)."""
         info = self.get_leader_info()
         return info.webdav_url if info else None
 
     def get_webdav_url_internal(self) -> Optional[str]:
-        """Get internal WebDAV URL from leader info (direct to port 9000).
+        """Internal WebDAV URL — use for container-to-container access."""
-        Use this for container-to-container communication.
-        """
         info = self.get_leader_info()
         return info.webdav_url_internal if info else None
 
     def get_api_url(self) -> Optional[str]:
-        """Get meta-core API URL from leader info."""
         info = self.get_leader_info()
         return info.api_url if info else None
 
-    def get_urls(self) -> Optional[URLsResponse]:
-        """Call meta-core /urls API to get current URLs."""
-        # First try using configured meta_core_url
-        api_url = self.meta_core_url
-
-        # Fall back to reading from file
-        if not api_url:
-            api_url = self._get_api_url_from_file()
-
-        if not api_url:
-            print("[LeaderClient] No meta-core API URL available")
-            return None
-
-        return self._fetch_urls(api_url)
-
     def wait_for_leader(self, timeout_ms: int = 30000) -> LeaderLockInfo:
-        """Wait for leader info to be available."""
+        """Block until meta-core is reachable.
-        import time
-        start_time = time.time()
-        poll_interval = 0.5  # 500ms
 
-        while (time.time() - start_time) * 1000 < timeout_ms:
+        Probes immediately then every 500ms, matching the previous
+        file-polling loop's timing.
+        """
+        self._ensure_started()
+        timeout_s = None if timeout_ms <= 0 else timeout_ms / 1000.0
+        deadline = None if timeout_s is None else time.time() + timeout_s
+
+        if self._locator.wait_for_core(timeout_s) is None:
+            raise TimeoutError(f"No meta-core found within {timeout_ms}ms")
+
+        while deadline is None or time.time() < deadline:
             info = self.get_leader_info()
             if info:
-                print(f"[LeaderClient] Leader found: {info.hostname} at {info.redis_url}")
+                print(f"[LeaderClient] meta-core found: {info.hostname} at {info.api_url}")
                 return info
+            time.sleep(0.5)
+        raise TimeoutError(f"No meta-core found within {timeout_ms}ms")
 
-            print("[LeaderClient] Waiting for leader...")
+    # -- change notification ----------------------------------------------
-            time.sleep(poll_interval)
-
-        raise TimeoutError(f"No leader found within {timeout_ms}ms")
 
     def start_watching(self) -> None:
-        """Start watching for leader changes."""
+        """Previously a watchdog on the lock directory.
-        if not WATCHDOG_AVAILABLE:
-            print("[LeaderClient] watchdog not available, file watching disabled")
-            return
 
-        try:
+        Discovery is always listening, so this only has to make sure the node
-            lock_dir = os.path.dirname(self.info_file_path)
+        is running.
-
+        """
-            handler = LockFileHandler(
+        self._ensure_started()
-                callback=self._on_file_change,
-                lock_filename='kv-leader.info'
-            )
-
-            self._file_observer = Observer()
-            self._file_observer.schedule(handler, lock_dir, recursive=False)
-            self._file_observer.start()
-            print(f"[LeaderClient] Watching for leader changes in {lock_dir}")
-
-        except Exception as e:
-            print(f"[LeaderClient] Could not start file watcher: {e}")
 
     def stop_watching(self) -> None:
-        """Stop watching for leader changes."""
+        """No-op: the locator keeps listening until close()."""
-        if self._file_observer:
-            self._file_observer.stop()
-            self._file_observer.join(timeout=2.0)
-            self._file_observer = None
-            print("[LeaderClient] Stopped watching for leader changes")
 
-    def _on_file_change(self) -> None:
+    def on_change(self, callback: Callable[[], None]) -> "LeaderClient":
-        """Handle file change event."""
-        print("[LeaderClient] Leader info changed, invalidating cache...")
-        # Invalidate cache to force fresh API call
-        self._cached_urls = None
-        self._urls_cache_time = 0
-
-        self.get_leader_info()
-        self._notify_change()
-
-    def on_change(self, callback: Callable[[], None]) -> 'LeaderClient':
-        """Register callback for leader changes."""
         self._on_change_callbacks.append(callback)
         return self
 
     def _notify_change(self) -> None:
-        """Notify all change callbacks."""
+        print("[LeaderClient] meta-core changed, invalidating cache...")
-        for callback in self._on_change_callbacks:
+        for cb in list(self._on_change_callbacks):
             try:
-                callback()
+                cb()
             except Exception as e:
                 print(f"[LeaderClient] Error in change callback: {e}")
 
     def get_cached_leader_info(self) -> Optional[LeaderLockInfo]:
-        """Get cached leader info (without re-reading file)."""
+        return self._leader_info
-        return self.leader_info
+
+    # -- nav menu ----------------------------------------------------------
+
+    def neighbors(self, all: bool = False, cap: Optional[str] = None) -> List[dict]:
+        """Neighbours for /api/neighbors: one row per name (every instance with
+        ``all``), optionally only nodes with a capability matching ``cap``."""
+        self._ensure_started()
+        return self._locator.neighbors(all=all, cap=cap)
+
+    def self_announce(self) -> dict:
+        """This service's own announce, so the menu can show itself."""
+        self._ensure_started()
+        return self._locator.self_announce()
 
     def close(self) -> None:
-        """Clean up resources."""
-        self._stop_event.set()
-        self.stop_watching()
         self._on_change_callbacks = []
+        self._locator.stop()
+        self._started = False
 
 
-# Singleton instance
 _leader_client: Optional[LeaderClient] = None
 
 
 def get_leader_client() -> Optional[LeaderClient]:
-    """Get or create the LeaderClient singleton.
+    """Process-wide singleton, so one UDP socket serves every caller."""
-
-    Returns None if META_CORE_PATH is not set.
-    """
     global _leader_client
     if _leader_client is None:
-        meta_core_path = os.environ.get('META_CORE_PATH', '/meta-core')
+        _leader_client = LeaderClient(
-        meta_core_url = os.environ.get('META_CORE_URL')
+            meta_core_path=os.environ.get("META_CORE_PATH"),
-        _leader_client = LeaderClient(meta_core_path, meta_core_url)
+            meta_core_url=os.environ.get("META_CORE_URL"),
+            # PUBLIC_URL (a browser-reachable debug-direct port) wins over
+            # BASE_URL (the Caddy URL), same ladder as meta-core. Announcing a
+            # Caddy URL on a stack with no Caddy puts a dead link in every menu.
+            base_url=os.environ.get("PUBLIC_URL") or os.environ.get("BASE_URL") or None,
+        )
     return _leader_client

src/storage/leader_storage.py

@@ -18,8 +18,9 @@ import time
 from typing import List, Optional, Callable
 
 from .provider import StorageProvider, VideoMetadata
-from .leader_client import LeaderClient, LeaderLockInfo
+from .leader_client import LeaderClient, LeaderLockInfo, get_leader_client
 from .meta_consumer import MetaConsumer
+from .meta_core_api_client import MetaCoreApiClient
 
 # Import webdav_client to configure it after leader discovery
 import webdav_client
@@ -70,6 +71,7 @@ class LeaderStorage(StorageProvider):
 
         self._leader_client: Optional[LeaderClient] = None
         self._client: Optional[redis.Redis] = None
+        self._api_client: Optional[MetaCoreApiClient] = None
         self._connected = False
         self._lock = threading.Lock()
 
@@ -95,10 +97,11 @@ class LeaderStorage(StorageProvider):
             self._connect_to_redis(self._override_url)
             return
 
-        # Use leader client
+        # Use the process-wide leader client so meta-stremio holds ONE UDP
-        self._leader_client = LeaderClient(
+        # discovery socket, shared with server.py's neighbours endpoint.
-            meta_core_path=self._meta_core_path
+        # META_CORE_URL, when set, pins meta-core and discovery never
-        )
+        # overrides it — see docs/project-architecture/service-discovery.md.
+        self._leader_client = get_leader_client()
 
         # Set up change callback
         self._leader_client.on_change(self._on_leader_change)
@@ -122,9 +125,14 @@ class LeaderStorage(StorageProvider):
                 # Try to get leader info (short timeout per attempt)
                 info = self._leader_client.get_leader_info()
 
-                if info and info.redis_url:
+                # api-mediated-access PR D: when redis_url is absent, run in
-                    print(f"[LeaderStorage] Found leader at {info.redis_url} (attempt {attempt})")
+                # HTTP-only mode against meta-core's /meta/* surface. Redis
-                    self._connect_to_redis(info.redis_url)
+                # access has been retired; the API client covers every read
+                # this storage provider needs. Only fall back to Redis when
+                # both redis_url and api_url are present (transition state).
+                if info and info.api_url:
+                    print(f"[LeaderStorage] Found leader at {info.api_url} (attempt {attempt})")
+                    self._connect_via_api(info)
 
                     if self._connected:
                         # Configure WebDAV client with leader's internal WebDAV URL (for container-to-container access)
@@ -140,7 +148,7 @@ class LeaderStorage(StorageProvider):
                         self._leader_client.start_watching()
                         return
                     else:
-                        print(f"[LeaderStorage] Failed to connect to Redis, retrying...")
+                        print(f"[LeaderStorage] Failed to connect to meta-core, retrying...")
                 else:
                     print(f"[LeaderStorage] Waiting for meta-core leader... (attempt {attempt})")
 
@@ -174,9 +182,9 @@ class LeaderStorage(StorageProvider):
             try:
                 info = self._leader_client.get_leader_info()
 
-                if info and info.redis_url:
+                if info and info.api_url:
-                    print(f"[LeaderStorage] Background: Found leader at {info.redis_url}")
+                    print(f"[LeaderStorage] Background: Found leader at {info.api_url}")
-                    self._connect_to_redis(info.redis_url)
+                    self._connect_via_api(info)
 
                     if self._connected:
                         # Configure WebDAV client
@@ -214,9 +222,30 @@ class LeaderStorage(StorageProvider):
         self._disconnect_redis()
 
     def is_connected(self) -> bool:
-        """Check if connected to the storage backend."""
+        """Check if connected to the storage backend (Redis or HTTP API).
+
+        HTTP-only mode auto-recovers from transient meta-core outages: if
+        the health probe fails we drop the connected flag but keep the
+        api_client so the *next* call probes again. The background
+        reconnect thread is only needed for Redis-mode where the
+        connection itself is sticky.
+        """
         with self._lock:
-            if not self._connected or not self._client:
+            # HTTP-only mode: probe meta-core's /health on every call. This
+            # is cheap (a single HTTP GET) and naturally recovers when
+            # meta-core comes back without needing a separate reconnect
+            # loop. The api_client object stays valid through transient
+            # failures — it's stateless aside from the configured URL.
+            if self._api_client is not None and self._client is None:
+                if self._api_client.health():
+                    self._connected = True
+                    return True
+                self._connected = False
+                return False
+            # Legacy Redis mode.
+            if not self._connected:
+                return False
+            if not self._client:
                 return False
             try:
                 self._client.ping()
@@ -241,9 +270,9 @@ class LeaderStorage(StorageProvider):
                 try:
                     info = self._leader_client.get_leader_info()
 
-                    if info and info.redis_url:
+                    if info and info.api_url:
-                        print(f"[LeaderStorage] Reconnecting to leader at {info.redis_url}...")
+                        print(f"[LeaderStorage] Reconnecting to leader at {info.api_url}...")
-                        self._connect_to_redis(info.redis_url)
+                        self._connect_via_api(info)
 
                         if self._connected:
                             # Reconfigure WebDAV client with new leader's internal WebDAV URL
@@ -260,8 +289,46 @@ class LeaderStorage(StorageProvider):
 
             print(f"[LeaderStorage] Failed to reconnect after {max_attempts} attempts")
 
+    def _connect_via_api(self, info: LeaderLockInfo) -> None:
+        """Connect to meta-core via its HTTP API. The api_client covers every
+        read this storage provider does — Redis access is no longer required
+        (api-mediated-access PR D). Also starts the SSE meta-events consumer
+        so cache invalidation still flows in real time.
+        """
+        with self._lock:
+            try:
+                # Drop any prior state (Redis or otherwise)
+                self._disconnect_redis_unlocked()
+
+                # Set up the HTTP read client + probe meta-core.
+                api = MetaCoreApiClient(api_url=info.api_url)
+                if not api.health():
+                    raise RuntimeError(f"meta-core health probe failed at {info.api_url}")
+                self._api_client = api
+                self._connected = True
+
+                print(f"[LeaderStorage] Connected via meta-core API at {info.api_url}")
+
+                # Start SSE consumer for cache invalidation.
+                try:
+                    self._meta_consumer = MetaConsumer(api_url=info.api_url)
+                    self._meta_consumer.on_change(self._on_metadata_change)
+                    self._meta_consumer.start()
+                except Exception as e:
+                    print(f"[LeaderStorage] Warning: failed to start meta consumer: {e}")
+
+                self._cache_dirty = True
+                self._notify_ready()
+
+            except Exception as e:
+                print(f"[LeaderStorage] Failed to connect via API: {e}")
+                self._api_client = None
+                self._connected = False
+
     def _connect_to_redis(self, url: str) -> None:
-        """Connect to Redis."""
+        """Connect to Redis. Legacy path — kept for direct REDIS_URL overrides
+        and ad-hoc tests. Production discovery goes through _connect_via_api.
+        """
         with self._lock:
             try:
                 # Disconnect existing client
@@ -274,11 +341,19 @@ class LeaderStorage(StorageProvider):
 
                 print(f"[LeaderStorage] Connected to Redis at {url}")
 
-                # Start meta events consumer for real-time updates
+                # Start meta events consumer (HTTP SSE against meta-core).
+                # Prefer the api URL we discovered via the LeaderClient; fall
+                # back to META_CORE_URL env so STORAGE_MODE=direct deployments
+                # (no leader discovery) still get live updates.
                 try:
-                    self._meta_consumer = MetaConsumer(self._client, self._prefix)
+                    info = self._leader_client.get_cached_leader_info() if self._leader_client else None
+                    api_url = (info.api_url if info else None) or os.environ.get("META_CORE_URL")
+                    if api_url:
+                        self._meta_consumer = MetaConsumer(api_url=api_url)
                         self._meta_consumer.on_change(self._on_metadata_change)
                         self._meta_consumer.start()
+                    else:
+                        print("[LeaderStorage] Warning: api_url unavailable; meta consumer not started")
                 except Exception as e:
                     print(f"[LeaderStorage] Warning: failed to start meta consumer: {e}")
 
@@ -299,7 +374,7 @@ class LeaderStorage(StorageProvider):
             self._disconnect_redis_unlocked()
 
     def _disconnect_redis_unlocked(self) -> None:
-        """Disconnect from Redis (without lock)."""
+        """Disconnect from Redis and/or API (without lock)."""
         # Stop meta consumer
         if self._meta_consumer:
             try:
@@ -314,6 +389,10 @@ class LeaderStorage(StorageProvider):
             except Exception:
                 pass
             self._client = None
+
+        # The API client has no socket; just drop the reference.
+        self._api_client = None
+
         self._connected = False
 
         # Clear cache
@@ -377,10 +456,14 @@ class LeaderStorage(StorageProvider):
         return f"{self._prefix}file:{hash_id}/"
 
     def _get_file_metadata(self, hash_id: str) -> dict:
-        """Get all metadata for a file using flat key scan."""
+        """Get all metadata for a file. Uses meta-core HTTP API when in
+        HTTP-only mode; falls back to Redis SCAN+GET otherwise."""
+        if self._api_client is not None:
+            flat = self._api_client.get_metadata_flat(hash_id)
+            return dict(flat) if flat else {}
+
         prefix = self._get_file_key_prefix(hash_id)
         data = {}
-
         for key in self._client.scan_iter(f"{prefix}*", count=100):
             key_str = key if isinstance(key, str) else key.decode()
             field = key_str[len(prefix):]
@@ -494,10 +577,9 @@ class LeaderStorage(StorageProvider):
         # Parse release date
         release_date = data.get('releasedate', data.get('releaseDate'))
 
-        # Parse episode title - try episodeTitle or extract from titles/eng
+        # Parse episode title. `titles/*` is the work's name set, never an
+        # episode title (METADATA_KEYS.md §3).
         episode_title = data.get('episodeTitle', data.get('episodeName'))
-        if not episode_title:
-            episode_title = data.get('titles/eng', data.get('titles/en'))
 
         # Parse tagline
         tagline = data.get('tagline')
@@ -577,6 +659,15 @@ class LeaderStorage(StorageProvider):
     # StorageProvider Interface
     # ========================================================================
 
+    def _list_hash_ids(self) -> List[str]:
+        """Enumerate every hashId. HTTP API in api-mediated mode, SMEMBERS
+        on file:__index__ in legacy Redis mode."""
+        if self._api_client is not None:
+            return self._api_client.list_hash_ids()
+        index_key = f"{self._prefix}file:__index__"
+        ids = self._client.smembers(index_key)
+        return [h if isinstance(h, str) else h.decode() for h in ids]
+
     def get_all_videos(self) -> List[VideoMetadata]:
         """Get all videos from storage."""
         if not self.is_connected():
@@ -586,8 +677,7 @@ class LeaderStorage(StorageProvider):
 
         try:
             # Get all hash IDs from index
-            index_key = f"{self._prefix}file:__index__"
+            hash_ids = self._list_hash_ids()
-            hash_ids = self._client.smembers(index_key)
 
             for hash_id in hash_ids:
                 hash_id_str = hash_id if isinstance(hash_id, str) else hash_id.decode()
@@ -648,16 +738,9 @@ class LeaderStorage(StorageProvider):
 
         try:
             # Get from index and check each file's imdbid
-            index_key = f"{self._prefix}file:__index__"
+            for hash_id_str in self._list_hash_ids():
-            for hash_id in self._client.smembers(index_key):
+                stored_imdb = self._get_imdb_field(hash_id_str)
-                hash_id_str = hash_id if isinstance(hash_id, str) else hash_id.decode()
+                if stored_imdb and stored_imdb.lower() == imdb_id:
-                prefix = self._get_file_key_prefix(hash_id_str)
-
-                # Check imdbid field directly
-                stored_imdb = self._client.get(f"{prefix}imdbid") or self._client.get(f"{prefix}imdbId")
-                if stored_imdb:
-                    stored_imdb_str = stored_imdb if isinstance(stored_imdb, str) else stored_imdb.decode()
-                    if stored_imdb_str.lower() == imdb_id:
                     data = self._get_file_metadata(hash_id_str)
                     return self._parse_video(hash_id_str, data)
 
@@ -666,13 +749,28 @@ class LeaderStorage(StorageProvider):
 
         return None
 
+    def _get_imdb_field(self, hash_id: str) -> Optional[str]:
+        """Read imdbid (or imdbId) for a file. HTTP API when in api-mediated
+        mode, direct Redis GET otherwise."""
+        if self._api_client is not None:
+            v = self._api_client.get_property(hash_id, "imdbid")
+            if v:
+                return v
+            return self._api_client.get_property(hash_id, "imdbId")
+        prefix = self._get_file_key_prefix(hash_id)
+        stored_imdb = self._client.get(f"{prefix}imdbid") or self._client.get(f"{prefix}imdbId")
+        if not stored_imdb:
+            return None
+        return stored_imdb if isinstance(stored_imdb, str) else stored_imdb.decode()
+
     def get_file_path_by_cid(self, cid: str) -> Optional[str]:
         """
         Get the file path for a file by its CID.
 
-        Looks up file:{cid}/* in Redis and returns the path.
+        In api-mediated mode this hits meta-core's GET /api/file/{cid}/info
-        Tries 'path' field first, then falls back to 'filePath'.
+        which already does the reverse-index lookup + canonical-path
-        This works for any file including poster images.
+        resolution. In Redis mode we fall back to reading file:{cid}/path
+        and file:{cid}/filePath directly.
 
         Returns the relative path if found, None otherwise.
         """
@@ -680,6 +778,11 @@ class LeaderStorage(StorageProvider):
             return None
 
         try:
+            if self._api_client is not None:
+                # /api/file/{cid}/info already returns a path relative to
+                # FILES_PATH, normalised. No further trimming required.
+                return self._api_client.resolve_file_path(cid)
+
             prefix = self._get_file_key_prefix(cid)
             # Try 'path' field first (relative path)
             path = self._client.get(f"{prefix}path")
@@ -704,6 +807,8 @@ class LeaderStorage(StorageProvider):
             return 0
 
         try:
+            if self._api_client is not None:
+                return self._api_client.count_hash_ids()
             # Use index set for accurate count
             index_key = f"{self._prefix}file:__index__"
             return self._client.scard(index_key)
@@ -740,8 +845,11 @@ class LeaderStorage(StorageProvider):
         if status['connected']:
             try:
                 status['video_count'] = self.get_video_count()
-                info = self._client.info('memory')
+                # Memory stats only available in Redis-direct mode; the HTTP
-                status['memory_used'] = info.get('used_memory_human', 'N/A')
+                # API doesn't surface them. Skip cleanly when unavailable.
+                if self._client is not None:
+                    mem = self._client.info('memory')
+                    status['memory_used'] = mem.get('used_memory_human', 'N/A')
             except Exception:
                 pass
 

src/storage/meshdisco.py

@@ -0,0 +1,531 @@
+"""Beacon v2 — local resource advertise / discover over UDP multicast
+(239.255.99.1:9099).
+
+Every node advertises resources tagged with capability strings
+(``metamesh.core``, ``metamesh.service/meta-stremio``,
+``metamesh.transport/nzb@1``) and keeps a live view of everyone else's. This
+service uses it to locate meta-core (the ``metamesh.core`` resource carries its
+/urls block) and to feed the nav menu, with no shared volume.
+
+The normative spec is docs/project-architecture/beacon-v2.md.
+
+Two details this module must get right, both of which fail silently:
+
+1. **Per-interface fan-out.** A socket that joins with ``INADDR_ANY`` and sends
+   via the default route covers exactly one interface. Containers here are
+   routinely attached to two docker networks, so we enumerate interfaces and
+   join/send on each using a hand-packed ``ip_mreqn``.
+2. **The pin wins.** When ``META_CORE_URL`` is set, the wire is never consulted
+   for core selection — otherwise a container that can see two cores could
+   latch onto the wrong one.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import socket
+import struct
+import threading
+import time
+from typing import Callable, Dict, List, Optional
+
+PROTO = "beacon"
+VERSION = 2
+DEFAULT_GROUP = "239.255.99.1"
+DEFAULT_PORT = 9099
+DEFAULT_INTERVAL = 10.0
+# A node unheard for interval * this is dropped (evaluated at read time).
+LIVENESS_FACTOR = 3
+READ_BUFFER = 8192
+# Senders stay under this so nothing fragments.
+MAX_DATAGRAM = 1400
+
+TYPE_PROBE = "probe"
+TYPE_ADVERTISE = "advertise"
+TYPE_BYE = "bye"
+
+CAP_CORE = "metamesh.core"
+CAP_SERVICE_PREFIX = "metamesh.service/"
+CAP_ANY_SERVICE = "metamesh.service/*"
+
+
+def cap_matches(pattern: str, cap: str) -> bool:
+    """Does capability ``cap`` satisfy ``pattern``?
+
+    ``*`` matches everything; an ``@N`` on the pattern must equal the cap's
+    contract (none ignores it); ``x/*`` matches any variant of ``x``; otherwise
+    the bases must be equal.
+    """
+    if pattern == "*":
+        return True
+    pbase, _, pc = pattern.rpartition("@") if "@" in pattern else (pattern, "", None)
+    cbase, _, cc = cap.rpartition("@") if "@" in cap else (cap, "", None)
+    if pc is not None and cc != pc:
+        return False
+    if pbase.endswith("/*"):
+        prefix = pbase[:-2]
+        if not cbase.startswith(prefix):
+            return False
+        rest = cbase[len(prefix):]
+        return rest.startswith("/") and len(rest) > 1
+    return pbase == cbase
+
+
+def resource_matches(resource: dict, pattern: str) -> bool:
+    return any(cap_matches(pattern, c) for c in resource.get("caps") or [])
+
+
+def resource_endpoint(resource: dict, name: str) -> Optional[str]:
+    """The named endpoint as an absolute URL ('/…' joins onto endpoints.http)."""
+    endpoints = resource.get("endpoints") or {}
+    v = endpoints.get(name)
+    if not v:
+        return None
+    if not v.startswith("/"):
+        return v
+    base = endpoints.get("http")
+    return base.rstrip("/") + v if base else None
+
+
+def parse_message(data: bytes) -> Optional[dict]:
+    """Decode a datagram; None for anything that is not well-formed beacon v2
+    (beacon v1, meta-discovery v1, other versions, junk)."""
+    try:
+        m = json.loads(data.decode("utf-8"))
+    except (ValueError, UnicodeDecodeError):
+        return None
+    if not isinstance(m, dict) or m.get("proto") != PROTO or m.get("v") != VERSION:
+        return None
+    t = m.get("type")
+    if t == TYPE_PROBE:
+        return m
+    if t in (TYPE_ADVERTISE, TYPE_BYE):
+        node = m.get("node")
+        if not isinstance(node, dict) or not node.get("name") or not node.get("instance"):
+            return None
+        return m
+    return None
+
+
+def _neighbor_row(node: dict, resources: List[dict], addr: str, last_seen: float) -> dict:
+    caps: List[str] = []
+    base_url = None
+    for r in resources:
+        for c in r.get("caps") or []:
+            if c not in caps:
+                caps.append(c)
+        if base_url is None and resource_matches(r, CAP_ANY_SERVICE):
+            base_url = resource_endpoint(r, "ui")
+    row = dict(node)
+    row.update({"resources": resources, "caps": caps, "addr": addr, "lastSeen": last_seen})
+    if base_url:
+        row["baseUrl"] = base_url
+    return row
+
+
+def _env_flag(name: str, default: bool = True) -> bool:
+    raw = os.environ.get(name)
+    if raw is None or raw == "":
+        return default
+    return raw.strip().lower() not in ("false", "0", "no", "off")
+
+
+def _interface_indexes() -> List[tuple]:
+    """(name, index, ipv4) for every up, non-loopback interface.
+
+    Falls back to a single unspecified entry when the platform cannot
+    enumerate, so discovery still works on the default route.
+    """
+    out: List[tuple] = []
+    try:
+        for idx, name in socket.if_nameindex():
+            if name == "lo":
+                continue
+            try:
+                # Cheap way to get the interface's IPv4 without netifaces.
+                addr = _ipv4_of(name)
+            except OSError:
+                continue
+            if addr:
+                out.append((name, idx, addr))
+    except (AttributeError, OSError):
+        pass
+    return out
+
+
+def _ipv4_of(ifname: str) -> Optional[str]:
+    """SIOCGIFADDR — avoids a netifaces dependency."""
+    import fcntl
+
+    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    try:
+        packed = struct.pack("256s", ifname[:15].encode("utf-8"))
+        return socket.inet_ntoa(fcntl.ioctl(s.fileno(), 0x8915, packed)[20:24])
+    except OSError:
+        return None
+    finally:
+        s.close()
+
+
+def local_ipv4() -> str:
+    """First non-loopback IPv4, else the hostname. Matches the Go helper."""
+    ifaces = _interface_indexes()
+    if ifaces:
+        return ifaces[0][2]
+    return socket.gethostname()
+
+
+class MeshNode:
+    """One beacon v2 participant: advertises its resources, answers probes,
+    and keeps a TTL map of every other node."""
+
+    def __init__(
+        self,
+        name: str,
+        instance: Optional[str] = None,
+        version: str = "",
+        group: Optional[str] = None,
+        port: Optional[int] = None,
+        interval: Optional[float] = None,
+        enabled: bool = True,
+        payload: Optional[Callable[[], dict]] = None,
+    ) -> None:
+        self.name = name
+        self.instance = instance or socket.gethostname()
+        self.version = version
+        self.group = group or os.environ.get("BEACON_GROUP") or DEFAULT_GROUP
+        self.port = port or int(os.environ.get("BEACON_PORT") or DEFAULT_PORT)
+        self.interval = interval or (float(os.environ.get("BEACON_INTERVAL_MS") or 0) / 1000.0) or DEFAULT_INTERVAL
+        self.enabled = enabled
+        # payload() -> {"status": str?, "resources": [resource, ...]}
+        self._payload = payload
+
+        self._sock: Optional[socket.socket] = None
+        self._ifaces: List[tuple] = []
+        self._nodes: Dict[str, dict] = {}
+        self._lock = threading.Lock()
+        self._send_lock = threading.Lock()
+        self._running = False
+        self._threads: List[threading.Thread] = []
+        self._listeners: List[Callable[[dict], None]] = []
+
+    # -- lifecycle ---------------------------------------------------------
+
+    def start(self) -> None:
+        if self._running or not self.enabled:
+            if not self.enabled:
+                print("[beacon] Disabled by configuration")
+            return
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        try:
+            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
+        except (AttributeError, OSError):
+            pass
+        sock.bind(("0.0.0.0", self.port))
+        # TTL 1: link-local only, never routed.
+        sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 1)
+
+        self._ifaces = _interface_indexes()
+        group_bin = socket.inet_aton(self.group)
+        joined = 0
+        for ifname, idx, addr in self._ifaces:
+            try:
+                # ip_mreqn: multiaddr, address (INADDR_ANY), ifindex. The
+                # ifindex is what makes this a per-interface join instead of
+                # "whatever the routing table picks".
+                mreq = group_bin + socket.inet_aton("0.0.0.0") + struct.pack("@i", idx)
+                sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
+                joined += 1
+            except OSError as e:
+                print(f"[beacon] Warning: join {self.group} on {ifname} failed: {e}")
+
+        self._sock = sock
+        self._running = True
+        print(
+            f"[beacon] Listening on {self.group}:{self.port} as "
+            f"{self.name}/{self.instance} (joined {joined} of {len(self._ifaces)} interfaces)"
+        )
+
+        for target in (self._read_loop, self._advertise_loop):
+            t = threading.Thread(target=target, daemon=True)
+            t.start()
+            self._threads.append(t)
+
+    def stop(self) -> None:
+        if not self._running:
+            return
+        self._running = False
+        # Bye, so neighbours drop us immediately rather than waiting out the
+        # staleness window.
+        try:
+            self._send({"proto": PROTO, "v": VERSION, "type": TYPE_BYE, "node": self._node_info()})
+        except OSError:
+            pass
+        if self._sock:
+            try:
+                self._sock.close()
+            except OSError:
+                pass
+        self._sock = None
+
+    # -- api ---------------------------------------------------------------
+
+    def probe(self, want: Optional[List[str]] = None) -> None:
+        """Multicast a probe. Every node owning a resource matching one of
+        ``want`` (every node, when empty) replies immediately."""
+        msg = {"proto": PROTO, "v": VERSION, "type": TYPE_PROBE, "from": self.instance}
+        if want:
+            msg["want"] = list(want)
+        self._send(msg)
+
+    def neighbors(self) -> List[dict]:
+        """Every live node, sorted by (name, instance)."""
+        cutoff = time.time() - self.interval * LIVENESS_FACTOR
+        with self._lock:
+            for key in [k for k, v in self._nodes.items() if v["lastSeen"] < cutoff]:
+                del self._nodes[key]
+            out = [_neighbor_row(v["node"], v["resources"], v["addr"], v["lastSeen"]) for v in self._nodes.values()]
+        return sorted(out, key=lambda n: (n.get("name") or "", n.get("instance") or ""))
+
+    def neighbors_by_name(self) -> List[dict]:
+        """One row per node name, most recently seen instance wins."""
+        best: Dict[str, dict] = {}
+        for nb in self.neighbors():
+            cur = best.get(nb.get("name") or "")
+            if cur is None or nb["lastSeen"] > cur["lastSeen"]:
+                best[nb.get("name") or ""] = nb
+        return sorted(best.values(), key=lambda n: n.get("name") or "")
+
+    def self_announce(self) -> dict:
+        """This node as a row, so a UI can render itself without waiting."""
+        status, resources = self._get_payload()
+        return _neighbor_row(self._node_info(status), resources, "", time.time())
+
+    def on_advertise(self, cb: Callable[[dict], None]) -> None:
+        self._listeners.append(cb)
+
+    # -- internals ---------------------------------------------------------
+
+    def _get_payload(self) -> tuple:
+        if not self._payload:
+            return None, []
+        try:
+            p = self._payload() or {}
+        except Exception:
+            p = {}
+        return p.get("status"), list(p.get("resources") or [])
+
+    def _node_info(self, status: Optional[str] = None) -> dict:
+        n = {"name": self.name, "instance": self.instance}
+        if self.version:
+            n["version"] = self.version
+        if status:
+            n["status"] = status
+        return n
+
+    def _advertise_msg(self) -> dict:
+        status, resources = self._get_payload()
+        return {
+            "proto": PROTO, "v": VERSION, "type": TYPE_ADVERTISE,
+            "node": self._node_info(status), "resources": resources,
+        }
+
+    def _send(self, msg: dict, dest: Optional[tuple] = None) -> None:
+        sock = self._sock
+        if sock is None:
+            return
+        body = json.dumps(msg).encode("utf-8")
+        if len(body) > MAX_DATAGRAM:
+            print(f"[beacon] Warning: {len(body)}-byte datagram exceeds the fragmentation-safe {MAX_DATAGRAM}")
+        with self._send_lock:
+            if dest is not None:
+                try:
+                    sock.sendto(body, dest)
+                except OSError as e:
+                    print(f"[beacon] Reply to {dest} failed: {e}")
+                return
+            # One write per interface — the default route alone reaches a
+            # single network.
+            for ifname, idx, addr in self._ifaces:
+                try:
+                    sock.setsockopt(
+                        socket.IPPROTO_IP,
+                        socket.IP_MULTICAST_IF,
+                        socket.inet_aton(addr) + socket.inet_aton("0.0.0.0") + struct.pack("@i", idx),
+                    )
+                    sock.sendto(body, (self.group, self.port))
+                except OSError as e:
+                    print(f"[beacon] Warning: send on {ifname} failed: {e}")
+
+    def _read_loop(self) -> None:
+        while self._running:
+            sock = self._sock
+            if sock is None:
+                return
+            try:
+                data, addr = sock.recvfrom(READ_BUFFER)
+            except OSError:
+                if not self._running:
+                    return
+                continue
+            msg = parse_message(data)
+            if msg is None:
+                continue  # v1 or foreign traffic on the shared group
+
+            if msg["type"] == TYPE_PROBE:
+                if msg.get("from") == self.instance:
+                    continue
+                ad = self._advertise_msg()
+                want = msg.get("want") or []
+                if not want or any(resource_matches(r, p) for r in ad["resources"] for p in want):
+                    self._send(ad, dest=addr)
+                continue
+
+            node = msg["node"]
+            if node["instance"] == self.instance:
+                continue  # our own multicast echo
+            if msg["type"] == TYPE_BYE:
+                with self._lock:
+                    self._nodes.pop(node["instance"], None)
+                continue
+
+            # Source address comes from the packet, never the payload.
+            entry = {"node": node, "resources": msg.get("resources") or [], "addr": addr[0], "lastSeen": time.time()}
+            with self._lock:
+                self._nodes[node["instance"]] = entry
+            row = _neighbor_row(entry["node"], entry["resources"], entry["addr"], entry["lastSeen"])
+            for cb in list(self._listeners):
+                try:
+                    cb(row)
+                except Exception as e:
+                    print(f"[beacon] Listener raised: {e}")
+
+    def _advertise_loop(self) -> None:
+        self._send(self._advertise_msg())
+        self.probe()
+        while self._running:
+            time.sleep(self.interval)
+            if not self._running:
+                return
+            try:
+                self._send(self._advertise_msg())
+            except OSError:
+                pass
+
+
+class MetaCoreLocator:
+    """Locates meta-core over beacon v2 (the first ``metamesh.core`` resource
+    carrying ``data.urls``) and advertises this service as
+    ``metamesh.service/<service_name>``."""
+
+    def __init__(
+        self,
+        service_name: str,
+        base_url: Optional[str] = None,
+        version: str = "",
+        meta_core_url: Optional[str] = None,
+        enabled: Optional[bool] = None,
+    ) -> None:
+        self._pin = (meta_core_url or "").strip() or None
+        self._current: Optional[dict] = None
+        self._change_callbacks: List[Callable[[], None]] = []
+
+        self._node = MeshNode(
+            name=service_name,
+            version=version,
+            enabled=_env_flag("ENABLE_UDP_DISCOVERY", True) if enabled is None else enabled,
+            payload=lambda: {"resources": [{
+                "id": "service",
+                "caps": [CAP_SERVICE_PREFIX + service_name],
+                "endpoints": {"ui": base_url or f"http://{local_ipv4()}"},
+            }]},
+        )
+        self._node.on_advertise(self._on_advertise)
+
+    def _on_advertise(self, nb: dict) -> None:
+        if self._pin:
+            return  # pinned: the wire cannot move us
+        core = next((r for r in nb.get("resources") or [] if resource_matches(r, CAP_CORE)), None)
+        urls = ((core or {}).get("data") or {}).get("urls")
+        if not isinstance(urls, dict) or not urls.get("apiUrl"):
+            return
+
+        prev = self._current
+        if prev is None:
+            self._current = urls
+            print(f"[beacon] meta-core discovered at {urls['apiUrl']} ({nb.get('addr')})")
+            for cb in list(self._change_callbacks):
+                try:
+                    cb()
+                except Exception as e:
+                    print(f"[beacon] Change callback raised: {e}")
+            return
+        if prev.get("apiUrl") == urls.get("apiUrl"):
+            self._current = urls
+            return
+        # A second, different core. Do not flap — keep the first and say so,
+        # because on a PCS box this means a client container can see a core it
+        # must not use.
+        print(
+            f"[beacon] Ignoring a second meta-core at {urls['apiUrl']} "
+            f"({nb.get('addr')}); staying with {prev.get('apiUrl')}. "
+            f"Set META_CORE_URL to pin this explicitly."
+        )
+
+    def start(self) -> None:
+        if self._pin:
+            print(f"[beacon] meta-core pinned to {self._pin}; discovery is advisory")
+        self._node.start()
+
+    def stop(self) -> None:
+        self._node.stop()
+        self._change_callbacks = []
+
+    def get_api_url(self) -> Optional[str]:
+        if self._pin:
+            return self._pin
+        if self._current:
+            return self._current.get("apiUrl")
+        return None
+
+    def get_urls(self) -> Optional[dict]:
+        return self._current
+
+    def wait_for_core(self, timeout: Optional[float] = 30.0) -> Optional[str]:
+        """Block until meta-core is located.
+
+        ``timeout=None`` waits forever, which is meta-stremio's default
+        (LEADER_WAIT_TIMEOUT=0). Re-probes every 500ms rather than waiting out
+        an advertise interval.
+        """
+        deadline = None if timeout is None else time.time() + timeout
+        logged = False
+        while deadline is None or time.time() < deadline:
+            url = self.get_api_url()
+            if url:
+                return url
+            if not logged:
+                print("[beacon] Waiting for meta-core...")
+                logged = True
+            self._node.probe([CAP_CORE])
+            time.sleep(0.5)
+        return None
+
+    def on_change(self, cb: Callable[[], None]) -> None:
+        self._change_callbacks.append(cb)
+
+    def neighbors(self, all: bool = False, cap: Optional[str] = None) -> List[dict]:
+        """One row per name (every instance with ``all``), optionally only
+        nodes with a resource matching ``cap``."""
+        rows = self._node.neighbors() if all else self._node.neighbors_by_name()
+        if cap:
+            rows = [r for r in rows if any(resource_matches(res, cap) for res in r.get("resources") or [])]
+        return rows
+
+    def self_announce(self) -> dict:
+        return self._node.self_announce()
+
+    def probe(self, want: Optional[List[str]] = None) -> None:
+        self._node.probe(want)

src/storage/meta_consumer.py

@@ -1,153 +1,249 @@
 """
 Meta Events Consumer for meta-stremio.
 
-Consumes the meta:events Redis stream published by meta-core.
+Consumes the meta:events stream published by meta-core. Originally used a
-Provides real-time notifications when metadata changes.
+Redis Streams XREADGROUP loop; migrated to meta-core's HTTP SSE endpoint at
+/api/events/meta as part of the api-mediated-access lockdown (see
+meta-core docs/api-mediated-access.md, PR C).
+
+The callback signature is unchanged so leader_storage.py needs no changes:
+each event is delivered as (key, event_type) where key has the form
+"file:<hashId>/<field>" and event_type is "set" / "del" / "expire".
 """
 from __future__ import annotations
 
-import threading
+import json
 import logging
+import os
+import threading
+import time
+from pathlib import Path
 from typing import Callable, List, Optional
+from urllib.parse import urlparse
 
-try:
+import urllib.request
-    from redis import Redis
+import urllib.error
-    REDIS_AVAILABLE = True
-except ImportError:
-    REDIS_AVAILABLE = False
-    Redis = None
 
 logger = logging.getLogger(__name__)
 
 
 class MetaConsumer:
     """
-    Consumes meta:events stream for real-time metadata updates.
+    Streams meta:events from meta-core over SSE and dispatches a filtered
+    subset of changes to registered callbacks.
 
-    Uses Redis consumer groups for reliable delivery.
+    Cursor durability: the last-seen event id is persisted to a file (default
-    Filters events to only notify about interesting field changes.
+    /meta-core/cursors/meta-stremio-meta.cursor) so the consumer resumes
+    cleanly after a restart. On a gap event the cursor is reset to the
+    server-supplied resumeFrom value.
     """
 
-    STREAM = "meta:events"
+    STREAM_PATH = "/api/events/meta"
-    GROUP = "stremio-consumer"
-    CONSUMER = "stremio-1"
 
-    # Fields that trigger cache invalidation
+    # Fields whose changes invalidate cached Stremio responses.
     INTERESTING_FIELDS = [
         "tmdb", "tmdbId", "title", "poster", "backdrop",
         "imdbId", "imdbid", "year", "type", "rating",
-        "description", "plot", "genres", "fileType"
+        "description", "plot", "genres", "fileType",
     ]
 
-    def __init__(self, redis_client: Redis, prefix: str = ""):
+    def __init__(
-        if not REDIS_AVAILABLE:
+        self,
-            raise ImportError("redis package not installed")
+        api_url: str,
-
+        cursor_path: Optional[str] = None,
-        self._client = redis_client
+        request_timeout: int = 35,
-        self._prefix = prefix
+    ) -> None:
+        self._api_url = api_url.rstrip("/")
+        # The SSE resume cursor used to default under /meta-core, which is not
+        # mounted any more (meta-discovery v1 removed the shared volume). It
+        # belongs with the rest of this service's local state: CACHE_FOLDER_PATH
+        # (/data/cache in the dev stack). Without a writable home the consumer
+        # silently replays from `$` on every restart.
+        self._cursor_path = cursor_path or os.path.join(
+            os.environ.get("CACHE_FOLDER_PATH", "/data/cache"),
+            "meta-stremio-meta.cursor",
+        )
+        self._request_timeout = request_timeout
         self._running = False
         self._thread: Optional[threading.Thread] = None
         self._callbacks: List[Callable[[str, str], None]] = []
+        self._cursor: str = self._load_cursor()
+        self._stop_event = threading.Event()
 
-    def on_change(self, callback: Callable[[str, str], None]) -> 'MetaConsumer':
+    def on_change(self, callback: Callable[[str, str], None]) -> "MetaConsumer":
-        """
+        """Register a callback. See module docstring for the contract."""
-        Register callback for metadata changes.
-
-        callback(key, event_type): Called when a file's metadata changes.
-            - key: The Redis key that changed (e.g., "file:abc123/tmdb")
-            - event_type: The operation type ("set", "del", etc.)
-        """
         self._callbacks.append(callback)
         return self
 
     def start(self) -> None:
-        """Start consuming events."""
         if self._running:
             return
-
-        # Create consumer group if not exists
-        try:
-            self._client.xgroup_create(
-                self.STREAM, self.GROUP, id="$", mkstream=True
-            )
-            logger.info(f"[MetaConsumer] Created consumer group '{self.GROUP}'")
-        except Exception as e:
-            if "BUSYGROUP" not in str(e):
-                logger.warning(f"[MetaConsumer] Group creation error: {e}")
-
         self._running = True
+        self._stop_event.clear()
         self._thread = threading.Thread(target=self._consume_loop, daemon=True)
         self._thread.start()
-        logger.info("[MetaConsumer] Started consuming meta:events stream")
+        logger.info(
+            "[MetaConsumer] Started SSE consumer for %s (cursor=%s)",
+            f"{self._api_url}{self.STREAM_PATH}",
+            self._cursor or "$",
+        )
 
     def stop(self) -> None:
-        """Stop consuming events."""
         if not self._running:
             return
-
         self._running = False
+        self._stop_event.set()
         if self._thread:
             self._thread.join(timeout=5)
             self._thread = None
+        # Best-effort flush of any unsaved cursor advance.
+        self._save_cursor()
         logger.info("[MetaConsumer] Stopped")
 
     def is_running(self) -> bool:
-        """Check if consumer is running."""
         return self._running
 
-    def _consume_loop(self) -> None:
+    # ---------- internals -----------------------------------------------
-        """Main consumer loop."""
+
-        while self._running:
+    def _load_cursor(self) -> str:
         try:
-                # Read from stream with 2s block timeout
+            with open(self._cursor_path, "r", encoding="utf-8") as f:
-                entries = self._client.xreadgroup(
+                value = f.read().strip()
-                    self.GROUP,
+                if value:
-                    self.CONSUMER,
+                    logger.info("[MetaConsumer] Resuming from cursor %s", value)
-                    {self.STREAM: ">"},
+                return value
-                    count=100,
+        except FileNotFoundError:
-                    block=2000
+            return ""
+        except Exception as e:
+            logger.warning(
+                "[MetaConsumer] Could not read cursor %s: %s", self._cursor_path, e
+            )
+            return ""
+
+    def _save_cursor(self) -> None:
+        if not self._cursor:
+            return
+        try:
+            Path(self._cursor_path).parent.mkdir(parents=True, exist_ok=True)
+            with open(self._cursor_path, "w", encoding="utf-8") as f:
+                f.write(self._cursor)
+        except Exception as e:
+            logger.warning(
+                "[MetaConsumer] Could not write cursor %s: %s", self._cursor_path, e
             )
 
-                if not entries:
+    def _consume_loop(self) -> None:
-                    continue
+        backoff = 0.5
-
+        max_backoff = 30.0
-                for stream_name, messages in entries:
+        while self._running:
-                    for msg_id, fields in messages:
+            try:
-                        self._process_event(msg_id, fields)
+                self._stream_once()
-                        # Acknowledge
+                backoff = 0.5  # successful read → reset backoff
-                        self._client.xack(self.STREAM, self.GROUP, msg_id)
-
             except Exception as e:
-                if self._running:
+                if not self._running:
-                    logger.error(f"[MetaConsumer] Error consuming: {e}")
+                    return
+                logger.warning(
+                    "[MetaConsumer] SSE error, reconnecting in %.1fs: %s",
+                    backoff, e,
+                )
+                # interruptible sleep
+                self._stop_event.wait(timeout=backoff)
+                backoff = min(backoff * 2, max_backoff)
 
-    def _process_event(self, msg_id: str, fields: dict) -> None:
+    def _stream_once(self) -> None:
-        """Process a single event from the stream."""
+        url = f"{self._api_url}{self.STREAM_PATH}"
-        # Handle both bytes and string keys
+        req = urllib.request.Request(url, method="GET")
-        key = fields.get(b"key") or fields.get("key")
+        req.add_header("Accept", "text/event-stream")
-        event_type = fields.get(b"type") or fields.get("type")
+        req.add_header("Cache-Control", "no-cache")
+        if self._cursor:
+            req.add_header("Last-Event-ID", self._cursor)
 
-        if key is None or event_type is None:
+        # urllib.request doesn't natively grok SSE; we open the socket and
+        # read line-by-line until the connection drops or the thread is
+        # asked to stop.
+        with urllib.request.urlopen(req, timeout=self._request_timeout) as resp:
+            buffer = ""
+            since_save = time.time()
+            for chunk in resp:
+                if not self._running:
+                    return
+                buffer += chunk.decode("utf-8", errors="replace")
+                # Process complete events (separated by blank lines).
+                while "\n\n" in buffer:
+                    event_chunk, buffer = buffer.split("\n\n", 1)
+                    self._handle_chunk(event_chunk)
+                # Flush cursor periodically.
+                if time.time() - since_save > 2.0:
+                    self._save_cursor()
+                    since_save = time.time()
+
+    def _handle_chunk(self, chunk: str) -> None:
+        event_id = ""
+        event_type = ""
+        data_lines: List[str] = []
+        for line in chunk.split("\n"):
+            if not line or line.startswith(":"):
+                continue  # comment / heartbeat
+            colon = line.find(":")
+            if colon == -1:
+                continue
+            field = line[:colon]
+            value = line[colon + 1:]
+            if value.startswith(" "):
+                value = value[1:]
+            if field == "id":
+                event_id = value
+            elif field == "event":
+                event_type = value
+            elif field == "data":
+                data_lines.append(value)
+
+        if not data_lines:
             return
 
-        # Decode bytes if needed
+        try:
-        key = key.decode() if isinstance(key, bytes) else key
+            data = json.loads("\n".join(data_lines))
-        event_type = event_type.decode() if isinstance(event_type, bytes) else event_type
+        except json.JSONDecodeError as e:
+            logger.warning(
+                "[MetaConsumer] Skipping unparseable event %s: %s",
+                event_id or "(no id)", e,
+            )
+            return
 
-        # Filter: only care about interesting field changes
+        # `gap` is a synthetic event signalling that the server trimmed the
+        # cursor out of retention. Reset to the resumeFrom value and
+        # continue. Stremio doesn't bootstrap from the stream so loss is
+        # acceptable — it just means the cache may be stale for a moment.
+        if event_type == "gap":
+            resume = data.get("resumeFrom")
+            if isinstance(resume, str) and resume:
+                self._cursor = resume
+            logger.warning(
+                "[MetaConsumer] Stream cursor trimmed; resuming from %s",
+                resume,
+            )
+            return
+
+        key = data.get("key")
+        if not key or not event_type:
+            return
         if not self._is_interesting_field(key):
+            # Still advance the cursor so we don't replay this event after
+            # a restart.
+            if event_id:
+                self._cursor = event_id
             return
 
-        # Notify callbacks
         for callback in self._callbacks:
             try:
                 callback(key, event_type)
             except Exception as e:
-                logger.error(f"[MetaConsumer] Callback error: {e}")
+                logger.error("[MetaConsumer] Callback error: %s", e)
+
+        if event_id:
+            self._cursor = event_id
 
     def _is_interesting_field(self, key: str) -> bool:
-        """Check if the key represents an interesting field."""
-        # Key format: file:{hashId}/{field}
         for field in self.INTERESTING_FIELDS:
             if key.endswith(f"/{field}"):
                 return True

src/storage/meta_core_api_client.py

@@ -0,0 +1,120 @@
+"""
+MetaCoreApiClient — HTTP read client against meta-core.
+
+Mirrors the subset of meta-stremio's storage-read surface that previously
+spoke ioredis directly. Backs the LeaderStorage when meta-core's leader info
+omits redis_url (post api-mediated-access PR D), so the addon keeps working
+without direct Redis access.
+
+Endpoints used:
+  - GET /meta                  → enumerate hashIds
+  - GET /meta/{hash}           → flat metadata map for one file
+  - GET /meta/{hash}/{key}     → single property (text/plain)
+  - GET /api/file/{cid}/info   → resolve CID to file path
+  - GET /health                → connectivity probe
+"""
+from __future__ import annotations
+
+import json
+import logging
+from typing import Optional, List, Dict
+from urllib.parse import quote
+from urllib.request import Request, urlopen
+from urllib.error import HTTPError, URLError
+
+logger = logging.getLogger(__name__)
+
+
+class MetaCoreApiClient:
+    """HTTP client for meta-core's /meta/* read surface."""
+
+    def __init__(self, api_url: str, timeout: int = 30):
+        self._api_url = api_url.rstrip('/')
+        self._timeout = timeout
+
+    # ----- enumeration ---------------------------------------------------
+
+    def list_hash_ids(self) -> List[str]:
+        """GET /meta — returns every hashId in file:__index__."""
+        body = self._get_json('/meta')
+        if not body:
+            return []
+        ids = body.get('hashIds')
+        return ids if isinstance(ids, list) else []
+
+    def count_hash_ids(self) -> int:
+        body = self._get_json('/meta')
+        if not body:
+            return 0
+        return int(body.get('count', 0))
+
+    # ----- per-file -----------------------------------------------------
+
+    def get_metadata_flat(self, hash_id: str) -> Optional[Dict[str, str]]:
+        """GET /meta/{hash}. Returns the flat property map or None on 404."""
+        body = self._get_json(f'/meta/{quote(hash_id, safe="")}', allow_404=True)
+        if not body:
+            return None
+        meta = body.get('metadata')
+        return meta if isinstance(meta, dict) else None
+
+    def get_property(self, hash_id: str, key: str) -> Optional[str]:
+        """GET /meta/{hash}/{key}. meta-core returns text/plain (or 404)."""
+        path = f'/meta/{quote(hash_id, safe="")}/{quote(key, safe="/")}'
+        url = f"{self._api_url}{path}"
+        try:
+            req = Request(url, headers={'Accept': 'text/plain'})
+            with urlopen(req, timeout=self._timeout) as resp:
+                text = resp.read().decode('utf-8', errors='replace')
+                return text if text != '' else None
+        except HTTPError as e:
+            if e.code == 404:
+                return None
+            logger.warning("[MetaCoreApiClient] GET %s → %s", path, e)
+            return None
+        except URLError as e:
+            logger.warning("[MetaCoreApiClient] GET %s failed: %s", path, e)
+            return None
+
+    # ----- CID resolution -----------------------------------------------
+
+    def resolve_file_path(self, cid: str) -> Optional[str]:
+        """GET /api/file/{cid}/info → relative path (or None on miss)."""
+        path = f'/api/file/{quote(cid, safe="")}/info'
+        body = self._get_json(path)
+        if not body or not body.get('exists'):
+            return None
+        return body.get('filePath') or None
+
+    # ----- health -------------------------------------------------------
+
+    def health(self) -> bool:
+        try:
+            req = Request(f"{self._api_url}/health", headers={'Accept': 'application/json'})
+            with urlopen(req, timeout=min(self._timeout, 5)) as resp:
+                return resp.status == 200
+        except Exception:
+            return False
+
+    # ----- internals ----------------------------------------------------
+
+    def _get_json(self, path: str, allow_404: bool = False) -> Optional[dict]:
+        url = f"{self._api_url}{path}"
+        try:
+            req = Request(url, headers={'Accept': 'application/json'})
+            with urlopen(req, timeout=self._timeout) as resp:
+                raw = resp.read().decode('utf-8', errors='replace')
+                if not raw:
+                    return None
+                return json.loads(raw)
+        except HTTPError as e:
+            if allow_404 and e.code == 404:
+                return None
+            logger.warning("[MetaCoreApiClient] GET %s → %s", path, e)
+            return None
+        except URLError as e:
+            logger.warning("[MetaCoreApiClient] GET %s failed: %s", path, e)
+            return None
+        except json.JSONDecodeError as e:
+            logger.warning("[MetaCoreApiClient] GET %s — invalid JSON: %s", path, e)
+            return None

src/storage/service_registration.py

@@ -1,297 +0,0 @@
-"""
-Service Registration for meta-stremio.
-
-Python implementation of the service registration pattern used in meta-sort/meta-fuse.
-Each service registers itself in /meta-core/services/{service-name}-{hostname}.json.
-
-Service discovery is centralized in meta-core - this module only handles registration.
-For service discovery, use the /api/services endpoint (fetches from meta-core).
-
-Features:
-- Service registration with heartbeat
-- Automatic heartbeat loop
-"""
-from __future__ import annotations
-
-import os
-import json
-import socket
-import threading
-import time
-from dataclasses import dataclass, asdict
-from datetime import datetime
-from pathlib import Path
-from typing import List, Optional, Dict, Callable
-from urllib.request import urlopen
-from urllib.error import URLError
-
-
-# Configuration
-META_CORE_PATH = os.environ.get('META_CORE_PATH', '/meta-core')
-SERVICE_NAME = os.environ.get('SERVICE_NAME', 'meta-stremio')
-SERVICE_VERSION = os.environ.get('SERVICE_VERSION', '1.0.0')
-BASE_URL = os.environ.get('BASE_URL', '')
-
-
-@dataclass
-class ServiceInfo:
-    """
-    Service registration info stored in JSON file.
-
-    Simplified format matching TypeScript ServiceInfo interface:
-    - name: Service name (e.g., 'meta-stremio')
-    - hostname: Container/host hostname
-    - baseUrl: Base URL for the service (e.g., 'http://localhost:8182')
-    - status: Current status ('running' | 'stale' | 'stopped')
-    - lastHeartbeat: ISO timestamp of last heartbeat
-    - role: Optional role ('leader' | 'follower') - only used by meta-core
-    """
-    name: str
-    hostname: str
-    baseUrl: str
-    status: str = 'running'  # running | stale | stopped
-    lastHeartbeat: str = ''
-    role: Optional[str] = None  # "leader", "follower", or None (for non-meta-core services)
-
-    def to_dict(self) -> dict:
-        """Convert to dictionary for JSON serialization.
-        Excludes None values to avoid polluting JSON with null fields.
-        """
-        result = asdict(self)
-        # Remove None values (e.g., role should only appear for meta-core)
-        return {k: v for k, v in result.items() if v is not None}
-
-
-class ServiceRegistration:
-    """
-    Service Discovery using shared filesystem.
-
-    Each service registers itself in /meta-core/services/{service-name}-{hostname}.json
-    and can discover other registered services.
-    """
-
-    def __init__(
-        self,
-        meta_core_path: str = None,
-        service_name: str = None,
-        version: str = None,
-        api_url: str = None,
-        base_url: str = None,
-        capabilities: List[str] = None,
-        endpoints: Dict[str, str] = None,
-        heartbeat_interval: float = 30.0,
-        stale_threshold: float = 60.0
-    ):
-        self.meta_core_path = meta_core_path or META_CORE_PATH
-        self.service_name = service_name or SERVICE_NAME
-        # Use base_url if provided, then api_url, then env BASE_URL, then auto-detect
-        self.base_url = base_url or api_url or BASE_URL or self._get_default_api_url()
-        self.heartbeat_interval = heartbeat_interval
-        self.stale_threshold = stale_threshold
-
-        # Keep for backwards compatibility but not used in simplified format
-        self._version = version or SERVICE_VERSION
-        self._capabilities = capabilities or []
-        self._endpoints = endpoints or {}
-
-        self.current_hostname = socket.gethostname()
-        self.services_dir = os.path.join(self.meta_core_path, 'services')
-        # Use hostname-based file naming like TypeScript services
-        self.service_file_path = os.path.join(self.services_dir, f'{self.service_name}-{self.current_hostname}.json')
-
-        self._heartbeat_thread: Optional[threading.Thread] = None
-        self._stop_event = threading.Event()
-        self._is_started = False
-
-    def _get_default_api_url(self) -> str:
-        """Get default API URL using local IP and PORT."""
-        port = os.environ.get('PORT', '7000')
-        ip = self._get_local_ip()
-        return f'http://{ip}:{port}'
-
-    def _get_local_ip(self) -> str:
-        """Get the local machine's IP address."""
-        try:
-            # Create a socket to determine the outbound IP
-            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-            s.connect(('8.8.8.8', 80))
-            ip = s.getsockname()[0]
-            s.close()
-            return ip
-        except Exception:
-            return socket.gethostname() or 'localhost'
-
-    def _ensure_services_dir(self) -> None:
-        """Ensure services directory exists."""
-        Path(self.services_dir).mkdir(parents=True, exist_ok=True)
-
-    def _build_service_info(self, status: str = 'running') -> ServiceInfo:
-        """Build service info for this service (simplified format)."""
-        return ServiceInfo(
-            name=self.service_name,
-            hostname=self.current_hostname,
-            baseUrl=self.base_url,
-            status=status,
-            lastHeartbeat=datetime.utcnow().isoformat() + 'Z'
-        )
-
-    def register(self) -> None:
-        """Register this service."""
-        self._ensure_services_dir()
-
-        info = self._build_service_info('starting')
-
-        with open(self.service_file_path, 'w') as f:
-            json.dump(info.to_dict(), f, indent=2)
-
-        print(f'[ServiceRegistration] Registered {self.service_name}')
-
-    def update_status(self, status: str) -> None:
-        """Update service status."""
-        try:
-            with open(self.service_file_path, 'r') as f:
-                info = json.load(f)
-
-            info['status'] = status
-            info['lastHeartbeat'] = datetime.utcnow().isoformat() + 'Z'
-
-            with open(self.service_file_path, 'w') as f:
-                json.dump(info, f, indent=2)
-
-        except Exception as e:
-            print(f'[ServiceRegistration] Failed to update status: {e}')
-
-    def heartbeat(self) -> None:
-        """Send heartbeat (update lastHeartbeat timestamp)."""
-        try:
-            with open(self.service_file_path, 'r') as f:
-                info = json.load(f)
-
-            info['lastHeartbeat'] = datetime.utcnow().isoformat() + 'Z'
-
-            with open(self.service_file_path, 'w') as f:
-                json.dump(info, f, indent=2)
-
-        except FileNotFoundError:
-            # If file was deleted, re-register
-            self.register()
-            self.update_status('running')
-        except Exception as e:
-            print(f'[ServiceRegistration] Heartbeat failed: {e}')
-
-    def _heartbeat_loop(self) -> None:
-        """Background heartbeat loop."""
-        while not self._stop_event.is_set():
-            try:
-                self.heartbeat()
-            except Exception as e:
-                print(f'[ServiceRegistration] Heartbeat error: {e}')
-
-            self._stop_event.wait(self.heartbeat_interval)
-
-    def start_heartbeat(self) -> None:
-        """Start heartbeat loop in background thread."""
-        if self._heartbeat_thread and self._heartbeat_thread.is_alive():
-            return
-
-        self._stop_event.clear()
-        self._heartbeat_thread = threading.Thread(
-            target=self._heartbeat_loop,
-            daemon=True,
-            name='service-heartbeat'
-        )
-        self._heartbeat_thread.start()
-
-    def stop_heartbeat(self) -> None:
-        """Stop heartbeat loop."""
-        self._stop_event.set()
-        if self._heartbeat_thread and self._heartbeat_thread.is_alive():
-            self._heartbeat_thread.join(timeout=2.0)
-
-    def unregister(self) -> None:
-        """Unregister this service (on shutdown)."""
-        self.stop_heartbeat()
-
-        try:
-            self.update_status('stopped')
-            print(f'[ServiceRegistration] Unregistered {self.service_name}')
-        except Exception as e:
-            print(f'[ServiceRegistration] Failed to unregister: {e}')
-
-    # ========================================================================
-    # Lifecycle
-    # ========================================================================
-
-    def start(self) -> None:
-        """Full startup sequence."""
-        if self._is_started:
-            return
-
-        # Try to register, but don't fail if filesystem is read-only
-        # This allows discovery-only mode when we can't write
-        can_register = os.access(self.services_dir, os.W_OK) if os.path.exists(self.services_dir) else False
-
-        if can_register:
-            try:
-                self.register()
-                self.update_status('running')
-                self.start_heartbeat()
-                print(f'[ServiceRegistration] Started {self.service_name}-{self.current_hostname} at {self.base_url}')
-            except Exception as e:
-                print(f'[ServiceRegistration] Registration failed (read-only?): {e}')
-                print(f'[ServiceRegistration] Running in discovery-only mode')
-        else:
-            print(f'[ServiceRegistration] Services directory not writable, running in discovery-only mode')
-
-        self._is_started = True
-
-    def stop(self) -> None:
-        """Full shutdown sequence."""
-        if not self._is_started:
-            return
-
-        self.unregister()
-        self._is_started = False
-
-    # ========================================================================
-    # Getters
-    # ========================================================================
-
-    def get_service_file_path(self) -> str:
-        return self.service_file_path
-
-    def get_services_dir(self) -> str:
-        return self.services_dir
-
-    def is_started(self) -> bool:
-        return self._is_started
-
-
-# Singleton instance
-_service_registration: Optional[ServiceRegistration] = None
-
-
-def get_service_discovery() -> ServiceRegistration:
-    """Get or create the service registration singleton.
-
-    Note: Function name kept as 'get_service_discovery' for backward compatibility.
-    """
-    global _service_registration
-    if _service_registration is None:
-        _service_registration = ServiceRegistration()
-    return _service_registration
-
-
-def init_service_discovery(
-    api_url: str = None,
-    base_url: str = None,
-    **kwargs
-) -> ServiceRegistration:
-    """Initialize and start the service registration singleton.
-
-    Note: Function name kept as 'init_service_discovery' for backward compatibility.
-    """
-    global _service_registration
-    _service_registration = ServiceRegistration(api_url=api_url, base_url=base_url, **kwargs)
-    _service_registration.start()
-    return _service_registration

www/index.html

@@ -292,6 +292,14 @@
         footer a:hover { text-decoration: underline; }
 
         /* Services Navigation */
+        /* <meta-service-menu> is in shadow DOM and cannot read the vars
+           above; map this UI's palette onto the four tokens it does read. */
+        meta-service-menu {
+            --mm-nav-fg: var(--text-primary);
+            --mm-nav-bg: var(--bg-tertiary);
+            --mm-nav-border: var(--border-color);
+            --mm-nav-accent: var(--accent-primary);
+        }
         .services-nav {
             background: var(--bg-secondary);
             border-radius: 12px;
@@ -362,16 +370,13 @@
             .services-nav { flex-direction: column; align-items: flex-start; }
         }
     </style>
+    <script type="module" src="meta-service-menu.js"></script>
 </head>
 <body>
     <div class="container">
-        <!-- Services Navigation -->
+        <!-- Neighbouring meta-* services (meta-discovery v1). The element
-        <nav class="services-nav" id="servicesNav" style="display: none;">
+             fetches /api/neighbors from this origin and renders a burger menu. -->
-            <span class="services-nav-label">Services:</span>
+        <meta-service-menu id="servicesNav" current="meta-stremio"></meta-service-menu>
-            <div class="services-nav-items" id="servicesNavItems">
-                <!-- Populated by JavaScript -->
-            </div>
-        </nav>
 
         <header class="header">
             <div>
@@ -548,78 +553,18 @@
             }
         }
 
-        // Service icons mapping
+        // Service nav is <meta-service-menu> now — no local icon map,
-        const serviceIcons = {
+        // name formatter or poller lives here any more.
-            'meta-sort': '📁',
-            'meta-fuse': '🗂️',
-            'meta-stremio': '🎬',
-            'meta-core': '⚙️',
-            'meta-dup': '🔍',
-            'meta-orbit': '🌐',
-            'default': '📦'
-        };
-
-        // Fetch and display discovered services
-        async function fetchServices() {
-            try {
-                const res = await fetch('/api/services');
-                const data = await res.json();
-
-                const navEl = document.getElementById('servicesNav');
-                const itemsEl = document.getElementById('servicesNavItems');
-
-                if (data.services && data.services.length > 0) {
-                    // Sort services by name
-                    const services = data.services.sort((a, b) => a.name.localeCompare(b.name));
-
-                    // Build navigation items
-                    itemsEl.innerHTML = services.map(svc => {
-                        const icon = serviceIcons[svc.name] || serviceIcons.default;
-                        const isActive = svc.name === data.current;
-                        const activeClass = isActive ? ' active' : '';
-
-                        // Use external URL for other services, current path for self
-                        const href = isActive ? '#' : svc.baseUrl;
-
-                        return `
-                            <a href="${href}" class="service-link${activeClass}" onclick="event.preventDefault(); ${isActive ? '' : `window.location.href='${svc.baseUrl}'`}">
-                                <span class="service-icon">${icon}</span>
-                                <span>${formatServiceName(svc.name)}</span>
-                                <span class="service-status"></span>
-                            </a>
-                        `;
-                    }).join('');
-
-                    navEl.style.display = 'flex';
-                } else {
-                    navEl.style.display = 'none';
-                }
-            } catch (e) {
-                console.error('Failed to fetch services:', e);
-                document.getElementById('servicesNav').style.display = 'none';
-            }
-        }
-
-        function formatServiceName(name) {
-            // Convert 'meta-stremio' to 'Meta Stremio'
-            return name.split('-').map(word =>
-                word.charAt(0).toUpperCase() + word.slice(1)
-            ).join(' ');
-        }
 
         // Initial fetch
         fetchStats();
         fetchMetrics();
-        fetchServices();
 
         // Refresh metrics every 5 seconds
         setInterval(fetchMetrics, 5000);
 
         // Refresh stats every 30 seconds
         setInterval(fetchStats, 30000);
-
-        // Refresh services every 60 seconds
-        setInterval(fetchServices, 60000);
     </script>
 </body>
 </html>

www/meta-service-menu.js

@@ -0,0 +1,308 @@
+/**
+ * <meta-service-menu> — the neighbouring meta-* services, as a burger menu.
+ *
+ * Replaces the eight divergent "services bar" implementations (four copies of
+ * a React `ServiceNav.tsx`, two mirrored `services.js`, two bespoke inline
+ * ones) with a single framework-free custom element. It drops into the React
+ * dashboards as a plain tag — no React dependency — and into the vanilla UIs
+ * as an ES module.
+ *
+ * Data comes from the host service's own `GET /api/neighbors`, same-origin.
+ * Never cross-origin: the dev stack's self-signed certs and the OIDC redirects
+ * make a fetch to a neighbour a dead end. Links are plain navigations.
+ *
+ * ⚠ MIRRORED FILE. Byte-identical copies live in every meta-* UI tree; see
+ * `scripts/check-mirrors.sh`. Edit one → edit all.
+ *
+ * Shadow DOM is not optional here. The eight UIs use four incompatible token
+ * vocabularies, and two of them define *different colours under the same
+ * names* (`--bg-secondary` is #12121a in meta-sort, #16213e in meta-core,
+ * #161b22 in meta-dup). meta-search and meta-share are also the only
+ * light-mode-capable UIs, so a hardcoded dark palette would break them.
+ * The element therefore styles itself from four tokens with dark defaults:
+ *
+ *     --mm-nav-fg, --mm-nav-bg, --mm-nav-border, --mm-nav-accent
+ *
+ * Each host maps its own variables onto those, e.g.
+ *     meta-service-menu { --mm-nav-bg: var(--bg-secondary); }
+ *
+ * Attributes:
+ *   current   name of the host service, so it can be marked and not linked
+ *   endpoint  override the default "/api/neighbors"
+ *   label     button tooltip (default "Services")
+ *
+ * Property:
+ *   rewriteUrl  optional (name, url, neighbour) => url, for hosts that must
+ *               translate internal hostnames into browser-reachable ones
+ *               (meta-gateway does this for its dev/prod split).
+ */
+
+const ICONS = {
+  "meta-core": "⚙️",
+  "meta-sort": "📁",
+  "meta-fuse": "🗂️",
+  "meta-stremio": "🎬",
+  "meta-dup": "🔍",
+  "meta-search": "🔎",
+  "meta-share": "🔗",
+  "meta-gateway": "🌉",
+  "meta-watch": "📺",
+  "meta-listen": "🎧",
+  "meta-read": "📖",
+};
+const DEFAULT_ICON = "📦";
+const POLL_MS = 30000;
+
+function titleCase(name) {
+  return String(name || "")
+    .split("-")
+    .map((w) => w.charAt(0).toUpperCase() + w.slice(1))
+    .join(" ");
+}
+
+const TEMPLATE = `
+<style>
+  :host {
+    --_fg: var(--mm-nav-fg, #e8eaed);
+    --_bg: var(--mm-nav-bg, #1a1a2e);
+    --_border: var(--mm-nav-border, rgba(255, 255, 255, 0.14));
+    --_accent: var(--mm-nav-accent, #4ecdc4);
+    position: relative;
+    display: inline-block;
+    font-family: inherit;
+    font-size: 14px;
+    line-height: 1.4;
+  }
+  button.trigger {
+    display: inline-flex;
+    align-items: center;
+    justify-content: center;
+    gap: 0.4rem;
+    width: 2.1rem;
+    height: 2.1rem;
+    padding: 0;
+    cursor: pointer;
+    color: var(--_fg);
+    background: transparent;
+    border: 1px solid var(--_border);
+    border-radius: 8px;
+    font: inherit;
+    transition: border-color 0.15s, background 0.15s;
+  }
+  button.trigger:hover { border-color: var(--_accent); }
+  button.trigger:focus-visible {
+    outline: 2px solid var(--_accent);
+    outline-offset: 2px;
+  }
+  button.trigger[aria-expanded="true"] { border-color: var(--_accent); }
+  .bars { display: block; width: 15px; }
+  .bars span {
+    display: block;
+    height: 2px;
+    margin: 3px 0;
+    border-radius: 2px;
+    background: currentColor;
+  }
+  .panel {
+    position: absolute;
+    top: calc(100% + 6px);
+    right: 0;
+    z-index: 1000;
+    min-width: 13rem;
+    max-height: 70vh;
+    overflow-y: auto;
+    padding: 0.3rem;
+    background: var(--_bg);
+    border: 1px solid var(--_border);
+    border-radius: 10px;
+    box-shadow: 0 10px 30px rgba(0, 0, 0, 0.35);
+  }
+  .panel[hidden] { display: none; }
+  a.item, .item {
+    display: flex;
+    align-items: center;
+    gap: 0.55rem;
+    padding: 0.45rem 0.6rem;
+    border-radius: 6px;
+    color: var(--_fg);
+    text-decoration: none;
+    white-space: nowrap;
+  }
+  a.item:hover { background: color-mix(in srgb, var(--_accent) 18%, transparent); }
+  .item.active { color: var(--_accent); font-weight: 600; cursor: default; }
+  .item.dead { opacity: 0.55; cursor: not-allowed; }
+  .icon { font-size: 1rem; line-height: 1; }
+  .dot {
+    width: 6px;
+    height: 6px;
+    margin-left: auto;
+    border-radius: 50%;
+    background: var(--_accent);
+  }
+  .empty {
+    padding: 0.6rem;
+    opacity: 0.7;
+    font-size: 0.85em;
+    white-space: nowrap;
+  }
+  /* Respect a host that opts out of motion. */
+  @media (prefers-reduced-motion: reduce) {
+    button.trigger { transition: none; }
+  }
+</style>
+<button class="trigger" type="button" aria-haspopup="menu" aria-expanded="false">
+  <span class="bars" aria-hidden="true"><span></span><span></span><span></span></span>
+</button>
+<div class="panel" role="menu" hidden></div>
+`;
+
+class MetaServiceMenu extends HTMLElement {
+  constructor() {
+    super();
+    this.attachShadow({ mode: "open" });
+    this.shadowRoot.innerHTML = TEMPLATE;
+    this._btn = this.shadowRoot.querySelector("button.trigger");
+    this._panel = this.shadowRoot.querySelector(".panel");
+    this._neighbors = [];
+    this._timer = null;
+    this._open = false;
+
+    this._onDocClick = (e) => {
+      if (!this._open) return;
+      if (e.composedPath().includes(this)) return;
+      this._setOpen(false);
+    };
+    this._onKey = (e) => {
+      if (e.key === "Escape" && this._open) {
+        this._setOpen(false);
+        this._btn.focus();
+      }
+    };
+  }
+
+  connectedCallback() {
+    this._btn.title = this.getAttribute("label") || "Services";
+    this._btn.setAttribute("aria-label", this._btn.title);
+    this._btn.addEventListener("click", () => this._setOpen(!this._open));
+    document.addEventListener("click", this._onDocClick);
+    document.addEventListener("keydown", this._onKey);
+
+    this.refresh();
+    this._timer = setInterval(() => this.refresh(), POLL_MS);
+  }
+
+  disconnectedCallback() {
+    if (this._timer) clearInterval(this._timer);
+    this._timer = null;
+    document.removeEventListener("click", this._onDocClick);
+    document.removeEventListener("keydown", this._onKey);
+  }
+
+  get endpoint() {
+    return this.getAttribute("endpoint") || "/api/neighbors";
+  }
+
+  get current() {
+    return this.getAttribute("current") || "";
+  }
+
+  async refresh() {
+    try {
+      const res = await fetch(this.endpoint, {
+        headers: { Accept: "application/json" },
+      });
+      if (!res.ok) throw new Error(String(res.status));
+      const data = await res.json();
+      // `services` is the legacy alias some backends still emit.
+      this._neighbors = data.neighbors || data.services || [];
+    } catch {
+      // Silent: the menu is navigation, not a health indicator. Keep whatever
+      // we last had rather than blanking the list on one failed poll.
+    }
+    this._render();
+  }
+
+  _setOpen(open) {
+    this._open = open;
+    this._panel.hidden = !open;
+    this._btn.setAttribute("aria-expanded", String(open));
+    if (open) this.refresh();
+  }
+
+  _render() {
+    const panel = this._panel;
+    panel.textContent = "";
+
+    const seen = new Set();
+    const rows = [];
+    for (const n of this._neighbors) {
+      const name = n.name || "";
+      if (!name || seen.has(name)) continue;
+      // Beacon v2 neighbours include plugins; the nav lists services only.
+      // (A row without `caps` comes from a pre-v2 backend — keep it.)
+      const isService = (c) => String(c).startsWith("metamesh.service/");
+      if (Array.isArray(n.caps) && !n.caps.some(isService)) continue;
+      seen.add(name);
+      rows.push(n);
+    }
+    // Always show the host itself, even before its own echo comes back.
+    if (this.current && !seen.has(this.current)) {
+      rows.push({ name: this.current, baseUrl: "" });
+    }
+    rows.sort((a, b) => String(a.name).localeCompare(String(b.name)));
+
+    if (rows.length === 0) {
+      const empty = document.createElement("div");
+      empty.className = "empty";
+      empty.textContent = "No services found";
+      panel.appendChild(empty);
+      return;
+    }
+
+    for (const n of rows) {
+      const isActive = n.name === this.current;
+      let url = n.baseUrl || "";
+      if (typeof this.rewriteUrl === "function") {
+        try {
+          url = this.rewriteUrl(n.name, url, n) || "";
+        } catch {
+          /* a bad hook must not blank the menu */
+        }
+      }
+
+      // A neighbour with no reachable URL is shown but not linked — better
+      // than a link that goes nowhere.
+      const el = document.createElement(isActive || !url ? "span" : "a");
+      el.className =
+        "item" + (isActive ? " active" : "") + (!isActive && !url ? " dead" : "");
+      if (el.tagName === "A") {
+        el.href = url;
+        el.setAttribute("role", "menuitem");
+      }
+      if (!isActive && !url) el.title = "No reachable URL advertised";
+
+      const icon = document.createElement("span");
+      icon.className = "icon";
+      icon.textContent = ICONS[n.name] || DEFAULT_ICON;
+
+      const label = document.createElement("span");
+      label.textContent = titleCase(n.name);
+
+      el.append(icon, label);
+      // Beacon v2 omits `status` when running.
+      if ((n.status || "running") === "running" || isActive) {
+        const dot = document.createElement("span");
+        dot.className = "dot";
+        dot.title = n.status || "running";
+        el.appendChild(dot);
+      }
+      panel.appendChild(el);
+    }
+  }
+}
+
+if (!customElements.get("meta-service-menu")) {
+  customElements.define("meta-service-menu", MetaServiceMenu);
+}
+
+export default MetaServiceMenu;