feat: add workspace-scoped Neo4j query service

.dockerignore

+.git
+.venv
+__pycache__
+.pytest_cache
+.ruff_cache
+tests
+*.db*
+.env
+config.local.yaml
+acceptance.local.yaml
+secrets

.gitignore

+.venv/
+__pycache__/
+.pytest_cache/
+.ruff_cache/
+*.egg-info/
+dist/
+.env
+*.db*
+config.local.yaml
+acceptance.local.yaml
+secrets/

ACCEPTANCE.md

+# V1 验收记录
+
+验收日期：2026-09-10。项目版本：0.1.0。
+
+## 结果
+
+- 自动化测试：**92 passed，0 failed，0 skipped**，包含真实 Neo4j 5.26.0 Community 集成测试。
+- Docker 端到端：**10 项检查全部通过**。
+- Ruff 静态检查、格式检查通过。
+- 最终服务健康，地址 `http://127.0.0.1:8080`，Workspace 为 `local`。
+- 镜像：`neo4j-query-service:0.1.0`，Linux ARM64。
+- 镜像 ID：`sha256:0c98c2e7ef400a190338fa313e4c582f6d617e207237017c919889d421e5fa93`。
+
+## 要求与验证证据
+
+| 要求 | 验证 |
+|---|---|
+| Key 绑定 Workspace，路由指定 Workspace | 鉴权、跨 Workspace 资源与外键测试；容器请求返回正确 401/403/404 |
+| Workspace 仅启动配置 | 冻结配置、源文件修改不影响已生成配置、移除 Workspace 后旧 Key 不可用 |
+| Key 脚本实时新增/撤销 | 容器内 seed/revoke 前后服务 PID 相同，下一次请求立即生效 |
+| 本地 db 文件持久化 | SQLite WAL、并发写入、容器重启保留数据、在线备份 integrity_check=ok |
+| 环境变量生成配置 | 优先级、错误配置不覆盖旧文件；环境变量读取只出现在 bootstrap.py |
+| 非 root Docker | UID 10001、只读根文件系统、0600 运行配置、9080 端口覆盖后健康检查正常 |
+| 可信维护者管理模板 | reader 写模板被拒绝；更新生成版本，冲突返回 409，禁用阻止历史版本执行 |
+| 参数校验与只读边界 | 参数默认值、类型/范围/未知字段、注入字符串、写入与过程/管理/跨库语句拒绝 |
+| 快捷查询 | Workspace 共享、固定参数/版本、更新冲突、模板更新不改变旧快捷查询 |
+| 超时与并发限制 | 全局/Workspace/Key 上限、真实事务超时、真实 HTTP 断连取消、名额回收 |
+| 结果上限与图数据 | 截断标志、超大结果错误、小字节预算、节点/关系/路径/大整数/时间等类型 |
+| 访问日志 | 不含 Key/参数/结果、按数量与时间清理、审计失败返回 503、断连记录 499 |
+| 启动失败边界 | 损坏 SQLite 阻止服务启动、重复进程被文件锁拒绝 |
+| 依赖故障隔离 | 一个 Workspace 数据库不可达不影响其他 Workspace |
+
+## 真实数据库验收
+
+写入拒绝和压力测试运行在独立的 `neoquery-acceptance-db`，测试前后核对图数据和索引未被拒绝的语句改变。超时和取消测试后，检查剩余服务事务为 0。
+
+现有 `neo4j-agent` 仅用于限量读取：通过代理返回 3 个节点和 1 条路径。没有对现有库创建测试节点、修改图数据、索引或配置。APOC 的现有部署配置保持不变，代理仍拒绝过程调用。
+
+交付容器更新到最终镜像后，原 Key 和已保存快捷查询仍可用。初始维护者/只读 Key 保存在项目的 `secrets/bootstrap-keys.json`（0600；父目录 0700），不进入 Git、镜像或验收报告。
+
+## 复现入口
+
+- `uv run pytest -q --integration`：需要 README 所述的隔离 Neo4j 实例。
+- `tests/docker_acceptance.py`：Docker 端到端脚本；具体三 Workspace 环境要求见 README。
+- `query-service healthcheck`：读取生成配置中的监听端口检查交付服务。
+
+本轮未进行公网部署、多实例运行、APOC 放行或 AMD64 架构验收。导出的镜像为本机验证的 ARM64 版本；其他架构可从附带 Dockerfile 和锁文件构建。

Dockerfile

+FROM python:3.12-slim@sha256:78387bc3881b8273120a12ebe6c1ab22b018ccc2c9adf565ae1ac9b536e184ea
+WORKDIR /app
+RUN groupadd --gid 10001 queryservice && useradd --uid 10001 --gid queryservice --no-create-home queryservice
+COPY --from=ghcr.io/astral-sh/uv:0.8.22@sha256:9874eb7afe5ca16c363fe80b294fe700e460df29a55532bbfea234a0f12eddb1 /uv /usr/local/bin/uv
+COPY pyproject.toml uv.lock ./
+RUN uv sync --frozen --no-dev --no-install-project
+COPY src ./src
+COPY migrations ./migrations
+COPY alembic.ini ./
+RUN uv sync --frozen --no-dev && mkdir -p /data /run/query-service && chown -R queryservice:queryservice /data /run/query-service
+ENV PATH="/app/.venv/bin:$PATH" PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1
+USER 10001:10001
+EXPOSE 8080
+HEALTHCHECK --interval=30s --timeout=3s --start-period=15s CMD query-service healthcheck --config "${NQ_CONFIG_OUTPUT:-/run/query-service/config.yaml}"
+ENTRYPOINT ["query-service-bootstrap"]

alembic.ini

+[alembic]
+script_location = %(here)s/migrations

compose.yaml

+services:
+  query-api:
+    build: .
+    image: neo4j-query-service:0.1.0
+    ports:
+      - "127.0.0.1:8080:8080"
+    volumes:
+      - ./config.local.yaml:/etc/query-service/workspaces.yaml:ro
+      - query-data:/data
+    environment:
+      NQ_PORT: "8080"
+    read_only: true
+    tmpfs:
+      - /run/query-service:uid=10001,gid=10001,mode=0700
+      - /tmp
+    security_opt:
+      - no-new-privileges:true
+    cap_drop: [ALL]
+    restart: unless-stopped
+    networks: [neo4j]
+networks:
+  neo4j:
+    external: true
+    name: neo4jagent_default
+volumes:
+  query-data:

examples/template.json

+{
+  "name": "Entity neighbors",
+  "description": "Return one-hop paths for an entity with a business ID",
+  "cypher": "MATCH p=(n {entity_id: $entity_id})-[r]-(m) RETURN p LIMIT $limit",
+  "parameter_schema": {
+    "type": "object",
+    "additionalProperties": false,
+    "properties": {
+      "entity_id": {"type": "string", "minLength": 1, "maxLength": 128},
+      "limit": {"type": "integer", "minimum": 1, "maximum": 100, "default": 50}
+    },
+    "required": ["entity_id"]
+  },
+  "example_params": {"entity_id": "example"}
+}

examples/workspaces.yaml

+config_version: 1
+db_path: /data/service.db
+workspaces:
+  - id: team
+    name: Team graph
+    neo4j:
+      uri: bolt://neo4j:7687
+      database: neo4j
+      username: neo4j
+      password: replace-with-your-password

migrations/env.py

+from alembic import context
+
+with context.config.attributes["connection"].begin():
+    context.configure(connection=context.config.attributes["connection"], render_as_batch=True)
+    with context.begin_transaction():
+        context.run_migrations()

migrations/versions/0001_initial.py

+"""Initial local metadata schema; no Neo4j migrations."""
+
+from alembic import op
+
+revision = "0001"
+down_revision = None
+
+DDL = [
+    """CREATE TABLE workspaces (
+        id TEXT PRIMARY KEY, name TEXT NOT NULL, created_at TEXT NOT NULL)""",
+    """CREATE TABLE api_keys (
+        id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL REFERENCES workspaces(id),
+        name TEXT NOT NULL, secret_hash TEXT NOT NULL, role TEXT NOT NULL CHECK(role IN ('reader','maintainer')),
+        created_at TEXT NOT NULL, expires_at TEXT, revoked_at TEXT,
+        UNIQUE(workspace_id,id))""",
+    """CREATE TABLE templates (
+        id TEXT NOT NULL, workspace_id TEXT NOT NULL REFERENCES workspaces(id), name TEXT NOT NULL,
+        description TEXT NOT NULL, current_version INTEGER NOT NULL, enabled INTEGER NOT NULL DEFAULT 1,
+        created_at TEXT NOT NULL, PRIMARY KEY(workspace_id,id))""",
+    """CREATE TABLE template_versions (
+        workspace_id TEXT NOT NULL, template_id TEXT NOT NULL, version INTEGER NOT NULL,
+        cypher TEXT NOT NULL, parameter_schema TEXT NOT NULL, example_params TEXT NOT NULL,
+        created_by TEXT NOT NULL, created_at TEXT NOT NULL,
+        PRIMARY KEY(workspace_id,template_id,version),
+        FOREIGN KEY(workspace_id,template_id) REFERENCES templates(workspace_id,id),
+        FOREIGN KEY(workspace_id,created_by) REFERENCES api_keys(workspace_id,id))""",
+    """CREATE TABLE shortcuts (
+        id TEXT NOT NULL, workspace_id TEXT NOT NULL, name TEXT NOT NULL,
+        template_id TEXT NOT NULL, template_version INTEGER NOT NULL,
+        params TEXT NOT NULL, revision INTEGER NOT NULL DEFAULT 1,
+        created_by TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL,
+        PRIMARY KEY(workspace_id,id),
+        FOREIGN KEY(workspace_id,template_id,template_version)
+            REFERENCES template_versions(workspace_id,template_id,version),
+        FOREIGN KEY(workspace_id,created_by) REFERENCES api_keys(workspace_id,id))""",
+    """CREATE TABLE access_logs (
+        id INTEGER PRIMARY KEY AUTOINCREMENT, request_id TEXT NOT NULL, created_at TEXT NOT NULL,
+        workspace_id TEXT, key_id TEXT, operation TEXT NOT NULL, template_id TEXT,
+        template_version INTEGER, shortcut_id TEXT, status INTEGER NOT NULL,
+        duration_ms INTEGER NOT NULL, row_count INTEGER, error_code TEXT)""",
+    "CREATE INDEX ix_keys_workspace ON api_keys(workspace_id)",
+    "CREATE INDEX ix_logs_created ON access_logs(created_at)",
+]
+
+
+def upgrade():
+    for sql in DDL:
+        op.execute(sql)
+
+
+def downgrade():
+    raise RuntimeError("Restore a database backup instead of destructive downgrade")

