68 lines
2.5 KiB
Python
68 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
from agents.progeny import ProgenyService
|
|
from control_plane.agents.models import Agent, AgentPlan, AgentRole, AgentVersion, PromotionStatus
|
|
from control_plane.events.models import Event, EventType
|
|
|
|
|
|
def test_m5_candidate_benchmark_and_promotion() -> None:
|
|
agent = Agent.objects.create(name="Reviewer", role=AgentRole.REVIEWER)
|
|
champion = AgentVersion.objects.create(
|
|
agent=agent,
|
|
version=1,
|
|
model="qwen",
|
|
system_contract="Review quality.",
|
|
promotion_status=PromotionStatus.CHAMPION,
|
|
)
|
|
agent.champion_version = champion
|
|
agent.save(update_fields=["champion_version", "updated_at"])
|
|
plan = AgentPlan.objects.create(
|
|
agent=agent,
|
|
name="Reviewer",
|
|
role=AgentRole.REVIEWER,
|
|
model="qwen",
|
|
system_contract="Review quality. Do not self-certify implementation.",
|
|
benchmarks=[{"task": "review diff"}],
|
|
success_criteria={"completion_rate": 1.0},
|
|
)
|
|
|
|
service = ProgenyService()
|
|
challenger = service.create_candidate_from_plan(plan)
|
|
run = service.replay_benchmark(champion, challenger, plan.benchmarks)
|
|
decision = service.promote_or_reject(run)
|
|
|
|
agent.refresh_from_db()
|
|
assert decision.decision == "PROMOTED"
|
|
assert agent.champion_version_id == challenger.id
|
|
assert challenger.promotion_status == PromotionStatus.CHAMPION
|
|
assert Event.objects.filter(event_type=EventType.AGENT_PROMOTED).exists()
|
|
|
|
|
|
def test_m5_rejects_weaker_challenger() -> None:
|
|
agent = Agent.objects.create(name="Judge", role=AgentRole.PROJECT_JUDGE)
|
|
champion = AgentVersion.objects.create(
|
|
agent=agent,
|
|
version=1,
|
|
model="qwen",
|
|
system_contract="Judge acceptance. Do not self-certify implementation.",
|
|
promotion_status=PromotionStatus.CHAMPION,
|
|
)
|
|
agent.champion_version = champion
|
|
agent.save(update_fields=["champion_version", "updated_at"])
|
|
plan = AgentPlan.objects.create(
|
|
agent=agent,
|
|
name="Judge",
|
|
role=AgentRole.PROJECT_JUDGE,
|
|
model="qwen",
|
|
system_contract="Judge acceptance.",
|
|
benchmarks=[{"task": "judge contract"}],
|
|
)
|
|
|
|
service = ProgenyService()
|
|
challenger = service.create_candidate_from_plan(plan)
|
|
run = service.replay_benchmark(champion, challenger, plan.benchmarks)
|
|
decision = service.promote_or_reject(run)
|
|
|
|
assert decision.decision == "REJECTED"
|
|
challenger.refresh_from_db()
|
|
assert challenger.promotion_status == PromotionStatus.REJECTED
|