aps-agent/server/agent_core/feature_flags.py

119 lines
5.1 KiB
Python
Raw Permalink Normal View History

# ============================================================
# 功能开关(Feature Flags)文件化配置(moduleId: feature-flags, 可重生 ✅)
# 用一个 JSON 文件控制哪些功能上线可见、哪些功能隐藏不用。
#
# 语义(务必牢记):
# - 这是「可用性/可见性」开关,不是权限边界;隐藏不等于鉴权,
# P2/P3 权力路径仍由 Harness 门禁独立保证。
# - fail-open 诚实回退:文件缺失/损坏/版本不符 → 回退默认值(默认全开,
# 但 _DEFAULT_OFF_KEYS 集合内的功能如 fallback 默认关),
# 并在响应里显式携带 source/error/defaultOff,绝不静默吞掉配置错误,
# 也绝不因配置损坏把系统锁死。
# - 未知键忽略但如实列入 unknown;非 bool 值该键回退各自默认值并记录错误。
#
# 配置文件(默认 server/data/features.json,APS_FEATURES_PATH 可覆盖):
# {"version": 1, "features": {"orders": true, "drawing": false, ...}}
# ============================================================
from __future__ import annotations
import json
import os
from typing import Any
_SCHEMA_VERSION = 1
_PATH_ENV = "APS_FEATURES_PATH"
# 功能目录:key → 中文标签(顺序即展示顺序;新增功能必须在此登记)
FEATURE_CATALOG: dict[str, str] = {
"visual": "会话结果视口(甘特/负荷/交期)",
"orders": "订单管理",
"master": "主数据",
"flex": "柔性工作台",
"timeline": "排产结构(时间线)",
"drawing": "图纸识别(DXF)",
"mesh": "多智能体编排",
"knowledge": "知识平台",
"skills": "算法 Skill 管理",
"gov": "审批治理",
"regen": "重生中心",
"settings": "设置",
"fallback": "智能兜底(Pi Agent)",
}
# 默认关闭集:这些键在「文件缺失/损坏/未显式配置/取值非 bool」时默认 False,
# 与其余键的 fail-open 相反。原因:fallback 会拉起外部 LLM 子进程并产生 token
# 成本,默认开会在无模型/离线现场制造意外副作用(方案 §5 部署坑、§7 F5)。
# 兜底是增强能力而非主干功能,关了只回到现状话术,系统不失能(有意取舍)。
_DEFAULT_OFF_KEYS = frozenset({"fallback"})
def default_features_path() -> str:
configured = os.environ.get(_PATH_ENV)
if configured:
return configured
from server.aps_home import path_under_data
return str(path_under_data("features.json"))
def _default_enabled() -> dict[str, bool]:
return {key: key not in _DEFAULT_OFF_KEYS for key in FEATURE_CATALOG}
def load_feature_flags(file_path: str | None = None) -> dict[str, Any]:
"""读取并解析功能开关配置。
Returns: {"version": 1, "source": "default"|"file", "path": str,
"error": str | None, "unknown": [str],
"features": {key: {"enabled": bool, "label": str}}}
"""
path = file_path or default_features_path()
enabled = _default_enabled()
source = "default"
error: str | None = None
unknown: list[str] = []
if not os.path.exists(path):
pass # 文件缺失:回退默认值(fallback 默认关,其余默认开)
else:
try:
with open(path, "r", encoding="utf-8") as handle:
raw = json.load(handle)
except (json.JSONDecodeError, OSError, UnicodeDecodeError) as exc:
error = f"功能开关配置无法解析({type(exc).__name__}),已回退默认值(fallback 默认关,其余默认开)"
else:
if not isinstance(raw, dict) or raw.get("version") != _SCHEMA_VERSION:
error = "功能开关配置顶层结构或 version 非法,已回退默认值(fallback 默认关,其余默认开)"
elif not isinstance(raw.get("features"), dict):
error = "功能开关配置 features 段不是对象,已回退默认值(fallback 默认关,其余默认开)"
else:
source = "file"
value_errors: list[str] = []
for key, value in raw["features"].items():
if key not in FEATURE_CATALOG:
unknown.append(str(key))
continue
if not isinstance(value, bool):
value_errors.append(key) # 非 bool:该键回退默认值(fallback 默认关)
continue
enabled[key] = value
if value_errors:
error = (
"功能开关配置存在非 bool 取值("
+ ",".join(sorted(value_errors))
+ "),对应功能已回退默认值(fallback 默认关,其余默认开)"
)
return {
"version": _SCHEMA_VERSION,
"source": source,
"path": str(path),
"error": error,
"unknown": unknown,
"defaultOff": sorted(_DEFAULT_OFF_KEYS),
"features": {
key: {"enabled": enabled[key], "label": label}
for key, label in FEATURE_CATALOG.items()
},
}