scense/app/_archive_disabled_20261028_151505/entry_semantics_boot.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""入口 HTTP 语义健壮装配引导（Bug Wp6gXrrEquzg3njONz60b / INT-404-neg）。

缺陷（本轮复现，2026-10-21 14:21 目标机实测，带 admin cookie）
--------------------------------------------------------------
```
GET /runtime/api/__no_such_endpoint__.dspy   -> 500  （期望 404）
GET /nonexist/x.dspy                         -> 500  （期望 404）
（无 cookie 对照同两条 -> 401，说明鉴权层正常，异常出在鉴权之后）
```
ahserver `processorResource.py:399` 对「URL 无法映射到磁盘资源」抛的是**通用 `Exception`**
（`f'{str(request.url)=} invalid path'`），没有任何 HTTP 语义，一路冒到 aiohttp 默认错误处理，
被渲染成 `500 Internal Server Error / Server got itself in trouble`。

R17-2（Bug 28PuqnfpWzB2ypJoUTU_N）已在 `app/semantic_wiring.py` 写好 404/405/400 中间件，并由
`app/scense.py` 在 `ahapp_built` 钩子里装配。但那条装配链路有一个**结构性静默失败点**：

    try:
        from app.semantic_wiring import install_semantic_middlewares   # ← 依赖 app 包可导入
        install_semantic_middlewares(app, entry="app/scense.py")
    except Exception:
        log.exception(...)        # 只打日志，进程照常起，/healthz 照常 200

只要 `app` 包在目标机不可导入（`python app/scense.py` 的 sys.path[0]=<deploy>/app 遮蔽、
投递缺文件、pyc 过期、包名冲突），语义中间件就**一个都装不上**，而系统看起来「全部正常」，
线上 500 原样存在——这正是 R17-2「修了但等于没生效」的机制。

本模块的职责
------------
把装配做成**与 app 包可导入性无关**的两层结构，并让装配结果可远程核验：

* 第 1 层 `entry_semantics`（零应用内依赖）：用**绝对文件路径 + importlib** 加载
  `app/entry_semantics.py`，装 notfound_guard（→404）+ method_guard（→405）。
  这一层不 import 本应用任何包，只要文件在投递包里就一定能装上。
* 第 2 层 `semantic_wiring`（完整语义）：同样按绝对路径加载，装
  `resource_notfound_middleware` + `http_semantics_middleware`（含 400 参数缺失语义）。
  失败仅降级为「第 1 层单层生效」，不再影响 404 语义。

返回结构（供入口写进 app["http_semantics"] 并由 `/healthz` 暴露）::

    {"installed": bool,          # 至少一层装上且含 notfound 守卫
     "degraded": bool,           # 仅第 1 层生效（完整语义未装上）
     "layers": {"entry_semantics": {...}, "semantic_wiring": {...}},
     "middlewares": [...], "added": int, "error": str|None}