pyproject.toml

+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "neo4j-query-service"
+version = "0.1.0"
+description = "Workspace-scoped, template-only Neo4j query API"
+requires-python = ">=3.12,<3.14"
+dependencies = [
+  "fastapi>=0.115,<1", "uvicorn>=0.34,<1", "neo4j>=5.26,<6",
+  "sqlalchemy>=2.0,<3", "alembic>=1.14,<2", "pydantic>=2.10,<3",
+  "pyyaml>=6.0,<7", "jsonschema>=4.23,<5", "typer>=0.15,<1",
+]
+
+[dependency-groups]
+dev = ["pytest>=8,<10", "pytest-asyncio>=0.25,<2", "httpx>=0.28,<1", "ruff>=0.11,<1"]
+
+[project.scripts]
+query-service = "neoquery.cli:app"
+query-service-bootstrap = "neoquery.bootstrap:main"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/neoquery"]
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+testpaths = ["tests"]
+
+[tool.ruff]
+target-version = "py312"
+line-length = 110
+
+[tool.ruff.lint]
+select = ["E4", "E7", "E9", "F", "I"]

src/neoquery/__init__.py

+"""Neo4j workspace query service."""

src/neoquery/bootstrap.py

+"""The ONLY module that reads environment variables; never imported by the server."""
+
+import json
+import os
+import tempfile
+from pathlib import Path
+
+import yaml
+
+from neoquery.config import Settings
+
+SCALARS = {
+    "NQ_LISTEN_HOST": ("listen_host", str),
+    "NQ_PORT": ("port", int),
+    "NQ_DB_PATH": ("db_path", str),
+    "NQ_LOG_RETENTION_DAYS": ("log_retention_days", int),
+    "NQ_LOG_MAX_RECORDS": ("log_max_records", int),
+}
+LIMITS = {
+    "NQ_QUERY_TIMEOUT_SECONDS": ("query_timeout_seconds", float),
+    "NQ_REQUEST_TIMEOUT_SECONDS": ("request_timeout_seconds", float),
+    "NQ_MAX_CONCURRENCY": ("max_concurrency", int),
+    "NQ_WORKSPACE_CONCURRENCY": ("workspace_concurrency", int),
+    "NQ_KEY_CONCURRENCY": ("key_concurrency", int),
+    "NQ_MAX_ROWS": ("max_rows", int),
+    "NQ_MAX_RESPONSE_BYTES": ("max_response_bytes", int),
+    "NQ_MAX_REQUEST_BYTES": ("max_request_bytes", int),
+}
+
+
+def merge_config(source: dict, env: dict) -> Settings:
+    data = json.loads(json.dumps(source))
+    known = (
+        set(SCALARS) | set(LIMITS) | {"NQ_CONFIG_SOURCE", "NQ_CONFIG_OUTPUT", "NQ_WORKSPACE_OVERRIDES_JSON"}
+    )
+    if any(key.startswith("NQ_") and key not in known for key in env):
+        raise ValueError("Unknown NQ_ environment setting")
+    for key, (field, cast) in SCALARS.items():
+        if key in env:
+            data[field] = cast(env[key])
+    for key, (field, cast) in LIMITS.items():
+        if key in env:
+            data.setdefault("limits", {})[field] = cast(env[key])
+    overrides = json.loads(env.get("NQ_WORKSPACE_OVERRIDES_JSON", "{}"))
+    by_id = {w["id"]: w for w in data.get("workspaces", [])}
+    for workspace_id, patch in overrides.items():
+        if (
+            workspace_id not in by_id
+            or not isinstance(patch, dict)
+            or set(patch) - {"neo4j", "limits", "name"}
+        ):
+            raise ValueError("Workspace overrides must reference existing IDs and allowed fields")
+        for field, value in patch.items():
+            if field in ("neo4j", "limits"):
+                by_id[workspace_id].setdefault(field, {}).update(value)
+            else:
+                by_id[workspace_id][field] = value
+    return Settings.model_validate(data)
+
+
+def generate(source: Path, output: Path, env: dict) -> Settings:
+    if source.resolve() == output.resolve():
+        raise ValueError("Source and generated config must be different files")
+    config = merge_config(yaml.safe_load(source.read_text()), env)
+    data = config.model_dump(mode="json")
+    for ws, original in zip(data["workspaces"], config.workspaces, strict=True):
+        ws["neo4j"]["password"] = original.neo4j.password.get_secret_value()
+    output.parent.mkdir(parents=True, exist_ok=True)
+    fd, name = tempfile.mkstemp(dir=output.parent, prefix=".config-")
+    try:
+        with os.fdopen(fd, "w") as file:
+            os.fchmod(file.fileno(), 0o600)
+            yaml.safe_dump(data, file, sort_keys=False)
+            file.flush()
+            os.fsync(file.fileno())
+        os.replace(name, output)
+    finally:
+        if Path(name).exists():
+            Path(name).unlink()
+    return config
+
+
+def main():
+    env = dict(os.environ)
+    output = Path(env.get("NQ_CONFIG_OUTPUT", "/run/query-service/config.yaml"))
+    try:
+        generate(Path(env.get("NQ_CONFIG_SOURCE", "/etc/query-service/workspaces.yaml")), output, env)
+    except Exception:
+        # Configuration validation may contain credentials; do not print raw input/exception.
+        raise SystemExit("Invalid startup configuration; check source and NQ_ settings") from None
+    os.execvp("query-service", ["query-service", "serve", "--config", str(output)])

src/neoquery/cli.py

+import fcntl
+import json
+import urllib.request
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+import typer
+import uvicorn
+
+from neoquery.config import load_settings
+from neoquery.db import Store
+
+app = typer.Typer(no_args_is_help=True, pretty_exceptions_show_locals=False)
+keys = typer.Typer(no_args_is_help=True, pretty_exceptions_show_locals=False)
+db = typer.Typer(no_args_is_help=True, pretty_exceptions_show_locals=False)
+app.add_typer(keys, name="keys")
+app.add_typer(db, name="db")
+
+
+@app.command()
+def healthcheck(config: Path = typer.Option(Path("/run/query-service/config.yaml"), exists=True)):
+    settings = load_settings(config)
+    with urllib.request.urlopen(f"http://127.0.0.1:{settings.port}/health/ready", timeout=2) as response:
+        if response.status != 200:
+            raise typer.Exit(1)
+
+
+def open_store(config: Path):
+    return Store(load_settings(config))
+
+
+@app.command()
+def serve(config: Path = typer.Option(..., exists=True)):
+    from neoquery.api import create_app
+
+    settings = load_settings(config)
+    # Prevent multiple servers with process-local limits using the same metadata DB.
+    lock_path = Path(settings.db_path + ".server.lock")
+    lock_path.parent.mkdir(parents=True, exist_ok=True)
+    with lock_path.open("a") as lock:
+        try:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            raise typer.BadParameter("A server is already using this metadata database") from None
+        store = Store(settings)
+        store.migrate()
+        uvicorn.run(
+            create_app(settings, store=store),
+            host=settings.listen_host,
+            port=settings.port,
+            workers=1,
+            access_log=False,
+            timeout_graceful_shutdown=15,
+        )
+
+
+@db.command()
+def migrate(config: Path = typer.Option(..., exists=True)):
+    store = open_store(config)
+    try:
+        store.migrate()
+    finally:
+        store.close()
+    typer.echo("Migration complete")
+
+
+@keys.command()
+def seed(
+    config: Path = typer.Option(..., exists=True),
+    workspace: str = typer.Option(...),
+    name: str = typer.Option(...),
+    role: str = typer.Option("reader"),
+    expires_in_days: int | None = typer.Option(None, min=1),
+):
+    store = open_store(config)
+    try:
+        expires = (
+            (datetime.now(UTC) + timedelta(days=expires_in_days)).isoformat() if expires_in_days else None
+        )
+        result = store.seed(workspace, name, role, expires)
+        typer.echo(json.dumps(result))
+    finally:
+        store.close()
+
+
+@keys.command("list")
+def list_keys(config: Path = typer.Option(..., exists=True), workspace: str = typer.Option(...)):
+    store = open_store(config)
+    try:
+        typer.echo(json.dumps(store.keys(workspace)))
+    finally:
+        store.close()
+
+
+@keys.command()
+def revoke(config: Path = typer.Option(..., exists=True), key_id: str = typer.Option(...)):
+    store = open_store(config)
+    try:
+        store.revoke(key_id)
+        typer.echo("Key revoked")
+    finally:
+        store.close()
+
+
+@db.command()
+def backup(config: Path = typer.Option(..., exists=True), output: Path = typer.Option(...)):
+    store = open_store(config)
+    try:
+        store.backup(str(output))
+        typer.echo("Backup complete")
+    finally:
+        store.close()

src/neoquery/codec.py

+import base64
+import math
+from datetime import date, datetime, time, timedelta
+
+from neo4j.graph import Node, Path, Relationship
+from neo4j.spatial import Point
+from neo4j.time import Date, DateTime, Duration, Time
+
+from neoquery.errors import ServiceError
+
+
+def encode(value, depth=0):
+    if depth > 32:
+        raise ServiceError(422, "result_nesting_limit", "Result nesting exceeds limit")
+
+    def child(v):
+        return encode(v, depth + 1)
+
+    if value is None:
+        kind, data = "null", None
+    elif isinstance(value, bool):
+        kind, data = "boolean", value
+    elif isinstance(value, int):
+        kind, data = "integer", str(value)
+    elif isinstance(value, float):
+        kind, data = "float", value if math.isfinite(value) else str(value)
+    elif isinstance(value, str):
+        kind, data = "string", value
+    elif isinstance(value, Node):
+        kind, data = (
+            "node",
+            {
+                "element_id": value.element_id,
+                "labels": sorted(value.labels),
+                "properties": {k: child(v) for k, v in value.items()},
+            },
+        )
+    elif isinstance(value, Relationship):
+        kind, data = (
+            "relationship",
+            {
+                "element_id": value.element_id,
+                "relationship_type": value.type,
+                "start_node_element_id": value.start_node.element_id,
+                "end_node_element_id": value.end_node.element_id,
+                "properties": {k: child(v) for k, v in value.items()},
+            },
+        )
+    elif isinstance(value, Path):
+        kind, data = (
+            "path",
+            {
+                "nodes": [child(v) for v in value.nodes],
+                "relationships": [child(v) for v in value.relationships],
+            },
+        )
+    elif isinstance(value, Point):
+        kind, data = "point", {"srid": value.srid, "coordinates": list(value)}
+    elif isinstance(value, Duration):
+        kind, data = (
+            "duration",
+            {
+                "months": value.months,
+                "days": value.days,
+                "seconds": value.seconds,
+                "nanoseconds": value.nanoseconds,
+            },
+        )
+    elif isinstance(value, (DateTime, datetime)):
+        kind, data = "datetime", value.iso_format() if isinstance(value, DateTime) else value.isoformat()
+    elif isinstance(value, (Date, date)):
+        kind, data = "date", value.iso_format() if isinstance(value, Date) else value.isoformat()
+    elif isinstance(value, (Time, time)):
+        kind, data = "time", value.iso_format() if isinstance(value, Time) else value.isoformat()
+    elif isinstance(value, timedelta):
+        kind, data = (
+            "duration",
+            {
+                "months": 0,
+                "days": value.days,
+                "seconds": value.seconds,
+                "nanoseconds": value.microseconds * 1000,
+            },
+        )
+    elif isinstance(value, (bytes, bytearray)):
+        kind, data = "bytes", base64.b64encode(value).decode("ascii")
+    elif isinstance(value, dict):
+        kind, data = "map", {k: child(v) for k, v in value.items()}
+    elif isinstance(value, (tuple, list)):
+        kind, data = "list", [child(v) for v in value]
+    else:
+        raise ServiceError(422, "unsupported_result_type", "Unsupported Neo4j result type")
+    return {"type": kind, "value": data}

