Files
PlantAI-rag-new/agent/legacy-rag/tools/runtime_surface.py
jerryandClaude Fable 5 2f82c4a8dc P4/M5: agent/legacy-rag 源码迁移(449 模块 191K 行)+ 影子 7911/7912 冒烟
- 按生产 7900 容器黄金清单迁移(compile.manifest 449 条目,0 缺失)
- 入口双态:api.app(生产同源 3 路由)/ api.main(裁剪 11 路由族 91 路径)
- 修复:advanced/__init__ 空文件(包标记误覆盖,已从主项目恢复)
- 修复:api/rag_routes.py 跨环境 site-packages 注入改 LEGACY_GF_SITE_PACKAGES 门控
- 修复:scripts/utils 缺失(动态加载 reasoning_engine)
- 影子验证:kg/stats 数据连通(6.2s,与 7900 同源 normalized 后端)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-31 14:20:41 +08:00

94 lines
3.5 KiB
Python

#!/usr/bin/env python3
"""Runtime surface manifest helpers for PlantAI-rag cleanup audits."""
from __future__ import annotations
import json
from functools import lru_cache
from pathlib import Path
from typing import Any
PROJECT_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_MANIFEST_PATH = PROJECT_ROOT / "config" / "runtime_surface_manifest.json"
def _as_string_set(value: Any) -> frozenset[str]:
if not isinstance(value, list):
return frozenset()
return frozenset(str(item) for item in value)
def _section(manifest: dict[str, Any], name: str) -> dict[str, Any]:
value = manifest.get(name)
return value if isinstance(value, dict) else {}
@lru_cache(maxsize=4)
def load_runtime_surface(manifest_path: str | Path = DEFAULT_MANIFEST_PATH) -> dict[str, Any]:
path = Path(manifest_path)
with path.open("r", encoding="utf-8") as handle:
manifest = json.load(handle)
if not isinstance(manifest, dict):
raise ValueError(f"Runtime surface manifest must be a JSON object: {path}")
if int(manifest.get("schema_version") or 0) != 1:
raise ValueError(f"Unsupported runtime surface manifest schema: {path}")
return manifest
def current_runtime_files(manifest: dict[str, Any] | None = None) -> frozenset[str]:
manifest = manifest or load_runtime_surface()
return _as_string_set(_section(manifest, "current_runtime").get("files"))
def current_runtime_prefixes(manifest: dict[str, Any] | None = None) -> tuple[str, ...]:
manifest = manifest or load_runtime_surface()
return tuple(_as_string_set(_section(manifest, "current_runtime").get("prefixes")))
def required_tracked_files(manifest: dict[str, Any] | None = None) -> frozenset[str]:
manifest = manifest or load_runtime_surface()
return _as_string_set(manifest.get("required_tracked_files"))
def legacy_prefixes(manifest: dict[str, Any] | None = None) -> tuple[str, ...]:
manifest = manifest or load_runtime_surface()
return tuple(_as_string_set(_section(manifest, "legacy").get("prefixes")))
def legacy_files(manifest: dict[str, Any] | None = None) -> frozenset[str]:
manifest = manifest or load_runtime_surface()
return _as_string_set(_section(manifest, "legacy").get("files"))
def docs_history_files(manifest: dict[str, Any] | None = None) -> frozenset[str]:
manifest = manifest or load_runtime_surface()
return _as_string_set(_section(manifest, "docs_history").get("files"))
def docs_history_prefixes(manifest: dict[str, Any] | None = None) -> tuple[str, ...]:
manifest = manifest or load_runtime_surface()
return tuple(_as_string_set(_section(manifest, "docs_history").get("prefixes")))
def test_fixture_prefixes(manifest: dict[str, Any] | None = None) -> tuple[str, ...]:
manifest = manifest or load_runtime_surface()
return tuple(_as_string_set(_section(manifest, "test_fixture").get("prefixes")))
def classify_path(path: str, manifest: dict[str, Any] | None = None) -> str:
manifest = manifest or load_runtime_surface()
if path in current_runtime_files(manifest) or path.startswith(current_runtime_prefixes(manifest)):
return "current_runtime"
if path.startswith(test_fixture_prefixes(manifest)):
return "test_fixture"
if (
path in docs_history_files(manifest)
or path.startswith(docs_history_prefixes(manifest))
or path.endswith(".md")
):
return "docs_history"
if path in legacy_files(manifest) or path.startswith(legacy_prefixes(manifest)):
return "legacy_script"
return "other_tracked"