feat(#182): 模型注册表后端化 TemplateRegistry 服务化 HTTP API（E1）

新增 core/model-framework/registry_api.py（标准库 http.server，对齐 auth_api
风格，零依赖），把已交付的 TemplateRegistry 暴露为 HTTP 服务：

- GET  /api/v1/registry/models                 列表（可按 stage 过滤）
- POST /api/v1/registry/models                 注册新版本（服务端校验）
- POST /api/v1/registry/models/{name}/{version}/promote   dev→staging→prod
- POST /api/v1/registry/models/{name}/rollback?version=&stage=  回滚指定 stage

服务端持久化：PersistedRegistry 每次写操作原子落盘到
deploy/data/registry/registry.json（可挂卷），重启不丢；首次启动自动播种
4 条演示模型（对齐 web/admin/admin.js seedModels）。

鉴权双轨：读接 FBA JWT（core/auth/fba_jwt.py），写要求权限码 iaop:admin
或 iaop:studio；未配置 FBA_TOKEN_SECRET_KEY 降级旧 core/auth 会话；
两者均无时为本地/CI 演示模式（读免鉴权、写需 token 占位）。

测试：tests/test_registry_api.py（16 例，含四接口回环联调、401/403 路径、
promote/rollback/持久化/种子）+ _fba_stub.py（FBA 桩）。全 284 例通过。

部署：deploy/fba/README.md 补 nginx 反代 /api/v1/registry/ → :8090 约定。

core/model-framework/registry_api.py

@@ -0,0 +1,513 @@
+# -*- coding: utf-8 -*-
+"""模型注册表 HTTP API（issue #182 / PRD 5.3 ③）。
+
+把已交付的 ``TemplateRegistry``（``core/model-framework/template_registry.py``）
+通过标准库 ``http.server`` 暴露为 HTTP 服务，对外提供四类接口：
+
+- ``GET  /api/v1/registry/models``                    列表（可按 stage 过滤）
+- ``POST /api/v1/registry/models``                    注册新版本（服务端校验）
+- ``POST /api/v1/registry/models/{name}/{version}/promote``  dev→staging→prod
+- ``POST /api/v1/registry/models/{name}/rollback?version=&stage=``  回滚到指定 stage 版本
+
+设计要点
+--------
+
+- **零依赖**：与 ``core/auth/auth_api.py`` 同款风格（``http.server.ThreadingHTTPServer``），
+  纯标准库，CI 可加载、可单测。
+- **服务端持久化**：注册表底层 ``TemplateRegistry.save/load`` JSON 落盘到
+  ``deploy/data/registry/registry.json``（生产可挂卷），重启不丢数据。
+- **鉴权双轨**：读接口接 FBA JWT（``core/auth/fba_jwt.py`` 的 ``FbaAuth``），写操作
+  要求权限码 ``iaop:admin`` 或 ``iaop:studio``；未配置 ``FBA_TOKEN_SECRET_KEY`` 时
+  降级旧 ``core/auth`` 会话（``auth_api.require_auth``）。两者均未配置时，
+  仅本地/CI 演示用——读接口免鉴权、写接口要求带任意有效 FBA/会话 token，
+  否则写接口返回 401（见 ``_require_write``）。
+- **种子**：首次启动数据文件不存在时，用旧版 4 条演示模型初始化
+  （与 ``web/admin/admin.js`` 的 ``seedModels()`` 对齐，便于前端无缝迁移）。
+
+可独立运行冒烟：``python -m core.model_framework.registry_api``（默认 :8090）。
+nginx 反代约定见 ``deploy/fba/README.md``。
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+import threading
+import time
+from http import HTTPStatus
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Any, Dict, List, Optional, Tuple
+from urllib.parse import parse_qs, urlsplit
+
+# 兼容两种导入位置：包内 ``from .template_registry`` 与
+# 单测 / 独立运行时 ``from model_framework.template_registry``。
+try:  # pragma: no cover - 取决于运行入口
+    from .template_registry import (  # type: ignore
+        ALLOWED_BACKBONES,
+        ModelTemplate,
+        Stage,
+        TemplateRegistry,
+        TemplateRegistryError,
+        is_valid_version,
+    )
+except ImportError:  # pragma: no cover
+    from model_framework.template_registry import (  # type: ignore
+        ALLOWED_BACKBONES,
+        ModelTemplate,
+        Stage,
+        TemplateRegistry,
+        TemplateRegistryError,
+        is_valid_version,
+    )
+
+# 写操作要求的权限码（与 deploy/fba/seed_iaop_menus.py 播种一致）
+WRITE_PERMS = ("iaop:admin", "iaop:studio")
+
+# 默认服务端数据目录（生产可挂卷）。相对仓库根。
+DEFAULT_DATA_DIR = os.path.normpath(
+    os.path.join(os.path.dirname(os.path.abspath(__file__)),
+                 "..", "..", "deploy", "data", "registry"))
+DEFAULT_DATA_FILE = "registry.json"
+
+#: 旧版演示模型种子（与 web/admin/admin.js seedModels 对齐，便于前端无缝迁移）
+SEED_MODELS: List[Dict[str, Any]] = [
+    {"name": "quality_forecast", "version": "1.2.0", "stage": "prod",
+     "backbone": "quality_forecast", "author": "bot_dev1",
+     "registered_at": "2026-08-01T09:00:00Z"},
+    {"name": "anomaly_detection", "version": "1.0.3", "stage": "staging",
+     "backbone": "anomaly_detection", "author": "bot_dev1",
+     "registered_at": "2026-08-02T09:00:00Z"},
+    {"name": "process_optimizer", "version": "0.9.1", "stage": "dev",
+     "backbone": "recipe_opt", "author": "engineer",
+     "registered_at": "2026-08-03T09:00:00Z"},
+    {"name": "cross_process_optimizer", "version": "0.5.0", "stage": "dev",
+     "backbone": "cross_process_opt", "author": "engineer",
+     "registered_at": "2026-08-04T09:00:00Z"},
+]
+
+
+class RegistryApiError(Exception):
+    """注册表 HTTP API 统一异常（携带 HTTP 状态码）。"""
+
+    def __init__(self, status: int, message: str) -> None:
+        super().__init__(message)
+        self.status = status
+        self.message = message
+
+
+# ---------------------------------------------------------------------------
+# 持久化注册表：写操作自动落盘
+# ---------------------------------------------------------------------------
+
+class PersistedRegistry:
+    """包一层 ``TemplateRegistry``，每次写操作后自动 JSON 落盘。
+
+    线程安全：所有公共方法持锁（``ThreadingHTTPServer`` 多线程服务）。
+    """
+
+    def __init__(self, data_path: str, *, autoload: bool = True) -> None:
+        self._path = data_path
+        self._reg = TemplateRegistry()
+        self._lock = threading.RLock()
+        os.makedirs(os.path.dirname(data_path), exist_ok=True)
+        if autoload and os.path.exists(data_path):
+            try:
+                self._reg = TemplateRegistry.load(data_path)
+            except (ValueError, OSError, TemplateRegistryError):
+                # 损坏的 JSON 不致命：退回空表，重新种子即可
+                self._reg = TemplateRegistry()
+        if autoload and len(self._reg) == 0:
+            self._seed(SEED_MODELS)
+
+    # -- 查询（只读）-------------------------------------------------------
+
+    def list_models(self, stage: Optional[str] = None) -> List[Dict[str, Any]]:
+        with self._lock:
+            names = self._reg.list_names()
+            out: List[Dict[str, Any]] = []
+            for n in names:
+                for v in self._reg.list_versions(n):
+                    tpl = self._reg.get(n, v)
+                    if stage and tpl.stage.value != stage:
+                        continue
+                    out.append(self._tpl_view(tpl))
+            return out
+
+    def get(self, name: str, version: Optional[str] = None) -> Dict[str, Any]:
+        with self._lock:
+            try:
+                tpl = self._reg.get(name, version)
+            except TemplateRegistryError as exc:
+                raise RegistryApiError(HTTPStatus.NOT_FOUND, str(exc))
+            return self._tpl_view(tpl)
+
+    # -- 写操作（落盘）-----------------------------------------------------
+
+    def register(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        with self._lock:
+            tpl = self._build_template(payload)
+            try:
+                self._reg.register(tpl)
+            except TemplateRegistryError as exc:
+                raise RegistryApiError(HTTPStatus.CONFLICT, str(exc))
+            self._persist()
+            return self._tpl_view(tpl)
+
+    def promote(self, name: str, version: str) -> Dict[str, Any]:
+        with self._lock:
+            try:
+                tpl = self._reg.promote(name, version)
+            except TemplateRegistryError as exc:
+                # 版本不存在 → 404；已在 prod → 409
+                status = (HTTPStatus.NOT_FOUND if "无版本" in str(exc) or "未注册" in str(exc)
+                          else HTTPStatus.CONFLICT)
+                raise RegistryApiError(status, str(exc))
+            self._persist()
+            return self._tpl_view(tpl)
+
+    def rollback(self, name: str, version: str, stage: str) -> Dict[str, Any]:
+        with self._lock:
+            try:
+                st = Stage.from_str(stage)
+            except TemplateRegistryError as exc:
+                raise RegistryApiError(HTTPStatus.BAD_REQUEST, str(exc))
+            try:
+                tpl = self._reg.rollback(name, st, version)
+            except TemplateRegistryError as exc:
+                status = (HTTPStatus.NOT_FOUND if "无版本" in str(exc) or "未注册" in str(exc)
+                          else HTTPStatus.BAD_REQUEST)
+                raise RegistryApiError(status, str(exc))
+            self._persist()
+            return self._tpl_view(tpl)
+
+    def history(self, name: Optional[str] = None) -> List[Dict[str, Any]]:
+        with self._lock:
+            return self._reg.history(name)
+
+    # -- 内部 --------------------------------------------------------------
+
+    def _persist(self) -> None:
+        tmp = self._path + ".tmp"
+        self._reg.save(tmp)
+        os.replace(tmp, self._path)  # 原子替换，避免半写
+
+    def _seed(self, seeds: List[Dict[str, Any]]) -> None:
+        for s in seeds:
+            try:
+                ts = s.get("registered_at")
+                # template_registry 用 epoch；字符串时间转 epoch，失败回退当前
+                ra: float
+                if isinstance(ts, (int, float)):
+                    ra = float(ts)
+                elif isinstance(ts, str):
+                    try:
+                        from datetime import datetime
+                        ra = datetime.fromisoformat(
+                            ts.replace("Z", "+00:00")).timestamp()
+                    except ValueError:
+                        ra = time.time()
+                else:
+                    ra = time.time()
+                tpl = ModelTemplate(
+                    name=s["name"], version=s["version"],
+                    backbone=s.get("backbone", "generic"),
+                    stage=Stage.from_str(s.get("stage", "dev")),
+                    registered_at=ra,
+                    description=s.get("description", ""),
+                    extra={"author": s.get("author", "")},
+                )
+                self._reg.register(tpl, force=True)
+            except TemplateRegistryError:
+                continue  # 脏种子跳过，不阻断启动
+        if len(self._reg) > 0:
+            self._persist()
+
+    def _build_template(self, payload: Dict[str, Any]) -> ModelTemplate:
+        name = str(payload.get("name", "")).strip()
+        version = str(payload.get("version", "")).strip()
+        if not name:
+            raise RegistryApiError(HTTPStatus.BAD_REQUEST, "name 不能为空")
+        if not is_valid_version(version):
+            raise RegistryApiError(
+                HTTPStatus.BAD_REQUEST,
+                f"非法版本号 {version!r}（例：v1 / 1.0.0 / v1.2-rc1）")
+        backbone = str(payload.get("backbone", "generic")).strip() or "generic"
+        if backbone not in ALLOWED_BACKBONES:
+            raise RegistryApiError(
+                HTTPStatus.BAD_REQUEST,
+                f"非法主干 {backbone!r}，允许：{list(ALLOWED_BACKBONES)}")
+        try:
+            stage = Stage.from_str(str(payload.get("stage", "dev")))
+        except TemplateRegistryError as exc:
+            raise RegistryApiError(HTTPStatus.BAD_REQUEST, str(exc))
+        metrics = payload.get("metrics") or {}
+        if not isinstance(metrics, dict):
+            raise RegistryApiError(HTTPStatus.BAD_REQUEST, "metrics 必须是对象")
+        try:
+            return ModelTemplate(
+                name=name, version=version, backbone=backbone, stage=stage,
+                hyperparams=dict(payload.get("hyperparams") or {}),
+                feature_columns=tuple(payload.get("feature_columns") or []),
+                target_column=str(payload.get("target_column", "") or ""),
+                metrics={k: float(v) for k, v in metrics.items()},
+                description=str(payload.get("description", "") or ""),
+                extra=dict(payload.get("extra") or {}),
+            )
+        except TemplateRegistryError as exc:
+            raise RegistryApiError(HTTPStatus.BAD_REQUEST, str(exc))
+
+    @staticmethod
+    def _tpl_view(tpl: ModelTemplate) -> Dict[str, Any]:
+        d = tpl.to_dict()
+        # 补 author（种子放进 extra），便于前端展示与迁移
+        d["author"] = (tpl.extra.get("author") or "")
+        return d
+
+
+# ---------------------------------------------------------------------------
+# 鉴权（双轨：FBA JWT / 旧 core-auth 会话；均未配置时仅读免鉴权）
+# ---------------------------------------------------------------------------
+
+class AuthService:
+    """双轨鉴权封装。读接口要求有效 token；写接口额外要求权限码。
+
+    - 配置了 ``FBA_TOKEN_SECRET_KEY``：走 ``FbaAuth.require_perm``；
+    - 否则若提供了 ``UserStore``：走 ``auth_api.require_auth``（旧会话，
+      写权限由 ``can_write`` 角色判定，对齐 iaop:admin/studio 语义）；
+    - 两者都无：本地/CI 演示模式——读接口免鉴权；写接口必须有 token
+      （``_require_write`` 在无任何鉴权后端时直接 401，避免裸奔写）。
+    """
+
+    def __init__(self, fba=None, user_store=None) -> None:
+        self._fba = fba
+        self._store = user_store
+
+    def require_read(self, headers) -> None:
+        # 读接口：有鉴权后端则校验；无后端（演示）放行
+        if self._fba is not None:
+            self._verify_fba(headers)
+            return
+        if self._store is not None:
+            self._require_session(headers)
+            return
+        # 无后端：读放行
+
+    def require_write(self, headers) -> None:
+        # 写接口：必须鉴权 + 权限码
+        if self._fba is not None:
+            for perm in WRITE_PERMS:
+                # iaop:admin 或 iaop:studio 任一即可
+                try:
+                    self._fba.require_perm(self._authorization(headers), perm)
+                    return
+                except Exception as exc:  # noqa: BLE001
+                    status = getattr(exc, "status", None)
+                    if status == HTTPStatus.FORBIDDEN:
+                        continue  # 试下一个权限码
+                    raise  # 401 / 503 直接抛
+            raise RegistryApiError(
+                HTTPStatus.FORBIDDEN,
+                f"权限不足：需要 {list(WRITE_PERMS)} 之一")
+        if self._store is not None:
+            user = self._require_session(headers)
+            role = getattr(user, "role", "")
+            if role not in ("admin", "engineer"):
+                raise RegistryApiError(
+                    HTTPStatus.FORBIDDEN,
+                    "权限不足：当前角色不可执行写操作")
+            return
+        # 无任何后端：写操作必须带 token 才放行（演示模式仍需 token 占位）
+        if not self._authorization(headers):
+            raise RegistryApiError(HTTPStatus.UNAUTHORIZED,
+                                   "写操作需要认证（配置 FBA_TOKEN_SECRET_KEY 或旧会话）")
+
+    # -- 内部 --------------------------------------------------------------
+
+    def _verify_fba(self, headers) -> None:
+        try:
+            self._fba.verify(self._authorization(headers))
+        except Exception as exc:  # noqa: BLE001
+            raise RegistryApiError(getattr(exc, "status", HTTPStatus.UNAUTHORIZED),
+                                   str(exc))
+
+    def _require_session(self, headers):
+        try:
+            from core.auth.auth_api import require_auth  # type: ignore
+        except ImportError:  # pragma: no cover
+            from .auth_api import require_auth  # type: ignore
+        try:
+            user, _ = require_auth(headers, self._store)
+        except Exception as exc:  # noqa: BLE001
+            raise RegistryApiError(getattr(exc, "status", HTTPStatus.UNAUTHORIZED),
+                                   str(exc))
+        return user
+
+    @staticmethod
+    def _authorization(headers) -> str:
+        if not headers:
+            return ""
+        return headers.get("Authorization", "") or ""
+
+
+# ---------------------------------------------------------------------------
+# HTTP 端点
+# ---------------------------------------------------------------------------
+
+#: 路由：GET /api/v1/registry/models
+_ROUTES_GET = (
+    re.compile(r"^/api/v1/registry/models/?$"),
+    re.compile(r"^/api/v1/registry/models/history/?$"),
+)
+#: POST .../models/{name}/{version}/promote
+_PROMOTE_RE = re.compile(
+    r"^/api/v1/registry/models/([^/]+)/([^/]+)/promote/?$")
+#: POST .../models/{name}/rollback
+_ROLLBACK_RE = re.compile(
+    r"^/api/v1/registry/models/([^/]+)/rollback/?$")
+
+
+class RegistryAPIHandler(BaseHTTPRequestHandler):
+    """注册表 HTTP 端点。``registry`` / ``auth`` 与 server 共享（见 make_server）。"""
+
+    server_version = "iAOP-RegistryAPI/1.0"
+    # HTTP/1.1 + Content-Length 让客户端能稳定读取错误响应体（401/403 等）
+    protocol_version = "HTTP/1.1"
+
+    def log_message(self, fmt, *args):  # 安静日志
+        pass
+
+    def handle_one_request(self):  # type: ignore[override]
+        # 每个请求后关闭连接：避免 keep-alive 下 ThreadingHTTPServer 对
+        # 缺 Content-Length 的客户端请求误判，从而中断错误响应。
+        super().handle_one_request()
+        self.close_connection = True
+
+    @property
+    def registry(self) -> PersistedRegistry:
+        return self.server.registry_store  # type: ignore[attr-defined]
+
+    @property
+    def auth(self) -> AuthService:
+        return self.server.auth_service  # type: ignore[attr-defined]
+
+    # -- 通用响应 ----------------------------------------------------------
+
+    def _json(self, status: int, body: Any) -> None:
+        payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(payload)))
+        self.end_headers()
+        self.wfile.write(payload)
+
+    def _read_json(self) -> Dict[str, Any]:
+        length = int(self.headers.get("Content-Length", 0) or 0)
+        if length <= 0 or length > 65536:
+            return {}
+        raw = self.rfile.read(length)
+        try:
+            data = json.loads(raw.decode("utf-8"))
+            return data if isinstance(data, dict) else {}
+        except (ValueError, UnicodeDecodeError):
+            raise RegistryApiError(HTTPStatus.BAD_REQUEST, "请求体不是合法 JSON")
+
+    def _handle_error(self, exc: Exception) -> None:
+        if isinstance(exc, RegistryApiError):
+            self._json(exc.status, {"error": exc.message})
+        else:
+            self._json(HTTPStatus.INTERNAL_SERVER_ERROR,
+                       {"error": "server error: %s" % exc})
+
+    # -- GET ---------------------------------------------------------------
+
+    def do_GET(self) -> None:  # noqa: N802 - http.server 约定
+        try:
+            split = urlsplit(self.path)
+            path = split.path
+            qs = parse_qs(split.query)
+            if _ROUTES_GET[0].match(path):
+                self.auth.require_read(self.headers)
+                stage = (qs.get("stage", [None])[0])
+                models = self.registry.list_models(stage)
+                self._json(HTTPStatus.OK, {"models": models, "total": len(models)})
+            elif _ROUTES_GET[1].match(path):
+                self.auth.require_read(self.headers)
+                name = qs.get("name", [None])[0]
+                self._json(HTTPStatus.OK,
+                           {"history": self.registry.history(name)})
+            elif path in ("/api/v1/registry/health", "/health"):
+                self._json(HTTPStatus.OK, {"status": "ok",
+                                           "models": len(self.registry.list_models())})
+            else:
+                self._json(HTTPStatus.NOT_FOUND, {"error": "not found"})
+        except Exception as exc:  # noqa: BLE001
+            self._handle_error(exc)
+
+    # -- POST --------------------------------------------------------------
+
+    def do_POST(self) -> None:  # noqa: N802 - http.server 约定
+        try:
+            path = urlsplit(self.path).path
+            qs = parse_qs(urlsplit(self.path).query)
+            if _ROUTES_GET[0].match(path):  # POST .../models → 注册
+                self.auth.require_write(self.headers)
+                body = self._read_json()
+                view = self.registry.register(body)
+                self._json(HTTPStatus.CREATED, view)
+            elif _PROMOTE_RE.match(path):
+                self.auth.require_write(self.headers)
+                m = _PROMOTE_RE.match(path)
+                view = self.registry.promote(m.group(1), m.group(2))
+                self._json(HTTPStatus.OK, view)
+            elif _ROLLBACK_RE.match(path):
+                self.auth.require_write(self.headers)
+                m = _ROLLBACK_RE.match(path)
+                name = m.group(1)
+                version = (qs.get("version", [None])[0] or "").strip()
+                stage = (qs.get("stage", [None])[0] or "prod").strip()
+                if not version:
+                    raise RegistryApiError(HTTPStatus.BAD_REQUEST,
+                                           "缺少 query 参数 version")
+                view = self.registry.rollback(name, version, stage)
+                self._json(HTTPStatus.OK, view)
+            else:
+                self._json(HTTPStatus.NOT_FOUND, {"error": "not found"})
+        except Exception as exc:  # noqa: BLE001
+            self._handle_error(exc)
+
+
+def make_server(data_path: Optional[str] = None, host: str = "127.0.0.1",
+                port: int = 8090, *, fba=None, user_store=None,
+                autoload: bool = True) -> ThreadingHTTPServer:
+    """构造注册表 HTTP 服务。
+
+    - ``data_path``：注册表 JSON 落盘路径，默认 ``deploy/data/registry/registry.json``。
+    - ``fba``：``FbaAuth`` 实例（读 ``FBA_TOKEN_SECRET_KEY``），None 则不启用 FBA 轨。
+    - ``user_store``：旧 ``core/auth`` ``UserStore``，None 则不启用旧会话轨。
+    """
+    if data_path is None:
+        data_path = os.path.join(DEFAULT_DATA_DIR, DEFAULT_DATA_FILE)
+    srv = ThreadingHTTPServer((host, port), RegistryAPIHandler)
+    srv.registry_store = PersistedRegistry(data_path, autoload=autoload)  # type: ignore[attr-defined]
+    srv.auth_service = AuthService(fba=fba, user_store=user_store)  # type: ignore[attr-defined]
+    return srv
+
+
+def _main() -> None:
+    # FBA 轨：读环境变量
+    fba = None
+    try:
+        from core.auth.fba_jwt import FbaAuth
+        fba = FbaAuth.from_env()
+    except Exception:  # noqa: BLE001 - 无 FBA 时降级
+        fba = None
+    srv = make_server(fba=fba)
+    print("iAOP RegistryAPI on http://%s:%d (data=%s, fba=%s)" % (
+        srv.server_address[0], srv.server_address[1],
+        srv.registry_store._path,  # noqa: SLF001
+        "on" if fba else "off"))  # pragma: no cover
+    srv.serve_forever()
+
+
+if __name__ == "__main__":  # pragma: no cover
+    _main()

core/model-framework/tests/_fba_stub.py

@@ -0,0 +1,52 @@
+# -*- coding: utf-8 -*-
+"""测试用 FBA 鉴权桩（issue #182 注册表 API 测试）。
+
+避免真实回源 FBA /auth/codes，本地固定权限码映射：
+
+- token == "good"     → 拥有 iaop:admin（写权限）
+- token == "readonly" → 仅读，无任何写权限码
+- 其它 / 缺失         → 401
+
+模拟 ``core/auth/fba_jwt.FbaAuth`` 的 ``verify`` / ``require_perm`` 接口形状。
+"""
+from __future__ import annotations
+
+from http import HTTPStatus
+
+
+class _FbaError(Exception):
+    def __init__(self, status: int, message: str) -> None:
+        super().__init__(message)
+        self.status = status
+        self.message = message
+
+
+class StubFba:
+    """FbaAuth 桩：固定权限码，不回源网络。"""
+
+    _PERMS = {
+        "good": {"iaop:admin", "iaop:studio"},
+        "readonly": set(),
+    }
+
+    def verify(self, authorization: str) -> dict:
+        token = self._extract(authorization)
+        if token not in self._PERMS:
+            raise _FbaError(HTTPStatus.UNAUTHORIZED, "Token 无效")
+        return {"sub": token}
+
+    def require_perm(self, authorization: str, perm: str) -> dict:
+        claims = self.verify(authorization)
+        token = self._extract(authorization)
+        if perm not in self._PERMS.get(token, set()):
+            raise _FbaError(HTTPStatus.FORBIDDEN, f"缺少权限：{perm}")
+        return claims
+
+    @staticmethod
+    def _extract(authorization: str) -> str:
+        if not authorization:
+            raise _FbaError(HTTPStatus.UNAUTHORIZED, "未携带认证信息")
+        parts = authorization.split(None, 1)
+        if len(parts) != 2 or parts[0].lower() != "bearer":
+            raise _FbaError(HTTPStatus.UNAUTHORIZED, "认证头格式应为 Bearer <token>")
+        return parts[1].strip()

core/model-framework/tests/test_registry_api.py

@@ -0,0 +1,267 @@
+# -*- coding: utf-8 -*-
+"""模型注册表 HTTP API 单元测试（issue #182）。
+
+覆盖验收要求：
+- 四接口 curl 联调（通过 make_server + urllib 真实起服务回环联调）；
+- 写操作 403 / 401 路径（无 token / 权限码缺失）；
+- promote / rollback 正确流转 + 落盘持久化（重启不丢）；
+- 种子首次启动初始化 4 条演示模型。
+
+运行（仓库根）：PYTHONPATH=. python -m unittest core.model_framework.tests.test_registry_api -v
+"""
+from __future__ import annotations
+
+import json
+import os
+import sys
+import tempfile
+import time
+import unittest
+import urllib.error
+import urllib.request
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+if HERE not in sys.path:
+    sys.path.insert(0, HERE)
+
+import _bootstrap  # noqa: E402  加载 model_framework 包
+
+from model_framework.registry_api import (  # noqa: E402
+    AuthService,
+    PersistedRegistry,
+    RegistryApiError,
+    make_server,
+)
+
+
+def _free_port() -> int:
+    import socket
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.bind(("127.0.0.1", 0))
+    port = s.getsockname()[1]
+    s.close()
+    return port
+
+
+class _ServerCtx:
+    """起一个回环 RegistryAPI 服务，返回 (base_url, stop)。"""
+
+    def __init__(self, data_path: str, *, fba=None, user_store=None,
+                 autoload: bool = True):
+        self.port = _free_port()
+        self.srv = make_server(data_path, "127.0.0.1", self.port,
+                               fba=fba, user_store=user_store, autoload=autoload)
+        import threading
+        self._t = threading.Thread(target=self.srv.serve_forever, daemon=True)
+        self._t.start()
+        self.base = f"http://127.0.0.1:{self.port}"
+
+    def stop(self):
+        self.srv.shutdown()
+        self.srv.server_close()
+
+
+def _req(method: str, url: str, body=None, headers=None):
+    data = None
+    h = dict(headers or {})
+    if body is not None:
+        data = json.dumps(body).encode("utf-8")
+        h.setdefault("Content-Type", "application/json")
+    req = urllib.request.Request(url, data=data, method=method, headers=h)
+    try:
+        with urllib.request.urlopen(req, timeout=5) as r:
+            return r.status, json.loads(r.read().decode("utf-8"))
+    except urllib.error.HTTPError as e:
+        payload = e.read().decode("utf-8")
+        try:
+            return e.code, json.loads(payload)
+        except ValueError:
+            return e.code, {"raw": payload}
+
+
+class PersistedRegistryTest(unittest.TestCase):
+    """底层 PersistedRegistry：注册 / promote / rollback / 持久化 / 种子。"""
+
+    def test_seed_on_first_start(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = os.path.join(d, "registry.json")
+            reg = PersistedRegistry(path)
+            models = reg.list_models()
+            self.assertEqual(len(models), 4)
+            names = {m["name"] for m in models}
+            self.assertIn("quality_forecast", names)
+            # 文件已落盘
+            self.assertTrue(os.path.exists(path))
+
+    def test_persist_across_restart(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = os.path.join(d, "registry.json")
+            reg = PersistedRegistry(path)
+            reg.register({"name": "quality_forecast", "version": "9.9.9",
+                          "stage": "dev"})
+            # 重新加载同一文件
+            reg2 = PersistedRegistry(path)
+            versions = reg2._reg.list_versions("quality_forecast")  # noqa: SLF001
+            self.assertIn("9.9.9", versions)
+
+    def test_promote_and_rollback(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = os.path.join(d, "registry.json")
+            reg = PersistedRegistry(path)
+            # 种子 quality_forecast 1.2.0 在 prod；用新 dev 版本走完整流转
+            reg.register({"name": "quality_forecast", "version": "2.0.0",
+                          "stage": "dev"})
+            promoted = reg.promote("quality_forecast", "2.0.0")
+            self.assertEqual(promoted["stage"], "staging")
+            # rollback staging 指针回到 1.2.0
+            rb = reg.rollback("quality_forecast", "1.2.0", "staging")
+            self.assertEqual(rb["version"], "1.2.0")
+            self.assertEqual(
+                reg._reg.stage_pointer("quality_forecast",  # noqa: SLF001
+                                       __import__(
+                                           "model_framework.template_registry",
+                                           fromlist=["Stage"]).Stage.STAGING),
+                "1.2.0")
+
+    def test_register_validates_bad_input(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = os.path.join(d, "registry.json")
+            reg = PersistedRegistry(path)
+            with self.assertRaises(RegistryApiError):
+                reg.register({"name": "", "version": "1.0.0"})
+            with self.assertRaises(RegistryApiError):
+                reg.register({"name": "x", "version": "not-a-version"})
+            with self.assertRaises(RegistryApiError):
+                reg.register({"name": "x", "version": "1.0.0",
+                              "backbone": "unknown_backbone"})
+
+    def test_duplicate_register_conflict(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = os.path.join(d, "registry.json")
+            reg = PersistedRegistry(path)
+            with self.assertRaises(RegistryApiError) as ctx:
+                reg.register({"name": "quality_forecast", "version": "1.2.0",
+                              "stage": "dev"})
+            self.assertEqual(ctx.exception.status, 409)
+
+
+class RegistryHttpApiTest(unittest.TestCase):
+    """四接口 HTTP 回环联调 + 鉴权 401/403 路径。"""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.path = os.path.join(self._tmp.name, "registry.json")
+        self.ctx = _ServerCtx(self.path)
+
+    def tearDown(self):
+        self.ctx.stop()
+        self._tmp.cleanup()
+
+    def test_list_and_seed(self):
+        status, body = _req("GET", self.ctx.base + "/api/v1/registry/models")
+        self.assertEqual(status, 200)
+        self.assertEqual(body["total"], 4)
+        # stage 过滤
+        status, body = _req("GET",
+                            self.ctx.base + "/api/v1/registry/models?stage=prod")
+        self.assertEqual(status, 200)
+        self.assertTrue(all(m["stage"] == "prod" for m in body["models"]))
+        self.assertEqual(body["total"], 1)
+
+    def test_health(self):
+        status, body = _req("GET", self.ctx.base + "/api/v1/registry/health")
+        self.assertEqual(status, 200)
+        self.assertEqual(body["status"], "ok")
+
+    def test_register_then_promote_then_rollback(self):
+        # 演示模式（无鉴权后端）：写接口需要 token 占位
+        st, bd = _req("POST", self.ctx.base + "/api/v1/registry/models",
+                      body={"name": "demo_model", "version": "0.1.0",
+                            "stage": "dev", "backbone": "generic"},
+                      headers={"Authorization": "Bearer demo-token"})
+        self.assertEqual(st, 201)
+        self.assertEqual(bd["name"], "demo_model")
+        # promote dev→staging
+        st, bd = _req("POST",
+                      self.ctx.base + "/api/v1/registry/models/demo_model/0.1.0/promote",
+                      headers={"Authorization": "Bearer demo-token"})
+        self.assertEqual(st, 200)
+        self.assertEqual(bd["stage"], "staging")
+        # rollback staging → 重新指向自身（验证回滚路径无错）
+        st, bd = _req("POST",
+                      self.ctx.base + "/api/v1/registry/models/demo_model/rollback?version=0.1.0&stage=staging",
+                      headers={"Authorization": "Bearer demo-token"})
+        self.assertEqual(st, 200)
+        self.assertEqual(bd["version"], "0.1.0")
+
+    def test_write_requires_token_in_demo_mode(self):
+        # 无 token → 401
+        st, bd = _req("POST", self.ctx.base + "/api/v1/registry/models",
+                      body={"name": "x", "version": "1.0.0"})
+        self.assertEqual(st, 401)
+
+    def test_promote_not_found(self):
+        st, bd = _req("POST",
+                      self.ctx.base + "/api/v1/registry/models/no_such/9.9.9/promote",
+                      headers={"Authorization": "Bearer demo-token"})
+        self.assertEqual(st, 404)
+
+    def test_rollback_missing_version_param(self):
+        st, bd = _req("POST",
+                      self.ctx.base + "/api/v1/registry/models/demo_model/rollback",
+                      headers={"Authorization": "Bearer demo-token"})
+        self.assertEqual(st, 400)
+
+    def test_persist_after_register(self):
+        _req("POST", self.ctx.base + "/api/v1/registry/models",
+             body={"name": "persist_model", "version": "1.0.0", "stage": "dev"},
+             headers={"Authorization": "Bearer demo-token"})
+        # 重启服务读同一文件
+        ctx2 = _ServerCtx(self.path)
+        try:
+            st, bd = _req("GET", ctx2.base + "/api/v1/registry/models")
+            names = {m["name"] for m in bd["models"]}
+            self.assertIn("persist_model", names)
+        finally:
+            ctx2.stop()
+
+
+class AuthWithFbaTest(unittest.TestCase):
+    """配置 FBA 轨时：读需有效 token；写需权限码，401/403 路径。"""
+
+    def setUp(self):
+        from model_framework.tests._fba_stub import StubFba
+        self._tmp = tempfile.TemporaryDirectory()
+        self.path = os.path.join(self._tmp.name, "registry.json")
+        # StubFba：token == "good" 有 iaop:admin；"readonly" 无写权限；无 token 401
+        self.ctx = _ServerCtx(self.path, fba=StubFba())
+
+    def tearDown(self):
+        self.ctx.stop()
+        self._tmp.cleanup()
+
+    def test_read_without_token_unauthorized(self):
+        st, _ = _req("GET", self.ctx.base + "/api/v1/registry/models")
+        self.assertEqual(st, 401)
+
+    def test_read_with_token_ok(self):
+        st, bd = _req("GET", self.ctx.base + "/api/v1/registry/models",
+                      headers={"Authorization": "Bearer good"})
+        self.assertEqual(st, 200)
+        self.assertEqual(bd["total"], 4)
+
+    def test_write_without_perm_forbidden(self):
+        st, bd = _req("POST", self.ctx.base + "/api/v1/registry/models",
+                      body={"name": "x", "version": "1.0.0"},
+                      headers={"Authorization": "Bearer readonly"})
+        self.assertEqual(st, 403)
+
+    def test_write_with_perm_ok(self):
+        st, bd = _req("POST", self.ctx.base + "/api/v1/registry/models",
+                      body={"name": "y", "version": "1.0.0", "stage": "dev"},
+                      headers={"Authorization": "Bearer good"})
+        self.assertEqual(st, 201)
+
+
+if __name__ == "__main__":
+    unittest.main()

deploy/fba/README.md

@@ -150,3 +150,41 @@ docker exec -it fba_postgres psql -U postgres -d fba       # 进数据库
 ### 旧架构存档
 - `web/` 静态演示外壳保留（FBA 停掉时兜底；/index.html、/auth/login.html 已加 C1 自动收口跳转）
 - 旧登录页三轨会话已精简为 FBA 单轨（session.js，FBA 接入版）
+
+---
+
+## 模型注册表 API 反代（issue #182，PRD 5.3 ③）
+
+iAOP 后端的模型模板注册表（`core/model-framework/registry_api.py`）以独立
+`http.server` 服务运行（默认 `127.0.0.1:8090`，零依赖标准库实现）。前端模型管理 /
+版本发布（E2，issue #183）通过 nginx 把 `/api/v1/registry/` 反代到该端口：
+
+```
+location /api/v1/registry/ {
+    proxy_pass http://127.0.0.1:8090/api/v1/registry/;
+    proxy_set_header Host              $host;
+    proxy_set_header X-Real-IP         $remote_addr;
+    proxy_set_header X-Forwarded-For   $proxy_add_x_forwarded_for;
+    proxy_set_header X-Forwarded-Proto $scheme;
+    proxy_http_version 1.1;
+    proxy_set_header Connection "";
+}
+```
+
+服务启动：
+
+```bash
+cd /opt/apps/iAOP
+# 配置 FBA 密钥后（与 fba.env 的 TOKEN_SECRET_KEY 一致）写操作接 FBA JWT 鉴权
+export FBA_TOKEN_SECRET_KEY=<同 fba.env>
+python -m core.model_framework.registry_api
+# → iAOP RegistryAPI on http://127.0.0.1:8090 (data=deploy/data/registry/registry.json, fba=on)
+```
+
+数据落盘到 `deploy/data/registry/registry.json`（生产可挂卷），重启不丢。
+首次启动自动播种 4 条演示模型（与 `web/admin/admin.js` 的 `seedModels()` 对齐）。
+
+鉴权双轨：读接口接 FBA JWT（`FbaAuth.verify`）；写接口要求权限码 `iaop:admin` 或
+`iaop:studio`。未配置 `FBA_TOKEN_SECRET_KEY` 时降级旧 `core/auth` 会话（写权限由
+角色 admin/engineer 判定）；两者均未配置时为本地/CI 演示模式——读免鉴权、写需
+带任意 token 占位（避免裸奔写）。