src/neoquery/config.py

+from pathlib import Path
+from typing import Annotated
+
+import yaml
+from pydantic import BaseModel, ConfigDict, Field, SecretStr, model_validator
+
+Identifier = Annotated[str, Field(pattern=r"^[a-z][a-z0-9_-]{0,63}$")]
+
+
+class StrictModel(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+
+class Limits(StrictModel):
+    query_timeout_seconds: float = Field(default=10, gt=0, le=300)
+    request_timeout_seconds: float = Field(default=12, gt=0, le=360)
+    max_concurrency: int = Field(default=16, ge=1, le=1024)
+    workspace_concurrency: int = Field(default=4, ge=1, le=1024)
+    key_concurrency: int = Field(default=2, ge=1, le=1024)
+    max_rows: int = Field(default=1000, ge=1, le=100000)
+    max_response_bytes: int = Field(default=5 * 1024 * 1024, ge=1024, le=100 * 1024 * 1024)
+    max_request_bytes: int = Field(default=256 * 1024, ge=1024, le=10 * 1024 * 1024)
+
+
+class WorkspaceLimits(StrictModel):
+    query_timeout_seconds: float | None = Field(default=None, gt=0)
+    request_timeout_seconds: float | None = Field(default=None, gt=0)
+    max_concurrency: int | None = Field(default=None, ge=1)
+    key_concurrency: int | None = Field(default=None, ge=1)
+    max_rows: int | None = Field(default=None, ge=1)
+    max_response_bytes: int | None = Field(default=None, ge=1024)
+
+
+class Neo4jConfig(StrictModel):
+    uri: str = Field(pattern=r"^(bolt|neo4j)(\+s|\+ssc)?://[^\s]+$")
+    database: str = Field(min_length=1, max_length=63)
+    username: str = Field(min_length=1)
+    password: SecretStr
+
+    @model_validator(mode="after")
+    def check_target(self):
+        from urllib.parse import urlsplit
+
+        uri = urlsplit(self.uri)
+        if uri.username or uri.password or uri.query or uri.fragment or uri.path not in ("", "/"):
+            raise ValueError("Neo4j URI must contain only scheme, host and port")
+        if not uri.hostname or self.database.lower() == "system" or not self.password.get_secret_value():
+            raise ValueError("A non-system database, host and password are required")
+        return self
+
+
+class Workspace(StrictModel):
+    id: Identifier
+    name: str = Field(min_length=1, max_length=128)
+    neo4j: Neo4jConfig
+    limits: WorkspaceLimits = Field(default_factory=WorkspaceLimits)
+
+
+class Settings(StrictModel):
+    config_version: int = Field(default=1, ge=1, le=1)
+    listen_host: str = "0.0.0.0"
+    port: int = Field(default=8080, ge=1, le=65535)
+    db_path: str = "/data/service.db"
+    limits: Limits = Field(default_factory=Limits)
+    log_retention_days: int = Field(default=14, ge=1, le=3650)
+    log_max_records: int = Field(default=100000, ge=100)
+    workspaces: tuple[Workspace, ...] = Field(min_length=1)
+
+    @model_validator(mode="after")
+    def validate_settings(self):
+        ids = [w.id for w in self.workspaces]
+        if len(ids) != len(set(ids)):
+            raise ValueError("Duplicate workspace ID")
+        if not Path(self.db_path).is_absolute():
+            raise ValueError("db_path must be absolute")
+        return self
+
+    def workspace(self, workspace_id: str) -> Workspace:
+        return next(w for w in self.workspaces if w.id == workspace_id)
+
+    def effective_limits(self, workspace: Workspace) -> Limits:
+        values = self.limits.model_dump()
+        for name, value in workspace.limits.model_dump(exclude_none=True).items():
+            target = "workspace_concurrency" if name == "max_concurrency" else name
+            values[target] = min(values[target], value)
+        return Limits(**values)
+
+
+def load_settings(path: str | Path) -> Settings:
+    return Settings.model_validate(yaml.safe_load(Path(path).read_text()))

src/neoquery/errors.py

+class ServiceError(Exception):
+    def __init__(self, status: int, code: str, message: str):
+        self.status = status
+        self.code = code
+        self.message = message
+        super().__init__(message)

src/neoquery/executor.py

+import asyncio
+import json
+import time
+from collections import Counter
+from contextlib import contextmanager
+
+from neo4j import READ_ACCESS, AsyncGraphDatabase
+from neo4j.exceptions import DriverError, Neo4jError
+
+from neoquery.codec import encode
+from neoquery.errors import ServiceError
+from neoquery.validation import check_cypher
+
+
+class Admission:
+    def __init__(self, maximum):
+        self.maximum = maximum
+        self.total = 0
+        self.workspaces = Counter()
+        self.keys = Counter()
+
+    @contextmanager
+    def enter(self, workspace, key, limits):
+        # All callers run on one event loop. No await between checking and incrementing.
+        if (
+            self.total >= self.maximum
+            or self.workspaces[workspace] >= limits.workspace_concurrency
+            or self.keys[key] >= limits.key_concurrency
+        ):
+            raise ServiceError(429, "concurrency_limit", "Query concurrency limit reached")
+        self.total += 1
+        self.workspaces[workspace] += 1
+        self.keys[key] += 1
+        try:
+            yield
+        finally:
+            self.total -= 1
+            self.workspaces[workspace] -= 1
+            self.keys[key] -= 1
+            if not self.workspaces[workspace]:
+                del self.workspaces[workspace]
+            if not self.keys[key]:
+                del self.keys[key]
+
+
+class Executor:
+    def __init__(self, settings):
+        self.settings = settings
+        self.admission = Admission(settings.limits.max_concurrency)
+        self.drivers = {}
+        for workspace in settings.workspaces:
+            limits = settings.effective_limits(workspace)
+            self.drivers[workspace.id] = AsyncGraphDatabase.driver(
+                workspace.neo4j.uri,
+                auth=(workspace.neo4j.username, workspace.neo4j.password.get_secret_value()),
+                max_connection_pool_size=min(limits.workspace_concurrency, settings.limits.max_concurrency),
+                connection_acquisition_timeout=limits.request_timeout_seconds,
+                connection_timeout=limits.request_timeout_seconds,
+                max_transaction_retry_time=0,
+            )
+
+    async def close(self):
+        await asyncio.gather(*(driver.close() for driver in self.drivers.values()))
+
+    async def run(
+        self, workspace_id, key_id, cypher, params, *, request_id, max_rows=None, explain_only=False
+    ):
+        check_cypher(cypher)
+        workspace = self.settings.workspace(workspace_id)
+        limits = self.settings.effective_limits(workspace)
+        count_limit = min(max_rows or limits.max_rows, limits.max_rows)
+        started = time.monotonic()
+        session = None
+        try:
+            with self.admission.enter(workspace_id, key_id, limits):
+                async with asyncio.timeout(limits.request_timeout_seconds):
+                    session = self.drivers[workspace_id].session(
+                        database=workspace.neo4j.database, default_access_mode=READ_ACCESS, fetch_size=100
+                    )
+                    async with session:
+                        tx = await session.begin_transaction(
+                            timeout=limits.query_timeout_seconds,
+                            metadata={"request_id": request_id, "workspace": workspace_id},
+                        )
+                        try:
+                            plan = await tx.run("EXPLAIN " + cypher, params)
+                            summary = await plan.consume()
+                            if summary.query_type != "r":
+                                raise ServiceError(
+                                    422, "not_read_only", "Only explicitly read-only queries are allowed"
+                                )
+                            if explain_only:
+                                return {"validated": True}
+                            result = await tx.run(cypher, params)
+                            columns = list(result.keys())
+                            rows, truncated = [], False
+                            # Streaming fetch; never consume the full unbounded query into a list.
+                            # Track actual payload bytes; the HTTP boundary checks the complete envelope.
+                            size = len(
+                                json.dumps(columns, ensure_ascii=False, separators=(",", ":")).encode()
+                            )
+                            async for record in result:
+                                if len(rows) == count_limit:
+                                    truncated = True
+                                    break
+                                row = [encode(v) for v in record.values()]
+                                size += (
+                                    len(json.dumps(row, ensure_ascii=False, separators=(",", ":")).encode())
+                                    + 1
+                                )
+                                if size > limits.max_response_bytes:
+                                    raise ServiceError(
+                                        422, "response_too_large", "Query response exceeds byte limit"
+                                    )
+                                rows.append(row)
+                            return {
+                                "columns": columns,
+                                "rows": rows,
+                                "row_count": len(rows),
+                                "truncated": truncated,
+                                "duration_ms": int((time.monotonic() - started) * 1000),
+                            }
+                        except asyncio.CancelledError:
+                            session.cancel()
+                            raise
+                        finally:
+                            if not tx.closed():
+                                await tx.rollback()
+        except TimeoutError:
+            if session:
+                session.cancel()
+            raise ServiceError(504, "query_timeout", "Query deadline exceeded") from None
+        except asyncio.CancelledError:
+            if session:
+                session.cancel()
+            raise
+        except Neo4jError as error:
+            code = error.code or ""
+            if "TimedOut" in code or "Terminated" in code:
+                raise ServiceError(504, "query_timeout", "Neo4j transaction deadline exceeded") from None
+            if "Statement" in code:
+                raise ServiceError(
+                    422, "invalid_query", "Neo4j rejected the query or parameter types"
+                ) from None
+            raise ServiceError(503, "neo4j_unavailable", "Workspace database unavailable") from None
+        except (DriverError, OSError):
+            raise ServiceError(503, "neo4j_unavailable", "Workspace database unavailable") from None

src/neoquery/validation.py

+import math
+import re
+from copy import deepcopy
+
+from jsonschema import Draft202012Validator, validators
+from jsonschema.exceptions import SchemaError
+
+from neoquery.errors import ServiceError
+
+DENIED = set(
+    "CREATE MERGE SET DELETE DETACH REMOVE DROP ALTER GRANT DENY REVOKE "
+    "CALL LOAD USE SHOW TERMINATE START STOP RENAME FOREACH INSERT "
+    "EXPLAIN PROFILE CYPHER FINISH TRANSACTIONS".split()
+)
+FUNCTIONS = set(
+    "abs acos asin atan atan2 avg ceil ceiling coalesce collect cos cot count date datetime "
+    "degrees duration e elementid endnode exp floor haversin head id isempty isnan keys labels "
+    "last left length linenumber localdatetime localtime log log10 ltrim max min nodes normalize "
+    "nullif pi point properties radians rand range reduce relationships replace reverse right "
+    "round rtrim sign sin size split sqrt startnode stdev stdevp substring sum tail tan time "
+    "timestamp toboolean tobooleanlist tofloat tofloatlist tointeger tointegerlist tolower "
+    "tostring tostringlist toupper trim type valueType percentilecont percentiledisc "
+    "all any none single exists shortestpath allshortestpaths".lower().split()
+)
+PAREN_KEYWORDS = set(
+    "MATCH OPTIONAL WHERE WITH RETURN DISTINCT AND OR XOR NOT IN AS CASE WHEN THEN ELSE "
+    "UNWIND ORDER BY SKIP LIMIT UNION ALL EXISTS".split()
+)
+TOKEN = re.compile(
+    r"(?P<space>\s+)|(?P<line>//[^\n]*)|(?P<comment>/\*[\s\S]*?\*/)|"
+    r"(?P<string>'(?:\\.|''|[^'\\])*'|\"(?:\\.|\"\"|[^\"\\])*\")|"
+    r"(?P<quoted>`(?:``|[^`])*`)|(?P<param>\$[A-Za-z_][A-Za-z0-9_]*)|"
+    r"(?P<word>[A-Za-z_][A-Za-z0-9_]*)|(?P<number>\d+(?:\.\d+)?(?:[eE][+-]?\d+)?)|"
+    r"(?P<symbol>[^\w\s])"
+)
+
+
+def reject(message, code="invalid_template"):
+    raise ServiceError(422, code, message)
+
+
+def check_cypher(cypher: str) -> set[str]:
+    tokens, pos = [], 0
+    while pos < len(cypher):
+        match = TOKEN.match(cypher, pos)
+        if not match:
+            reject("Unsupported Cypher token; quote non-ASCII identifiers")
+        kind, value = match.lastgroup, match.group()
+        pos = match.end()
+        if kind not in ("space", "line", "comment"):
+            tokens.append((kind, value))
+    if not tokens:
+        reject("Empty query")
+    for index, (kind, value) in enumerate(tokens):
+        if kind == "word" and value.upper() in DENIED:
+            reject("Write, administration, procedure, database switching and query-prefix syntax is disabled")
+        if kind == "symbol" and value in (";", "'", '"', "`", "$", "\\"):
+            reject("Multiple statements or unsupported quoting")
+        if kind not in ("word", "quoted") or index + 1 >= len(tokens) or tokens[index + 1][1] != "(":
+            continue
+        name = value[1:-1].replace("``", "`") if kind == "quoted" else value
+        if kind == "word" and name.upper() in PAREN_KEYWORDS:
+            continue
+        if index > 0 and tokens[index - 1][1] == ".":
+            reject("Namespaced and custom functions are disabled")
+        if name.lower() not in FUNCTIONS:
+            reject("Function is outside the built-in allowlist")
+    return {value[1:] for kind, value in tokens if kind == "param"}
+
+
+ALLOWED_SCHEMA = {
+    "type",
+    "properties",
+    "required",
+    "additionalProperties",
+    "items",
+    "enum",
+    "default",
+    "description",
+    "minimum",
+    "maximum",
+    "exclusiveMinimum",
+    "exclusiveMaximum",
+    "minLength",
+    "maxLength",
+    "minItems",
+    "maxItems",
+    "minProperties",
+    "maxProperties",
+}
+TYPES = {"string", "integer", "number", "boolean", "null", "array", "object"}
+
+
+def check_schema(schema: dict, depth=0):
+    if depth > 12 or not isinstance(schema, dict) or set(schema) - ALLOWED_SCHEMA:
+        reject("Parameter schema uses unsupported keywords or exceeds nesting limit")
+    try:
+        Draft202012Validator.check_schema(schema)
+    except SchemaError:
+        reject("Invalid parameter schema")
+    kind = schema.get("type")
+    if not isinstance(kind, str) or kind not in TYPES:
+        reject("Each parameter schema must have one explicit supported type")
+    if kind == "object":
+        if schema.get("additionalProperties") is not False:
+            reject("Object schemas require additionalProperties=false")
+        for child in schema.get("properties", {}).values():
+            check_schema(child, depth + 1)
+    if kind == "array":
+        if "items" not in schema:
+            reject("Arrays require an item schema")
+        check_schema(schema["items"], depth + 1)
+    if "default" in schema:
+        validate_params(schema, schema["default"], root=False)
+
+
+def _defaults(schema, value):
+    if schema["type"] == "object" and isinstance(value, dict):
+        result = deepcopy(value)
+        for name, child in schema.get("properties", {}).items():
+            if name not in result and "default" in child:
+                result[name] = deepcopy(child["default"])
+            if name in result:
+                result[name] = _defaults(child, result[name])
+        return result
+    if schema["type"] == "array" and isinstance(value, list):
+        return [_defaults(schema["items"], item) for item in value]
+    return value
+
+
+def _finite(value, depth=0):
+    if depth > 32:
+        reject("Parameter nesting exceeds limit", "invalid_params")
+    if isinstance(value, float) and not math.isfinite(value):
+        reject("Non-finite parameter value", "invalid_params")
+    if isinstance(value, int) and not isinstance(value, bool) and not -(2**63) <= value < 2**63:
+        reject("Integer parameter outside Neo4j 64-bit range", "invalid_params")
+    if isinstance(value, dict):
+        for child in value.values():
+            _finite(child, depth + 1)
+    if isinstance(value, list):
+        for child in value:
+            _finite(child, depth + 1)
+
+
+StrictValidator = validators.extend(
+    Draft202012Validator,
+    type_checker=Draft202012Validator.TYPE_CHECKER.redefine("integer", lambda _, value: type(value) is int),
+)
+
+
+def validate_params(schema, params, root=True):
+    if root and (schema.get("type") != "object" or not isinstance(params, dict)):
+        reject("Parameters must be an object", "invalid_params")
+    value = _defaults(schema, params)
+    _finite(value)
+    errors = list(StrictValidator(schema).iter_errors(value))
+    if errors:
+        # Do not echo sensitive parameter values in error responses or logs.
+        reject("Parameters do not match the template schema", "invalid_params")
+    return value
+
+
+def validate_definition(cypher, schema, examples):
+    params = check_cypher(cypher)
+    check_schema(schema)
+    if schema.get("type") != "object" or params != set(schema.get("properties", {})):
+        reject("Declared parameter names must exactly match Cypher placeholders")
+    for name, child in schema.get("properties", {}).items():
+        if name not in schema.get("required", []) and "default" not in child:
+            reject("Each Cypher parameter must be required or have a default")
+    return validate_params(schema, examples)

tests/conftest.py

+from copy import deepcopy
+
+import httpx
+import pytest
+import pytest_asyncio
+
+from neoquery.api import create_app
+from neoquery.config import Settings
+from neoquery.db import Store
+
+EMPTY_SCHEMA = {"type": "object", "additionalProperties": False, "properties": {}}
+
+
+def pytest_addoption(parser):
+    parser.addoption("--integration", action="store_true", help="Use the disposable Neo4j at localhost:17687")
+
+
+@pytest.fixture
+def settings(tmp_path):
+    return Settings.model_validate(
+        {
+            "db_path": str(tmp_path / "service.db"),
+            "workspaces": [
+                {
+                    "id": name,
+                    "name": name,
+                    "neo4j": {
+                        "uri": "bolt://localhost:7687",
+                        "database": "neo4j",
+                        "username": "neo4j",
+                        "password": "test-secret",
+                    },
+                }
+                for name in ("alpha", "beta")
+            ],
+        }
+    )
+
+
+@pytest.fixture
+def store(settings):
+    store = Store(settings)
+    store.migrate()
+    yield store
+    store.close()
+
+
+@pytest.fixture
+def tokens(store):
+    return {
+        name: store.seed(workspace, name, role)
+        for name, workspace, role in (
+            ("admin", "alpha", "maintainer"),
+            ("reader", "alpha", "reader"),
+            ("peer", "alpha", "reader"),
+            ("outsider", "beta", "maintainer"),
+        )
+    }
+
+
+class FakeExecutor:
+    def __init__(self):
+        self.calls = []
+
+    async def run(self, workspace_id, key_id, cypher, params, **kwargs):
+        self.calls.append((workspace_id, cypher, deepcopy(params), kwargs))
+        if kwargs.get("explain_only"):
+            return {"validated": True}
+        return {
+            "columns": ["value"],
+            "rows": [[{"type": "integer", "value": "1"}]],
+            "row_count": 1,
+            "truncated": False,
+            "duration_ms": 0,
+        }
+
+    async def close(self):
+        pass
+
+
+@pytest.fixture
+def executor():
+    return FakeExecutor()
+
+
+@pytest_asyncio.fixture
+async def client(settings, store, executor):
+    app = create_app(settings, store=store, executor=executor)
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+        yield client
+
+
+def headers(token):
+    return {"Authorization": "Bearer " + token["key"]}
+
+
+def definition(**kwargs):
+    return {
+        "name": "Sample",
+        "description": "",
+        "cypher": "RETURN 1 AS value",
+        "parameter_schema": deepcopy(EMPTY_SCHEMA),
+        "example_params": {},
+        **kwargs,
+    }

tests/docker_acceptance.py

+"""End-to-end Docker acceptance. Local Workspace queries are strictly read-only.
+
+Run against the disposable API container configured with alpha/beta/local workspaces.
+Never prints credentials, query parameters or returned graph data.
+"""
+
+import argparse
+import concurrent.futures
+import json
+import subprocess
+import time
+from pathlib import Path
+
+import httpx
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--container", default="neoquery-acceptance-api")
+    parser.add_argument("--base-url", default="http://127.0.0.1:18081")
+    parser.add_argument("--report", type=Path, required=True)
+    args = parser.parse_args()
+    report = {"checks": {}, "local_queries": []}
+
+    def command(*parts):
+        result = subprocess.run(["docker", "exec", args.container, *parts], capture_output=True, text=True)
+        if result.returncode:
+            raise AssertionError("Docker management command failed (output withheld)")
+        return result.stdout
+
+    config = "/run/query-service/config.yaml"
+
+    def seed(workspace, role="reader"):
+        return json.loads(
+            command(
+                "query-service",
+                "keys",
+                "seed",
+                "--config",
+                config,
+                "--workspace",
+                workspace,
+                "--name",
+                "acceptance",
+                "--role",
+                role,
+            )
+        )
+
+    def request(method, path, key, body=None):
+        with httpx.Client(base_url=args.base_url, timeout=20) as client:
+            return client.request(method, path, headers={"Authorization": "Bearer " + key["key"]}, json=body)
+
+    def base(ws):
+        return f"/api/v1/workspaces/{ws}"
+
+    def template(ws, key, query, schema=None, examples=None):
+        response = request(
+            "POST",
+            base(ws) + "/templates",
+            key,
+            {
+                "name": "Acceptance query",
+                "cypher": query,
+                "parameter_schema": schema
+                or {"type": "object", "additionalProperties": False, "properties": {}},
+                "example_params": examples or {},
+            },
+        )
+        assert response.status_code == 201, (response.status_code, response.json().get("error"))
+        return response.json()
+
+    for _ in range(50):
+        try:
+            if httpx.get(args.base_url + "/health/ready").status_code == 200:
+                break
+        except httpx.HTTPError:
+            pass
+        time.sleep(0.2)
+    else:
+        raise AssertionError("API did not become ready")
+
+    initial_pid = subprocess.check_output(
+        ["docker", "inspect", args.container, "--format", "{{.State.Pid}}"], text=True
+    ).strip()
+    admin, reader, outsider = seed("alpha", "maintainer"), seed("alpha"), seed("beta")
+    assert request("GET", base("alpha") + "/templates", reader).status_code == 200
+    assert request("GET", base("alpha") + "/templates", outsider).status_code == 403
+    report["checks"]["workspace_isolation"] = True
+
+    schema = {
+        "type": "object",
+        "additionalProperties": False,
+        "properties": {"limit": {"type": "integer", "default": 3, "minimum": 1, "maximum": 20}},
+    }
+    tpl = template("alpha", admin, "UNWIND range(1,100) AS n RETURN n LIMIT $limit", schema)
+    saved = request(
+        "POST",
+        base("alpha") + "/shortcuts",
+        reader,
+        {"name": "Saved acceptance query", "template_id": tpl["id"]},
+    ).json()
+    assert saved["params"] == {"limit": 3} and saved["template_version"] == 1
+    peer = seed("alpha")
+    url = base("alpha") + "/shortcuts/" + saved["id"] + "/execute"
+    assert request("POST", url, peer, {}).json()["row_count"] == 3
+    assert request("POST", url, peer, {"params": {"limit": 1}}).status_code == 422
+    report["checks"]["shared_fixed_shortcut"] = True
+
+    def concurrent_work(index):
+        new_key = seed("alpha")
+        response = request(
+            "POST", base("alpha") + "/shortcuts", new_key, {"name": str(index), "template_id": tpl["id"]}
+        )
+        assert response.status_code == 201
+        assert request("GET", base("alpha") + "/templates", new_key).status_code == 200
+        return new_key
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
+        generated = list(pool.map(concurrent_work, range(8)))
+    command("query-service", "keys", "revoke", "--config", config, "--key-id", reader["id"])
+    assert request("GET", base("alpha") + "/templates", reader).status_code == 401
+    current_pid = subprocess.check_output(
+        ["docker", "inspect", args.container, "--format", "{{.State.Pid}}"], text=True
+    ).strip()
+    assert initial_pid == current_pid
+    report["checks"]["live_seed_revoke_without_restart"] = True
+    report["checks"]["concurrent_seed_shortcuts_logs"] = True
+
+    # Compile-only registration, then disconnect a real HTTP socket from an expensive read query.
+    expensive = template(
+        "alpha", admin, "UNWIND range(1,100000) AS x UNWIND range(1,100000) AS y RETURN sum(x*y)"
+    )
+    try:
+        httpx.post(
+            args.base_url + base("alpha") + f"/templates/{expensive['id']}/execute",
+            headers={"Authorization": "Bearer " + peer["key"]},
+            json={},
+            timeout=0.05,
+        )
+    except httpx.TimeoutException:
+        pass
+    else:
+        raise AssertionError("Expensive request should outlast the client timeout")
+    for _ in range(30):
+        raw = command(
+            "python",
+            "-c",
+            "import sqlite3,json; c=sqlite3.connect('/data/service.db'); "
+            'print(json.dumps(c.execute("SELECT count(*) FROM access_logs WHERE status=499").fetchone()[0]))',
+        )
+        if json.loads(raw) > 0:
+            break
+        time.sleep(0.1)
+    else:
+        raise AssertionError("Client disconnection not audited")
+    assert request("POST", url, peer, {}).status_code == 200
+    report["checks"]["real_http_disconnect_cancellation"] = True
+
+    # Existing local data is only sampled, never seeded/updated/deleted.
+    local_key = seed("local", "maintainer")
+    for cypher in ("MATCH (n) RETURN n LIMIT 3", "MATCH p=()-[]->() RETURN p LIMIT 1"):
+        local_tpl = template("local", local_key, cypher)
+        response = request("POST", base("local") + f"/templates/{local_tpl['id']}/execute", local_key, {})
+        assert response.status_code == 200, (response.status_code, response.json().get("error"))
+        payload = response.json()
+        report["local_queries"].append(
+            {
+                "rows": payload["row_count"],
+                "types": sorted({value["type"] for row in payload["rows"] for value in row}),
+            }
+        )
+    report["checks"]["existing_local_neo4j_read_only_queries"] = True
+
+    backup_path = f"/data/acceptance-backup-{time.time_ns()}.db"
+    command("query-service", "db", "backup", "--config", config, "--output", backup_path)
+    assert (
+        command(
+            "python",
+            "-c",
+            f"import sqlite3; print(sqlite3.connect({backup_path!r}).execute('PRAGMA integrity_check').fetchone()[0])",
+        ).strip()
+        == "ok"
+    )
+    report["checks"]["online_backup"] = True
+    source = command(
+        "python",
+        "-c",
+        "from pathlib import Path; import stat; p=Path('/run/query-service/config.yaml'); print(oct(stat.S_IMODE(p.stat().st_mode)))",
+    )
+    assert source.strip() == "0o600"
+    command("query-service", "healthcheck", "--config", config)
+    assert command("id", "-u").strip() == "10001"
+    report["checks"]["nonroot_private_config_environment_port_healthcheck"] = True
+
+    subprocess.run(["docker", "restart", args.container], capture_output=True, check=True)
+    for _ in range(40):
+        try:
+            response = request("POST", url, peer, {})
+            if response.status_code == 200:
+                break
+        except httpx.HTTPError:
+            pass
+        time.sleep(0.2)
+    else:
+        raise AssertionError("Persisted shortcut did not survive restart")
+    report["checks"]["restart_persistence_and_idempotent_migration"] = True
+
+    # Check log tables without printing their contents.
+    all_tokens = [admin, reader, outsider, peer, local_key, *generated]
+    raw_logs = command(
+        "python",
+        "-c",
+        "import sqlite3,json; print(json.dumps(sqlite3.connect('/data/service.db').execute('SELECT * FROM access_logs').fetchall()))",
+    )
+    assert all(item["key"].split(".")[1] not in raw_logs for item in all_tokens)
+    report["checks"]["log_redaction"] = True
+    args.report.write_text(json.dumps(report, indent=2))
+    print(json.dumps(report, indent=2))
+
+
+if __name__ == "__main__":
+    main()

tests/test_api.py

+import asyncio
+
+from conftest import definition, headers
+from sqlalchemy import text
+
+BASE = "/api/v1/workspaces/alpha"
+
+
+async def create(client, tokens, payload=None):
+    response = await client.post(
+        BASE + "/templates", headers=headers(tokens["admin"]), json=payload or definition()
+    )
+    assert response.status_code == 201, response.text
+    return response.json()
+
+
+async def test_auth_roles_and_cross_workspace(client, tokens):
+    assert (await client.get(BASE + "/templates")).status_code == 401
+    assert (await client.get(BASE + "/templates", headers=headers(tokens["outsider"]))).status_code == 403
+    assert (
+        await client.post(BASE + "/templates", headers=headers(tokens["reader"]), json=definition())
+    ).status_code == 403
+    template = await create(client, tokens)
+    foreign = "/api/v1/workspaces/beta/templates/" + template["id"]
+    assert (await client.get(foreign, headers=headers(tokens["outsider"]))).status_code == 404
+
+
+async def test_template_shortcut_version_fixed_shared_and_disabled(client, tokens, executor):
+    template = await create(client, tokens)
+    shortcut_response = await client.post(
+        BASE + "/shortcuts",
+        headers=headers(tokens["reader"]),
+        json={"name": "saved", "template_id": template["id"]},
+    )
+    assert shortcut_response.status_code == 201, shortcut_response.text
+    shortcut = shortcut_response.json()
+    assert shortcut["template_version"] == 1 and shortcut["params"] == {}
+    response = await client.put(
+        BASE + "/templates/" + template["id"],
+        headers=headers(tokens["admin"]),
+        json=definition(cypher="RETURN 2 AS value", expected_version=1),
+    )
+    assert response.status_code == 200, response.text
+    response = await client.post(
+        BASE + "/shortcuts/" + shortcut["id"] + "/execute", headers=headers(tokens["peer"]), json={}
+    )
+    assert response.status_code == 200, response.text
+    assert executor.calls[-1][1] == "RETURN 1 AS value"
+    assert response.json()["template_version"] == 1
+    assert (
+        await client.post(
+            BASE + "/shortcuts/" + shortcut["id"] + "/execute",
+            headers=headers(tokens["peer"]),
+            json={"params": {}},
+        )
+    ).status_code == 422
+    assert (
+        await client.put(
+            BASE + "/templates/" + template["id"],
+            headers=headers(tokens["admin"]),
+            json=definition(expected_version=1),
+        )
+    ).status_code == 409
+    assert (
+        await client.post(
+            BASE + "/templates/" + template["id"] + "/disable", headers=headers(tokens["admin"])
+        )
+    ).status_code == 200
+    assert (
+        await client.post(
+            BASE + "/shortcuts/" + shortcut["id"] + "/execute", headers=headers(tokens["peer"]), json={}
+        )
+    ).status_code == 409
+
+
+async def test_shortcut_crud_and_cross_reference(client, tokens):
+    template = await create(client, tokens)
+    payload = {"name": "saved", "template_id": template["id"]}
+    response = await client.post(
+        "/api/v1/workspaces/beta/shortcuts", headers=headers(tokens["outsider"]), json=payload
+    )
+    assert response.status_code == 404
+    result = (await client.post(BASE + "/shortcuts", headers=headers(tokens["reader"]), json=payload)).json()
+    path = BASE + "/shortcuts/" + result["id"]
+    assert (
+        await client.put(path, headers=headers(tokens["peer"]), json={**payload, "expected_revision": 1})
+    ).status_code == 200
+    assert (
+        await client.put(path, headers=headers(tokens["peer"]), json={**payload, "expected_revision": 1})
+    ).status_code == 409
+    assert len((await client.get(BASE + "/shortcuts", headers=headers(tokens["peer"]))).json()["items"]) == 1
+    assert (await client.delete(path, headers=headers(tokens["peer"]))).status_code == 200
+    assert (await client.get(path, headers=headers(tokens["reader"]))).status_code == 404
+
+
+async def test_seed_revoke_immediate_and_no_log_secrets(client, store, tokens):
+    fresh = await asyncio.to_thread(store.seed, "alpha", "live", "reader")
+    assert (await client.get(BASE + "/templates", headers=headers(fresh))).status_code == 200
+    store.revoke(fresh["id"])
+    assert (await client.get(BASE + "/templates", headers=headers(fresh))).status_code == 401
+    with store.engine.connect() as connection:
+        logs = str(connection.execute(text("SELECT * FROM access_logs")).all())
+    assert fresh["key"] not in logs and fresh["key"].split(".")[1] not in logs
+
+
+async def test_rejected_template_never_reaches_neo4j(client, tokens, executor):
+    response = await client.post(
+        BASE + "/templates", headers=headers(tokens["admin"]), json=definition(cypher="CREATE (n) RETURN n")
+    )
+    assert response.status_code == 422
+    assert not executor.calls
+
+
+async def test_body_limit_and_unknown_params(client, settings, tokens):
+    response = await client.post(
+        BASE + "/templates",
+        headers=headers(tokens["admin"]),
+        content=b"x" * (settings.limits.max_request_bytes + 1),
+    )
+    assert response.status_code == 413
+    template = await create(client, tokens)
+    response = await client.post(
+        BASE + "/templates/" + template["id"] + "/execute",
+        headers=headers(tokens["reader"]),
+        json={"params": {"secret": "dont-echo"}},
+    )
+    assert response.status_code == 422 and "dont-echo" not in response.text
+
+
+async def test_audit_failure_not_reported_as_success(client, tokens, store, monkeypatch):
+    def broken(data):
+        raise RuntimeError("disk full")
+
+    monkeypatch.setattr(store, "audit", broken)
+    response = await client.get(BASE + "/templates", headers=headers(tokens["reader"]))
+    assert response.status_code == 503 and response.json()["error"]["code"] == "audit_unavailable"
+
+
+async def test_health_and_openapi(client):
+    assert (await client.get("/health/live")).status_code == 200
+    assert (await client.get("/health/ready")).status_code == 200
+    assert (
+        "/api/v1/workspaces/{workspace_id}/templates" in (await client.get("/openapi.json")).json()["paths"]
+    )
+
+
+async def test_simultaneous_key_seed_shortcuts_and_logs(client, tokens, store):
+    template = await create(client, tokens)
+
+    async def save(index):
+        key = await asyncio.to_thread(store.seed, "alpha", str(index), "reader")
+        response = await client.post(
+            BASE + "/shortcuts",
+            headers=headers(key),
+            json={"name": str(index), "template_id": template["id"]},
+        )
+        assert response.status_code == 201, response.text
+
+    await asyncio.gather(*(save(i) for i in range(20)))

tests/test_boundaries.py

+import asyncio
+from datetime import date, datetime, time, timedelta
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+from conftest import definition, headers
+
+from neoquery.api import create_app
+from neoquery.codec import encode
+from neoquery.config import Limits, Settings
+from neoquery.errors import ServiceError
+from neoquery.executor import Admission, Executor
+
+
+def test_concurrency_all_three_boundaries_and_recovery():
+    limits = Limits(max_concurrency=3, workspace_concurrency=2, key_concurrency=1)
+    admission = Admission(3)
+    with admission.enter("a", "a1", limits):
+        with pytest.raises(ServiceError):
+            with admission.enter("a", "a1", limits):
+                pass
+        with admission.enter("a", "a2", limits):
+            with pytest.raises(ServiceError):
+                with admission.enter("a", "a3", limits):
+                    pass
+            with admission.enter("b", "b1", limits):
+                with pytest.raises(ServiceError):
+                    with admission.enter("c", "c1", limits):
+                        pass
+    assert admission.total == 0 and not admission.keys and not admission.workspaces
+    with pytest.raises(ValueError):
+        with admission.enter("a", "a1", limits):
+            raise ValueError()
+    assert admission.total == 0
+
+
+@pytest.mark.parametrize(
+    "value,kind",
+    [
+        (2**63 - 1, "integer"),
+        (True, "boolean"),
+        (None, "null"),
+        (b"\x00\xff", "bytes"),
+        (date(2026, 9, 9), "date"),
+        (datetime(2026, 9, 9, 12), "datetime"),
+        (time(12, 30), "time"),
+        (timedelta(days=1, microseconds=2), "duration"),
+        ({"type": "node", "value": [1, "x"]}, "map"),
+        (float("inf"), "float"),
+    ],
+)
+def test_codec(value, kind):
+    assert encode(value)["type"] == kind
+
+
+async def test_disconnect_cancels_endpoint_and_audits(settings, store, tokens):
+    entered, cancelled = asyncio.Event(), asyncio.Event()
+
+    class SlowExecutor:
+        async def run(self, *args, **kwargs):
+            entered.set()
+            try:
+                await asyncio.Event().wait()
+            finally:
+                cancelled.set()
+
+    template = store.save_template("alpha", tokens["admin"]["id"], definition())
+    app = create_app(settings, store=store, executor=SlowExecutor())
+    queue = asyncio.Queue()
+    await queue.put({"type": "http.request", "body": b"{}", "more_body": False})
+    responses = []
+
+    async def send(message):
+        responses.append(message)
+
+    scope = {
+        "type": "http",
+        "http_version": "1.1",
+        "method": "POST",
+        "scheme": "http",
+        "path": f"/api/v1/workspaces/alpha/templates/{template['id']}/execute",
+        "raw_path": b"/",
+        "query_string": b"",
+        "root_path": "",
+        "server": ("localhost", 80),
+        "client": ("localhost", 1000),
+        "headers": [
+            (b"content-type", b"application/json"),
+            (b"authorization", ("Bearer " + tokens["reader"]["key"]).encode()),
+        ],
+    }
+    task = asyncio.create_task(app(scope, queue.get, send))
+    await asyncio.wait_for(entered.wait(), timeout=2)
+    await queue.put({"type": "http.disconnect"})
+    await asyncio.wait_for(task, timeout=2)
+    assert cancelled.is_set() and responses == []
+    from sqlalchemy import text
+
+    with store.engine.connect() as connection:
+        row = connection.execute(
+            text("SELECT status,error_code FROM access_logs ORDER BY id DESC LIMIT 1")
+        ).first()
+    assert tuple(row) == (499, "client_disconnected")
+
+
+async def test_request_timeout_cancels_work(settings, store, tokens):
+    data = settings.model_dump()
+    data["limits"]["request_timeout_seconds"] = 0.05
+    settings = Settings.model_validate(data)
+    executor = AsyncMock()
+    cancelled = asyncio.Event()
+
+    async def slow(*args, **kwargs):
+        try:
+            await asyncio.Event().wait()
+        finally:
+            cancelled.set()
+
+    executor.run.side_effect = slow
+    template = store.save_template("alpha", tokens["admin"]["id"], definition())
+    app = create_app(settings, store=store, executor=executor)
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+        response = await client.post(
+            f"/api/v1/workspaces/alpha/templates/{template['id']}/execute",
+            json={},
+            headers=headers(tokens["reader"]),
+        )
+    assert response.status_code == 504 and cancelled.is_set()
+
+
+@pytest.mark.parametrize("query_type", [None, "rw", "w", "s", "unknown"])
+async def test_explain_unknown_or_write_type_never_executes(settings, query_type):
+    executor = Executor(settings)
+    driver = AsyncMock()
+    session = AsyncMock()
+    transaction = AsyncMock()
+    transaction.closed = lambda: False
+    session.__aenter__.return_value = session
+    session.begin_transaction.return_value = transaction
+    result = AsyncMock()
+    result.consume.return_value.query_type = query_type
+    transaction.run.return_value = result
+    # session() is a synchronous factory in the Neo4j driver.
+    driver.session = lambda **kwargs: session
+    original = executor.drivers["alpha"]
+    executor.drivers["alpha"] = driver
+    await original.close()
+    try:
+        with pytest.raises(ServiceError) as error:
+            await executor.run("alpha", "key", "RETURN 1", {}, request_id="test")
+        assert error.value.code == "not_read_only"
+        assert transaction.run.await_count == 1
+        transaction.rollback.assert_awaited_once()
+        assert executor.admission.total == 0
+    finally:
+        await executor.close()

tests/test_cli.py

+import fcntl
+from pathlib import Path
+from unittest.mock import patch
+
+import yaml
+from typer.testing import CliRunner
+
+from neoquery.cli import app
+
+
+def config_path(settings, tmp_path):
+    data = settings.model_dump(mode="json")
+    for workspace in data["workspaces"]:
+        workspace["neo4j"]["password"] = "test-secret"
+    path = tmp_path / "config.yaml"
+    path.write_text(yaml.safe_dump(data))
+    return path
+
+
+def test_server_refuses_second_process(settings, tmp_path):
+    config = config_path(settings, tmp_path)
+    lock_path = Path(settings.db_path + ".server.lock")
+    with lock_path.open("a") as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        with patch("neoquery.cli.uvicorn.run") as run:
+            result = CliRunner().invoke(app, ["serve", "--config", str(config)])
+    assert result.exit_code != 0
+    run.assert_not_called()
+
+
+def test_migration_failure_prevents_server_start(settings, tmp_path):
+    config = config_path(settings, tmp_path)
+    Path(settings.db_path).write_bytes(b"This is not a SQLite database")
+    with patch("neoquery.cli.uvicorn.run") as run:
+        result = CliRunner().invoke(app, ["serve", "--config", str(config)])
+    assert result.exit_code != 0
+    run.assert_not_called()

tests/test_config_store.py

+import json
+import sqlite3
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+import yaml
+from pydantic import ValidationError
+from sqlalchemy import text
+
+from neoquery.bootstrap import generate, merge_config
+from neoquery.config import Settings, load_settings
+from neoquery.db import Store
+from neoquery.errors import ServiceError
+
+
+def source(settings):
+    data = settings.model_dump(mode="json")
+    for ws in data["workspaces"]:
+        ws["neo4j"]["password"] = "test-secret"
+    return data
+
+
+def test_config_precedence_private_atomic_and_immutable(settings, tmp_path):
+    src, dest = tmp_path / "source.yaml", tmp_path / "generated.yaml"
+    src.write_text(yaml.safe_dump(source(settings)))
+    generated = generate(
+        src,
+        dest,
+        {
+            "NQ_PORT": "9000",
+            "NQ_MAX_ROWS": "10",
+            "NQ_WORKSPACE_OVERRIDES_JSON": json.dumps(
+                {"alpha": {"neo4j": {"password": "override-secret"}, "limits": {"max_rows": 10000}}}
+            ),
+        },
+    )
+    assert generated.port == 9000 and generated.limits.max_rows == 10
+    assert generated.effective_limits(generated.workspace("alpha")).max_rows == 10
+    assert generated.workspace("alpha").neo4j.password.get_secret_value() == "override-secret"
+    assert dest.stat().st_mode & 0o777 == 0o600
+    src.write_text("broken")
+    assert load_settings(dest) == generated
+    with pytest.raises(ValidationError):
+        generated.port = 1
+
+
+def test_bad_config_does_not_replace_existing(settings, tmp_path):
+    src, dest = tmp_path / "source.yaml", tmp_path / "config.yaml"
+    src.write_text(yaml.safe_dump(source(settings)))
+    generate(src, dest, {})
+    before = dest.read_bytes()
+    with pytest.raises(ValueError):
+        generate(src, dest, {"NQ_UNKNOWN": "not-supported"})
+    assert dest.read_bytes() == before
+
+
+@pytest.mark.parametrize(
+    "mutation",
+    [
+        lambda s: s["workspaces"].append(s["workspaces"][0]),
+        lambda s: s["workspaces"][0]["neo4j"].update(database="system"),
+        lambda s: s["workspaces"][0]["neo4j"].update(uri="bolt://admin:secret@localhost:7687"),
+        lambda s: s.update(db_path="relative.db"),
+        lambda s: s.update(unknown="value"),
+    ],
+)
+def test_invalid_config(settings, mutation):
+    data = source(settings)
+    mutation(data)
+    with pytest.raises(ValidationError):
+        Settings.model_validate(data)
+
+
+def test_unknown_workspace_override(settings):
+    with pytest.raises(ValueError):
+        merge_config(source(settings), {"NQ_WORKSPACE_OVERRIDES_JSON": '{"new":{"name":"no"}}'})
+
+
+def test_live_keys_revoke_expiry_and_no_plaintext(store, tokens):
+    fresh = store.seed("alpha", "new", "reader")
+    assert store.authenticate(fresh["key"], "alpha")["id"] == fresh["id"]
+    store.revoke(fresh["id"])
+    with pytest.raises(ServiceError) as exc:
+        store.authenticate(fresh["key"], "alpha")
+    assert exc.value.status == 401
+    expired = store.seed("alpha", "expired", "reader", "2020-01-01T00:00:00+00:00")
+    with pytest.raises(ServiceError):
+        store.authenticate(expired["key"], "alpha")
+    with store.engine.connect() as connection:
+        dump = json.dumps([dict(r) for r in connection.execute(text("SELECT * FROM api_keys")).mappings()])
+    assert fresh["key"].split(".")[1] not in dump
+
+
+def test_migration_repeat_persistence_and_backup(store, tokens, tmp_path):
+    store.migrate()
+    other = Store(store.settings)
+    try:
+        assert other.authenticate(tokens["reader"]["key"], "alpha")["role"] == "reader"
+        target = tmp_path / "backup.db"
+        other.backup(str(target))
+        with sqlite3.connect(target) as connection:
+            assert connection.execute("PRAGMA integrity_check").fetchone() == ("ok",)
+            assert connection.execute("SELECT COUNT(*) FROM api_keys").fetchone()[0] == 4
+    finally:
+        other.close()
+
+
+def test_seed_and_log_writes_concurrent(store):
+    def work(index):
+        key = store.seed("alpha", str(index), "reader")
+        store.audit({"request_id": str(index), "operation": "test", "status": 200})
+        return store.authenticate(key["key"], "alpha")
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        assert len(list(pool.map(work, range(40)))) == 40
+
+
+def test_audit_retention_count(store):
+    values = store.settings.model_dump()
+    values["log_max_records"] = 100
+    store.settings = Settings.model_validate(values)
+    for index in range(120):
+        store.audit({"request_id": str(index), "operation": "test", "status": 200})
+    store.prune_logs()
+    with store.engine.connect() as connection:
+        assert connection.execute(text("SELECT COUNT(*) FROM access_logs")).scalar() == 100
+
+
+def test_audit_retention_age(store):
+    store.audit({"request_id": "old", "operation": "test", "status": 200})
+    store.audit({"request_id": "new", "operation": "test", "status": 200})
+    with store.engine.begin() as connection:
+        connection.execute(
+            text("UPDATE access_logs SET created_at='2000-01-01T00:00:00+00:00' WHERE request_id='old'")
+        )
+    store.prune_logs()
+    with store.engine.connect() as connection:
+        assert connection.execute(text("SELECT request_id FROM access_logs")).scalars().all() == ["new"]
+
+
+def test_removed_workspace_key_disabled_without_deleting_metadata(store, tokens):
+    values = store.settings.model_dump()
+    values["workspaces"] = [values["workspaces"][1]]
+    store.settings = Settings.model_validate(values)
+    with pytest.raises(ServiceError) as error:
+        store.authenticate(tokens["reader"]["key"], "alpha")
+    assert error.value.code == "workspace_disabled"
+    assert len(store.keys("alpha")) == 3

tests/test_integration.py

+"""Real tests, opt in with --integration; target is deliberately fixed to disposable port 17687."""
+
+import asyncio
+from copy import deepcopy
+
+import httpx
+import pytest
+import pytest_asyncio
+from conftest import definition, headers
+from neo4j import GraphDatabase
+
+from neoquery.api import create_app
+from neoquery.config import Settings
+from neoquery.db import Store
+from neoquery.errors import ServiceError
+from neoquery.executor import Executor
+
+URI = "bolt://127.0.0.1:17687"
+AUTH = ("neo4j", "acceptance-only-2026")
+BASE = "/api/v1/workspaces/alpha"
+
+
+@pytest.fixture(scope="module")
+def graph(request):
+    if not request.config.getoption("--integration"):
+        pytest.skip("requires disposable Neo4j: --integration")
+    with GraphDatabase.driver(URI, auth=AUTH) as driver:
+        driver.verify_connectivity()
+        with driver.session(database="neo4j") as session:
+            session.run("MATCH (n:NQAcceptance) DETACH DELETE n").consume()
+            session.run(
+                "CREATE (a:NQAcceptance {entity_id:'a', name:'Alpha', huge:9223372036854775807}), "
+                "(b:NQAcceptance {entity_id:'b', name:'Beta'}), (a)-[:NQ_LINK {weight:1.5}]->(b)"
+            ).consume()
+        yield driver
+        with driver.session(database="neo4j") as session:
+            session.run("MATCH (n:NQAcceptance) DETACH DELETE n").consume()
+
+
+@pytest.fixture
+def real_settings(settings, graph):
+    data = settings.model_dump()
+    for workspace in data["workspaces"]:
+        workspace["neo4j"].update(uri=URI, password=AUTH[1])
+    return Settings.model_validate(data)
+
+
+@pytest_asyncio.fixture
+async def real(real_settings):
+    store = Store(real_settings)
+    store.migrate()
+    tokens = {role: store.seed("alpha", role, role) for role in ("reader", "maintainer")}
+    executor = Executor(real_settings)
+    app = create_app(real_settings, store=store, executor=executor)
+    async with app.router.lifespan_context(app):
+        async with httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=app), base_url="http://test"
+        ) as client:
+            yield client, tokens, executor, store
+
+
+async def post_template(client, tokens, cypher, **kwargs):
+    response = await client.post(
+        BASE + "/templates", headers=headers(tokens["maintainer"]), json=definition(cypher=cypher, **kwargs)
+    )
+    assert response.status_code == 201, response.text
+    return response.json()
+
+
+async def test_real_graph_types_and_query_contract(real, graph):
+    client, tokens, _, _ = real
+    template = await post_template(
+        client,
+        tokens,
+        "MATCH p=(a:NQAcceptance)-[r:NQ_LINK]->(b) "
+        "RETURN a,r,p,a.huge AS big,date('2026-09-09') AS d,datetime('2026-09-09T10:00:00Z') AS dt,"
+        "duration({days:2,seconds:3}) AS duration,point({x:1,y:2}) AS pt,"
+        "{items:[1,true,null]} AS nested",
+    )
+    response = await client.post(
+        BASE + f"/templates/{template['id']}/execute", headers=headers(tokens["reader"]), json={}
+    )
+    assert response.status_code == 200, response.text
+    values = response.json()["rows"][0]
+    assert [v["type"] for v in values] == [
+        "node",
+        "relationship",
+        "path",
+        "integer",
+        "date",
+        "datetime",
+        "duration",
+        "point",
+        "map",
+    ]
+    assert values[3]["value"] == "9223372036854775807"
+    assert values[1]["value"]["start_node_element_id"] == values[0]["value"]["element_id"]
+    assert len(values[2]["value"]["nodes"]) == 2
+
+
+async def test_real_safe_parameter_binding_and_fixed_shortcut(real, graph):
+    client, tokens, executor, store = real
+    schema = {
+        "type": "object",
+        "additionalProperties": False,
+        "properties": {"name": {"type": "string", "default": "Alpha"}},
+    }
+    template = await post_template(
+        client, tokens, "MATCH (n:NQAcceptance) WHERE n.name=$name RETURN n", parameter_schema=schema
+    )
+    response = await client.post(
+        BASE + f"/templates/{template['id']}/execute",
+        headers=headers(tokens["reader"]),
+        json={"params": {"name": "Alpha' DELETE n //"}},
+    )
+    assert response.status_code == 200 and response.json()["row_count"] == 0
+    saved = (
+        await client.post(
+            BASE + "/shortcuts",
+            headers=headers(tokens["reader"]),
+            json={"name": "fixed", "template_id": template["id"]},
+        )
+    ).json()
+    assert saved["params"] == {"name": "Alpha"}
+    response = await client.post(
+        BASE + f"/shortcuts/{saved['id']}/execute", headers=headers(tokens["reader"]), json={}
+    )
+    assert response.status_code == 200 and response.json()["row_count"] == 1
+    assert executor.admission.total == 0
+
+
+async def test_real_row_limit_and_exact_truncation(real):
+    client, tokens, _, _ = real
+    template = await post_template(client, tokens, "UNWIND range(1,1000000) AS n RETURN n")
+    for maximum in (3, 1000, 99999):
+        response = await client.post(
+            BASE + f"/templates/{template['id']}/execute",
+            headers=headers(tokens["reader"]),
+            json={"max_rows": maximum},
+        )
+        assert response.status_code == 200, response.text
+        assert response.json()["row_count"] == min(maximum, 1000) and response.json()["truncated"]
+    exact = await post_template(client, tokens, "UNWIND range(1,3) AS n RETURN n")
+    result = (
+        await client.post(
+            BASE + f"/templates/{exact['id']}/execute",
+            headers=headers(tokens["reader"]),
+            json={"max_rows": 3},
+        )
+    ).json()
+    assert result["row_count"] == 3 and not result["truncated"]
+
+
+async def test_real_write_rejection_preserves_graph_and_schema(real, graph):
+    client, tokens, _, _ = real
+
+    def snapshot():
+        with graph.session(database="neo4j") as session:
+            counts = session.run("MATCH (n) RETURN count(n) AS n").single()["n"]
+            rels = session.run("MATCH ()-[r]->() RETURN count(r) AS r").single()["r"]
+            indexes = sorted(r["name"] for r in session.run("SHOW INDEXES YIELD name RETURN name"))
+            names = sorted(r["name"] for r in session.run("MATCH (n:NQAcceptance) RETURN n.name AS name"))
+            return counts, rels, indexes, names
+
+    before = snapshot()
+    for cypher in [
+        "CREATE (n:NQBad) RETURN n",
+        "MATCH (n) DETACH DELETE n",
+        "MATCH (n) SET n.name='bad' RETURN n",
+        "CREATE INDEX forbidden FOR (n:NQBad) ON (n.x)",
+        "CALL db.labels()",
+        "USE system SHOW USERS",
+    ]:
+        response = await client.post(
+            BASE + "/templates", headers=headers(tokens["maintainer"]), json=definition(cypher=cypher)
+        )
+        assert response.status_code == 422
+    assert snapshot() == before
+
+
+async def test_real_timeout_cancel_and_capacity_recovery(real_settings, graph):
+    values = real_settings.model_dump()
+    values["limits"].update(request_timeout_seconds=0.25, query_timeout_seconds=0.2, key_concurrency=1)
+    settings = Settings.model_validate(values)
+    executor = Executor(settings)
+    expensive = "UNWIND range(1,100000) AS x UNWIND range(1,100000) AS y RETURN sum(x*y)"
+    try:
+        with pytest.raises(ServiceError) as exc:
+            await executor.run("alpha", "a", expensive, {}, request_id="timeout-case")
+        assert exc.value.status == 504
+        assert executor.admission.total == 0
+        normal = await executor.run("alpha", "a", "RETURN 1 AS value", {}, request_id="after-timeout")
+        assert normal["row_count"] == 1
+        task = asyncio.create_task(executor.run("alpha", "a", expensive, {}, request_id="cancel-case"))
+        await asyncio.sleep(0.05)
+        with pytest.raises(ServiceError) as exc:
+            await executor.run("alpha", "a", "RETURN 1", {}, request_id="over-cap")
+        assert exc.value.status == 429
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        assert executor.admission.total == 0
+        assert (await executor.run("alpha", "a", "RETURN 1", {}, request_id="after-cancel"))["row_count"] == 1
+    finally:
+        await executor.close()
+
+
+async def test_real_byte_limit_and_error_recovery(real_settings):
+    values = real_settings.model_dump()
+    values["limits"]["max_response_bytes"] = 2048
+    executor = Executor(Settings.model_validate(values))
+    try:
+        with pytest.raises(ServiceError) as exc:
+            await executor.run("alpha", "a", "RETURN $large AS x", {"large": "x" * 10000}, request_id="large")
+        assert exc.value.code == "response_too_large"
+        assert executor.admission.total == 0
+        with pytest.raises(ServiceError) as exc:
+            await executor.run("alpha", "a", "MATCH this is invalid RETURN 1", {}, request_id="syntax")
+        assert exc.value.status == 422
+        assert (await executor.run("alpha", "a", "RETURN 1", {}, request_id="after-error"))["row_count"] == 1
+    finally:
+        await executor.close()
+
+
+async def test_unavailable_workspace_does_not_break_others(real_settings):
+    values = deepcopy(real_settings.model_dump())
+    values["workspaces"][1]["neo4j"]["uri"] = "bolt://127.0.0.1:17688"
+    executor = Executor(Settings.model_validate(values))
+    try:
+        with pytest.raises(ServiceError) as exc:
+            await executor.run("beta", "b", "RETURN 1", {}, request_id="offline")
+        assert exc.value.status == 503
+        assert (await executor.run("alpha", "a", "RETURN 1", {}, request_id="online"))["row_count"] == 1
+    finally:
+        await executor.close()
+
+
+async def test_small_valid_byte_budget_still_allows_small_results(real_settings):
+    values = real_settings.model_dump()
+    values["limits"]["max_response_bytes"] = 1024
+    executor = Executor(Settings.model_validate(values))
+    try:
+        result = await executor.run("alpha", "key", "RETURN 1", {}, request_id="small-budget")
+        assert result["row_count"] == 1
+    finally:
+        await executor.close()

