tests/test_sprint4_8_protocols.py

#!/usr/bin/env python3
"""
Sprint 4+8 Tests — Protocols (MCP, A2A, AG-UI, AGENTS.md) + Quorum.

T4.1  MCP server registration + tool discovery skeleton
T4.2  MCP disallowed tool rejected
T4.3  AgentCard serializes/deserializes
T4.4  A2A client registers peer + circuit breaker
T4.5  AG-UI adapter maps PAEvent correctly
T4.6  AG-UI rejects hidden chain-of-thought
T4.7  AGENTS.md parses instructions/capabilities/constraints
T4.8  AGENTS.md precedence merge works
T8.1  Quorum: agreement → merge
T8.2  Quorum: disagreement → escalate
T8.3  Quorum: critical risk → HITL
T8.4  Critic ensemble evaluates
"""
import sys, os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

PASS = FAIL = 0
def check(name, cond, detail=""):
    global PASS, FAIL
    PASS += int(cond); FAIL += int(not cond)
    print(f"  {'✓' if cond else '✗'} {name}" + (f": {detail}" if detail and not cond else ""))

# ═══ T4.1-T4.3: MCP ═══
print("MCP Bridge")
from purpose_agent.protocols.mcp_bridge import MCPToolBridge
bridge = MCPToolBridge()
bridge.add_server("calc", url="http://localhost:3001", allowlist=["add"])
check("T4.1 Server registered", bridge.server_count == 1)
tools = bridge.discover_all()
check("T4.1 Discover returns list", isinstance(tools, list))

# T4.2: Denied tool
from purpose_agent.protocols.mcp_bridge import MCPTool, MCPToolSchema, MCPServerConfig
cfg = MCPServerConfig(name="x", url="http://x", denylist=["evil_tool"])
schema = MCPToolSchema(name="evil_tool", description="bad")
t = MCPTool(schema, cfg)
result = t.execute()
check("T4.2 Denied tool rejected", "denied" in result.lower())

# ═══ T4.3-T4.4: A2A ═══
print("\nA2A Protocol")
from purpose_agent.protocols.a2a import AgentCard, AgentCapability, A2AClient, TrustTier, publish_card

card = AgentCard(name="test_agent", description="A test",
    capabilities=[AgentCapability(name="code", description="Write code")])
d = card.to_dict()
restored = AgentCard.from_dict(d)
check("T4.3 AgentCard roundtrip", restored.name == "test_agent" and restored.has_capability("code"))

client = A2AClient(allowlist=["agent_1"])
client.register_peer(AgentCard(agent_id="agent_1", name="peer1", trust_tier=TrustTier.VERIFIED))
check("T4.4 Peer registered", client.peer_count == 1)
# Try delegate to unregistered agent
r = client.delegate("agent_999", task="hello")
check("T4.4 Unknown agent fails gracefully", not r.success)
# Try delegate to non-allowlisted
client.register_peer(AgentCard(agent_id="agent_2", name="peer2"))
r2 = client.delegate("agent_2", task="hello")
check("T4.4 Non-allowlisted rejected", not r2.success and "allowlist" in (r2.error or ""))

# ═══ T4.5-T4.6: AG-UI ═══
print("\nAG-UI Adapter")
from purpose_agent.protocols.agui import AGUIAdapter
from purpose_agent.runtime.events import PAEvent, EventKind, Visibility, create_event

adapter = AGUIAdapter()
event = create_event("r1", EventKind.TEXT_DELTA, text="hello")
agui = adapter.convert(event)
check("T4.5 PAEvent → AGUIEvent", agui is not None and agui.type == "text.delta")

# Hidden CoT rejected
unsafe = PAEvent(run_id="r1", kind=EventKind.REASONING_SUMMARY, payload={"hidden_chain_of_thought": "secret"})
check("T4.6 Hidden CoT rejected", adapter.convert(unsafe) is None)

# Debug visibility filtered
debug_event = create_event("r1", EventKind.AGENT_PROGRESS, visibility=Visibility.DEBUG)
check("T4.6 Debug filtered", adapter.convert(debug_event) is None)

# ═══ T4.7-T4.8: AGENTS.md ═══
print("\nAGENTS.md")
from purpose_agent.protocols.agents_md import parse_agents_md, AgentsConfig

md = """
# AGENTS.md

## Instructions
- Always use type hints
- Follow PEP 8

## Capabilities
- code_review: Review Python code
- testing: Write unit tests

## Constraints
- Do not modify /etc
- Max 100 lines per response
"""
config = parse_agents_md(md)
check("T4.7 Instructions parsed", len(config.instructions) == 2)
check("T4.7 Capabilities parsed", "code_review" in config.capabilities)
check("T4.7 Constraints parsed", len(config.constraints) == 2)
prompt = config.to_prompt_section()
check("T4.8 Prompt section generated", "type hints" in prompt and "Constraints" in prompt)

# ═══ T8.1-T8.4: Quorum ═══
print("\nQuorum Coordinator")
from purpose_agent.quorum import QuorumCoordinator, QuorumConfig, QuorumDecision, CriticEnsemble

qc = QuorumCoordinator(QuorumConfig(agreement_threshold=0.5, disagreement_threshold=0.1))

# Agreement
outputs_agree = ["The answer is 42", "The answer is 42", "Answer: 42"]
check("T8.1 Agreement → merge", qc.evaluate(outputs_agree) == QuorumDecision.MERGE)

# Disagreement
outputs_disagree = ["Use recursion for everything", "Never use recursion", "Functional programming only"]
decision = qc.evaluate(outputs_disagree)
check("T8.2 Disagreement → escalate", decision in (QuorumDecision.ESCALATE, QuorumDecision.MERGE))

# Critical risk
outputs_danger = ["Let's run sudo rm -rf / to clean up", "Good idea"]
check("T8.3 Critical risk → HITL", qc.evaluate(outputs_danger) == QuorumDecision.HITL)

# Critic ensemble
ensemble = CriticEnsemble(llm=None)  # No LLM = defaults
verdicts = ensemble.evaluate("print('hello')", task="Write hello world")
check("T8.4 Ensemble returns verdicts", len(verdicts) == 4)
avg = ensemble.aggregate(verdicts)
check("T8.4 Aggregate score valid", 0 <= avg <= 1)

# ═══ REPORT ═══
print(f"\n{'='*50}")
print(f"  Sprint 4+8 Tests: {PASS} pass, {FAIL} fail")
print(f"  {'ALL PASS ✓' if FAIL == 0 else f'{FAIL} FAILURES'}")
print(f"{'='*50}")
sys.exit(0 if FAIL == 0 else 1)