xst
245 lines
9.0 KiB
Python
245 lines
9.0 KiB
Python
"""
|
||
Agentic QE Fleet — Manifest 管理模块
|
||
|
||
负责跨战区 manifest JSON 的读写、合并、验证和迁移。
|
||
每个战区产出一个 manifest_<zone>.json，fleet_runner.py 消费。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||
MANIFESTS_DIR = REPO_ROOT / "output" / "manifests"
|
||
|
||
# 战区顺序
|
||
ZONE_ORDER = ["prepare", "analyze", "design", "execute", "review", "monitor"]
|
||
|
||
|
||
def ensure_manifests_dir() -> Path:
|
||
MANIFESTS_DIR.mkdir(parents=True, exist_ok=True)
|
||
return MANIFESTS_DIR
|
||
|
||
|
||
def manifest_path(base_name: str, zone: str) -> Path:
|
||
"""返回战区 manifest 文件路径。"""
|
||
return MANIFESTS_DIR / f"{base_name}_{zone}.json"
|
||
|
||
|
||
def load_manifest(base_name: str, zone: str) -> dict[str, Any]:
|
||
"""加载指定战区的 manifest。"""
|
||
path = manifest_path(base_name, zone)
|
||
if not path.exists():
|
||
raise FileNotFoundError(f"Manifest 不存在: {path}")
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
|
||
|
||
def load_manifest_safe(base_name: str, zone: str) -> dict[str, Any] | None:
|
||
"""安全加载 manifest，不存在时返回 None。"""
|
||
path = manifest_path(base_name, zone)
|
||
if not path.exists():
|
||
return None
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
|
||
|
||
def save_manifest(base_name: str, zone: str, data: dict[str, Any]) -> Path:
|
||
"""保存战区 manifest。"""
|
||
ensure_manifests_dir()
|
||
path = manifest_path(base_name, zone)
|
||
data.setdefault("_meta", {})
|
||
data["_meta"]["zone"] = zone
|
||
data["_meta"]["base_name"] = base_name
|
||
data["_meta"]["updated_at"] = datetime.now(timezone.utc).isoformat()
|
||
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
return path
|
||
|
||
|
||
def build_merged_manifest(base_name: str, zones: list[str] | None = None) -> dict[str, Any]:
|
||
"""合并多个战区的 manifest 为统一视图。"""
|
||
if zones is None:
|
||
zones = ZONE_ORDER
|
||
|
||
merged: dict[str, Any] = {
|
||
"_meta": {
|
||
"base_name": base_name,
|
||
"merged_zones": [],
|
||
"merged_at": datetime.now(timezone.utc).isoformat(),
|
||
}
|
||
}
|
||
|
||
for zone in zones:
|
||
manifest = load_manifest_safe(base_name, zone)
|
||
if manifest is None:
|
||
continue
|
||
merged["_meta"]["merged_zones"].append(zone)
|
||
# 战区数据按 zone 命名空间隔离
|
||
merged[zone] = manifest
|
||
# 同时摊平顶层便捷字段（后者覆盖前者）
|
||
for key, value in manifest.items():
|
||
if key.startswith("_"):
|
||
continue
|
||
merged[key] = value
|
||
|
||
return merged
|
||
|
||
|
||
def save_merged_manifest(base_name: str, zones: list[str] | None = None) -> Path:
|
||
"""合并所有战区 manifest 并保存为统一清单文件（供导出等步骤使用）。"""
|
||
merged = build_merged_manifest(base_name, zones)
|
||
ensure_manifests_dir()
|
||
path = MANIFESTS_DIR / f"{base_name}.json"
|
||
merged.setdefault("_meta", {})
|
||
merged["_meta"]["updated_at"] = datetime.now(timezone.utc).isoformat()
|
||
path.write_text(json.dumps(merged, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
return path
|
||
|
||
|
||
def get_zone_status(base_name: str) -> dict[str, str]:
|
||
"""返回各战区完成状态。"""
|
||
status: dict[str, str] = {}
|
||
for zone in ZONE_ORDER:
|
||
manifest = load_manifest_safe(base_name, zone)
|
||
if manifest is None:
|
||
status[zone] = "pending"
|
||
elif manifest.get("_meta", {}).get("status") == "completed":
|
||
status[zone] = "completed"
|
||
else:
|
||
status[zone] = "in_progress"
|
||
return status
|
||
|
||
|
||
def mark_zone_completed(base_name: str, zone: str) -> Path:
|
||
"""标记战区为已完成。"""
|
||
manifest = load_manifest(base_name, zone)
|
||
manifest.setdefault("_meta", {})["status"] = "completed"
|
||
return save_manifest(base_name, zone, manifest)
|
||
|
||
|
||
def get_confirmation_gate(base_name: str) -> dict[str, Any] | None:
|
||
"""获取当前确认门禁状态（分析战区产出）。"""
|
||
manifest = load_manifest_safe(base_name, "analyze")
|
||
if manifest is None:
|
||
return None
|
||
return manifest.get("confirmation_gate")
|
||
|
||
|
||
def get_quality_verdict(base_name: str) -> dict[str, Any] | None:
|
||
"""获取质量裁决（评审战区产出）。"""
|
||
manifest = load_manifest_safe(base_name, "review")
|
||
if manifest is None:
|
||
return None
|
||
return manifest.get("quality_verdict")
|
||
|
||
|
||
def get_latest_zone(base_name: str) -> str | None:
|
||
"""返回最新已完成的战区名称。"""
|
||
for zone in reversed(ZONE_ORDER):
|
||
manifest = load_manifest_safe(base_name, zone)
|
||
if manifest and manifest.get("_meta", {}).get("status") == "completed":
|
||
return zone
|
||
return None
|
||
|
||
|
||
def get_next_zone(base_name: str) -> str | None:
|
||
"""返回下一个待执行的战区名称。"""
|
||
for zone in ZONE_ORDER:
|
||
manifest = load_manifest_safe(base_name, zone)
|
||
if manifest is None or manifest.get("_meta", {}).get("status") != "completed":
|
||
return zone
|
||
# 全部完成
|
||
return None
|
||
|
||
|
||
def migrate_legacy_manifest(base_name: str) -> dict[str, Any]:
|
||
"""从旧的单 manifest JSON 迁移到新的多战区 manifest 结构。"""
|
||
legacy_path = MANIFESTS_DIR / f"{base_name}.json"
|
||
if not legacy_path.exists():
|
||
raise FileNotFoundError(f"旧版 manifest 不存在: {legacy_path}")
|
||
|
||
legacy = json.loads(legacy_path.read_text(encoding="utf-8"))
|
||
|
||
# 从旧 manifest 推断 prepare 战区数据
|
||
prepare_data = {
|
||
"_meta": {"zone": "prepare", "base_name": base_name, "status": "completed",
|
||
"migrated_from_legacy": True},
|
||
"base_name": legacy.get("base_name", base_name),
|
||
"normalized_requirement_file": legacy.get("normalized_requirement_file"),
|
||
"normalized_technical_solution_files": legacy.get("normalized_technical_solution_files", []),
|
||
"technical_solution_files": legacy.get("technical_solution_files", []),
|
||
"project_profile_file": legacy.get("project_profile_file"),
|
||
"effective_terminology_files": legacy.get("effective_terminology_files", []),
|
||
"optional_terminology_files": legacy.get("optional_terminology_files", []),
|
||
"knowledge_base_files": legacy.get("knowledge_base_files", []),
|
||
}
|
||
save_manifest(base_name, "prepare", prepare_data)
|
||
|
||
# 从旧 manifest 推断 analyze 战区数据
|
||
analyze_data = {
|
||
"_meta": {"zone": "analyze", "base_name": base_name, "status": "completed",
|
||
"migrated_from_legacy": True},
|
||
"related_requirements": legacy.get("related_requirements", []),
|
||
"conflict_candidates_count": legacy.get("conflict_candidates_count", 0),
|
||
"confirmation_gate": legacy.get("confirmation_gate", {}),
|
||
"analysis_file": legacy.get("analysis_file"),
|
||
"relation_report_file": legacy.get("relation_report_file"),
|
||
}
|
||
save_manifest(base_name, "analyze", analyze_data)
|
||
|
||
return build_merged_manifest(base_name)
|
||
|
||
|
||
def get_pending_ai_agents(base_name: str) -> list[dict[str, Any]]:
|
||
"""扫描所有战区 manifest，返回状态为 'pending_ai' 的 Agent 列表。
|
||
|
||
返回按战区依赖顺序排列的待执行 AI Agent 信息。
|
||
"""
|
||
pending: list[dict[str, Any]] = []
|
||
for zone in ZONE_ORDER:
|
||
manifest = load_manifest_safe(base_name, zone)
|
||
if manifest is None:
|
||
continue
|
||
agent_notes = manifest.get("agent_notes", {})
|
||
for agent_id, note in agent_notes.items():
|
||
if isinstance(note, dict) and note.get("status") == "pending_ai":
|
||
pending.append({
|
||
"agent_id": agent_id,
|
||
"zone": zone,
|
||
"prompt_file": note.get("prompt_file", ""),
|
||
"output_file": note.get("output_file", ""),
|
||
"message": note.get("message", ""),
|
||
})
|
||
elif isinstance(note, str) and "待 AI Agent" in note:
|
||
pending.append({
|
||
"agent_id": agent_id,
|
||
"zone": zone,
|
||
"prompt_file": f"agents/{zone}/{agent_id.replace('-', '_')}.md",
|
||
"output_file": "",
|
||
"message": note,
|
||
})
|
||
return pending
|
||
|
||
|
||
def update_agent_status(base_name: str, zone: str, agent_id: str,
|
||
status: str, message: str = "") -> bool:
|
||
"""更新 manifest 中 Agent 的状态。
|
||
|
||
返回 True 表示更新成功，False 表示 manifest 不存在或 agent 不存在。
|
||
"""
|
||
manifest = load_manifest_safe(base_name, zone)
|
||
if manifest is None:
|
||
return False
|
||
agent_notes = manifest.setdefault("agent_notes", {})
|
||
if agent_id in agent_notes:
|
||
if isinstance(agent_notes[agent_id], dict):
|
||
agent_notes[agent_id]["status"] = status
|
||
agent_notes[agent_id]["message"] = message
|
||
else:
|
||
agent_notes[agent_id] = {"status": status, "message": message}
|
||
else:
|
||
agent_notes[agent_id] = {"status": status, "message": message}
|
||
save_manifest(base_name, zone, manifest)
|
||
return True
|