diff --git a/README.md b/README.md index 6392106..4603cf0 100644 --- a/README.md +++ b/README.md @@ -60,12 +60,13 @@ uvicorn meridian.api.main:app --host 0.0.0.0 --port 8080 # meridian --config configs/local_gpu.yaml ``` -Published images: `ghcr.io/imv-in/meridian:0.9.3` / `:latest` +Published images: `ghcr.io/imv-in/meridian:0.12.0` / `:latest` + ```bash docker run --rm -p 8080:8080 \ -v "$(pwd)/configs/mock_demo.yaml:/app/config.yaml:ro" \ -e MERIDIAN_CONFIG=/app/config.yaml \ - ghcr.io/imv-in/meridian:0.9.3 + ghcr.io/imv-in/meridian:0.12.0 ``` (You’ll still need reachable backends in that config.) @@ -77,7 +78,22 @@ export MERIDIAN_CONFIG=configs/local_gpu.yaml uvicorn meridian.api.main:app --host 0.0.0.0 --port 8080 ``` -Load / overhead numbers: [`docs/LOAD.md`](docs/LOAD.md) +### Measured real-engine overhead + +Meridian v0.12.0 was tested against Ollama 0.31.1 with `qwen2.5:0.5b` on +an RTX 4060 Laptop GPU. All direct and gateway sync/stream checks passed with +zero benchmark errors. + +| Concurrency | Requests | Direct p50 | Meridian p50 | p50 delta | RPS change | +|------------:|---------:|-----------:|-------------:|----------:|-----------:| +| 1 | 30 | 174.1 ms | 180.5 ms | +6.4 ms | -3.9% | +| 4 | 40 | 368.5 ms | 375.5 ms | +7.0 ms | -2.2% | +| 8 | 80 | 243.0 ms | 247.2 ms | +4.1 ms | -1.7% | + +These are sequential single-host runs, not a cross-row scaling curve. Ollama +GPU warm state and dynamic batching affect absolute latency. See the +[`methodology and raw evidence`](docs/LOAD.md) and +[`reproducible Ollama/vLLM validation`](docs/REAL_ENGINE_VALIDATION.md). ## Documentation map diff --git a/docs/LOAD.md b/docs/LOAD.md index 9042a1f..1448d92 100644 --- a/docs/LOAD.md +++ b/docs/LOAD.md @@ -1,4 +1,4 @@ -# Load & overhead numbers (0.9.3) +# Load and overhead numbers How much latency Meridian adds in front of a backend, and how to re-measure on your hardware for ~1k-user capacity planning. @@ -24,33 +24,13 @@ The script: ### Recipe ```bash -# 1) Backend -ollama pull qwen2.5:0.5b -ollama serve # default http://127.0.0.1:11434 - -# 2) Meridian (config already points at Ollama) -MERIDIAN_CONFIG=configs/local_gpu.yaml \ - uvicorn meridian.api.main:app --host 127.0.0.1 --port 18080 - -# 3) Smoke (stream + non-stream + headers) -python scripts/smoke_test.py --url http://127.0.0.1:18080 --model qwen2.5:0.5b - -# 4) Overhead (direct Ollama vs via Meridian) -python scripts/bench_overhead.py \ - --backend-url http://127.0.0.1:11434 \ - --gateway-url http://127.0.0.1:18080 \ - --model qwen2.5:0.5b \ - --requests 30 --concurrency 1 --warmup 3 - -# Optional light concurrency: -python scripts/bench_overhead.py \ - --backend-url http://127.0.0.1:11434 \ - --gateway-url http://127.0.0.1:18080 \ - --model qwen2.5:0.5b \ - --requests 20 --concurrency 4 --warmup 2 +PYTHON=.venv/bin/python sh scripts/validate_ollama.sh ``` -If `auth.enabled`, add `--auth mrdn_...` to the bench and smoke commands. +The profile handles backend readiness, a disposable Meridian process, smoke +checks, and the serial overhead run. See +[`REAL_ENGINE_VALIDATION.md`](./REAL_ENGINE_VALIDATION.md) for the complete +Ollama/vLLM release matrix and environment overrides. > Real-model **absolute** latency is dominated by the engine. Use these numbers > to confirm **gateway overhead stays small relative to generation time**, not @@ -92,40 +72,59 @@ assert absolute ms (hardware variance). ## Reference numbers (Ollama, real path) -Recorded **2026-07-10** on the same Linux host: +Recorded **2026-07-30**. Complete machine-readable results: +[`serial`](./validation/ollama-v0.12.0.json), +[`concurrency 4`](./validation/ollama-v0.12.0-c4.json), and +[`concurrency 8`](./validation/ollama-v0.12.0-c8.json). | Host detail | Value | |-------------|--------| | GPU | NVIDIA GeForce RTX 4060 Laptop (8 GiB) | | Backend | Ollama `qwen2.5:0.5b` on `127.0.0.1:11434` | -| Meridian | **v0.9.3**, `configs/local_gpu.yaml`, port **18080** (no auth/budgets/cost) | +| Meridian | **v0.12.0**, generated validation config, port **18080** (no auth/budgets/cost) | +| Ollama | **0.31.1** | +| Python | **3.12.11** | +| NVIDIA driver | **580.159.03** | | Request shape | non-stream chat, `max_tokens=8`, message `"bench"` | ### Serial isolation (`n=30`, `concurrency=1`) | Path | p50 (ms) | p95 (ms) | mean (ms) | RPS | errors | |------|----------|----------|-----------|-----|--------| -| Direct → Ollama | 151.0 | 153.2 | 151.2 | 6.6 | 0 | -| Via Meridian | 153.0 | 155.3 | 153.3 | 6.5 | 0 | -| **Overhead** | **~1.9 ms** | **~2.1 ms** | **~2.1 ms** | — | — | +| Direct → Ollama | 174.1 | 185.0 | 173.9 | 5.75 | 0 | +| Via Meridian | 180.5 | 186.2 | 181.1 | 5.52 | 0 | +| **Overhead** | **6.4 ms** | **1.3 ms** | **7.2 ms** | — | — | -**Takeaway:** gateway adds ~**2 ms** (~**1.3%** of end-to-end p50). Engine time is the budget. +**Takeaway:** this run added **6.4 ms** at p50, about **3.7%** of direct +end-to-end p50. Engine generation remained the dominant latency component. -### Light concurrent (`n=20`, `concurrency=4`) +### Concurrent load (`n=40`, `concurrency=4`) | Path | p50 (ms) | p95 (ms) | mean (ms) | RPS | errors | |------|----------|----------|-----------|-----|--------| -| Direct → Ollama | 180.7 | 243.1 | 189.1 | 20.2 | 0 | -| Via Meridian | 186.7 | 235.5 | 192.8 | 19.9 | 0 | -| **Delta p50** | **~5.9 ms** | (noisy) | — | ~same RPS | — | +| Direct -> Ollama | 368.5 | 400.5 | 361.9 | 10.89 | 0 | +| Via Meridian | 375.5 | 401.4 | 367.1 | 10.66 | 0 | +| **Delta / ratio** | **7.0 ms** | **0.9 ms** | **5.2 ms** | **-2.2%** | — | -Under concurrency, engine queueing dominates; Meridian RPS tracks direct within ~2%. +### Concurrent load (`n=80`, `concurrency=8`) + +| Path | p50 (ms) | p95 (ms) | mean (ms) | RPS | errors | +|------|----------|----------|-----------|-----|--------| +| Direct -> Ollama | 243.0 | 321.5 | 252.9 | 30.44 | 0 | +| Via Meridian | 247.2 | 337.8 | 257.4 | 29.92 | 0 | +| **Delta / ratio** | **4.1 ms** | **16.3 ms** | **4.6 ms** | **-1.7%** | — | + +The concurrency runs show no material throughput loss through Meridian. The +concurrency-8 p95 increase is more variable than the serial and concurrency-4 +runs, so it should be treated as a host and engine observation rather than a +gateway capacity limit. Repeat this matrix on the target deployment hardware +before making capacity commitments. ### Functional proof (same stack) -`scripts/smoke_test.py --url http://127.0.0.1:18080 --model qwen2.5:0.5b` — pass -(`/meridian/status`, `/meridian/version` **0.9.3**, non-stream + stream/`[DONE]`, -`x-meridian-backend=ollama-4070`). +The v0.12.0 profile passed direct models, non-stream, and stream checks, then +passed gateway status/version/models, non-stream, stream/`[DONE]`, and required +`x-request-id` / `x-meridian-backend` header checks. ### How to interpret for ~1000 users diff --git a/docs/REAL_ENGINE_VALIDATION.md b/docs/REAL_ENGINE_VALIDATION.md new file mode 100644 index 0000000..0115b3c --- /dev/null +++ b/docs/REAL_ENGINE_VALIDATION.md @@ -0,0 +1,96 @@ +# Real-engine release validation + +Meridian release candidates are validated against live Ollama and vLLM servers, +not only the in-process mock backend. The validation harness checks the backend +first, starts a disposable Meridian process, runs the existing gateway smoke +test, and records a direct-versus-gateway benchmark as JSON. + +## Validation matrix + +| Engine | Runtime | Model | Status for v0.12.0 | Evidence | +|--------|---------|-------|--------------------|----------| +| Ollama | 0.31.1 | `qwen2.5:0.5b` | Passed at concurrency 1, 4, and 8 | [`serial`](./validation/ollama-v0.12.0.json), [`c4`](./validation/ollama-v0.12.0-c4.json), [`c8`](./validation/ollama-v0.12.0-c8.json) | +| vLLM | `vllm/vllm-openai:v0.10.2` | `Qwen/Qwen2.5-0.5B-Instruct` | Pending local GPU run | Evidence added after a passing run | + +The vLLM profile pins model revision +`7ae557604adf67be50417f59c2c2f167def9a775` so a later model update cannot +silently change release evidence. + +## Checks performed + +Each profile verifies: + +1. Direct `GET /v1/models` returns at least one model. +2. Direct non-stream chat returns at least one choice. +3. Direct streaming chat ends with `data: [DONE]`. +4. Meridian starts from a generated, isolated configuration. +5. `scripts/smoke_test.py` passes models, sync, stream, and response-header checks. +6. `scripts/bench_overhead.py` completes with no direct or gateway errors. +7. Evidence records Meridian, engine, Python, platform, GPU, and benchmark details. + +Prompts and generated text are not written to the evidence file. + +## Ollama + +Prerequisites are Ollama, `curl`, and a Meridian development installation. The +profile uses an existing Ollama server when available. Otherwise, it starts one +for the validation and stops it afterward. + +```bash +python -m venv .venv +. .venv/bin/activate +pip install -e ".[dev]" +PYTHON=.venv/bin/python sh scripts/validate_ollama.sh +``` + +Override the defaults with environment variables: + +```bash +MODEL=llama3.2:3b REQUESTS=20 CONCURRENCY=1 \ + OUTPUT=/tmp/ollama-validation.json \ + PYTHON=.venv/bin/python sh scripts/validate_ollama.sh +``` + +## vLLM + +Prerequisites are Docker, the NVIDIA Container Toolkit, an NVIDIA GPU, `curl`, +and enough disk space for the pinned image and model. The profile removes its +container on exit and reuses the host Hugging Face cache. + +```bash +PYTHON=.venv/bin/python sh scripts/validate_vllm.sh +``` + +Useful overrides for constrained GPUs: + +```bash +GPU_MEMORY_UTILIZATION=0.65 MAX_MODEL_LEN=2048 \ + PYTHON=.venv/bin/python sh scripts/validate_vllm.sh +``` + +## Generic OpenAI-compatible backend + +For an already-running server, call the common harness directly: + +```bash +.venv/bin/python scripts/validate_real_backend.py \ + --engine custom \ + --engine-version 1.0.0 \ + --backend-url http://127.0.0.1:8000 \ + --model my-model \ + --output /tmp/custom-validation.json +``` + +Use a free `--gateway-port` if port `18080` is occupied. + +## Release procedure + +1. Check out the exact release tag on the GPU validation host. +2. Run both profiles without changing their pinned defaults. +3. Confirm every check is `passed` and both benchmark error counts are zero. +4. Review the environment metadata and benchmark values for obvious anomalies. +5. Commit the evidence files with the release documentation. + +Latency is informational because it varies by host, engine state, thermals, and +driver version. Functional failures and non-zero request errors block release; +an absolute latency threshold does not. diff --git a/docs/validation/ollama-v0.12.0-c4.json b/docs/validation/ollama-v0.12.0-c4.json new file mode 100644 index 0000000..2260549 --- /dev/null +++ b/docs/validation/ollama-v0.12.0-c4.json @@ -0,0 +1,52 @@ +{ + "schema_version": 1, + "recorded_at": "2026-07-30T17:56:45.403619+00:00", + "meridian_version": "0.12.0", + "engine": "ollama", + "engine_version": "0.31.1", + "backend_url": "http://127.0.0.1:11434", + "model": "qwen2.5:0.5b", + "model_revision": null, + "host": { + "platform": "Linux-7.0.0-28-generic-x86_64-with-glibc2.39", + "python": "3.12.11", + "gpu": { + "name": "NVIDIA GeForce RTX 4060 Laptop GPU", + "memory_mib": "8188", + "driver": "580.159.03" + } + }, + "checks": { + "direct_models": "passed", + "direct_chat": "passed", + "direct_stream": "passed", + "gateway_smoke": "passed" + }, + "benchmark": { + "mode": "external", + "backend_url": "http://127.0.0.1:11434", + "gateway_url": "http://127.0.0.1:18080", + "requests": 40, + "concurrency": 4, + "direct": { + "n": 40, + "p50_ms": 368.46687649995147, + "p95_ms": 400.45164840039433, + "p99_ms": 409.4095587598713, + "mean_ms": 361.8813320751542, + "rps": 10.89368797209658, + "errors": 0 + }, + "via_meridian": { + "n": 40, + "p50_ms": 375.4659959995479, + "p95_ms": 401.36569595015317, + "p99_ms": 406.304408490123, + "mean_ms": 367.06462375004776, + "rps": 10.656981113434352, + "errors": 0 + }, + "overhead_p50_ms": 6.999, + "overhead_p95_ms": 0.914 + } +} diff --git a/docs/validation/ollama-v0.12.0-c8.json b/docs/validation/ollama-v0.12.0-c8.json new file mode 100644 index 0000000..a1381de --- /dev/null +++ b/docs/validation/ollama-v0.12.0-c8.json @@ -0,0 +1,52 @@ +{ + "schema_version": 1, + "recorded_at": "2026-07-30T17:57:42.790348+00:00", + "meridian_version": "0.12.0", + "engine": "ollama", + "engine_version": "0.31.1", + "backend_url": "http://127.0.0.1:11434", + "model": "qwen2.5:0.5b", + "model_revision": null, + "host": { + "platform": "Linux-7.0.0-28-generic-x86_64-with-glibc2.39", + "python": "3.12.11", + "gpu": { + "name": "NVIDIA GeForce RTX 4060 Laptop GPU", + "memory_mib": "8188", + "driver": "580.159.03" + } + }, + "checks": { + "direct_models": "passed", + "direct_chat": "passed", + "direct_stream": "passed", + "gateway_smoke": "passed" + }, + "benchmark": { + "mode": "external", + "backend_url": "http://127.0.0.1:11434", + "gateway_url": "http://127.0.0.1:18080", + "requests": 80, + "concurrency": 8, + "direct": { + "n": 80, + "p50_ms": 243.04335550004907, + "p95_ms": 321.52103000053097, + "p99_ms": 409.1220455409346, + "mean_ms": 252.8889539500824, + "rps": 30.441044313314393, + "errors": 0 + }, + "via_meridian": { + "n": 80, + "p50_ms": 247.18422350088076, + "p95_ms": 337.83950110109794, + "p99_ms": 430.61965636992664, + "mean_ms": 257.4414281375084, + "rps": 29.92078287226416, + "errors": 0 + }, + "overhead_p50_ms": 4.141, + "overhead_p95_ms": 16.318 + } +} diff --git a/docs/validation/ollama-v0.12.0.json b/docs/validation/ollama-v0.12.0.json new file mode 100644 index 0000000..a319cee --- /dev/null +++ b/docs/validation/ollama-v0.12.0.json @@ -0,0 +1,52 @@ +{ + "schema_version": 1, + "recorded_at": "2026-07-30T17:16:13.588388+00:00", + "meridian_version": "0.12.0", + "engine": "ollama", + "engine_version": "0.31.1", + "backend_url": "http://127.0.0.1:11434", + "model": "qwen2.5:0.5b", + "model_revision": null, + "host": { + "platform": "Linux-7.0.0-28-generic-x86_64-with-glibc2.39", + "python": "3.12.11", + "gpu": { + "name": "NVIDIA GeForce RTX 4060 Laptop GPU", + "memory_mib": "8188", + "driver": "580.159.03" + } + }, + "checks": { + "direct_models": "passed", + "direct_chat": "passed", + "direct_stream": "passed", + "gateway_smoke": "passed" + }, + "benchmark": { + "mode": "external", + "backend_url": "http://127.0.0.1:11434", + "gateway_url": "http://127.0.0.1:18080", + "requests": 30, + "concurrency": 1, + "direct": { + "n": 30, + "p50_ms": 174.07333799974367, + "p95_ms": 184.97197050010072, + "p99_ms": 194.15516243993807, + "mean_ms": 173.92129879996597, + "rps": 5.74870926568729, + "errors": 0 + }, + "via_meridian": { + "n": 30, + "p50_ms": 180.52277999959188, + "p95_ms": 186.22431705030067, + "p99_ms": 188.05540685048982, + "mean_ms": 181.07593136671008, + "rps": 5.521790883099198, + "errors": 0 + }, + "overhead_p50_ms": 6.449, + "overhead_p95_ms": 1.252 + } +} diff --git a/meridian_control/README.md b/meridian_control/README.md new file mode 100644 index 0000000..c72ab43 --- /dev/null +++ b/meridian_control/README.md @@ -0,0 +1,67 @@ +# meridian-control + +The central control plane for [`meridian-node`](https://github.com/IMV-IN/meridian-node) +GPU agents. It is a **separate service** from the Meridian gateway (it does not +import or modify gateway code); the gateway consumes its serving projection +read-only. + +It implements the node control protocol (`contracts/v1` in the meridian-node +repo): approval-gated enrollment with a built-in Ed25519 CA, epoch-fenced +sessions with restore-safe fencing, leases, desired-state generations, and +observations. State is durable via SQLAlchemy — **SQLite by default**, Postgres +via a connection URL — so control replicas are stateless. + +See the meridian-node `DESIGN.md`, decision 13 and sections 8, 10, 15, and 17. + +## Install & run + +```bash +pip install -e ".[control]" # from the Meridian repo root + +# Run the control plane (SQLite by default) +meridian-control run --host 0.0.0.0 --port 8443 +# Or point at Postgres: +MERIDIAN_CONTROL_DB_URL=postgresql+psycopg://user:pass@host/meridian_control \ + meridian-control run + +# Mint a one-time enrollment token for a node +meridian-control mint-token --auto-approve +``` + +## Endpoints + +| Method + path | Purpose | +|---|---| +| `POST /control/v1/enroll` | Node enrollment (Bearer token) | +| `GET /control/v1/enroll/claims/{id}` | Pending-approval possession handshake | +| `POST /control/v1/nodes/{id}/sessions` | Establish a fenced session | +| `POST /control/v1/nodes/{id}/heartbeat` | Renew lease, get desired generation | +| `GET /control/v1/nodes/{id}/desired-state` | Fetch the desired snapshot | +| `POST /control/v1/nodes/{id}/observations` | Report observed state | +| `POST /admin/tokens` · `/admin/claims/{id}/approve` · `/admin/nodes/{id}/desired` · `/admin/nodes/{id}/stop-authorize` · `/admin/nodes/{id}/revoke` · `/admin/restore` | Operator actions | +| `GET /admin/projection` | Serving projection (gateway consumes read-only) | + +## Restore safety + +`POST /admin/restore {"high_water_epoch": N}` is the restore-from-backup runbook +step (DESIGN 17.5): it increments the control-plane incarnation and raises the +epoch floor past the last-issued high-water mark, so a fencing epoch that +regressed in restored data can never re-admit a previously fenced agent. + +## Tests + +```bash +pytest tests/control +``` + +`tests/control/test_cross_repo.py` runs the **real meridian-node agent** against +this service over a real HTTP socket — the cross-repository compatibility target +from the node's DESIGN §20/§24. It is skipped unless `meridian-node` is +installed (`pip install -e ../meridian-node[http]`). + +## Not yet included + +- Alembic migrations (schema is created via `create_all` for now). +- Gateway wiring: the serving projection exists (`GET /admin/projection`); the + gateway registry integration is a separate, independently-reviewed change. +- Certificate rotation/revocation automation and CRL distribution. diff --git a/meridian_control/__init__.py b/meridian_control/__init__.py new file mode 100644 index 0000000..04e6b1f --- /dev/null +++ b/meridian_control/__init__.py @@ -0,0 +1,13 @@ +"""meridian-control: the central control plane for meridian-node GPU agents. + +Implements the node control protocol (contracts/v1 in the meridian-node repo): +enrollment with a built-in Ed25519 CA, epoch-fenced sessions, leases, desired +state generations, and observations. Durable via SQLAlchemy (SQLite by default, +Postgres via a connection URL). This is a separate service from the Meridian +gateway; it does not import or modify gateway code. + +See the meridian-node DESIGN.md sections 2, 8, 10, 15, and 17. +""" + +__version__ = "0.1.0" +PROTOCOL_VERSION = "v1" diff --git a/meridian_control/app.py b/meridian_control/app.py new file mode 100644 index 0000000..679160e --- /dev/null +++ b/meridian_control/app.py @@ -0,0 +1,37 @@ +"""Application factory for meridian-control. + +Separate FastAPI app from the Meridian gateway. Wires the durable service, the +CA, and the routes. The gateway consumes the serving projection read-only and is +not imported here. +""" + +from __future__ import annotations + +from typing import Callable, Optional + +from fastapi import FastAPI + +from .ca import NodeCA +from .config import ControlConfig +from .db import make_session_factory +from .routes import register_error_handler, router +from .service import ControlService + + +def create_app(config: Optional[ControlConfig] = None, now: Optional[Callable] = None) -> FastAPI: + config = config or ControlConfig.from_env() + session_factory = make_session_factory(config.db_url) + ca = NodeCA.load_or_create(config.ca_dir) + service_kwargs = {"now": now} if now is not None else {} + service = ControlService(session_factory, ca, config, **service_kwargs) + + app = FastAPI(title="meridian-control", version="0.1.0") + app.state.control_service = service + app.include_router(router) + register_error_handler(app) + + @app.get("/healthz") + def healthz(): + return {"status": "ok"} + + return app diff --git a/meridian_control/ca.py b/meridian_control/ca.py new file mode 100644 index 0000000..1aba372 --- /dev/null +++ b/meridian_control/ca.py @@ -0,0 +1,83 @@ +"""Built-in minimal CA (DESIGN.md 15.2, decision 14). + +Issues Ed25519 node certificates bound to node identity. Node-cert profile only. +Behind this narrow interface so an enterprise PKI or cert-manager can replace it +without protocol changes. Keys are stored owner-only; KMS/HSM-backed storage is +a later option. +""" + +from __future__ import annotations + +import datetime as dt +import os +from pathlib import Path + +from cryptography import x509 +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey +from cryptography.x509.oid import NameOID + + +class NodeCA: + def __init__(self, key: Ed25519PrivateKey, cert: x509.Certificate): + self._key = key + self._cert = cert + + @classmethod + def load_or_create(cls, ca_dir: str | Path) -> "NodeCA": + d = Path(ca_dir) + key_path, cert_path = d / "ca_key.pem", d / "ca_cert.pem" + if key_path.exists() and cert_path.exists(): + key = serialization.load_pem_private_key(key_path.read_bytes(), password=None) + assert isinstance(key, Ed25519PrivateKey) + cert = x509.load_pem_x509_certificate(cert_path.read_bytes()) + return cls(key, cert) + + d.mkdir(parents=True, exist_ok=True) + key = Ed25519PrivateKey.generate() + subject = issuer = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "meridian-control-ca")]) + now = dt.datetime.now(dt.timezone.utc) + cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - dt.timedelta(minutes=1)) + .not_valid_after(now + dt.timedelta(days=3650)) + .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True) + .add_extension(x509.KeyUsage( + digital_signature=True, key_cert_sign=True, crl_sign=True, + key_encipherment=False, content_commitment=False, data_encipherment=False, + key_agreement=False, encipher_only=False, decipher_only=False, + ), critical=True) + .sign(key, None) + ) + fd = os.open(key_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "wb") as f: + f.write(key.private_bytes( + serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption() + )) + cert_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM)) + return cls(key, cert) + + def issue_node_cert(self, node_id: str, node_public_key: bytes, lifetime_hours: int = 24) -> str: + pub = Ed25519PublicKey.from_public_bytes(node_public_key) + now = dt.datetime.now(dt.timezone.utc) + cert = ( + x509.CertificateBuilder() + .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, node_id)])) + .issuer_name(self._cert.subject) + .public_key(pub) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - dt.timedelta(minutes=1)) + .not_valid_after(now + dt.timedelta(hours=lifetime_hours)) + .add_extension(x509.SubjectAlternativeName([x509.UniformResourceIdentifier(f"meridian-node://{node_id}")]), + critical=False) + .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True) + .sign(self._key, None) + ) + return cert.public_bytes(serialization.Encoding.PEM).decode() + + def trust_bundle(self) -> str: + return self._cert.public_bytes(serialization.Encoding.PEM).decode() diff --git a/meridian_control/cli.py b/meridian_control/cli.py new file mode 100644 index 0000000..c40c444 --- /dev/null +++ b/meridian_control/cli.py @@ -0,0 +1,43 @@ +"""Entry point: run the control plane, or mint an enrollment token.""" + +from __future__ import annotations + +import argparse +import sys + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="meridian-control") + sub = parser.add_subparsers(dest="command", required=True) + + run = sub.add_parser("run", help="run the control-plane HTTP server") + run.add_argument("--host", default="127.0.0.1") + run.add_argument("--port", type=int, default=8443) + + tok = sub.add_parser("mint-token", help="create a one-time enrollment token") + tok.add_argument("--auto-approve", action="store_true") + tok.add_argument("--ttl", type=int, default=3600) + + args = parser.parse_args(argv) + + if args.command == "run": + import uvicorn + + from .app import create_app + + uvicorn.run(create_app(), host=args.host, port=args.port) + return 0 + + if args.command == "mint-token": + from .app import create_app + + app = create_app() + token = app.state.control_service.create_token(auto_approve=args.auto_approve, ttl_seconds=args.ttl) + print(token) + return 0 + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/meridian_control/config.py b/meridian_control/config.py new file mode 100644 index 0000000..637227a --- /dev/null +++ b/meridian_control/config.py @@ -0,0 +1,24 @@ +"""Control-plane settings. SQLite by default; Postgres via MERIDIAN_CONTROL_DB_URL.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path + + +@dataclass +class ControlConfig: + db_url: str = "sqlite:///meridian_control.db" + ca_dir: Path = Path("./ca") + lease_ttl_seconds: int = 30 + heartbeat_interval_seconds: int = 10 + cert_lifetime_hours: int = 24 + + @classmethod + def from_env(cls) -> "ControlConfig": + return cls( + db_url=os.environ.get("MERIDIAN_CONTROL_DB_URL", cls.db_url), + ca_dir=Path(os.environ.get("MERIDIAN_CONTROL_CA_DIR", "./ca")), + lease_ttl_seconds=int(os.environ.get("MERIDIAN_CONTROL_LEASE_TTL", "30")), + ) diff --git a/meridian_control/db.py b/meridian_control/db.py new file mode 100644 index 0000000..8287478 --- /dev/null +++ b/meridian_control/db.py @@ -0,0 +1,21 @@ +"""SQLAlchemy engine/session wiring. Sync sessions, one per request.""" + +from __future__ import annotations + +from sqlalchemy import create_engine +from sqlalchemy.orm import DeclarativeBase, sessionmaker + + +class Base(DeclarativeBase): + pass + + +def make_engine(db_url: str): + connect_args = {"check_same_thread": False} if db_url.startswith("sqlite") else {} + return create_engine(db_url, connect_args=connect_args, future=True) + + +def make_session_factory(db_url: str): + engine = make_engine(db_url) + Base.metadata.create_all(engine) + return sessionmaker(bind=engine, expire_on_commit=False, future=True) diff --git a/meridian_control/models.py b/meridian_control/models.py new file mode 100644 index 0000000..ebae0bb --- /dev/null +++ b/meridian_control/models.py @@ -0,0 +1,97 @@ +"""ORM models for durable control-plane state (DESIGN.md 17). + +Distinct resources, never collapsed into one Backend object: nodes, enrollment +tokens, claims, desired snapshots, observations, stop authorizations, and the +singleton control-plane incarnation used for restore-safe fencing (17.5). +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Optional + +from sqlalchemy import JSON, DateTime, ForeignKey, Integer, LargeBinary, String, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column + +from .db import Base + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc) + + +class Incarnation(Base): + __tablename__ = "incarnation" + id: Mapped[int] = mapped_column(Integer, primary_key=True, default=1) + value: Mapped[int] = mapped_column(Integer, default=1) + # Epochs never regress below this floor; a restore runbook raises it past the + # last-issued high-water mark so a fenced agent can never be re-admitted. + epoch_floor: Mapped[int] = mapped_column(Integer, default=0) + + +class EnrollmentToken(Base): + __tablename__ = "enrollment_tokens" + token_hash: Mapped[str] = mapped_column(String(64), primary_key=True) + auto_approve: Mapped[bool] = mapped_column(default=False) + used: Mapped[bool] = mapped_column(default=False) + expires_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) + + +class Claim(Base): + __tablename__ = "claims" + claim_id: Mapped[str] = mapped_column(String(128), primary_key=True) + node_id: Mapped[str] = mapped_column(String(128)) + public_key: Mapped[bytes] = mapped_column(LargeBinary) + nonce: Mapped[str] = mapped_column(String(128)) + status: Mapped[str] = mapped_column(String(16), default="pending") # pending|approved|denied + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow) + + +class Node(Base): + __tablename__ = "nodes" + node_id: Mapped[str] = mapped_column(String(128), primary_key=True) + public_key: Mapped[bytes] = mapped_column(LargeBinary) + certificate_pem: Mapped[str] = mapped_column(String, default="") + display_name: Mapped[str] = mapped_column(String(253), default="") + labels: Mapped[dict] = mapped_column(JSON, default=dict) + active_session_id: Mapped[Optional[str]] = mapped_column(String(128), nullable=True) + fencing_epoch: Mapped[int] = mapped_column(Integer, default=0) + incarnation: Mapped[int] = mapped_column(Integer, default=1) + highest_sequence: Mapped[int] = mapped_column(Integer, default=0) + lease_expires_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) + desired_generation: Mapped[int] = mapped_column(Integer, default=0) + revoked: Mapped[bool] = mapped_column(default=False) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow) + + +class DesiredSnapshotRow(Base): + __tablename__ = "desired_snapshots" + __table_args__ = (UniqueConstraint("node_id", "generation", name="uq_node_generation"),) + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + node_id: Mapped[str] = mapped_column(String(128), ForeignKey("nodes.node_id")) + generation: Mapped[int] = mapped_column(Integer) + snapshot: Mapped[dict] = mapped_column(JSON) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow) + + +class ObservationRow(Base): + __tablename__ = "observations" + node_id: Mapped[str] = mapped_column(String(128), ForeignKey("nodes.node_id"), primary_key=True) + sequence: Mapped[int] = mapped_column(Integer, default=0) + observation: Mapped[dict] = mapped_column(JSON, default=dict) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow) + + +class StopAuthorization(Base): + __tablename__ = "stop_authorizations" + node_id: Mapped[str] = mapped_column(String(128), primary_key=True) + engine_id: Mapped[str] = mapped_column(String(128), primary_key=True) + + +class AuditEvent(Base): + __tablename__ = "audit_events" + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + node_id: Mapped[str] = mapped_column(String(128), default="") + kind: Mapped[str] = mapped_column(String(64)) + detail: Mapped[str] = mapped_column(String(1024), default="") + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow) diff --git a/meridian_control/routes.py b/meridian_control/routes.py new file mode 100644 index 0000000..0b7a63e --- /dev/null +++ b/meridian_control/routes.py @@ -0,0 +1,123 @@ +"""FastAPI routes for the node control protocol (contracts/v1) plus operator +admin endpoints. Sync handlers run in FastAPI's threadpool, matching the sync +SQLAlchemy service.""" + +from __future__ import annotations + +import base64 +from datetime import datetime, timezone +from typing import Optional, cast + +from fastapi import APIRouter, Body, Header, Query, Request +from fastapi.responses import JSONResponse + +from .service import ControlService, ControlServiceError + + +def _b64url_decode(s: str) -> bytes: + return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4)) + + +def _svc(request: Request) -> ControlService: + return cast(ControlService, request.app.state.control_service) + + +router = APIRouter() + + +# --- node protocol ------------------------------------------------------ +@router.post("/control/v1/enroll") +def enroll(request: Request, body: dict = Body(...), authorization: Optional[str] = Header(None)): + token = authorization[7:] if authorization and authorization.startswith("Bearer ") else None + return _svc(request).enroll(token, body) + + +@router.get("/control/v1/enroll/claims/{claim_id}") +def get_claim(request: Request, claim_id: str, x_possession_proof: Optional[str] = Header(None)): + svc = _svc(request) + if x_possession_proof is None: + return {"status": "pending", "claim_id": claim_id, "nonce": svc.claim_nonce(claim_id)} + return svc.resolve_claim(claim_id, _b64url_decode(x_possession_proof)) + + +@router.post("/control/v1/nodes/{node_id}/sessions") +def establish_session(request: Request, node_id: str, body: dict = Body(...)): + return _svc(request).establish_session(node_id, body) + + +@router.post("/control/v1/nodes/{node_id}/heartbeat") +def heartbeat(request: Request, node_id: str, body: dict = Body(...)): + return _svc(request).heartbeat(node_id, body) + + +@router.get("/control/v1/nodes/{node_id}/desired-state") +def desired_state(request: Request, node_id: str, generation: int = Query(0)): + return _svc(request).fetch_desired_state(node_id, generation) + + +@router.post("/control/v1/nodes/{node_id}/observations") +def observations(request: Request, node_id: str, body: dict = Body(...)): + return _svc(request).post_observation(node_id, body) + + +# --- operator admin ----------------------------------------------------- +@router.post("/admin/tokens") +def create_token(request: Request, body: dict = Body(default={})): + token = _svc(request).create_token( + auto_approve=bool(body.get("auto_approve", False)), + ttl_seconds=int(body.get("ttl_seconds", 3600)), + ) + return {"token": token} + + +@router.post("/admin/claims/{claim_id}/approve") +def approve_claim(request: Request, claim_id: str): + _svc(request).approve_claim(claim_id) + return {"ok": True} + + +@router.post("/admin/claims/{claim_id}/deny") +def deny_claim(request: Request, claim_id: str): + _svc(request).deny_claim(claim_id) + return {"ok": True} + + +@router.post("/admin/nodes/{node_id}/desired") +def set_desired(request: Request, node_id: str, body: dict = Body(...)): + _svc(request).set_desired(node_id, body) + return {"ok": True} + + +@router.post("/admin/nodes/{node_id}/stop-authorize") +def stop_authorize(request: Request, node_id: str, body: dict = Body(...)): + _svc(request).authorize_stop(node_id, list(body.get("engine_ids", []))) + return {"ok": True} + + +@router.post("/admin/nodes/{node_id}/revoke") +def revoke(request: Request, node_id: str): + _svc(request).revoke_node(node_id) + return {"ok": True} + + +@router.post("/admin/restore") +def restore(request: Request, body: dict = Body(...)): + incarnation = _svc(request).record_restore(int(body.get("high_water_epoch", 0))) + return {"incarnation": incarnation} + + +@router.get("/admin/projection") +def projection(request: Request): + return {"endpoints": _svc(request).serving_projection()} + + +def register_error_handler(app) -> None: + @app.exception_handler(ControlServiceError) + async def _handle(request: Request, exc: ControlServiceError): + return JSONResponse( + status_code=exc.http_status, + content={"error": { + "code": exc.code, "message": exc.message, "retryable": exc.retryable, + "server_time": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), + }}, + ) diff --git a/meridian_control/service.py b/meridian_control/service.py new file mode 100644 index 0000000..560ab2b --- /dev/null +++ b/meridian_control/service.py @@ -0,0 +1,345 @@ +"""Control-plane service logic (DESIGN.md 8.1, 10, 15, 17.5). + +Durable version of the node repo's reference core: single-use hashed tokens, +approval gating, real Ed25519 certificate issuance, epoch-fenced sessions with a +restore-safe epoch floor, sequence-checked heartbeats, desired-state +generations, and observations. All state lives in SQLAlchemy; nothing is held in +process memory across requests, so control replicas are stateless. +""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import secrets +from datetime import datetime, timedelta, timezone +from typing import Callable, Optional, cast + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey +from sqlalchemy import select + +from .ca import NodeCA +from .config import ControlConfig +from .models import ( + AuditEvent, + Claim, + DesiredSnapshotRow, + EnrollmentToken, + Incarnation, + Node, + ObservationRow, + StopAuthorization, +) + + +class ControlServiceError(Exception): + def __init__(self, code: str, message: str, retryable: bool = False, http_status: int = 400): + super().__init__(f"{code}: {message}") + self.code = code + self.message = message + self.retryable = retryable + self.http_status = http_status + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc) + + +def _iso(d: datetime) -> str: + if d.tzinfo is None: + d = d.replace(tzinfo=timezone.utc) + return d.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + + +def _b64url_decode(s: str) -> bytes: + return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4)) + + +def _hash_token(token: str) -> str: + return hashlib.sha256(token.encode()).hexdigest() + + +class ControlService: + def __init__(self, session_factory, ca: NodeCA, config: ControlConfig, now: Callable[[], datetime] = _utcnow): + self._sf = session_factory + self._ca = ca + self._config = config + self._now = now + + def _ttl(self) -> timedelta: + return timedelta(seconds=self._config.lease_ttl_seconds) + + def _incarnation(self, session) -> Incarnation: + inc = session.get(Incarnation, 1) + if inc is None: + inc = Incarnation(id=1, value=1, epoch_floor=0) + session.add(inc) + session.flush() + return cast(Incarnation, inc) + + # --- admin ---------------------------------------------------------- + def create_token(self, auto_approve: bool = False, ttl_seconds: int = 3600) -> str: + token = "enr_" + secrets.token_urlsafe(24) + with self._sf() as s: + s.add(EnrollmentToken( + token_hash=_hash_token(token), auto_approve=auto_approve, + expires_at=self._now() + timedelta(seconds=ttl_seconds), + )) + s.commit() + return token + + def approve_claim(self, claim_id: str) -> None: + self._set_claim_status(claim_id, "approved") + + def deny_claim(self, claim_id: str) -> None: + self._set_claim_status(claim_id, "denied") + + def _set_claim_status(self, claim_id: str, status: str) -> None: + with self._sf() as s: + claim = s.get(Claim, claim_id) + if claim is None: + raise ControlServiceError("NODE_NOT_FOUND", "unknown claim", http_status=404) + claim.status = status + s.commit() + + def set_desired(self, node_id: str, snapshot: dict) -> None: + with self._sf() as s: + node = s.get(Node, node_id) + if node is None: + raise ControlServiceError("NODE_NOT_FOUND", "unknown node", http_status=404) + generation = int(snapshot["generation"]) + existing = s.scalar( + select(DesiredSnapshotRow).where( + DesiredSnapshotRow.node_id == node_id, DesiredSnapshotRow.generation == generation + ) + ) + if existing is None: + s.add(DesiredSnapshotRow(node_id=node_id, generation=generation, snapshot=snapshot)) + else: + existing.snapshot = snapshot + if generation > node.desired_generation: + node.desired_generation = generation + s.commit() + + def authorize_stop(self, node_id: str, engine_ids: list[str]) -> None: + with self._sf() as s: + s.query(StopAuthorization).filter(StopAuthorization.node_id == node_id).delete() + for eid in engine_ids: + s.add(StopAuthorization(node_id=node_id, engine_id=eid)) + s.commit() + + def revoke_node(self, node_id: str) -> None: + with self._sf() as s: + node = s.get(Node, node_id) + if node is None: + raise ControlServiceError("NODE_NOT_FOUND", "unknown node", http_status=404) + node.revoked = True + s.add(AuditEvent(node_id=node_id, kind="node_revoked")) + s.commit() + + def record_restore(self, high_water_epoch: int) -> int: + """Restore-from-backup runbook step (DESIGN.md 17.5). Increments the + control-plane incarnation and raises the epoch floor past the last-issued + high-water mark so a regressed epoch can never re-admit a fenced agent.""" + with self._sf() as s: + inc = self._incarnation(s) + inc.value += 1 + inc.epoch_floor = max(inc.epoch_floor, high_water_epoch) + s.add(AuditEvent(kind="control_restore", detail=f"incarnation={inc.value} floor={inc.epoch_floor}")) + s.commit() + return inc.value + + # --- enrollment ----------------------------------------------------- + def enroll(self, token: Optional[str], request: dict) -> dict: + with self._sf() as s: + row = s.get(EnrollmentToken, _hash_token(token)) if token else None + if row is None: + raise ControlServiceError("INVALID_CREDENTIAL", "unknown enrollment token", http_status=401) + if row.used: + raise ControlServiceError("INVALID_CREDENTIAL", "enrollment token already used", http_status=401) + if row.expires_at is not None and self._now() > _aware(row.expires_at): + raise ControlServiceError("INVALID_CREDENTIAL", "enrollment token expired", http_status=401) + row.used = True + + node_id = "node_" + secrets.token_hex(12) + public_key = _b64url_decode(request["node_public_key"]) + if row.auto_approve: + self._create_node(s, node_id, public_key, request) + resp = self._approved_response(s, node_id) + s.add(AuditEvent(node_id=node_id, kind="enrolled_auto")) + s.commit() + return resp + + claim_id = "claim_" + secrets.token_hex(12) + s.add(Claim(claim_id=claim_id, node_id=node_id, public_key=public_key, nonce=secrets.token_hex(32))) + s.add(AuditEvent(node_id=node_id, kind="claim_created", detail=claim_id)) + s.commit() + return {"status": "pending", "claim_id": claim_id} + + def claim_nonce(self, claim_id: str) -> str: + with self._sf() as s: + claim = s.get(Claim, claim_id) + if claim is None: + raise ControlServiceError("NODE_NOT_FOUND", "unknown claim", http_status=404) + return str(claim.nonce) + + def resolve_claim(self, claim_id: str, signature: bytes) -> dict: + with self._sf() as s: + claim = s.get(Claim, claim_id) + if claim is None: + raise ControlServiceError("NODE_NOT_FOUND", "unknown claim", http_status=404) + try: + Ed25519PublicKey.from_public_bytes(claim.public_key).verify(signature, claim.nonce.encode()) + except InvalidSignature as e: + raise ControlServiceError("INVALID_CREDENTIAL", "claim possession proof failed", http_status=401) from e + if claim.status == "denied": + return {"status": "denied"} + if claim.status != "approved": + return {"status": "pending", "claim_id": claim_id} + if s.get(Node, claim.node_id) is None: + self._create_node(s, claim.node_id, claim.public_key, {}) + resp = self._approved_response(s, claim.node_id) + s.add(AuditEvent(node_id=claim.node_id, kind="enrolled_approved")) + s.commit() + return resp + + def _create_node(self, s, node_id: str, public_key: bytes, request: dict) -> None: + cert = self._ca.issue_node_cert(node_id, public_key, self._config.cert_lifetime_hours) + host_claims = request.get("host_claims", {}) if request else {} + s.add(Node( + node_id=node_id, public_key=public_key, certificate_pem=cert, + display_name=host_claims.get("hostname", ""), + )) + s.flush() + + def _approved_response(self, s, node_id: str) -> dict: + node = s.get(Node, node_id) + return { + "status": "approved", + "node_id": node_id, + "certificate": node.certificate_pem, + "trust_bundle": self._ca.trust_bundle(), + "protocol_version": "v1", + "lease": { + "heartbeat_interval_seconds": self._config.heartbeat_interval_seconds, + "lease_ttl_seconds": self._config.lease_ttl_seconds, + }, + } + + # --- session -------------------------------------------------------- + def establish_session(self, node_id: str, request: dict) -> dict: + with self._sf() as s: + node = s.get(Node, node_id) + if node is None: + raise ControlServiceError("NODE_NOT_FOUND", "unknown node", http_status=404) + if node.revoked: + raise ControlServiceError("NODE_NOT_AUTHORIZED", "node certificate revoked", http_status=403) + inc = self._incarnation(s) + session_id = request["agent_session_id"] + lease_valid = node.lease_expires_at is not None and self._now() < _aware(node.lease_expires_at) + has_active = node.active_session_id is not None and node.active_session_id != session_id + if lease_valid and has_active and not request.get("handoff_token"): + raise ControlServiceError( + "CONFLICT", "active session holds a valid lease; wait for expiry or supply a handoff token", + http_status=409, + ) + new_epoch = max(node.fencing_epoch, inc.epoch_floor) + 1 + node.fencing_epoch = new_epoch + node.incarnation = inc.value + node.active_session_id = session_id + node.lease_expires_at = self._now() + self._ttl() + s.add(AuditEvent(node_id=node_id, kind="session_established", detail=f"epoch={new_epoch}")) + s.commit() + return { + "fencing_epoch": new_epoch, + "lease_expires_at": _iso(node.lease_expires_at), + "session_assertion": self._session_assertion(node_id, session_id, new_epoch, inc.value), + } + + def _session_assertion(self, node_id: str, session_id: str, epoch: int, incarnation: int) -> str: + # A server-issued binding of identity+epoch+incarnation. Opaque to the + # agent today; the helper's epoch record is the enforced boundary. + payload = {"node_id": node_id, "session_id": session_id, "epoch": epoch, "incarnation": incarnation} + return base64.urlsafe_b64encode(json.dumps(payload, sort_keys=True).encode()).decode().rstrip("=") + + # --- heartbeat ------------------------------------------------------ + def heartbeat(self, node_id: str, request: dict) -> dict: + with self._sf() as s: + node = s.get(Node, node_id) + if node is None: + raise ControlServiceError("NODE_NOT_FOUND", "unknown node", http_status=404) + if node.revoked: + raise ControlServiceError("NODE_NOT_AUTHORIZED", "node certificate revoked", http_status=403) + if request["agent_session_id"] != node.active_session_id or request["fencing_epoch"] != node.fencing_epoch: + raise ControlServiceError("STALE_AGENT_SESSION", "agent session is no longer active", http_status=409) + if request["sequence"] <= node.highest_sequence: + raise ControlServiceError("CONFLICT", "sequence is not greater than highest accepted", http_status=409) + node.highest_sequence = request["sequence"] + node.lease_expires_at = self._now() + self._ttl() + stop_ids = [ + r.engine_id for r in s.scalars( + select(StopAuthorization).where(StopAuthorization.node_id == node_id) + ) + ] + s.commit() + return { + "server_time": _iso(self._now()), + "accepted_sequence": request["sequence"], + "fencing_epoch": node.fencing_epoch, + "lease_expires_at": _iso(node.lease_expires_at), + "desired_generation": node.desired_generation, + "admission_projection_revision": 0, + "stop_authorized_engine_ids": sorted(stop_ids), + } + + # --- desired state / observations ---------------------------------- + def fetch_desired_state(self, node_id: str, generation: int) -> dict: + with self._sf() as s: + if s.get(Node, node_id) is None: + raise ControlServiceError("NODE_NOT_FOUND", "unknown node", http_status=404) + row = s.scalar( + select(DesiredSnapshotRow).where(DesiredSnapshotRow.node_id == node_id) + .order_by(DesiredSnapshotRow.generation.desc()) + ) + if row is None: + raise ControlServiceError("GENERATION_NOT_FOUND", "no desired state published", http_status=404) + return dict(row.snapshot) + + def post_observation(self, node_id: str, observation: dict) -> dict: + with self._sf() as s: + if s.get(Node, node_id) is None: + raise ControlServiceError("NODE_NOT_FOUND", "unknown node", http_status=404) + seq = int(observation.get("sequence", 0)) + row = s.get(ObservationRow, node_id) + if row is None: + s.add(ObservationRow(node_id=node_id, sequence=seq, observation=observation)) + elif seq >= row.sequence: + row.sequence = seq + row.observation = observation + s.commit() + return {"received": True} + + # --- read models (operator / gateway projection) ------------------- + def serving_projection(self) -> list[dict]: + """Read-only projection the gateway consumes (DESIGN.md 17). Routable + endpoints are those with a valid lease and a Ready engine observation.""" + out: list[dict] = [] + with self._sf() as s: + for node in s.scalars(select(Node)): + lease_valid = node.lease_expires_at is not None and self._now() < _aware(node.lease_expires_at) + obs = s.get(ObservationRow, node.node_id) + engines = (obs.observation.get("engines", []) if obs else []) + for e in engines: + if e.get("phase") == "Ready" and e.get("endpoint"): + out.append({ + "engine_id": e["engine_id"], "node_id": node.node_id, + "endpoint": e["endpoint"], "provider": "managed", + "routable": bool(lease_valid) and not node.revoked, + }) + return out + + +def _aware(d: datetime) -> datetime: + return d if d.tzinfo is not None else d.replace(tzinfo=timezone.utc) diff --git a/mkdocs.yml b/mkdocs.yml index 04e1e81..ee6e136 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -69,6 +69,7 @@ nav: - Deployment guide: DEPLOY.md - Air-gapped installs: AIRGAP.md - Load and overhead: LOAD.md + - Real-engine validation: REAL_ENGINE_VALIDATION.md - Operate: - Operations runbook: OPS_RUNBOOK.md - Cost and usage: ENTERPRISE_COST.md diff --git a/pyproject.toml b/pyproject.toml index 0315387..f2d3107 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,6 +28,12 @@ audit = [ docs = [ "mkdocs-material>=9.5,<10", ] +# meridian-control: the central node control plane (DESIGN in the meridian-node +# repo). Isolated from the gateway install; only pulled in for the control +# service. SQLite by default, Postgres via a connection URL. +control = [ + "sqlalchemy>=2.0", +] dev = [ "ruff", "mypy", @@ -39,10 +45,16 @@ dev = [ # keeps working on a plain `[dev]` install. "aiokafka", "boto3", + # Control-plane service is part of the test suite. + "sqlalchemy>=2.0", ] [project.scripts] meridian = "meridian.cli.main:cli" +meridian-control = "meridian_control.cli:main" + +[tool.hatch.build.targets.wheel] +packages = ["meridian", "meridian_control"] [tool.ruff] target-version = "py39" diff --git a/scripts/validate_ollama.sh b/scripts/validate_ollama.sh new file mode 100755 index 0000000..7ca2151 --- /dev/null +++ b/scripts/validate_ollama.sh @@ -0,0 +1,54 @@ +#!/usr/bin/env sh +set -eu + +ROOT=$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd) +BACKEND_URL=${BACKEND_URL:-http://127.0.0.1:11434} +MODEL=${MODEL:-qwen2.5:0.5b} +OUTPUT=${OUTPUT:-$ROOT/docs/validation/ollama-v0.12.0.json} +PYTHON=${PYTHON:-python3} +OLLAMA_PID= + +cleanup() { + if [ -n "$OLLAMA_PID" ]; then + kill "$OLLAMA_PID" 2>/dev/null || true + wait "$OLLAMA_PID" 2>/dev/null || true + fi +} +trap cleanup EXIT INT TERM + +command -v ollama >/dev/null 2>&1 || { + printf '%s\n' "ollama is required: https://ollama.com/download" >&2 + exit 1 +} +command -v curl >/dev/null 2>&1 || { + printf '%s\n' "curl is required" >&2 + exit 1 +} + +if ! curl --fail --silent "$BACKEND_URL/v1/models" >/dev/null 2>&1; then + ollama serve >"${TMPDIR:-/tmp}/meridian-ollama.log" 2>&1 & + OLLAMA_PID=$! + attempts=0 + until curl --fail --silent "$BACKEND_URL/v1/models" >/dev/null 2>&1; do + attempts=$((attempts + 1)) + if [ "$attempts" -ge 60 ]; then + printf '%s\n' "Ollama did not become ready; see ${TMPDIR:-/tmp}/meridian-ollama.log" >&2 + exit 1 + fi + sleep 1 + done +fi + +ollama pull "$MODEL" +ENGINE_VERSION=$(ollama --version 2>&1 | sed -n 's/.*version is //p' | tail -n 1) +ENGINE_VERSION=${ENGINE_VERSION:-unknown} + +"$PYTHON" "$ROOT/scripts/validate_real_backend.py" \ + --engine ollama \ + --engine-version "$ENGINE_VERSION" \ + --backend-url "$BACKEND_URL" \ + --model "$MODEL" \ + --requests "${REQUESTS:-30}" \ + --concurrency "${CONCURRENCY:-1}" \ + --warmup "${WARMUP:-3}" \ + --output "$OUTPUT" diff --git a/scripts/validate_real_backend.py b/scripts/validate_real_backend.py new file mode 100755 index 0000000..5e494e8 --- /dev/null +++ b/scripts/validate_real_backend.py @@ -0,0 +1,227 @@ +#!/usr/bin/env python3 +"""Validate a live OpenAI-compatible backend through a disposable Meridian process.""" + +from __future__ import annotations + +import argparse +import json +import os +import platform +import subprocess +import sys +import tempfile +import time +import urllib.error +import urllib.request +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Optional + +import yaml + +ROOT = Path(__file__).resolve().parent.parent + + +def request_json( + url: str, *, body: Optional[dict[str, Any]] = None, timeout: float = 60.0 +) -> dict[str, Any]: + data = json.dumps(body).encode() if body is not None else None + headers = {"Content-Type": "application/json"} if data else {} + request = urllib.request.Request(url, data=data, headers=headers, method="POST" if data else "GET") + with urllib.request.urlopen(request, timeout=timeout) as response: + assert response.status == 200, f"{url} returned HTTP {response.status}" + return json.loads(response.read()) + + +def check_direct_backend(base_url: str, model: str) -> None: + models = request_json(f"{base_url}/v1/models", timeout=10) + assert models.get("data"), f"backend returned no models: {models}" + + body = { + "model": model, + "messages": [{"role": "user", "content": "Reply with OK."}], + "max_tokens": 8, + } + completion = request_json(f"{base_url}/v1/chat/completions", body=body) + assert completion.get("choices"), f"backend returned no choices: {completion}" + + stream_body = {**body, "stream": True} + data = json.dumps(stream_body).encode() + request = urllib.request.Request( + f"{base_url}/v1/chat/completions", + data=data, + headers={"Content-Type": "application/json"}, + method="POST", + ) + saw_done = False + with urllib.request.urlopen(request, timeout=60) as response: + for line in response: + if line.decode(errors="replace").strip() == "data: [DONE]": + saw_done = True + break + assert saw_done, "backend stream did not end with data: [DONE]" + + +def wait_for_gateway(url: str, process: subprocess.Popen[str], timeout: float = 30.0) -> None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if process.poll() is not None: + raise RuntimeError(f"Meridian exited before readiness with status {process.returncode}") + try: + request_json(f"{url}/meridian/status", timeout=2) + return + except (OSError, ValueError, AssertionError, urllib.error.URLError): + time.sleep(0.5) + raise RuntimeError(f"Meridian was not ready at {url} after {timeout:.0f}s") + + +def command_output(command: list[str]) -> Optional[str]: + try: + return subprocess.run(command, check=True, capture_output=True, text=True).stdout.strip() + except (OSError, subprocess.CalledProcessError): + return None + + +def gpu_metadata() -> Optional[dict[str, str]]: + output = command_output([ + "nvidia-smi", + "--query-gpu=name,memory.total,driver_version", + "--format=csv,noheader,nounits", + ]) + if not output: + return None + name, memory_mib, driver = (part.strip() for part in output.splitlines()[0].split(",", 2)) + return {"name": name, "memory_mib": memory_mib, "driver": driver} + + +def run_checked(command: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + print("+", " ".join(command), file=sys.stderr) + return subprocess.run(command, check=True, text=True, **kwargs) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--engine", required=True) + parser.add_argument("--engine-version", default="unknown") + parser.add_argument("--backend-url", required=True) + parser.add_argument("--model", required=True) + parser.add_argument("--model-revision") + parser.add_argument("--gateway-port", type=int, default=18080) + parser.add_argument("--requests", type=int, default=20) + parser.add_argument("--concurrency", type=int, default=1) + parser.add_argument("--warmup", type=int, default=3) + parser.add_argument("--output", type=Path) + args = parser.parse_args() + + backend_url = args.backend_url.rstrip("/") + gateway_url = f"http://127.0.0.1:{args.gateway_port}" + check_direct_backend(backend_url, args.model) + + with tempfile.TemporaryDirectory(prefix="meridian-real-backend-") as directory: + workdir = Path(directory) + config_path = workdir / "config.yaml" + log_path = workdir / "meridian.log" + config = { + "gateway": {"host": "127.0.0.1", "port": args.gateway_port, "strategy": "least_inflight"}, + "health": {"interval_s": 5, "timeout_s": 2, "fail_threshold": 2, "success_threshold": 1}, + "logging": {"level": "WARNING", "jsonl_path": str(workdir / "requests.jsonl")}, + "backends": [{ + "name": f"{args.engine}-validation", + "url": backend_url, + "engine": args.engine, + "model": args.model, + "weight": 1, + "health_endpoint": "/v1/models", + }], + } + config_path.write_text(yaml.safe_dump(config, sort_keys=False), encoding="utf-8") + environment = {**os.environ, "MERIDIAN_CONFIG": str(config_path)} + + with log_path.open("w", encoding="utf-8") as log: + process = subprocess.Popen( + [ + sys.executable, + "-m", + "uvicorn", + "meridian.api.main:app", + "--host", + "127.0.0.1", + "--port", + str(args.gateway_port), + ], + cwd=ROOT, + env=environment, + stdout=log, + stderr=subprocess.STDOUT, + text=True, + ) + try: + wait_for_gateway(gateway_url, process) + run_checked([ + sys.executable, + str(ROOT / "scripts/smoke_test.py"), + "--url", + gateway_url, + "--model", + args.model, + "--wait", + "0", + ]) + benchmark = run_checked([ + sys.executable, + str(ROOT / "scripts/bench_overhead.py"), + "--backend-url", + backend_url, + "--gateway-url", + gateway_url, + "--model", + args.model, + "--requests", + str(args.requests), + "--concurrency", + str(args.concurrency), + "--warmup", + str(args.warmup), + "--json", + ], capture_output=True) + finally: + process.terminate() + try: + process.wait(timeout=10) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + + if process.returncode not in (0, -15): + raise RuntimeError(f"Meridian exited with status {process.returncode}; log: {log_path.read_text()}") + + evidence = { + "schema_version": 1, + "recorded_at": datetime.now(timezone.utc).isoformat(), + "meridian_version": command_output([sys.executable, "-c", "import meridian; print(meridian.__version__)"]), + "engine": args.engine, + "engine_version": args.engine_version, + "backend_url": backend_url, + "model": args.model, + "model_revision": args.model_revision, + "host": {"platform": platform.platform(), "python": platform.python_version(), "gpu": gpu_metadata()}, + "checks": { + "direct_models": "passed", + "direct_chat": "passed", + "direct_stream": "passed", + "gateway_smoke": "passed", + }, + "benchmark": json.loads(benchmark.stdout), + } + rendered = json.dumps(evidence, indent=2) + "\n" + if args.output: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(rendered, encoding="utf-8") + print(f"Evidence written to {args.output}") + else: + print(rendered, end="") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/validate_vllm.sh b/scripts/validate_vllm.sh new file mode 100755 index 0000000..95d2b3c --- /dev/null +++ b/scripts/validate_vllm.sh @@ -0,0 +1,59 @@ +#!/usr/bin/env sh +set -eu + +ROOT=$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd) +IMAGE=${IMAGE:-vllm/vllm-openai:v0.10.2} +MODEL=${MODEL:-Qwen/Qwen2.5-0.5B-Instruct} +MODEL_REVISION=${MODEL_REVISION:-7ae557604adf67be50417f59c2c2f167def9a775} +BACKEND_PORT=${BACKEND_PORT:-18000} +BACKEND_URL=http://127.0.0.1:$BACKEND_PORT +OUTPUT=${OUTPUT:-$ROOT/docs/validation/vllm-v0.12.0.json} +PYTHON=${PYTHON:-python3} +CONTAINER=meridian-vllm-validation + +cleanup() { + docker rm -f "$CONTAINER" >/dev/null 2>&1 || true +} +trap cleanup EXIT INT TERM + +command -v docker >/dev/null 2>&1 || { + printf '%s\n' "docker is required" >&2 + exit 1 +} +command -v curl >/dev/null 2>&1 || { + printf '%s\n' "curl is required" >&2 + exit 1 +} + +cleanup +docker run --detach --name "$CONTAINER" --gpus all \ + --ipc=host \ + -p "127.0.0.1:$BACKEND_PORT:8000" \ + -v "${HF_HOME:-$HOME/.cache/huggingface}:/root/.cache/huggingface" \ + "$IMAGE" \ + --model "$MODEL" \ + --revision "$MODEL_REVISION" \ + --served-model-name "$MODEL" \ + --gpu-memory-utilization "${GPU_MEMORY_UTILIZATION:-0.75}" \ + --max-model-len "${MAX_MODEL_LEN:-4096}" + +attempts=0 +until curl --fail --silent "$BACKEND_URL/v1/models" >/dev/null 2>&1; do + attempts=$((attempts + 1)) + if [ "$attempts" -ge 300 ]; then + docker logs "$CONTAINER" >&2 + exit 1 + fi + sleep 2 +done + +"$PYTHON" "$ROOT/scripts/validate_real_backend.py" \ + --engine vllm \ + --engine-version "${IMAGE##*:}" \ + --backend-url "$BACKEND_URL" \ + --model "$MODEL" \ + --model-revision "$MODEL_REVISION" \ + --requests "${REQUESTS:-30}" \ + --concurrency "${CONCURRENCY:-1}" \ + --warmup "${WARMUP:-3}" \ + --output "$OUTPUT" diff --git a/tests/control/conftest.py b/tests/control/conftest.py new file mode 100644 index 0000000..833a937 --- /dev/null +++ b/tests/control/conftest.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +import base64 +from datetime import datetime, timedelta, timezone + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from meridian_control.ca import NodeCA +from meridian_control.config import ControlConfig +from meridian_control.db import make_session_factory +from meridian_control.service import ControlService + + +class MutableClock: + def __init__(self, start: datetime | None = None): + self._now = start or datetime(2026, 8, 1, 12, 0, 0, tzinfo=timezone.utc) + + def __call__(self) -> datetime: + return self._now + + def advance(self, seconds: float) -> None: + self._now = self._now + timedelta(seconds=seconds) + + +@pytest.fixture +def clock(): + return MutableClock() + + +def make_service(tmp_path, clock) -> ControlService: + cfg = ControlConfig(db_url=f"sqlite:///{tmp_path}/control.db", ca_dir=tmp_path / "ca", lease_ttl_seconds=30) + session_factory = make_session_factory(cfg.db_url) + ca = NodeCA.load_or_create(cfg.ca_dir) + return ControlService(session_factory, ca, cfg, now=clock) + + +class NodeKey: + """A node keypair with helpers mirroring the agent's identity operations.""" + + def __init__(self): + self.key = Ed25519PrivateKey.generate() + + def public_b64url(self) -> str: + raw = self.key.public_key().public_bytes( + encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw + ) + return base64.urlsafe_b64encode(raw).decode().rstrip("=") + + def sign_b64url(self, message: bytes) -> str: + return base64.urlsafe_b64encode(self.key.sign(message)).decode().rstrip("=") + + +@pytest.fixture +def node_key(): + return NodeKey() diff --git a/tests/control/test_api.py b/tests/control/test_api.py new file mode 100644 index 0000000..9e0bd56 --- /dev/null +++ b/tests/control/test_api.py @@ -0,0 +1,76 @@ +"""HTTP surface via FastAPI TestClient: the full enroll/session/heartbeat flow, +the pending-approval possession handshake, and the error envelope.""" + +from __future__ import annotations + +import pytest +from conftest import NodeKey +from fastapi.testclient import TestClient + +from meridian_control.app import create_app +from meridian_control.config import ControlConfig + + +@pytest.fixture +def client(tmp_path): + cfg = ControlConfig(db_url=f"sqlite:///{tmp_path}/api.db", ca_dir=tmp_path / "ca", lease_ttl_seconds=30) + return TestClient(create_app(cfg)) + + +def _mint(client, auto_approve=True): + return client.post("/admin/tokens", json={"auto_approve": auto_approve}).json()["token"] + + +def test_enroll_session_heartbeat_over_http(client): + key = NodeKey() + token = _mint(client, auto_approve=True) + r = client.post("/control/v1/enroll", headers={"Authorization": f"Bearer {token}"}, + json={"node_public_key": key.public_b64url()}) + assert r.status_code == 200 and r.json()["status"] == "approved" + node_id = r.json()["node_id"] + + s = client.post(f"/control/v1/nodes/{node_id}/sessions", json={"agent_session_id": "s1"}) + epoch = s.json()["fencing_epoch"] + h = client.post(f"/control/v1/nodes/{node_id}/heartbeat", + json={"agent_session_id": "s1", "fencing_epoch": epoch, "sequence": 1}) + assert h.status_code == 200 and h.json()["accepted_sequence"] == 1 + + +def test_pending_approval_handshake_over_http(client): + key = NodeKey() + token = _mint(client, auto_approve=False) + r = client.post("/control/v1/enroll", headers={"Authorization": f"Bearer {token}"}, + json={"node_public_key": key.public_b64url()}) + claim_id = r.json()["claim_id"] + + # Step 1: fetch the nonce. Step 2: prove possession. + nonce = client.get(f"/control/v1/enroll/claims/{claim_id}").json()["nonce"] + proof = key.sign_b64url(nonce.encode()) + pending = client.get(f"/control/v1/enroll/claims/{claim_id}", headers={"X-Possession-Proof": proof}) + assert pending.json()["status"] == "pending" + + client.post(f"/admin/claims/{claim_id}/approve") + approved = client.get(f"/control/v1/enroll/claims/{claim_id}", headers={"X-Possession-Proof": proof}) + assert approved.json()["status"] == "approved" + + +def test_stale_session_error_envelope(client): + key = NodeKey() + token = _mint(client, auto_approve=True) + node_id = client.post("/control/v1/enroll", headers={"Authorization": f"Bearer {token}"}, + json={"node_public_key": key.public_b64url()}).json()["node_id"] + client.post(f"/control/v1/nodes/{node_id}/sessions", json={"agent_session_id": "s1"}) + # Wrong epoch -> 409 with a typed error envelope. + r = client.post(f"/control/v1/nodes/{node_id}/heartbeat", + json={"agent_session_id": "s1", "fencing_epoch": 999, "sequence": 1}) + assert r.status_code == 409 + assert r.json()["error"]["code"] == "STALE_AGENT_SESSION" + assert r.json()["error"]["retryable"] is False + + +def test_invalid_token_rejected(client): + key = NodeKey() + r = client.post("/control/v1/enroll", headers={"Authorization": "Bearer nope"}, + json={"node_public_key": key.public_b64url()}) + assert r.status_code == 401 + assert r.json()["error"]["code"] == "INVALID_CREDENTIAL" diff --git a/tests/control/test_ca.py b/tests/control/test_ca.py new file mode 100644 index 0000000..c160f85 --- /dev/null +++ b/tests/control/test_ca.py @@ -0,0 +1,35 @@ +"""The built-in CA issues real Ed25519 node certificates chained to the CA.""" + +from __future__ import annotations + +from conftest import NodeKey +from cryptography import x509 +from cryptography.hazmat.primitives import serialization + +from meridian_control.ca import NodeCA + + +def test_issue_node_cert_binds_identity_and_chains_to_ca(tmp_path): + ca = NodeCA.load_or_create(tmp_path / "ca") + key = NodeKey() + raw_pub = key.key.public_key().public_bytes( + encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw + ) + pem = ca.issue_node_cert("node_abc", raw_pub, lifetime_hours=24) + cert = x509.load_pem_x509_certificate(pem.encode()) + + # Subject is the node identity, and the cert carries the node's public key. + assert cert.subject.rfc4514_string() == "CN=node_abc" + assert cert.public_key().public_bytes( + encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw + ) == raw_pub + + # The CA signature verifies against the CA's public key (chain of trust). + ca_cert = x509.load_pem_x509_certificate(ca.trust_bundle().encode()) + ca_cert.public_key().verify(cert.signature, cert.tbs_certificate_bytes) + + +def test_ca_is_stable_across_reload(tmp_path): + ca1 = NodeCA.load_or_create(tmp_path / "ca") + ca2 = NodeCA.load_or_create(tmp_path / "ca") + assert ca1.trust_bundle() == ca2.trust_bundle() diff --git a/tests/control/test_cross_repo.py b/tests/control/test_cross_repo.py new file mode 100644 index 0000000..dceeb43 --- /dev/null +++ b/tests/control/test_cross_repo.py @@ -0,0 +1,112 @@ +"""Cross-repository compatibility: the real meridian-node agent + HttpTransport +drive the real meridian-control app over a real HTTP socket (DESIGN.md 20, 24). + +This is the closed loop the two repos were designed for: the control service +runs under uvicorn on an ephemeral port, and the node agent's own transport +speaks to it. Skipped if meridian-node is not installed. +""" + +from __future__ import annotations + +import socket +import threading +import time + +import pytest + +from meridian_control.app import create_app +from meridian_control.config import ControlConfig + +pytest.importorskip("meridian_node") +from meridian_node.agent import Agent # noqa: E402 +from meridian_node.config import Config # noqa: E402 +from meridian_node.control import ControlError # noqa: E402 +from meridian_node.http_transport import HttpTransport # noqa: E402 + + +@pytest.fixture +def live(tmp_path): + import uvicorn + + cfg = ControlConfig(db_url=f"sqlite:///{tmp_path}/x.db", ca_dir=tmp_path / "ca", lease_ttl_seconds=30) + app = create_app(cfg) + port = _free_port() + server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=port, log_level="warning")) + thread = threading.Thread(target=server.run, daemon=True) + thread.start() + deadline = time.time() + 10 + while not server.started and time.time() < deadline: + time.sleep(0.02) + assert server.started, "control server did not start" + try: + yield app, f"http://127.0.0.1:{port}" + finally: + server.should_exit = True + thread.join(timeout=5) + + +def _free_port() -> int: + s = socket.socket() + s.bind(("127.0.0.1", 0)) + port = s.getsockname()[1] + s.close() + return port + + +def _agent(tmp_path, url, token, mode="observe-only"): + transport = HttpTransport(url, enrollment_token=token) + node_cfg = Config(control_plane_url=url, state_dir=tmp_path / "node", mode=mode) + return Agent(node_cfg, transport), transport + + +def test_enroll_session_heartbeat_across_repos(live, tmp_path): + app, url = live + token = app.state.control_service.create_token(auto_approve=True) + agent, _ = _agent(tmp_path, url, token) + + node_id = agent.ensure_enrolled() + from meridian_control.models import Node + with app.state.control_service._sf() as s: + assert s.get(Node, node_id) is not None # persisted in the durable store + agent.establish_session() + resp = agent.heartbeat_once() + assert resp["accepted_sequence"] == 1 + assert resp["desired_generation"] == 0 + + +def test_pending_approval_across_repos(live, tmp_path): + app, url = live + svc = app.state.control_service + token = svc.create_token(auto_approve=False) + agent, _ = _agent(tmp_path, url, token) + + with pytest.raises(ControlError): + agent.ensure_enrolled(max_polls=2) # not approved yet + + from meridian_control.models import Claim + with svc._sf() as s: + claim_id = s.query(Claim).first().claim_id + svc.approve_claim(claim_id) + node_id = agent.ensure_enrolled(max_polls=2) # resumes the persisted claim + assert node_id.startswith("node_") + + +def test_desired_fetch_and_observation_across_repos(live, tmp_path): + app, url = live + svc = app.state.control_service + token = svc.create_token(auto_approve=True) + agent, transport = _agent(tmp_path, url, token) + node_id = agent.ensure_enrolled() + agent.establish_session() + + svc.set_desired(node_id, {"node_id": node_id, "generation": 2, + "snapshot_hash": "sha256:" + "0" * 64, "assignments": []}) + resp = agent.heartbeat_once() + assert resp["desired_generation"] == 2 + assert transport.fetch_desired_state(node_id, 2)["generation"] == 2 + + transport.post_observation(node_id, {"node_id": node_id, "agent_session_id": agent.session_id, + "sequence": 1, "observed_generation": 2, "engines": []}) + from meridian_control.models import ObservationRow + with svc._sf() as s: + assert s.get(ObservationRow, node_id).observation["observed_generation"] == 2 diff --git a/tests/control/test_service.py b/tests/control/test_service.py new file mode 100644 index 0000000..25bc901 --- /dev/null +++ b/tests/control/test_service.py @@ -0,0 +1,150 @@ +"""Durable control-service logic: enrollment, fencing, leases, sequence +monotonicity, restore-safe fencing, revocation, desired state, projection.""" + +from __future__ import annotations + +import base64 + +import pytest +from conftest import make_service + +from meridian_control.service import ControlServiceError + + +def _sig_bytes(node_key, nonce: str) -> bytes: + return base64.urlsafe_b64decode(node_key.sign_b64url(nonce.encode()) + "==") + + +def _enroll_auto(svc, node_key): + token = svc.create_token(auto_approve=True) + return svc.enroll(token, {"node_public_key": node_key.public_b64url(), "host_claims": {"hostname": "h"}}) + + +def test_enroll_auto_issues_cert_and_token_is_single_use(tmp_path, clock, node_key): + svc = make_service(tmp_path, clock) + token = svc.create_token(auto_approve=True) + resp = svc.enroll(token, {"node_public_key": node_key.public_b64url()}) + assert resp["status"] == "approved" + assert "BEGIN CERTIFICATE" in resp["certificate"] + with pytest.raises(ControlServiceError) as exc: + svc.enroll(token, {"node_public_key": node_key.public_b64url()}) + assert exc.value.code == "INVALID_CREDENTIAL" + + +def test_pending_approval_flow(tmp_path, clock, node_key): + svc = make_service(tmp_path, clock) + token = svc.create_token(auto_approve=False) + resp = svc.enroll(token, {"node_public_key": node_key.public_b64url()}) + assert resp["status"] == "pending" + claim_id = resp["claim_id"] + + nonce = svc.claim_nonce(claim_id) + # Not approved yet -> still pending. + assert svc.resolve_claim(claim_id, _sig_bytes(node_key, nonce))["status"] == "pending" + + svc.approve_claim(claim_id) + approved = svc.resolve_claim(claim_id, _sig_bytes(node_key, nonce)) + assert approved["status"] == "approved" + assert approved["node_id"].startswith("node_") + + +def test_resolve_claim_rejects_bad_signature(tmp_path, clock, node_key): + svc = make_service(tmp_path, clock) + token = svc.create_token(auto_approve=False) + claim_id = svc.enroll(token, {"node_public_key": node_key.public_b64url()})["claim_id"] + svc.approve_claim(claim_id) + with pytest.raises(ControlServiceError) as exc: + svc.resolve_claim(claim_id, b"\x00" * 64) + assert exc.value.code == "INVALID_CREDENTIAL" + + +def test_session_and_heartbeat_sequence(tmp_path, clock, node_key): + svc = make_service(tmp_path, clock) + node_id = _enroll_auto(svc, node_key)["node_id"] + sess = svc.establish_session(node_id, {"agent_session_id": "s1"}) + epoch = sess["fencing_epoch"] + assert epoch == 1 + + hb = svc.heartbeat(node_id, {"agent_session_id": "s1", "fencing_epoch": epoch, "sequence": 5}) + assert hb["accepted_sequence"] == 5 + with pytest.raises(ControlServiceError) as exc: + svc.heartbeat(node_id, {"agent_session_id": "s1", "fencing_epoch": epoch, "sequence": 5}) + assert exc.value.code == "CONFLICT" + + +def test_stale_session_rejected_after_takeover(tmp_path, clock, node_key): + svc = make_service(tmp_path, clock) + node_id = _enroll_auto(svc, node_key)["node_id"] + svc.establish_session(node_id, {"agent_session_id": "s1"}) + svc.heartbeat(node_id, {"agent_session_id": "s1", "fencing_epoch": 1, "sequence": 1}) + + clock.advance(31) # lease expires + svc.establish_session(node_id, {"agent_session_id": "s2"}) # epoch 2 + with pytest.raises(ControlServiceError) as exc: + svc.heartbeat(node_id, {"agent_session_id": "s1", "fencing_epoch": 1, "sequence": 2}) + assert exc.value.code == "STALE_AGENT_SESSION" + + +def test_session_blocked_while_lease_valid(tmp_path, clock, node_key): + svc = make_service(tmp_path, clock) + node_id = _enroll_auto(svc, node_key)["node_id"] + svc.establish_session(node_id, {"agent_session_id": "s1"}) + with pytest.raises(ControlServiceError) as exc: + svc.establish_session(node_id, {"agent_session_id": "s2"}) + assert exc.value.code == "CONFLICT" + clock.advance(31) + assert svc.establish_session(node_id, {"agent_session_id": "s2"})["fencing_epoch"] == 2 + + +def test_restore_raises_epoch_floor_and_fences_old_sessions(tmp_path, clock, node_key): + svc = make_service(tmp_path, clock) + node_id = _enroll_auto(svc, node_key)["node_id"] + svc.establish_session(node_id, {"agent_session_id": "s1"}) # epoch 1 + + # Simulate restore from backup: ops record the last-issued high-water epoch. + incarnation = svc.record_restore(high_water_epoch=9) + assert incarnation == 2 + + clock.advance(31) + sess = svc.establish_session(node_id, {"agent_session_id": "s2"}) + assert sess["fencing_epoch"] == 10 # max(1, floor 9) + 1; never regresses + + # The pre-restore session can never re-admit itself. + with pytest.raises(ControlServiceError) as exc: + svc.heartbeat(node_id, {"agent_session_id": "s1", "fencing_epoch": 1, "sequence": 1}) + assert exc.value.code == "STALE_AGENT_SESSION" + + +def test_revoked_node_cannot_heartbeat(tmp_path, clock, node_key): + svc = make_service(tmp_path, clock) + node_id = _enroll_auto(svc, node_key)["node_id"] + svc.establish_session(node_id, {"agent_session_id": "s1"}) + svc.revoke_node(node_id) + with pytest.raises(ControlServiceError) as exc: + svc.heartbeat(node_id, {"agent_session_id": "s1", "fencing_epoch": 1, "sequence": 1}) + assert exc.value.code == "NODE_NOT_AUTHORIZED" + + +def test_desired_state_generation_and_fetch(tmp_path, clock, node_key): + svc = make_service(tmp_path, clock) + node_id = _enroll_auto(svc, node_key)["node_id"] + svc.set_desired(node_id, {"node_id": node_id, "generation": 3, "snapshot_hash": "sha256:x", "assignments": []}) + svc.establish_session(node_id, {"agent_session_id": "s1"}) + hb = svc.heartbeat(node_id, {"agent_session_id": "s1", "fencing_epoch": 1, "sequence": 1}) + assert hb["desired_generation"] == 3 + assert svc.fetch_desired_state(node_id, 3)["generation"] == 3 + + +def test_serving_projection_reflects_ready_and_lease(tmp_path, clock, node_key): + svc = make_service(tmp_path, clock) + node_id = _enroll_auto(svc, node_key)["node_id"] + svc.establish_session(node_id, {"agent_session_id": "s1"}) + svc.heartbeat(node_id, {"agent_session_id": "s1", "fencing_epoch": 1, "sequence": 1}) + svc.post_observation(node_id, {"sequence": 1, "engines": [ + {"engine_id": "e1", "phase": "Ready", "endpoint": "https://10.0.0.5:8000"} + ]}) + proj = svc.serving_projection() + assert proj and proj[0]["engine_id"] == "e1" and proj[0]["routable"] is True + + clock.advance(31) # lease expired -> not routable + assert svc.serving_projection()[0]["routable"] is False