fengqun/examples/run_swarm_characteristics_acceptance.py

from pathlib import Path
import json
import subprocess
import sys


ROOT = Path(__file__).resolve().parents[1]


SCENARIOS = [
    {
        "id": "F01",
        "feature": "去中心化",
        "name": "decentralization_no_single_agent_control_node",
        "claim": "没有单一 Agent 控制节点；多个 Agent 基于共享任务池自主 claim，并且没有重复领取。",
        "metrics": {
            "participating_agents_min": 4,
            "duplicate_claims": 0,
            "control_keys": 0,
            "decision_records": 8,
        },
        "test": "tests.test_swarm_characteristics_acceptance.SwarmCharacteristicsAcceptanceTest.test_decentralization_has_no_single_agent_control_node",
    },
    {
        "id": "F02",
        "feature": "自组织",
        "name": "self_organization_from_local_interactions",
        "claim": "无预设全局分类结果时，Agent 通过局部环境更新形成有序 dominant cluster。",
        "metrics": {
            "dominant_cluster": "api",
            "local_interaction_count": 5,
            "preseeded_global_plan": False,
        },
        "test": "tests.test_swarm_characteristics_acceptance.SwarmCharacteristicsAcceptanceTest.test_self_organization_forms_order_from_local_interactions",
    },
    {
        "id": "F03",
        "feature": "涌现性",
        "name": "emergence_global_result_exceeds_single_signal",
        "claim": "群体聚合结果超过任何单个局部弱信号，最终全局输出不能由单个 Agent 的局部值直接推出。",
        "metrics": {
            "global_score_must_exceed_best_single_local_signal": True,
            "expected_global_candidate": "beta",
        },
        "test": "tests.test_swarm_characteristics_acceptance.SwarmCharacteristicsAcceptanceTest.test_emergence_global_result_exceeds_single_local_signal",
    },
    {
        "id": "F04",
        "feature": "鲁棒性",
        "name": "robustness_single_agent_failure_isolated",
        "claim": "单个 Agent 失败不影响整体功能，失败任务可观测，其他任务仍能收敛。",
        "metrics": {
            "failed_tasks": 1,
            "completed_tasks_min": 2,
            "negative_observation_required": True,
        },
        "test": "tests.test_swarm_characteristics_acceptance.SwarmCharacteristicsAcceptanceTest.test_robustness_single_agent_failure_does_not_stop_convergence",
    },
    {
        "id": "F05",
        "feature": "可扩展性",
        "name": "scalability_three_five_seven_agents_same_architecture",
        "claim": "3/5/7 Agent 增减不改变共享任务池架构，任务完成、失败恢复和收敛保持稳定。",
        "metrics": {
            "agent_counts": [3, 5, 7],
            "tasks_per_agent": 2,
            "duplicate_claims": 0,
            "failed_tasks": 0,
        },
        "test": "tests.test_swarm_characteristics_acceptance.SwarmCharacteristicsAcceptanceTest.test_scalability_three_five_seven_agents_keep_same_architecture",
    },
    {
        "id": "F06",
        "feature": "隐式协作",
        "name": "implicit_collaboration_by_environment",
        "claim": "Agent 不需要直接互发消息，而是通过信息素和 shared_state 环境变化间接协调。",
        "metrics": {
            "direct_message_keys": 0,
            "first_claim": "high-signal",
            "environment_trail_required": True,
        },
        "test": "tests.test_swarm_characteristics_acceptance.SwarmCharacteristicsAcceptanceTest.test_implicit_collaboration_uses_environment_not_direct_messages",
    },
]


def main() -> None:
    results = []
    for scenario in SCENARIOS:
        command = [sys.executable, "-B", "-m", "unittest", scenario["test"]]
        completed = subprocess.run(
            command,
            cwd=ROOT,
            text=True,
            capture_output=True,
            timeout=120,
        )
        results.append(
            {
                "id": scenario["id"],
                "feature": scenario["feature"],
                "name": scenario["name"],
                "claim": scenario["claim"],
                "metrics": scenario["metrics"],
                "command": " ".join(command),
                "passed": completed.returncode == 0,
                "evidence": summarize_output(completed.stdout, completed.stderr),
            }
        )
        if completed.returncode != 0:
            break

    report = {
        "standard": "swarm-six-characteristics-v1",
        "status": "PASS" if len(results) == len(SCENARIOS) and all(item["passed"] for item in results) else "FAIL",
        "features": [
            "去中心化",
            "自组织",
            "涌现性",
            "鲁棒性",
            "可扩展性",
            "隐式协作",
        ],
        "scope_note": (
            "当前验收限定在本地最小蜂群原型，证明 Agent 层共享环境和自主 claim 的六特征。"
            "它不等同于无协调器的生产级分布式 runtime。"
        ),
        "scenarios": results,
    }
    print(json.dumps(report, ensure_ascii=False, indent=2))
    if report["status"] != "PASS":
        raise SystemExit(1)


def summarize_output(stdout: str, stderr: str) -> dict[str, str]:
    combined = "\n".join(part.strip() for part in [stdout, stderr] if part.strip())
    return {"tail": combined[-1000:] if combined else "<no output>"}


if __name__ == "__main__":
    main()