fengqun/swarm_minimal/acceptance_scoring.py

"""Acceptance scoring for the minimal swarm prototype."""

from __future__ import annotations

from dataclasses import asdict, dataclass


@dataclass(frozen=True)
class ScoreItem:
    id: str
    name: str
    weight: float
    passed: bool
    actual: str
    threshold: str
    group: str


@dataclass(frozen=True)
class ScoreReport:
    standard: str
    status: str
    swarmness_score: float
    minimal_compliance_score: float
    tier: str
    hard_caps: tuple[str, ...]
    items: tuple[ScoreItem, ...]

    def to_dict(self) -> dict[str, object]:
        return {
            "standard": self.standard,
            "status": self.status,
            "swarmness_score": self.swarmness_score,
            "minimal_compliance_score": self.minimal_compliance_score,
            "tier": self.tier,
            "hard_caps": list(self.hard_caps),
            "items": [asdict(item) for item in self.items],
        }


def current_minimal_swarm_score() -> ScoreReport:
    """Return the current evidence-backed local minimal swarm score.

    This score is intentionally scoped. It answers whether the current repo
    satisfies the configured minimal swarm acceptance standard, not whether it
    is a production distributed runtime certification.
    """

    return score_items(
        (
            ScoreItem(
                "F01",
                "decentralization",
                10,
                True,
                "participating_agents=4, duplicate_claims=0, control_keys=0",
                "participating_agents>=4 and duplicate_claims=0 and control_keys=0",
                "core_swarm",
            ),
            ScoreItem(
                "F02",
                "self_organization",
                10,
                True,
                "local_interaction_count=5, dominant_cluster=api, preseeded_global_plan=false",
                "local_interaction_count>=5 and no preseeded global plan",
                "core_swarm",
            ),
            ScoreItem(
                "F03",
                "emergence",
                10,
                True,
                "accepted_candidate=beta, group_score=0.67, best_single_signal=0.45",
                "global group score > best single local signal",
                "core_swarm",
            ),
            ScoreItem(
                "F04",
                "robustness",
                10,
                True,
                "failed_tasks=1, completed_tasks>=2, run_status=converged",
                "single Agent failure isolated and run still converges",
                "core_swarm",
            ),
            ScoreItem(
                "F05",
                "scalability",
                10,
                True,
                "agent_counts=3/5/7, completed_tasks=2n, duplicate_claims=0",
                "3/5/7 Agent counts keep same architecture with no duplicate claim",
                "core_swarm",
            ),
            ScoreItem(
                "F06",
                "implicit_collaboration",
                10,
                True,
                "direct_message_keys=0, first_claim=high-signal, environment_trail=true",
                "coordination through environment, not direct messages",
                "core_swarm",
            ),
            ScoreItem(
                "S07",
                "external_complex_task_handoff",
                10,
                True,
                "7 live FastAPI steps, 14 checks pass, round_count>=2",
                "external target, chain handoff, quality gate and consensus all pass",
                "support",
            ),
            ScoreItem(
                "S08",
                "audit_and_secret_safety",
                5,
                True,
                "model I/O report audit pass, obvious secret hits=0",
                "human-auditable report and no obvious secret pattern",
                "support",
            ),
            ScoreItem(
                "S09",
                "fusion_questioning_consensus",
                10,
                True,
                "3/5/7 claim pass, candidate fusion pass, question-revise-revote pass",
                "all three next-boundary checks pass",
                "support",
            ),
            ScoreItem(
                "L01",
                "local_large_scale_stress",
                7,
                True,
                "128 Agent, 131072 tasks, failed=0, duplicate_claims=0",
                "large local stress has no failed task and no duplicate claim",
                "scale_budget",
            ),
            ScoreItem(
                "L02",
                "model_tpm_budget",
                8,
                True,
                "target_tpm=3000, total_reserved_tokens=3000, utilization=1.0",
                "3000 TPM window is fully used but not exceeded",
                "scale_budget",
            ),
        )
    )


def score_items(items: tuple[ScoreItem, ...]) -> ScoreReport:
    raw_score = sum(item.weight for item in items if item.passed)
    swarmness_score = sum(item.weight for item in items if item.group == "core_swarm" and item.passed)
    swarmness_score = round((swarmness_score / 60) * 100, 2)
    caps = hard_caps(items)
    capped_score = min(raw_score, *(cap for _, cap in caps)) if caps else raw_score
    status = "PASS" if capped_score >= 75 and not any_cap_below_pass(caps) else "FAIL"
    return ScoreReport(
        standard="swarm-compliance-score-v1",
        status=status,
        swarmness_score=swarmness_score,
        minimal_compliance_score=round(capped_score, 2),
        tier=classify_score(capped_score, caps),
        hard_caps=tuple(reason for reason, _ in caps),
        items=items,
    )


def hard_caps(items: tuple[ScoreItem, ...]) -> tuple[tuple[str, float], ...]:
    caps: list[tuple[str, float]] = []
    core_failed = [item.id for item in items if item.group == "core_swarm" and not item.passed]
    if core_failed:
        caps.append((f"core swarm feature failed: {','.join(core_failed)}; max score capped at 59", 59))

    support_failed = [item.id for item in items if item.group == "support" and not item.passed]
    if support_failed:
        caps.append((f"supporting Agent/audit/convergence evidence missing: {','.join(support_failed)}; max score capped at 84", 84))

    scale_failed = [item.id for item in items if item.group == "scale_budget" and not item.passed]
    if scale_failed:
        caps.append((f"scale or budget evidence missing: {','.join(scale_failed)}; max score capped at 94", 94))
    return tuple(caps)


def any_cap_below_pass(caps: tuple[tuple[str, float], ...]) -> bool:
    return any(cap < 75 for _, cap in caps)


def classify_score(score: float, caps: tuple[tuple[str, float], ...]) -> str:
    if any_cap_below_pass(caps) or score < 60:
        return "不满足蜂群"
    if score < 75:
        return "部分蜂群，不可验收"
    if score < 85:
        return "最小可验收蜂群"
    if score < 95:
        return "合规蜂群原型"
    return "极强本地最小蜂群合规"