铁律：**绝不抛出**——装配点在 ahapp_built 钩子里，抛错会让 run_app 失败、整机启不来。
"""

import datetime
import importlib.util
import json
import logging
import os
import traceback

log = logging.getLogger("scense.entry.semantics_boot")

APP_DIR = os.path.dirname(os.path.abspath(__file__))

#: 零应用内依赖层（404 + 405）
LAYER_ZERO_DEP = "entry_semantics"
#: 完整语义层（404 + 405 + 400）
LAYER_FULL = "semantic_wiring"

#: 判定「notfound 守卫是否生效」的中间件名集合（两层任一名命中即算生效）
LAYER_AUTH_GATE = "auth_route_gate"
AUTH_GATE_NAMES = ("auth_route_gate_middleware",)
NOTFOUND_GUARD_NAMES = ("entry_notfound_guard", "resource_notfound_middleware")
METHOD_GUARD_NAMES = ("entry_method_guard", "http_semantics_middleware")


def middleware_names(app):
    """取当前 app.middlewares 的函数名列表（用于装配后置核验）。"""
    return [getattr(m, "__name__", repr(m)) for m in (getattr(app, "middlewares", None) or [])]


def load_module_by_file(name, filename):
    """按绝对文件路径加载模块（不依赖 sys.path / 包导入）。

    `name` 只作为 sys.modules 里的键；加载失败抛异常，由调用方降级处理。
    """
    path = os.path.join(APP_DIR, filename)
    if not os.path.isfile(path):
        raise OSError("模块文件不存在: %s" % path)
    spec = importlib.util.spec_from_file_location(name, path)
    if spec is None or spec.loader is None:
        raise ImportError("无法为 %s 构造 import spec" % path)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def _install_zero_dep(app, web=None):
    """第 1 层：entry_semantics（零应用内依赖，绝对路径加载）。"""
    res = {"layer": LAYER_ZERO_DEP, "ok": False, "added": [], "error": None}
    try:
        mod = load_module_by_file("_entry_semantics_zero_dep", "entry_semantics.py")
    except Exception as exc:                                   # noqa: BLE001
        res["error"] = "%s: %s" % (type(exc).__name__, exc)
        res["traceback"] = traceback.format_exc()
        return res
    try:
        out = mod.install(app, web=web) if web is not None else mod.install(app)
        res["ok"] = bool(out.get("ok"))
        res["added"] = list(out.get("added") or [])
        res["error"] = out.get("error")
    except Exception as exc:                                   # noqa: BLE001
        res["error"] = "%s: %s" % (type(exc).__name__, exc)
        res["traceback"] = traceback.format_exc()
    return res


def _install_full(app, entry="app/scense.py"):
    """第 2 层：semantic_wiring（完整 404/405/400 语义，绝对路径加载）。"""
    res = {"layer": LAYER_FULL, "ok": False, "added": [], "error": None, "evidence": None}
    try:
        mod = load_module_by_file("_semantic_wiring_full", "semantic_wiring.py")
    except Exception as exc:                                   # noqa: BLE001
        res["error"] = "%s: %s" % (type(exc).__name__, exc)
        res["traceback"] = traceback.format_exc()
        return res
    try:
        out = mod.install_semantic_middlewares(app, entry=entry) or {}
        res["ok"] = bool(out.get("ok"))
        res["added"] = list(out.get("added") or out.get("added_names") or [])
        res["error"] = out.get("error")
        res["evidence"] = out.get("evidence")
    except Exception as exc:                                   # noqa: BLE001
        res["error"] = "%s: %s" % (type(exc).__name__, exc)
        res["traceback"] = traceback.format_exc()
    return res


def _install_auth_route_gate(app):
    """第 3 层：统一鉴权门禁（存在性→鉴权，404/401 三态互不覆盖）。

    绝对路径加载 auth_route_gate.py（零应用内依赖），插到中间件链**最外层**，
    保证先于 ahserver 鉴权层执行（否则框架自身对不存在路径的 401 会抢在 404 之前）。
    永不抛出。
    """
    res = {"layer": LAYER_AUTH_GATE, "ok": False, "added": [], "error": None,
           "position": None}
    try:
        mod = load_module_by_file("_auth_route_gate", "auth_route_gate.py")
    except Exception as exc:                                   # noqa: BLE001
        res["error"] = "%s: %s" % (type(exc).__name__, exc)
        res["traceback"] = traceback.format_exc()
        return res
    try:
        out = mod.install(app) or {}
        res["ok"] = bool(out.get("ok"))
        res["added"] = list(out.get("added") or [])
        res["position"] = out.get("position")
        res["error"] = out.get("error")
        res["middlewares"] = out.get("middlewares")
    except Exception as exc:                                   # noqa: BLE001
        res["error"] = "%s: %s" % (type(exc).__name__, exc)
        res["traceback"] = traceback.format_exc()
    return res


def _install_http_semantics_impl(app, entry="app/scense.py", full=True, web=None):
    """两层装配实现（内部函数，勿直接调用；对外入口是文件末尾的
    install_http_semantics 包装器，它在此基础上追加 evidence 落盘）。

    **永不抛出**。

    幂等：两层安装器自身按中间件名字去重，重复调用不会重复 append。
    """
    result = {"entry": entry, "installed": False, "degraded": False,
              "layers": {}, "middlewares": [], "added": 0, "error": None}
    try:
        before = middleware_names(app)

        zero = _install_zero_dep(app, web=web)
        result["layers"][LAYER_ZERO_DEP] = {k: v for k, v in zero.items() if k != "traceback"}
        if not zero["ok"]:
            log.error("[semantics_boot] 零依赖层装配失败：%s", zero.get("error"))

        if full:
            fll = _install_full(app, entry=entry)
            result["layers"][LAYER_FULL] = {k: v for k, v in fll.items() if k != "traceback"}
            if not fll["ok"]:
                log.error("[semantics_boot] 完整语义层装配失败（降级为 404/405 单层）：%s",
                          fll.get("error"))
        else:
            result["layers"][LAYER_FULL] = {"layer": LAYER_FULL, "ok": False,
                                            "skipped": True, "error": "switch_disabled"}

        gate = _install_auth_route_gate(app)
        result["layers"][LAYER_AUTH_GATE] = {k: v for k, v in gate.items() if k != "traceback"}
        if not gate.get("ok"):
            log.error("[semantics_boot] 统一鉴权门禁装配失败：%s", gate.get("error"))

        names = middleware_names(app)
        result["middlewares"] = names
        result["added"] = len([n for n in names if n not in before]) if before else len(names)

        has_auth_gate = any(n in AUTH_GATE_NAMES for n in names)
        has_notfound = any(n in NOTFOUND_GUARD_NAMES for n in names)
        has_method = any(n in METHOD_GUARD_NAMES for n in names)
        result["installed"] = bool(has_notfound and has_method and has_auth_gate)
        result["auth_gate"] = {"present": has_auth_gate, "layer": LAYER_AUTH_GATE}
        result["degraded"] = bool(result["installed"]
                                  and not result["layers"][LAYER_FULL].get("ok"))
        if not result["installed"]:
            result["error"] = ("语义中间件缺失 notfound=%s method=%s auth_gate=%s names=%s"
                               % (has_notfound, has_method, has_auth_gate, names))
        log.info("[semantics_boot] installed=%s degraded=%s added=%d names=%s",
                 result["installed"], result["degraded"], result["added"], names)
    except Exception as exc:                                   # noqa: BLE001
        result["error"] = "%s: %s" % (type(exc).__name__, exc)
        result["traceback"] = traceback.format_exc()
        log.error("[semantics_boot] 装配整体异常（不抛出，应用继续启动）：%s", result["error"])
    return result


# ── 装配结果落盘（Bug Wp6gXrrEquzg3njONz60b：装配是否真生效不能只靠日志）──
# 线上是否真装上语义中间件，此前只能从进程日志推断（日志 roll 掉就无证）。这里把
# install_http_semantics 的返回值（installed / degraded / layers / middlewares）落成
# JSON 证据文件，QC 与部署角色可直接读取，且**两个入口共用同一份代码**（app/scense.py
# 与 app/yuanjing.py 都走本函数），避免"只在一个入口修"的同类缺陷复发。
# 落盘失败绝不影响启动（语义装配优先），只在日志告警。
WIRING_EVIDENCE_REL = os.path.join("docs", "evidence", "http_semantics_wiring.json")


def persist_wiring(result, entry=None, base_dir=None):
    """把装配结果写入 <app dir>/docs/evidence/http_semantics_wiring.json（永不抛出）。"""
    try:
        root = base_dir or os.path.dirname(os.path.abspath(__file__))
        path = os.path.join(root, WIRING_EVIDENCE_REL)
        os.makedirs(os.path.dirname(path), exist_ok=True)
        payload = {
            "ts": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "pid": os.getpid(),
            "entry": entry or (result or {}).get("entry"),
            "installed": bool((result or {}).get("installed")),
            "degraded": bool((result or {}).get("degraded")),
            "middlewares": list((result or {}).get("middlewares") or []),
            "added": (result or {}).get("added"),
            "error": (result or {}).get("error"),
            "layers": (result or {}).get("layers") or {},
            "evidence_path": path,
        }
        tmp = path + ".tmp"
        with open(tmp, "w", encoding="utf-8") as fp:
            json.dump(payload, fp, ensure_ascii=False, indent=2, default=str)
        os.replace(tmp, path)
        log.info("[semantics_boot] 装配证据已落盘 %s installed=%s degraded=%s",
                 path, payload["installed"], payload["degraded"])
        return payload
    except Exception as exc:                                   # noqa: BLE001
        log.warning("[semantics_boot] 装配证据落盘失败（不影响启动）：%s: %s",
                    type(exc).__name__, exc)
        return None


def install_http_semantics(app, entry="app/scense.py", full=True, web=None):
    """install_http_semantics + 装配证据落盘（保持原签名与返回结构，永不抛出）。"""
    result = _install_http_semantics_impl(app, entry=entry, full=full, web=web)
    try:
        result["evidence"] = persist_wiring(result, entry=entry)
    except Exception:                                          # noqa: BLE001
        result["evidence"] = None
    return result