JOJO
b8a51c1c63
## 模型逻辑清理 早期把模型 API 端点/密钥/模型 ID 硬编码的残留（AGENT_API_* / THINKING_* / TITLE_* 三件套）已彻底移除。模型配置统一由 config/custom_models.json （经 model_profiles 解析为档案）描述，运行时通过 apply_profile 注入； 没有任何可用模型时按既定行为报错。 - config/api.py: 删 9 个三件套符号，仅保留 DEFAULT_RESPONSE_MAX_TOKENS - utils/api_client.py: client 改为空配置起步，移除三件套 import - server/chat_flow_helpers.py: 删死 import（TITLE_* 符号） ## 部署级配置外置（按"决策权归谁"分类） config/*.json 不再一概锚定源码树： - 程序能力（docker_risk_markers / skill_hints）留源码树，随版本演进 - 部署者自定义（custom_models / host_workspaces / auto_approval / goal_review / forbidden_commands / host_sandbox_policy）外置到 ~/.agents/<mode>/config/ 新增统一解析机制（config/paths.py）： - DEPLOY_CONFIG_DIR（默认 ~/.agents/<mode>/config，可用环境变量覆盖） - resolve_deploy_config()：只读，回退链 部署目录→源码树.json→源码树.example （开发环境不必先跑 setup 也能用源码树种子） - deploy_config_path()：写回用，稳定指向部署目录 6 个加载点改用上述解析；顺带修复 approval_agent / goal_review_agent 的 DEBUG_TRANSCRIPT_DIR 仍指旧源码树 logs/ 的漏迁问题。 ## 安全与运维 - 含密钥/机器特定的 5 个文件停止 git 追踪（git rm --cached，本地保留） 并加入 .gitignore，仓库仅留 .example 种子；forbidden_commands 保留 追踪作默认黑名单 - scripts/setup.py: 模型配置写到部署目录（用与 paths 一致的独立推导， 不 import config 以免过早锁定路径） - scripts/migrate_runtime_data.py: 新增 config/*.json → 部署目录迁移 （备份 + 不覆盖已存在） ## 关联：P1 配置收敛 + P2 首启向导 - config/server.py: Web 端口/监听地址/debug/NODE_BIN/PYTHON_BIN 单一事实源 - 消灭 8091 多处重复定义（state.py 死代码、app_legacy、main.py 各读各的） - 修复 sub_agent_manager 命令数组硬编码 "node" → NODE_BIN（便携包内置 Node 的前提） - scripts/setup.py: 终端首启向导（模式/端口/管理员/密钥/模型） ## 测试 test_config_paths_resolution 更新以反映新行为（host_workspaces 锚定部署 目录、新增 DEPLOY_CONFIG_DIR 用例）；全部离线用例通过。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
80 lines
2.9 KiB
Python
80 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import threading
|
|
from pathlib import Path
|
|
from typing import Dict, List
|
|
|
|
from config import HOST_SANDBOX_MACOS_WRITABLE_PATHS, deploy_config_path
|
|
|
|
_LOCK = threading.Lock()
|
|
# 部署级配置（机器特定可读写路径，会被运行时写回）→ ~/.agents/<mode>/config
|
|
_POLICY_PATH = Path(deploy_config_path("host_sandbox_policy.json"))
|
|
|
|
|
|
def _ensure_file() -> None:
|
|
if _POLICY_PATH.exists():
|
|
return
|
|
_POLICY_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
_POLICY_PATH.write_text(
|
|
json.dumps({"macos_writable_paths": [], "macos_readable_extra_paths": []}, ensure_ascii=False, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def load_policy() -> Dict:
|
|
with _LOCK:
|
|
_ensure_file()
|
|
try:
|
|
data = json.loads(_POLICY_PATH.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
data = {"macos_writable_paths": [], "macos_readable_extra_paths": []}
|
|
if not isinstance(data, dict):
|
|
data = {"macos_writable_paths": [], "macos_readable_extra_paths": []}
|
|
writable_items = data.get("macos_writable_paths")
|
|
if not isinstance(writable_items, list):
|
|
writable_items = []
|
|
readable_items = data.get("macos_readable_extra_paths")
|
|
if not isinstance(readable_items, list):
|
|
readable_items = []
|
|
data["macos_writable_paths"] = [str(x).strip() for x in writable_items if str(x).strip()]
|
|
data["macos_readable_extra_paths"] = [str(x).strip() for x in readable_items if str(x).strip()]
|
|
return data
|
|
|
|
|
|
def save_policy(policy: Dict) -> Dict:
|
|
payload = dict(policy or {})
|
|
writable_items = payload.get("macos_writable_paths")
|
|
if not isinstance(writable_items, list):
|
|
writable_items = []
|
|
readable_items = payload.get("macos_readable_extra_paths")
|
|
if not isinstance(readable_items, list):
|
|
readable_items = []
|
|
payload["macos_writable_paths"] = [str(x).strip() for x in writable_items if str(x).strip()]
|
|
payload["macos_readable_extra_paths"] = [str(x).strip() for x in readable_items if str(x).strip()]
|
|
with _LOCK:
|
|
_ensure_file()
|
|
_POLICY_PATH.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
return payload
|
|
|
|
|
|
def get_macos_writable_paths() -> List[str]:
|
|
file_items = load_policy().get("macos_writable_paths", [])
|
|
merged: List[str] = []
|
|
for raw in list(HOST_SANDBOX_MACOS_WRITABLE_PATHS or []) + list(file_items or []):
|
|
val = str(raw).strip()
|
|
if val and val not in merged:
|
|
merged.append(val)
|
|
return merged
|
|
|
|
|
|
def get_macos_readable_paths() -> List[str]:
|
|
data = load_policy()
|
|
readable_extra = data.get("macos_readable_extra_paths", [])
|
|
merged: List[str] = []
|
|
for raw in list(get_macos_writable_paths()) + list(readable_extra or []):
|
|
val = str(raw).strip()
|
|
if val and val not in merged:
|
|
merged.append(val)
|
|
return merged
|