宜博 / lemongraph

#!/usr/bin/env bash
set -euo pipefail

# SDD 一键闭环自检脚本（支持：py_compile + 4 场景重放 + schema/约束兼容 + 产物审计）
# 用法：
#   1) 全链路闭环（推荐）：bash scripts/sdd_selfcheck.sh [OUT_DIR]
#      - 支持环境变量：PRD_ROOT / PROJECT_NAME / VERSION / OWNER / PYTHON_BIN
#   2) 仅审计现有目录：bash scripts/sdd_selfcheck.sh check [OUT_DIR]
#
# 默认场景：
#   - default_graph：默认图模式
#   - disable_graph：--disable-graph
#   - custom_null：外部 manifest + constraints: null
#   - custom_bad：外部 manifest + constraints: bad 类型（字符串）

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
REPO_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"

PYTHON_BIN="${PYTHON_BIN:-python3}"
GENERATOR="$SCRIPT_DIR/sdd_mvcp_generator.py"
GRAPH_SCRIPT="$SCRIPT_DIR/sdd_exec_graph.py"
TEMPLATE_MANIFEST="$REPO_ROOT/templates/sdd_exec_graph_manifest.json"
SCHEMA_FILE="$REPO_ROOT/references/sdd_mvcp_contract_schema.json"

if ! command -v "$PYTHON_BIN" >/dev/null 2>&1; then
  echo "❌ missing python executable: $PYTHON_BIN"
  exit 1
fi

OUT_DIR="${OUT_DIR:-./SDD}"
MODE="full"

if [[ ${1:-} == "check" ]]; then
  MODE="check"
  shift
  OUT_DIR="${1:-$OUT_DIR}"
else
  # 全链路模式可选通过参数覆盖输出基目录
  OUT_DIR="${1:-$OUT_DIR}"
fi

PRD_ROOT="${PRD_ROOT:-/tmp/sdd_100_a}"
PROJECT_NAME="${PROJECT_NAME:-DemoProj}"
VERSION="${VERSION:-v1.0.0}"
OWNER="${OWNER:-ops}"

if [[ "$MODE" == "full" && ! -d "$PRD_ROOT" ]]; then
  echo "❌ PRD 根目录不存在：$PRD_ROOT"
  echo "请在 env 中设置 PRD_ROOT（例如 /tmp/sdd_100_a），或传入显式 PRD_ROOT="
  exit 1
fi

mkdir -p "$OUT_DIR"
OUT_DIR="$(cd "$OUT_DIR" && pwd)"

RUN_ROOT="$OUT_DIR/.sdd_selfcheck_runs"
mkdir -p "$RUN_ROOT"
LOG_DIR="$RUN_ROOT/logs"
mkdir -p "$LOG_DIR"

SC1="$RUN_ROOT/default_graph"
SC2="$RUN_ROOT/disable_graph"
SC3="$RUN_ROOT/custom_manifest_null"
SC4="$RUN_ROOT/custom_manifest_bad"

MAN_NULL="$RUN_ROOT/custom_manifest_null.json"
MAN_BAD="$RUN_ROOT/custom_manifest_bad.json"

failures=0

log() {
  echo "[$(date +%Y-%m-%dT%H:%M:%S)] $*"
}

assert_cmd() {
  local cmd="$1"
  if ! eval "$cmd"; then
    echo "❌ 失败：$cmd"
    return 1
  fi
}

run_py_compile() {
  log "步骤1/4：py_compile"
  "$PYTHON_BIN" -m py_compile "$GENERATOR" "$GRAPH_SCRIPT"
}

build_custom_manifest() {
  local src="$1"
  local dst="$2"
  local constraints_mode="$3"

  "$PYTHON_BIN" - "$src" "$dst" "$constraints_mode" "$PROJECT_NAME" "$VERSION" "$OWNER" <<'PY'
import json
import sys
from pathlib import Path

src, dst, constraints_mode, project, version, owner = sys.argv[1:7]
obj = json.loads(Path(src).read_text(encoding="utf-8"))

meta = obj.get("meta", {})
if not isinstance(meta, dict):
    meta = {}

meta.update(
    {
        "project": project,
        "version": version,
        "out_dir": str(Path(dst).parent),
        "requirements_count": 1,
        "parallel_mode": "e2e_check",
        "derived_rules": {
            "strict_contract_mode": False,
            "source": "selfcheck",
        },
        "generated_at": "1970-01-01T00:00:00Z",
    }
)
obj["meta"] = meta

if constraints_mode == "null":
    obj["constraints"] = None
elif constraints_mode == "bad":
    obj["constraints"] = "invalid-constraints-type"

obj.setdefault("agent_pool", ["agent-1", "agent-2", "agent-3", "agent-4", "agent-5"])
obj.setdefault("loop_rules", [])
obj.setdefault("nodes", [])

# 保证 owner/owner_pool 不影响测试过程
Path(dst).write_text(json.dumps(obj, ensure_ascii=False, indent=2), encoding="utf-8")
PY
}

