133 lines
7 KiB
Python
133 lines
7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
from django.core.exceptions import ValidationError
|
|
|
|
from agents.control_room import AgentControlRoomService
|
|
from agents.providers import DeterministicSolProvider
|
|
from control_plane.agents.management.commands.seed_core_agents import Command as SeedAgentsCommand
|
|
from control_plane.agents.models import Agent, AgentRole, AgentScope, AgentVersion, Competency, ImprovementCandidate, PromotionStatus, ProgenySignal
|
|
from control_plane.projects.models import Milestone, Project, ProjectPlan, Task, TaskAttempt, TaskStatus
|
|
from control_plane.resources.models import ModelRequest, Resource, ResourceKind
|
|
from control_plane.verification.models import Review, TestRun, Verification, VerificationLevel, VerificationResult
|
|
from graph.agent_control import agent_investigation_graph_v1, agent_investigation_registry
|
|
from graph.bootstrap import champion_agent_investigation_graph_v1
|
|
from graph.langgraph_runtime import LangGraphRuntime
|
|
from graph.models import GraphRun
|
|
from model_router.router import ModelRouter
|
|
|
|
|
|
def seed() -> AgentControlRoomService:
|
|
SeedAgentsCommand().handle()
|
|
return AgentControlRoomService()
|
|
|
|
|
|
def coder() -> AgentVersion:
|
|
return AgentVersion.objects.get(agent__role=AgentRole.CODER, promotion_status=PromotionStatus.CHAMPION)
|
|
|
|
|
|
def project_task(version: AgentVersion) -> Task:
|
|
project = Project.objects.create(name="Agent Control", goal="Inspect workforce")
|
|
plan = ProjectPlan.objects.create(project=project, version=1, goal=project.goal)
|
|
milestone = Milestone.objects.create(project=project, plan=plan, key="A", title="A", goal="A")
|
|
task = Task.objects.create(project=project, milestone=milestone, status=TaskStatus.COMPLETE, task_type="implementation", goal="Do work", retry_count=1)
|
|
attempt = TaskAttempt.objects.create(task=task, attempt_number=1, coder=version, status="COMPLETE")
|
|
TestRun.objects.create(project=project, task=task, command="pytest", status="PASS")
|
|
Review.objects.create(task=task, reviewer=AgentVersion.objects.get(agent__role=AgentRole.REVIEWER), status="PASS")
|
|
Verification.objects.create(project=project, task=task, level=VerificationLevel.TASK, result=VerificationResult.PASS)
|
|
ModelRequest.objects.create(project=project, agent_version=version, logical_role="CODING", model="qwen", status="COMPLETE", request={}, response={}, latency_ms=120, prompt_tokens=10, completion_tokens=20)
|
|
return task
|
|
|
|
|
|
def test_competencies_scope_and_team_fixture() -> None:
|
|
service = seed()
|
|
version = coder()
|
|
assignment = service.assign_competency(version, "django_backend", proficiency=0.8, confidence=0.9, evidence={"tests": "pass"})
|
|
team = service.create_software_feature_team()
|
|
|
|
assert Competency.objects.filter(key="django_backend").exists()
|
|
assert assignment.agent == version.agent
|
|
assert version.scope == AgentScope.GLOBAL
|
|
assert team.members.count() >= 4
|
|
assert any(member["role"] == "backend_coding" for member in service.get_team(team.id)["members"])
|
|
|
|
|
|
def test_champion_uniqueness_challenger_ancestry_and_fork() -> None:
|
|
service = seed()
|
|
version = coder()
|
|
with pytest.raises(ValidationError):
|
|
AgentVersion.objects.create(agent=version.agent, version=99, model="qwen", system_contract="bad", promotion_status=PromotionStatus.CHAMPION)
|
|
|
|
fork = service.fork_agent(version, name="Django Backend Engineer", scope=AgentScope.STUDIO)
|
|
|
|
assert fork.promotion_status == PromotionStatus.CHALLENGER
|
|
assert fork.parent_version == version
|
|
assert str(version.id) in fork.ancestry
|
|
assert fork.agent.scope == AgentScope.STUDIO
|
|
|
|
|
|
def test_performance_health_usage_and_model_visibility() -> None:
|
|
service = seed()
|
|
version = coder()
|
|
task = project_task(version)
|
|
Resource.objects.create(name="Qwen Worker Pool", kind=ResourceKind.MODEL_COMPUTE, provider="local_inference", roles=["qwen", "CODING"], health_status="AVAILABLE")
|
|
ProgenySignal.objects.create(agent_version=version, task=task, source="reviewer", severity="HIGH", failure_category="MALFORMED_OUTPUT", summary="Malformed structured output", grouping_key="agent-health")
|
|
|
|
performance = service.get_agent_performance(version.id)
|
|
health = service.get_agent_health(version.id)
|
|
usage = service.get_agent_usage(version.id)
|
|
detail = service.get_agent_version(version.id)
|
|
|
|
assert performance["quality"]["task_completion_rate"] == 1
|
|
assert performance["robustness"]["retry_rate"] == 1
|
|
assert health["status"] in {"WATCH", "DEGRADED"}
|
|
assert usage["task_types"] == ["implementation"]
|
|
assert detail["model"] == "qwen"
|
|
|
|
|
|
def test_investigate_extend_evolve_and_graph_lineage() -> None:
|
|
service = seed()
|
|
version = coder()
|
|
project_task(version)
|
|
graph_version = champion_agent_investigation_graph_v1()
|
|
graph_run = GraphRun.objects.create(execution_graph_version=graph_version, current_node=graph_version.graph_spec["entry"], metadata={"agent_version_id": str(version.id)})
|
|
|
|
LangGraphRuntime(agent_investigation_registry(service)).run_until_terminal_or_paused(graph_run)
|
|
graph_run.refresh_from_db()
|
|
plan = service.extend_agent(version, "accessibility")
|
|
candidate = service.evolve_agent(version, "Reduce malformed structured output", {"malformed_rate": 0.2})
|
|
|
|
assert graph_run.status == "COMPLETE"
|
|
assert "progeny_investigation_id" in graph_run.metadata
|
|
assert ImprovementCandidate.objects.exists()
|
|
assert plan.plan_type == "EXTEND"
|
|
assert candidate.target_type == "AGENT"
|
|
|
|
|
|
def test_frontend_agent_audit_bootstrap_and_design_artifact() -> None:
|
|
service = AgentControlRoomService(ModelRouter({"sol": DeterministicSolProvider(json.dumps({"InformationArchitecture": ["Workspace"], "FrontendImplementationPlan": {"framework_neutral": True, "django_compatible": True, "react_required": False}, "PageSpec": {}, "ComponentSpec": [], "InteractionSpec": [], "DesignTokens": {}, "ResponsiveRules": [], "AccessibilityRequirements": []}))}))
|
|
before = service.frontend_agent_audit()
|
|
versions = service.bootstrap_frontend_agents()
|
|
after = service.frontend_agent_audit()
|
|
project = Project.objects.create(name="Design Dogfood", goal="Design workspace")
|
|
artifact = service.create_frontend_design_artifact(project, versions["Product / Frontend Design Agent"])
|
|
|
|
assert before["missing"]
|
|
assert not after["missing"]
|
|
assert versions["Product / Frontend Design Agent"].permissions["design_only"] is True
|
|
assert artifact.content["FrontendImplementationPlan"]["framework_neutral"] is True
|
|
assert artifact.content["FrontendImplementationPlan"]["django_compatible"] is True
|
|
|
|
|
|
def test_machine_readable_api_shapes_and_challengers() -> None:
|
|
service = seed()
|
|
version = coder()
|
|
challenger = service.fork_agent(version, name="Coder Specialist")
|
|
|
|
assert service.list_agents()
|
|
assert service.get_agent(version.agent_id)["versions"]
|
|
challengers = service.get_agent_challengers(challenger.agent_id)
|
|
assert challengers["challengers"][0]["id"] == str(challenger.id)
|
|
assert agent_investigation_graph_v1().name == "agent_investigation"
|