tests/test_validation.py

+import pytest
+
+from neoquery.errors import ServiceError
+from neoquery.validation import check_cypher, check_schema, validate_definition, validate_params
+
+
+@pytest.mark.parametrize(
+    "cypher",
+    [
+        "CREATE (n)",
+        "MATCH (n) DELETE n",
+        "MATCH (n) SET n.x=1 RETURN n",
+        "MERGE (n)",
+        "MATCH (n) REMOVE n.x RETURN n",
+        "DROP INDEX something",
+        "SHOW USERS",
+        "CALL db.labels()",
+        "CALL { MATCH (n) RETURN n } RETURN n",
+        "USE system RETURN 1",
+        "LOAD CSV FROM 'http://localhost' AS row RETURN row",
+        "RETURN 1; RETURN 2",
+        "EXPLAIN CREATE (n)",
+        "PROFILE RETURN 1",
+        "CYPHER 5 RETURN 1",
+        "cAlL/*hi*/db.labels()",
+        "RETURN apoc.cypher.runFirstColumn('CREATE (n)', {}, true)",
+        "RETURN `apoc`.`load`.`json`('http://localhost')",
+        "RETURN customFunction(1)",
+        "RETURN COUNT { CREATE (n) RETURN n }",
+        "RETURN EXISTS { MATCH (n) SET n.x=1 RETURN n }",
+        "RETURN 'unterminated",
+        "RETURN `unterminated",
+        "RETURN $`parameter`",
+    ],
+)
+def test_reject_unsafe_or_unsupported(cypher):
+    with pytest.raises(ServiceError):
+        check_cypher(cypher)
+
+
+@pytest.mark.parametrize(
+    "cypher",
+    [
+        "MATCH (n) RETURN n LIMIT $limit",
+        "MATCH p=(n)-[r]-(m) RETURN p LIMIT 10",
+        "MATCH (n) WHERE n.name = $name RETURN n",
+        "RETURN 'CREATE (n); CALL db.labels()' AS text",
+        "// CREATE (n)\nMATCH (n) /* DELETE n */ RETURN properties(n)",
+        "MATCH (n:`中文标签`) RETURN n.`中文属性`",
+        "UNWIND range(1,10) AS n RETURN sum(n)",
+        "MATCH (n) RETURN collect({name:n.name, labels:labels(n)})",
+        "RETURN 'it\\'s safe' AS v",
+        "RETURN $create AS value",
+    ],
+)
+def test_allow_graph_queries(cypher):
+    check_cypher(cypher)
+
+
+SCHEMA = {
+    "type": "object",
+    "additionalProperties": False,
+    "properties": {
+        "name": {"type": "string", "minLength": 1, "maxLength": 100},
+        "limit": {"type": "integer", "minimum": 1, "maximum": 50, "default": 10},
+    },
+    "required": ["name"],
+}
+
+
+def test_defaults_and_injection_values():
+    value = "'; CREATE (n) //"
+    result = validate_definition(
+        "MATCH (n) WHERE n.name=$name RETURN n LIMIT $limit", SCHEMA, {"name": value}
+    )
+    assert result == {"name": value, "limit": 10}
+
+
+@pytest.mark.parametrize(
+    "params",
+    [
+        {},
+        {"name": "x", "limit": "10"},
+        {"name": "x", "limit": True},
+        {"name": "x", "limit": 51},
+        {"name": "x", "extra": 1},
+    ],
+)
+def test_invalid_params(params):
+    with pytest.raises(ServiceError):
+        validate_params(SCHEMA, params)
+
+
+def test_schema_rejects_refs_patterns_and_permissive_objects():
+    for schema in [
+        {"type": "string", "pattern": ".*"},
+        {"type": "string", "$ref": "http://example.com"},
+        {"type": "object"},
+        {"type": "array"},
+    ]:
+        with pytest.raises(ServiceError):
+            check_schema(schema)