run_generator() {
  local label="$1"
  local out="$2"
  local disable_graph="$3"
  local manifest="$4"

  local log_file="$LOG_DIR/${label}.log"
  rm -rf "$out"
  mkdir -p "$out"

  local cmd=("$PYTHON_BIN" "$GENERATOR" --project-name "$PROJECT_NAME" --version "$VERSION" --owner "$OWNER" --prd-root "$PRD_ROOT" --out "$out")

  if [[ "$disable_graph" == "1" ]]; then
    cmd+=("--disable-graph")
  fi
  if [[ -n "$manifest" ]]; then
    cmd+=("--manifest" "$manifest")
  fi

  log "运行场景 ${label}：${cmd[*]}"
  if ! ("${cmd[@]}" >"$log_file" 2>&1); then
    echo "❌ 场景 ${label} 生成失败，日志：$log_file"
    cat "$log_file"
    return 1
  fi

  log "场景 ${label} 生成完成，日志：$log_file"
}

validate_artifacts() {
  local out_dir="$1"
  local expect_graph="$2"  # 1/0
  local expect_manifest="$3"  # 可选，空串时不强制
  local scenario="$4"

  "$PYTHON_BIN" - "$out_dir" "$expect_graph" "$expect_manifest" "$SCHEMA_FILE" "$scenario" <<'PY'
import json
import os
import sys
from pathlib import Path

out = Path(sys.argv[1])
expect_graph = sys.argv[2] == "1"
expected_manifest = sys.argv[3].strip()
schema_path = Path(sys.argv[4])
scenario = sys.argv[5]

ok = True


def check(cond, msg):
    global ok
    if cond:
        print(f"✅ {msg}")
    else:
        print(f"❌ {msg}")
        ok = False


def check_schema(contract_path: Path) -> None:
    global ok
    if not schema_path.exists():
        print(f"ℹ️ schema 文件不存在，跳过 jsonschema 校验: {schema_path}")
        return

    try:
        import jsonschema
    except Exception:
        print("ℹ️ 未安装 jsonschema，跳过 jsonschema 校验")
        return

    schema = json.loads(schema_path.read_text(encoding="utf-8"))
    contract = json.loads(contract_path.read_text(encoding="utf-8"))
    try:
        jsonschema.validate(contract, schema)
    except Exception as exc:
        print(f"❌ jsonschema 校验失败: {exc}")
        ok = False
    else:
        print("✅ jsonschema 校验通过")


# contract
contract_path = out / "sdd-mvcp-contract.json"
check(contract_path.exists(), f"[{scenario}] 合约文件存在")
if contract_path.exists():
    contract = json.loads(contract_path.read_text(encoding="utf-8"))
    for k in ["model_contract", "view_contract", "controller_contract", "process_contract"]:
        check(isinstance(contract.get(k), list) and len(contract[k]) > 0, f"[{scenario}] contract[{k}] 非空")

    check(isinstance(contract.get("acceptance_matrix"), list) and len(contract.get("acceptance_matrix", [])) > 0, f"[{scenario}] acceptance_matrix 非空")
    check(isinstance(contract.get("assumptions"), list) and len(contract.get("assumptions", [])) > 0, f"[{scenario}] assumptions 非空")
    check(isinstance(contract.get("risks"), list) and len(contract.get("risks", [])) > 0, f"[{scenario}] risks 非空")
    check(isinstance(contract.get("non_goals"), list) and len(contract.get("non_goals", [])) > 0, f"[{scenario}] non_goals 非空")

    check_schema(contract_path)

runbook_path = out / "sdd-graph.runbook.json"
if expect_graph:
    check(runbook_path.exists(), f"[{scenario}] runbook 存在（图场景）")
    if runbook_path.exists():
        runbook = json.loads(runbook_path.read_text(encoding="utf-8"))
        check(isinstance(runbook.get("source_manifest"), str) and bool(runbook.get("source_manifest")), f"[{scenario}] runbook.source_manifest 存在")
        check(bool(runbook.get("sdd_dir")), f"[{scenario}] runbook.sdd_dir 存在")
        check(isinstance(runbook.get("manifest_meta"), dict), f"[{scenario}] runbook.manifest_meta 为 dict")
        check(isinstance(runbook.get("constraints"), dict), f"[{scenario}] runbook.constraints 为 dict")
        manifest_meta = runbook.get("manifest_meta") or {}
        check(isinstance(manifest_meta.get("derived_rules"), dict), f"[{scenario}] runbook.manifest_meta.derived_rules 为 dict")

        outcome = runbook.get("outcome") or {}
        waves = runbook.get("waves", [])
        check(isinstance(outcome, dict), f"[{scenario}] runbook.outcome 存在")
        if isinstance(outcome, dict):
            check(outcome.get("total_waves") == len(waves), f"[{scenario}] runbook.outcome.total_waves 与 waves 长度一致")
            check(isinstance(outcome.get("total_nodes"), int), f"[{scenario}] outcome.total_nodes 存在")

        um = out / "unit_matrix.json"
        hb = out / "handoff_bundle.md"
        check(um.exists(), f"[{scenario}] unit_matrix 存在")
        if um.exists():
            try:
                um_content = json.loads(um.read_text(encoding="utf-8"))
                check(bool(um_content.get("units")), f"[{scenario}] unit_matrix.units 非空")
            except Exception:
                check(False, f"[{scenario}] unit_matrix 可解析")
        check(hb.exists() and hb.stat().st_size > 0, f"[{scenario}] handoff_bundle.md 存在且非空")

        check("manifest_meta_keys" in (out / "00-SDD-生成记录.md").read_text(encoding="utf-8"), f"[{scenario}] 生成记录包含 manifest_meta_keys")
        check("manifest_constraint_keys" in (out / "00-SDD-生成记录.md").read_text(encoding="utf-8"), f"[{scenario}] 生成记录包含 manifest_constraint_keys")

        if expected_manifest:
            # 与旧版本兼容：优先保证写入的是可落盘 manifest 文件路径即可
            check(os.path.basename(str(runbook.get("source_manifest", "")) or "").endswith("sdd_exec_graph_manifest.json"), f"[{scenario}] source_manifest 可落盘")
else:
    check(not runbook_path.exists(), f"[{scenario}] disable-graph 场景下 runbook 不应存在")
    # 非图场景不要求 unit_matrix/handoff

if not ok:
    sys.exit(1)
PY
}

run_full() {
  run_py_compile

  build_custom_manifest "$TEMPLATE_MANIFEST" "$MAN_NULL" null
  build_custom_manifest "$TEMPLATE_MANIFEST" "$MAN_BAD" bad

  log "步骤2/4：场景1 default_graph（默认图）"
  if ! run_generator "default_graph" "$SC1" "0" ""; then
    failures=$((failures + 1))
    return 1
  fi

  log "步骤3/4：场景2 disable_graph（禁图）"
  if ! run_generator "disable_graph" "$SC2" "1" ""; then
    failures=$((failures + 1))
    return 1
  fi

  log "步骤4/4：场景3 custom_manifest_null（外部 manifest + constraints null）"
  if ! run_generator "custom_manifest_null" "$SC3" "0" "$MAN_NULL"; then
    failures=$((failures + 1))
    return 1
  fi

  log "步骤4+：场景4 custom_manifest_bad（外部 manifest + constraints 非法类型）"
  if ! run_generator "custom_manifest_bad" "$SC4" "0" "$MAN_BAD"; then
    failures=$((failures + 1))
    return 1
  fi

  validate_artifacts "$SC1" 1 "" "default_graph" && validate_artifacts "$SC2" 0 "" "disable_graph" && validate_artifacts "$SC3" 1 "$MAN_NULL" "custom_manifest_null" && validate_artifacts "$SC4" 1 "$MAN_BAD" "custom_manifest_bad"

  if [[ $? -ne 0 ]]; then
    failures=$((failures + 1))
    return 1
  fi

  return 0
}

run_check() {
  if [[ ! -d "$OUT_DIR" ]]; then
    echo "❌ 检查目录不存在：$OUT_DIR"
    return 1
  fi

  if [[ -f "$OUT_DIR/sdd-graph.runbook.json" ]]; then
    validate_artifacts "$OUT_DIR" 1 "" "check_existing"
  else
    validate_artifacts "$OUT_DIR" 0 "" "check_existing"
  fi
}

if [[ "$MODE" == "check" ]]; then
  if ! run_check; then
    echo "❌ 仅校验模式失败：$OUT_DIR"
    exit 1
  fi
  echo "✅ sdd_selfcheck 一键闭环通过（仅产物校验）"
  echo "检查目录：$OUT_DIR"
  exit 0
else
  if ! run_full; then
    echo "❌ sdd_selfcheck 一键闭环失败，请查看日志：$LOG_DIR"
    echo "  - 失败次数：$failures"
    exit 1
  fi
fi

echo "✅ sdd_selfcheck 一键闭环通过"
echo "输出目录："
echo "  default_graph:      $SC1"
echo "  disable_graph:      $SC2"
echo "  custom_manifest_n:  $SC3"
echo "  custom_manifest_b:  $SC4"