Add Agent Control Room V1

agents/control_room.py

@@ -0,0 +1,241 @@
+from __future__ import annotations
+
+import json
+from datetime import timedelta
+from statistics import median
+
+from django.utils import timezone
+
+from agents.progeny import ProgenyService
+from control_plane.agents.models import Agent, AgentCompetency, AgentPlan, AgentRole, AgentRun, AgentScope, AgentTeam, AgentTeamMember, AgentVersion, BenchmarkRun, Competency, ImprovementCandidate, ProgenySignal, PromotionStatus, ProgenyInvestigation, ProgenyExperiment, ReplayRun
+from control_plane.events.bus import EventBus
+from control_plane.projects.models import Artifact, CommitRecord, Project, Task, TaskAttempt
+from control_plane.resources.models import ModelRequest, Resource
+from control_plane.verification.models import Review, TestRun, Verification, VerificationLevel, VerificationResult
+from graph.models import GraphRun
+from model_router.router import ModelCapability, ModelRequestContract, ModelRouter
+
+
+COMPETENCIES = [
+    ("django_backend", "Django Backend", "backend"),
+    ("frontend_design", "Frontend/Product Design", "frontend"),
+    ("frontend_engineering", "Frontend Engineering", "frontend"),
+    ("testing", "Testing", "quality"),
+    ("security_review", "Security Review", "quality"),
+    ("architecture", "Architecture", "strategy"),
+    ("product_strategy", "Product Strategy", "strategy"),
+    ("repository_archaeology", "Repository Archaeology", "analysis"),
+    ("performance_analysis", "Performance Analysis", "quality"),
+    ("accessibility", "Accessibility", "frontend"),
+    ("model_training", "Model Training", "ml"),
+]
+
+
+class AgentControlRoomService:
+    def __init__(self, router: ModelRouter | None = None, bus: EventBus | None = None) -> None:
+        self.router = router
+        self.bus = bus or EventBus()
+
+    def ensure_competencies(self) -> list[Competency]:
+        competencies = []
+        for key, name, domain in COMPETENCIES:
+            competency, _ = Competency.objects.get_or_create(key=key, defaults={"name": name, "domain": domain, "description": f"{name} competency."})
+            competencies.append(competency)
+        return competencies
+
+    def assign_competency(self, version: AgentVersion, key: str, *, proficiency: float = 0.5, confidence: float = 0.5, evidence: dict[str, object] | None = None, source: str = "control_room") -> AgentCompetency:
+        competency, _ = Competency.objects.get_or_create(key=key, defaults={"name": key.replace("_", " ").title()})
+        assignment, _ = AgentCompetency.objects.update_or_create(agent_version=version, competency=competency, defaults={"agent": version.agent, "proficiency": proficiency, "confidence": confidence, "evidence": evidence or {}, "source": source, "last_evaluated_at": timezone.now()})
+        return assignment
+
+    def bootstrap_frontend_agents(self) -> dict[str, AgentVersion]:
+        self.ensure_competencies()
+        specs = [
+            ("Product / Frontend Design Agent", AgentRole.FRONTEND_DESIGNER, "sol", ["frontend_design", "product_strategy", "architecture", "accessibility"], "Produce stack-neutral InformationArchitecture, PageSpec, ComponentSpec, InteractionSpec, DesignTokens, ResponsiveRules, AccessibilityRequirements, and FrontendImplementationPlan. Do not assume React; preserve Django templates/HTMX/Alpine where appropriate."),
+            ("Frontend Engineer", AgentRole.FRONTEND_ENGINEER, "qwen", ["frontend_engineering", "django_backend", "testing", "accessibility"], "Implement frontend work after inspecting the project stack. Preserve existing architecture; prefer Django templates/HTMX/Alpine for Django projects unless a plan explicitly chooses another framework."),
+            ("UX / Accessibility Reviewer", AgentRole.UX_ACCESSIBILITY_REVIEWER, "qwen", ["accessibility", "frontend_design", "testing"], "Review UX, accessibility, responsive states, keyboard behavior, and implementation fit without forcing a framework."),
+            ("Visual Judge", AgentRole.VISUAL_JUDGE, "qwen", ["frontend_design", "accessibility"], "Placeholder visual evaluator until screenshot infrastructure is available; judge design contracts and static artifacts."),
+        ]
+        versions: dict[str, AgentVersion] = {}
+        for name, role, model, competencies, contract in specs:
+            agent, created = Agent.objects.get_or_create(name=name, defaults={"role": role, "description": contract, "purpose": contract, "scope": AgentScope.GLOBAL})
+            version, version_created = AgentVersion.objects.get_or_create(agent=agent, version=1, defaults={"model": model, "system_contract": contract, "capabilities": competencies, "context_policy": {"include_raw_secrets": False, "inspect_project_stack": True}, "retrieval_policy": {"project_stack_first": True}, "tool_policy": {"framework_neutral": True}, "retry_policy": {"max_retries": 1}, "permissions": {"design_only": role == AgentRole.FRONTEND_DESIGNER}, "promotion_status": PromotionStatus.CHALLENGER, "benchmark_status": "SEEDED", "creation_source": "agent_control_room", "scope": AgentScope.GLOBAL})
+            if version_created:
+                self.bus.publish("AGENT_VERSION_CREATED", actor="agent_control_room", payload={"agent": name, "version": version.version})
+            if created:
+                self.bus.publish("AGENT_CREATED", actor="agent_control_room", payload={"agent": name})
+            if agent.champion_version_id is None:
+                agent.champion_version = version
+                version.promotion_status = PromotionStatus.CHAMPION
+                version.immutable_since = timezone.now()
+                version.save(update_fields=["promotion_status", "immutable_since", "updated_at"])
+                agent.save(update_fields=["champion_version", "updated_at"])
+            for competency in competencies:
+                self.assign_competency(version, competency, proficiency=0.7, confidence=0.7, evidence={"bootstrap": True})
+            versions[name] = version
+        return versions
+
+    def frontend_agent_audit(self) -> dict[str, object]:
+        coverage = {
+            "Product / UX Design": AgentVersion.objects.filter(agent__role=AgentRole.FRONTEND_DESIGNER).exists(),
+            "Frontend Architecture": AgentCompetency.objects.filter(competency__key="architecture", agent_version__agent__role=AgentRole.FRONTEND_DESIGNER).exists(),
+            "Frontend Engineering": AgentVersion.objects.filter(agent__role=AgentRole.FRONTEND_ENGINEER).exists(),
+            "Accessibility Review": AgentVersion.objects.filter(agent__role=AgentRole.UX_ACCESSIBILITY_REVIEWER).exists(),
+            "Visual / UX Review": AgentVersion.objects.filter(agent__role__in=[AgentRole.UX_ACCESSIBILITY_REVIEWER, AgentRole.VISUAL_JUDGE]).exists(),
+        }
+        return {"coverage": coverage, "missing": [name for name, present in coverage.items() if not present]}
+
+    def create_software_feature_team(self) -> AgentTeam:
+        versions = self.bootstrap_frontend_agents()
+        coder = AgentVersion.objects.filter(agent__role=AgentRole.CODER, promotion_status=PromotionStatus.CHAMPION).first()
+        reviewer = AgentVersion.objects.filter(agent__role=AgentRole.REVIEWER, promotion_status=PromotionStatus.CHAMPION).first()
+        team, created = AgentTeam.objects.get_or_create(name="Software Feature Team", defaults={"purpose": "Design, implement, and review software features.", "scope": AgentScope.GLOBAL})
+        if created:
+            self.bus.publish("TEAM_CREATED", payload={"team_id": str(team.id), "name": team.name})
+        for role, version in [("product_frontend_design", versions["Product / Frontend Design Agent"]), ("frontend_engineering", versions["Frontend Engineer"]), ("backend_coding", coder), ("review_quality", reviewer)]:
+            if version is None:
+                continue
+            _, member_created = AgentTeamMember.objects.get_or_create(team=team, agent_version=version, role=role, defaults={"responsibilities": [role]})
+            if member_created:
+                self.bus.publish("TEAM_MEMBER_ADDED", payload={"team_id": str(team.id), "agent_version_id": str(version.id), "role": role})
+        return team
+
+    def list_agents(self) -> list[dict[str, object]]:
+        return [self._agent(agent) for agent in Agent.objects.select_related("champion_version").order_by("name")]
+
+    def get_agent(self, agent_id) -> dict[str, object]:
+        return self._agent(Agent.objects.get(id=agent_id), include_versions=True)
+
+    def get_agent_version(self, version_id) -> dict[str, object]:
+        version = AgentVersion.objects.select_related("agent", "parent_version").get(id=version_id)
+        return {"id": str(version.id), "agent_id": str(version.agent_id), "agent": version.agent.name, "version": version.version, "model": version.model, "model_config": version.model_config, "system_contract": version.system_contract, "context_policy": version.context_policy, "retrieval_policy": version.retrieval_policy, "tool_policy": version.tool_policy, "retry_policy": version.retry_policy, "workflow": version.workflow, "permissions": version.permissions, "resource_preferences": version.resource_preferences, "scope": version.scope, "parent_version_id": str(version.parent_version_id) if version.parent_version_id else None, "ancestry": version.ancestry, "creation_source": version.creation_source, "benchmark_status": version.benchmark_status, "promotion_status": version.promotion_status, "competencies": list(version.competencies.select_related("competency").values("competency__key", "competency__name", "proficiency", "confidence", "evidence", "source"))}
+
+    def get_agent_usage(self, version_id) -> dict[str, object]:
+        version = AgentVersion.objects.get(id=version_id)
+        role = version.agent.role
+        attempts = TaskAttempt.objects.filter(coder=version)
+        commits = CommitRecord.objects.filter(coder=version) | CommitRecord.objects.filter(reviewer=version) | CommitRecord.objects.filter(judge=version)
+        return {"execution_graph_versions": list(version.experiment_variants.exclude(execution_graph_version=None).values("execution_graph_version__graph__name", "execution_graph_version__version").distinct()), "graph_runs": GraphRun.objects.filter(task__attempts__coder=version).distinct().count(), "projects": list(attempts.exclude(task__project=None).values_list("task__project__name", flat=True).distinct()), "task_types": list(attempts.values_list("task__task_type", flat=True).distinct()), "recent_agent_runs": list(version.runs.order_by("-created_at").values("id", "status", "metrics")[:10]), "replay_experiments": ProgenyExperiment.objects.filter(variants__agent_version=version).distinct().count(), "commit_count": commits.distinct().count(), "logical_role": role}
+
+    def get_agent_performance(self, version_id, *, window: str = "lifetime", project_id=None, task_type: str | None = None, last_n: int | None = None) -> dict[str, object]:
+        version = AgentVersion.objects.get(id=version_id)
+        attempts = TaskAttempt.objects.filter(coder=version).select_related("task")
+        if project_id:
+            attempts = attempts.filter(task__project_id=project_id)
+        if task_type:
+            attempts = attempts.filter(task__task_type=task_type)
+        if window == "7d":
+            attempts = attempts.filter(created_at__gte=timezone.now() - timedelta(days=7))
+        if window == "30d":
+            attempts = attempts.filter(created_at__gte=timezone.now() - timedelta(days=30))
+        if last_n:
+            ids = list(attempts.order_by("-created_at").values_list("id", flat=True)[:last_n])
+            attempts = TaskAttempt.objects.filter(id__in=ids)
+        tasks = Task.objects.filter(attempts__in=attempts).distinct()
+        task_count = tasks.count()
+        complete = tasks.filter(status="COMPLETE").count()
+        test_runs = TestRun.objects.filter(task__in=tasks)
+        reviews = Review.objects.filter(task__in=tasks)
+        judges = Verification.objects.filter(task__in=tasks, level=VerificationLevel.TASK)
+        requests = ModelRequest.objects.filter(agent_version=version)
+        latencies = [item for item in requests.exclude(latency_ms=None).values_list("latency_ms", flat=True)]
+        telemetry = [run.metadata.get("telemetry", {}) for run in GraphRun.objects.filter(task__in=tasks)]
+        mutation_failures = sum(int(t.get("patch_mismatches", 0) or 0) + int(t.get("write_file_fallbacks", 0) or 0) for t in telemetry)
+        return {"window": window, "filters": {"project_id": str(project_id) if project_id else None, "task_type": task_type, "last_n": last_n}, "quality": {"task_count": task_count, "task_completion_rate": complete / task_count if task_count else None, "reviewer_pass": reviews.filter(status="PASS").count(), "reviewer_rework": reviews.filter(status="REWORK_REQUIRED").count(), "reviewer_reject": reviews.filter(status="REJECTED").count(), "judge_pass_rate": judges.filter(result=VerificationResult.PASS).count() / judges.count() if judges.count() else None, "test_pass_rate": test_runs.filter(status="PASS").count() / test_runs.count() if test_runs.count() else None}, "robustness": {"retry_rate": tasks.exclude(retry_count=0).count() / task_count if task_count else None, "retry_exhaustion": tasks.filter(status="FAILED").count(), "malformed_model_output": ProgenySignal.objects.filter(agent_version=version, failure_category__icontains="MALFORMED").count(), "mutation_failures": mutation_failures, "provider_failures": requests.exclude(failure_reason="").count()}, "efficiency": {"median_runtime_ms": median(latencies) if latencies else None, "model_requests_per_task": requests.count() / task_count if task_count else None, "tokens_per_task": (sum(r.prompt_tokens or 0 for r in requests) + sum(r.completion_tokens or 0 for r in requests)) / task_count if task_count else None, "mutation_operations_per_task": sum(int(t.get("mutation_operations", 0) or 0) for t in telemetry) / task_count if task_count else None}, "systemic": {"progeny_signals": ProgenySignal.objects.filter(agent_version=version).count(), "investigations": ProgenyInvestigation.objects.filter(signals__agent_version=version).distinct().count(), "scenario_failures": version.agent.project.scenario_findings.count() if version.agent.project_id else 0}}
+
+    def get_agent_health(self, version_id) -> dict[str, object]:
+        version = AgentVersion.objects.get(id=version_id)
+        perf = self.get_agent_performance(version_id, window="30d")
+        reasons = []
+        status = "HEALTHY"
+        completion = perf["quality"]["task_completion_rate"]
+        test_pass = perf["quality"]["test_pass_rate"]
+        unresolved = ProgenySignal.objects.filter(agent_version=version, status="OPEN").count()
+        provider_available = self._provider_health(version)
+        if version.promotion_status == PromotionStatus.CHALLENGER:
+            status = "CHALLENGED"
+            reasons.append("Version is a challenger awaiting benchmark/promotion evidence.")
+        if version.promotion_status == PromotionStatus.DISABLED or version.status == "DISABLED":
+            status = "DISABLED"
+            reasons.append("Version is disabled.")
+        if completion is not None and completion < 0.5:
+            status = "DEGRADED"
+            reasons.append("Recent task completion rate below 50%.")
+        if test_pass is not None and test_pass < 0.7 and status != "DEGRADED":
+            status = "WATCH"
+            reasons.append("Recent test pass rate below 70%.")
+        if unresolved:
+            status = "DEGRADED" if unresolved >= 3 else ("WATCH" if status == "HEALTHY" else status)
+            reasons.append(f"{unresolved} unresolved Progeny signal(s).")
+        if provider_available == "UNAVAILABLE":
+            status = "DEGRADED"
+            reasons.append("Configured provider is unavailable.")
+        return {"status": status, "reasons": reasons or ["No negative health evidence in selected window."], "evidence": {"performance": perf, "provider_health": provider_available, "benchmark_status": version.benchmark_status}}
+
+    def get_agent_progeny(self, version_id) -> dict[str, object]:
+        version = AgentVersion.objects.get(id=version_id)
+        return {"signals": list(ProgenySignal.objects.filter(agent_version=version).values("id", "source", "severity", "failure_category", "summary", "status")), "investigations": list(ProgenyInvestigation.objects.filter(signals__agent_version=version).distinct().values("id", "status", "recommended_target", "recommended_route", "confidence")), "improvement_candidates": list(version.improvement_candidates.values("id", "status", "hypothesis", "recommended_route"))}
+
+    def get_agent_challengers(self, agent_id) -> dict[str, object]:
+        agent = Agent.objects.get(id=agent_id)
+        return {"champion": str(agent.champion_version_id) if agent.champion_version_id else None, "challengers": [self.get_agent_version(v.id) for v in agent.versions.filter(promotion_status=PromotionStatus.CHALLENGER)], "benchmarks": list(BenchmarkRun.objects.filter(champion__agent=agent).values("id", "champion_id", "challenger_id", "metrics", "decision"))}
+
+    def investigate_agent(self, version: AgentVersion) -> ProgenyInvestigation:
+        signal = ProgenySignal.objects.create(agent_version=version, source="agent_control_room", severity="MEDIUM", failure_category="AGENT_HEALTH", summary=f"Investigate {version.agent.name} v{version.version} based on Control Room telemetry.", evidence={"performance": self.get_agent_performance(version.id), "health": self.get_agent_health(version.id)}, grouping_key=f"agent:{version.id}:health")
+        self.bus.publish("AGENT_INVESTIGATION_STARTED", actor="agent_control_room", payload={"agent_version_id": str(version.id), "signal_id": str(signal.id)})
+        investigation = ProgenyService(self.bus).create_smart_investigation(signal.grouping_key)
+        self.bus.publish("AGENT_INVESTIGATION_COMPLETED", actor="agent_control_room", payload={"agent_version_id": str(version.id), "investigation_id": str(investigation.id)})
+        return investigation
+
+    def create_improvement_candidate(self, investigation: ProgenyInvestigation) -> ImprovementCandidate:
+        return ProgenyService(self.bus).create_improvement_candidate(investigation)
+
+    def extend_agent(self, version: AgentVersion, competency_key: str) -> AgentPlan:
+        competency, _ = Competency.objects.get_or_create(key=competency_key, defaults={"name": competency_key.replace("_", " ").title()})
+        return AgentPlan.objects.create(agent=version.agent, name=f"Extend {version.agent.name} with {competency.name}", role=version.agent.role, model=version.model, system_contract=version.system_contract, capabilities=[*version.capabilities, competency_key], tools=version.tools, permissions=version.permissions, context_policy=version.context_policy, workflow=version.workflow, benchmarks=[{"competency": competency_key}], success_criteria={"competency_added": competency_key}, target_agent_version=version, plan_type="EXTEND", scope=version.scope, evidence={"parent_version_id": str(version.id)}, status="DRAFT", created_by="agent_control_room")
+
+    def evolve_agent(self, version: AgentVersion, objective: str, baseline: dict[str, object]) -> ImprovementCandidate:
+        investigation = ProgenyInvestigation.objects.create(signal_clusters=[], affected_agents=[str(version.id)], affected_projects=[], affected_graph_versions=[], affected_nodes=[], hypotheses=[{"hypothesis": objective, "baseline": baseline}], recommended_target="AGENT", confidence=0.7, recommended_route="ReplayArena", proposed_experiments=["Replay recent failures against challenger prompt/tool policy."], expected_impact=objective, estimated_cost="MEDIUM")
+        return ImprovementCandidate.objects.create(investigation=investigation, target_type="AGENT", target_id=str(version.id), target_label=f"{version.agent.name} v{version.version}", hypothesis=objective, recommended_route="ReplayArena", evidence={"baseline": baseline}, agent_version=version)
+
+    def fork_agent(self, version: AgentVersion, *, name: str, scope: str = AgentScope.GLOBAL, project: Project | None = None) -> AgentVersion:
+        agent = Agent.objects.create(name=name, role=version.agent.role, description=f"Specialized descendant of {version.agent.name}", purpose=version.agent.purpose, scope=scope, project=project)
+        child = AgentVersion.objects.create(agent=agent, version=1, model=version.model, model_config=version.model_config, system_contract=version.system_contract, capabilities=version.capabilities, tools=version.tools, permissions=version.permissions, context_policy=version.context_policy, retrieval_policy=version.retrieval_policy, tool_policy=version.tool_policy, workflow=version.workflow, retry_policy=version.retry_policy, resource_preferences=version.resource_preferences, graph_usage_policy=version.graph_usage_policy, scope=scope, parent_version=version, ancestry=[*version.ancestry, str(version.id)], creation_source="fork", promotion_status=PromotionStatus.CHALLENGER, benchmark_status="REQUIRED")
+        self.bus.publish("AGENT_CHALLENGER_CREATED", actor="agent_control_room", payload={"parent_version_id": str(version.id), "child_version_id": str(child.id)})
+        return child
+
+    def create_frontend_design_artifact(self, project: Project, version: AgentVersion) -> Artifact:
+        contract = {"InformationArchitecture": ["Project list", "Project workspace", "Agent Control Room", "Roadmap", "Scenario Lab"], "PageSpec": {"ProjectWorkspace": "Stack-neutral workspace for project status, lifecycle actions, evidence, and approvals."}, "ComponentSpec": ["LifecycleStatusPanel", "EvidenceTimeline", "ApprovalQueue", "AgentHealthCard"], "InteractionSpec": ["Filter lifecycle items", "Open evidence detail", "Approve gated action"], "DesignTokens": {"semantic": ["surface", "accent", "danger", "warning", "success"]}, "ResponsiveRules": ["Single-column mobile", "Two-column tablet", "Dashboard grid desktop"], "AccessibilityRequirements": ["Keyboard reachable actions", "Visible focus", "WCAG AA contrast", "ARIA labels for status badges"], "FrontendImplementationPlan": {"framework_neutral": True, "django_compatible": True, "preferred_for_django": ["Django templates", "HTMX", "Alpine", "project CSS/Tailwind if present"], "react_required": False}}
+        prompt = "Produce a stack-neutral UI specification for the future Artifex Project Workspace. Do not force React; support Django/HTMX."
+        if self.router:
+            try:
+                response = self.router.complete(ModelRequestContract(purpose=ModelCapability.PLANNING, model_hint=version.model, agent_version=version, project=project, prompt=prompt))
+                parsed = json.loads(response.content)
+                if isinstance(parsed, dict):
+                    contract = parsed
+            except Exception:
+                pass
+        return Artifact.objects.create(project=project, artifact_type="FRONTEND_DESIGN_SPEC", name="Artifex Project Workspace UI Spec", content=contract, generated_by=f"{version.agent.name} v{version.version}")
+
+    def list_teams(self) -> list[dict[str, object]]:
+        return [self._team(team) for team in AgentTeam.objects.order_by("name")]
+
+    def get_team(self, team_id) -> dict[str, object]:
+        return self._team(AgentTeam.objects.get(id=team_id))
+
+    def _agent(self, agent: Agent, *, include_versions: bool = False) -> dict[str, object]:
+        data = {"id": str(agent.id), "name": agent.name, "role": agent.role, "purpose": agent.purpose, "scope": agent.scope, "status": agent.status, "champion_version": str(agent.champion_version_id) if agent.champion_version_id else None, "competencies": list(agent.competencies.select_related("competency").values("competency__key", "proficiency", "confidence"))}
+        if include_versions:
+            data["versions"] = [self.get_agent_version(version.id) for version in agent.versions.order_by("version")]
+        return data
+
+    def _team(self, team: AgentTeam) -> dict[str, object]:
+        return {"id": str(team.id), "name": team.name, "purpose": team.purpose, "scope": team.scope, "status": team.status, "members": list(team.members.select_related("agent_version__agent").values("role", "responsibilities", "agent_version_id", "agent_version__agent__name", "agent_version__version", "status")), "metadata": team.metadata}
+
+    def _provider_health(self, version: AgentVersion) -> str:
+        resource = None
+        for candidate in Resource.objects.filter(is_active=True):
+            if version.model in candidate.roles or version.model.lower() in candidate.provider.lower():
+                resource = candidate
+                break
+        return resource.health_status if resource else "UNKNOWN"

artifex/urls.py

@@ -3,9 +3,10 @@ from __future__ import annotations
 from django.contrib import admin
 from django.urls import path
 
-from control_plane.projects.views import dashboard
+from control_plane.projects.views import agent_control_room, dashboard
 
 urlpatterns = [
     path("", dashboard, name="dashboard"),
+    path("agents/", agent_control_room, name="agent_control_room"),
     path("admin/", admin.site.urls),
 ]

control_plane/agents/admin.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from django.contrib import admin
 
-from control_plane.agents.models import Agent, AgentPlan, AgentRun, AgentVersion, BenchmarkRun
+from control_plane.agents.models import Agent, AgentCompetency, AgentPlan, AgentRun, AgentTeam, AgentTeamMember, AgentVersion, BenchmarkRun, Competency
 
 
 admin.site.register(Agent)
@@ -10,3 +10,7 @@ admin.site.register(AgentVersion)
 admin.site.register(AgentPlan)
 admin.site.register(AgentRun)
 admin.site.register(BenchmarkRun)
+admin.site.register(Competency)
+admin.site.register(AgentCompetency)
+admin.site.register(AgentTeam)
+admin.site.register(AgentTeamMember)

control_plane/agents/migrations/0008_agent_control_room_v1.py

@@ -0,0 +1,99 @@
+import uuid
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("agents", "0007_replay_arena"),
+        ("graph", "0004_unique_champion_graph_version"),
+        ("projects", "0006_roadmap_scenario_lab_v1"),
+    ]
+
+    operations = [
+        migrations.AlterField(model_name="agent", name="role", field=models.CharField(choices=[("PROJECT_ARCHAEOLOGIST", "Project Archaeologist"), ("PLANNER", "Planner"), ("CODER", "Coder"), ("REVIEWER", "Reviewer"), ("PROJECT_JUDGE", "Project Judge"), ("FRONTEND_DESIGNER", "Frontend Designer"), ("FRONTEND_ENGINEER", "Frontend Engineer"), ("UX_ACCESSIBILITY_REVIEWER", "Ux Accessibility Reviewer"), ("VISUAL_JUDGE", "Visual Judge")], max_length=80)),
+        migrations.AlterField(model_name="agentplan", name="role", field=models.CharField(choices=[("PROJECT_ARCHAEOLOGIST", "Project Archaeologist"), ("PLANNER", "Planner"), ("CODER", "Coder"), ("REVIEWER", "Reviewer"), ("PROJECT_JUDGE", "Project Judge"), ("FRONTEND_DESIGNER", "Frontend Designer"), ("FRONTEND_ENGINEER", "Frontend Engineer"), ("UX_ACCESSIBILITY_REVIEWER", "Ux Accessibility Reviewer"), ("VISUAL_JUDGE", "Visual Judge")], max_length=80)),
+        migrations.AddField(model_name="agent", name="metadata", field=models.JSONField(blank=True, default=dict)),
+        migrations.AddField(model_name="agent", name="project", field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="scoped_agents", to="projects.project")),
+        migrations.AddField(model_name="agent", name="purpose", field=models.TextField(blank=True)),
+        migrations.AddField(model_name="agent", name="scope", field=models.CharField(choices=[("GLOBAL", "Global"), ("STUDIO", "Studio"), ("PROJECT", "Project")], default="GLOBAL", max_length=32)),
+        migrations.AddField(model_name="agent", name="status", field=models.CharField(default="ACTIVE", max_length=32)),
+        migrations.AddField(model_name="agent", name="studio", field=models.CharField(blank=True, max_length=120)),
+        migrations.AddField(model_name="agentversion", name="ancestry", field=models.JSONField(blank=True, default=list)),
+        migrations.AddField(model_name="agentversion", name="creation_source", field=models.CharField(default="seed", max_length=80)),
+        migrations.AddField(model_name="agentversion", name="graph_usage_policy", field=models.JSONField(blank=True, default=dict)),
+        migrations.AddField(model_name="agentversion", name="immutable_since", field=models.DateTimeField(blank=True, null=True)),
+        migrations.AddField(model_name="agentversion", name="metadata", field=models.JSONField(blank=True, default=dict)),
+        migrations.AddField(model_name="agentversion", name="model_config", field=models.JSONField(blank=True, default=dict)),
+        migrations.AddField(model_name="agentversion", name="resource_preferences", field=models.JSONField(blank=True, default=dict)),
+        migrations.AddField(model_name="agentversion", name="retrieval_policy", field=models.JSONField(blank=True, default=dict)),
+        migrations.AddField(model_name="agentversion", name="scope", field=models.CharField(choices=[("GLOBAL", "Global"), ("STUDIO", "Studio"), ("PROJECT", "Project")], default="GLOBAL", max_length=32)),
+        migrations.AddField(model_name="agentversion", name="status", field=models.CharField(default="ACTIVE", max_length=32)),
+        migrations.AddField(model_name="agentversion", name="tool_policy", field=models.JSONField(blank=True, default=dict)),
+        migrations.AddField(model_name="agentversion", name="parent_version", field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="children", to="agents.agentversion")),
+        migrations.AlterField(model_name="agentversion", name="promotion_status", field=models.CharField(choices=[("DRAFT", "Draft"), ("CANDIDATE", "Candidate"), ("CHALLENGER", "Challenger"), ("CHAMPION", "Champion"), ("REJECTED", "Rejected"), ("RETIRED", "Retired"), ("DISABLED", "Disabled")], default="CANDIDATE", max_length=32)),
+        migrations.AddField(model_name="agentplan", name="evidence", field=models.JSONField(blank=True, default=dict)),
+        migrations.AddField(model_name="agentplan", name="plan_type", field=models.CharField(default="CREATE", max_length=32)),
+        migrations.AddField(model_name="agentplan", name="scope", field=models.CharField(choices=[("GLOBAL", "Global"), ("STUDIO", "Studio"), ("PROJECT", "Project")], default="GLOBAL", max_length=32)),
+        migrations.AddField(model_name="agentplan", name="status", field=models.CharField(default="DRAFT", max_length=32)),
+        migrations.AddField(model_name="agentplan", name="target_agent_version", field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="plans_targeting_version", to="agents.agentversion")),
+        migrations.CreateModel(
+            name="Competency",
+            fields=[
+                ("id", models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+                ("key", models.CharField(max_length=120, unique=True)),
+                ("name", models.CharField(max_length=200)),
+                ("description", models.TextField(blank=True)),
+                ("domain", models.CharField(blank=True, max_length=120)),
+                ("metadata", models.JSONField(blank=True, default=dict)),
+            ],
+        ),
+        migrations.CreateModel(
+            name="AgentTeam",
+            fields=[
+                ("id", models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+                ("name", models.CharField(max_length=200)),
+                ("purpose", models.TextField(blank=True)),
+                ("scope", models.CharField(choices=[("GLOBAL", "Global"), ("STUDIO", "Studio"), ("PROJECT", "Project")], default="GLOBAL", max_length=32)),
+                ("status", models.CharField(default="ACTIVE", max_length=32)),
+                ("metadata", models.JSONField(blank=True, default=dict)),
+                ("execution_graph_version", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="agent_teams", to="graph.executiongraphversion")),
+            ],
+        ),
+        migrations.CreateModel(
+            name="AgentCompetency",
+            fields=[
+                ("id", models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+                ("proficiency", models.FloatField(default=0.5)),
+                ("confidence", models.FloatField(default=0.5)),
+                ("evidence", models.JSONField(blank=True, default=dict)),
+                ("source", models.CharField(default="manual", max_length=80)),
+                ("last_evaluated_at", models.DateTimeField(blank=True, null=True)),
+                ("agent", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name="competencies", to="agents.agent")),
+                ("agent_version", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name="competencies", to="agents.agentversion")),
+                ("competency", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="agent_assignments", to="agents.competency")),
+            ],
+        ),
+        migrations.CreateModel(
+            name="AgentTeamMember",
+            fields=[
+                ("id", models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+                ("role", models.CharField(max_length=120)),
+                ("responsibilities", models.JSONField(blank=True, default=list)),
+                ("status", models.CharField(default="ACTIVE", max_length=32)),
+                ("metadata", models.JSONField(blank=True, default=dict)),
+                ("agent_version", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="team_memberships", to="agents.agentversion")),
+                ("team", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="members", to="agents.agentteam")),
+            ],
+            options={"constraints": [models.UniqueConstraint(fields=("team", "agent_version", "role"), name="unique_agent_team_member_role")]},
+        ),
+    ]

control_plane/agents/models.py

@@ -13,19 +13,38 @@ class AgentRole(models.TextChoices):
     CODER = "CODER"
     REVIEWER = "REVIEWER"
     PROJECT_JUDGE = "PROJECT_JUDGE"
+    FRONTEND_DESIGNER = "FRONTEND_DESIGNER"
+    FRONTEND_ENGINEER = "FRONTEND_ENGINEER"
+    UX_ACCESSIBILITY_REVIEWER = "UX_ACCESSIBILITY_REVIEWER"
+    VISUAL_JUDGE = "VISUAL_JUDGE"
 
 
 class PromotionStatus(models.TextChoices):
+    DRAFT = "DRAFT"
     CANDIDATE = "CANDIDATE"
     CHALLENGER = "CHALLENGER"
     CHAMPION = "CHAMPION"
     REJECTED = "REJECTED"
+    RETIRED = "RETIRED"
+    DISABLED = "DISABLED"
+
+
+class AgentScope(models.TextChoices):
+    GLOBAL = "GLOBAL"
+    STUDIO = "STUDIO"
+    PROJECT = "PROJECT"
 
 
 class Agent(TimestampedModel):
     name = models.CharField(max_length=200, unique=True)
     role = models.CharField(max_length=80, choices=AgentRole.choices)
     description = models.TextField(blank=True)
+    purpose = models.TextField(blank=True)
+    scope = models.CharField(max_length=32, choices=AgentScope.choices, default=AgentScope.GLOBAL)
+    studio = models.CharField(max_length=120, blank=True)
+    project = models.ForeignKey("projects.Project", on_delete=models.SET_NULL, null=True, blank=True, related_name="scoped_agents")
+    status = models.CharField(max_length=32, default="ACTIVE")
+    metadata = models.JSONField(default=dict, blank=True)
     champion_version = models.ForeignKey(
         "AgentVersion", on_delete=models.SET_NULL, null=True, blank=True, related_name="championed_by"
     )
@@ -35,20 +54,39 @@ class AgentVersion(TimestampedModel):
     agent = models.ForeignKey(Agent, on_delete=models.CASCADE, related_name="versions")
     version = models.PositiveIntegerField()
     model = models.CharField(max_length=120)
+    model_config = models.JSONField(default=dict, blank=True)
     system_contract = models.TextField()
     capabilities = models.JSONField(default=list, blank=True)
     tools = models.JSONField(default=list, blank=True)
     permissions = models.JSONField(default=dict, blank=True)
     context_policy = models.JSONField(default=dict, blank=True)
+    retrieval_policy = models.JSONField(default=dict, blank=True)
+    tool_policy = models.JSONField(default=dict, blank=True)
     workflow = models.JSONField(default=dict, blank=True)
     retry_policy = models.JSONField(default=dict, blank=True)
+    resource_preferences = models.JSONField(default=dict, blank=True)
+    graph_usage_policy = models.JSONField(default=dict, blank=True)
+    scope = models.CharField(max_length=32, choices=AgentScope.choices, default=AgentScope.GLOBAL)
+    parent_version = models.ForeignKey("self", on_delete=models.SET_NULL, null=True, blank=True, related_name="children")
+    ancestry = models.JSONField(default=list, blank=True)
+    creation_source = models.CharField(max_length=80, default="seed")
     evaluator = models.JSONField(default=dict, blank=True)
     benchmark_status = models.CharField(max_length=32, default="UNBENCHMARKED")
     promotion_status = models.CharField(max_length=32, choices=PromotionStatus.choices, default=PromotionStatus.CANDIDATE)
+    status = models.CharField(max_length=32, default="ACTIVE")
+    immutable_since = models.DateTimeField(null=True, blank=True)
+    metadata = models.JSONField(default=dict, blank=True)
 
     class Meta:
         constraints = [models.UniqueConstraint(fields=["agent", "version"], name="unique_agent_version")]
 
+    def save(self, *args: object, **kwargs: object) -> None:
+        if self.promotion_status == PromotionStatus.CHAMPION:
+            existing = AgentVersion.objects.filter(agent=self.agent, promotion_status=PromotionStatus.CHAMPION).exclude(pk=self.pk)
+            if existing.exists():
+                raise ValidationError("Only one champion AgentVersion is allowed per Agent.")
+        super().save(*args, **kwargs)
+
 
 class AgentPlan(TimestampedModel):
     agent = models.ForeignKey(Agent, on_delete=models.CASCADE, related_name="plans", null=True, blank=True)
@@ -63,9 +101,54 @@ class AgentPlan(TimestampedModel):
     workflow = models.JSONField(default=dict, blank=True)
     benchmarks = models.JSONField(default=list, blank=True)
     success_criteria = models.JSONField(default=dict, blank=True)
+    target_agent_version = models.ForeignKey(AgentVersion, on_delete=models.SET_NULL, null=True, blank=True, related_name="plans_targeting_version")
+    plan_type = models.CharField(max_length=32, default="CREATE")
+    scope = models.CharField(max_length=32, choices=AgentScope.choices, default=AgentScope.GLOBAL)
+    evidence = models.JSONField(default=dict, blank=True)
+    status = models.CharField(max_length=32, default="DRAFT")
     created_by = models.CharField(max_length=120, default="sol")
 
 
+class Competency(TimestampedModel):
+    key = models.CharField(max_length=120, unique=True)
+    name = models.CharField(max_length=200)
+    description = models.TextField(blank=True)
+    domain = models.CharField(max_length=120, blank=True)
+    metadata = models.JSONField(default=dict, blank=True)
+
+
+class AgentCompetency(TimestampedModel):
+    agent = models.ForeignKey(Agent, on_delete=models.CASCADE, related_name="competencies", null=True, blank=True)
+    agent_version = models.ForeignKey(AgentVersion, on_delete=models.CASCADE, related_name="competencies", null=True, blank=True)
+    competency = models.ForeignKey(Competency, on_delete=models.CASCADE, related_name="agent_assignments")
+    proficiency = models.FloatField(default=0.5)
+    confidence = models.FloatField(default=0.5)
+    evidence = models.JSONField(default=dict, blank=True)
+    source = models.CharField(max_length=80, default="manual")
+    last_evaluated_at = models.DateTimeField(null=True, blank=True)
+
+
+class AgentTeam(TimestampedModel):
+    name = models.CharField(max_length=200)
+    purpose = models.TextField(blank=True)
+    scope = models.CharField(max_length=32, choices=AgentScope.choices, default=AgentScope.GLOBAL)
+    status = models.CharField(max_length=32, default="ACTIVE")
+    execution_graph_version = models.ForeignKey("graph.ExecutionGraphVersion", on_delete=models.SET_NULL, null=True, blank=True, related_name="agent_teams")
+    metadata = models.JSONField(default=dict, blank=True)
+
+
+class AgentTeamMember(TimestampedModel):
+    team = models.ForeignKey(AgentTeam, on_delete=models.CASCADE, related_name="members")
+    agent_version = models.ForeignKey(AgentVersion, on_delete=models.CASCADE, related_name="team_memberships")
+    role = models.CharField(max_length=120)
+    responsibilities = models.JSONField(default=list, blank=True)
+    status = models.CharField(max_length=32, default="ACTIVE")
+    metadata = models.JSONField(default=dict, blank=True)
+
+    class Meta:
+        constraints = [models.UniqueConstraint(fields=["team", "agent_version", "role"], name="unique_agent_team_member_role")]
+
+
 class AgentRun(TimestampedModel):
     agent_version = models.ForeignKey(AgentVersion, on_delete=models.PROTECT, related_name="runs")
     project = models.ForeignKey("projects.Project", on_delete=models.CASCADE, null=True, blank=True)

control_plane/projects/views.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from django.shortcuts import render
 
+from agents.control_room import AgentControlRoomService
 from control_plane.events.models import Event
 from control_plane.projects.models import Project, TaskStatus
 
@@ -15,3 +16,12 @@ def dashboard(request):
         "blocked_tasks": sum(project.tasks.filter(status=TaskStatus.BLOCKED).count() for project in projects),
     }
     return render(request, "projects/dashboard.html", {"projects": projects, "summary": summary, "recent_events": recent_events})
+
+
+def agent_control_room(request):
+    service = AgentControlRoomService()
+    agents = service.list_agents()
+    for agent in agents:
+        champion_id = agent.get("champion_version")
+        agent["health"] = service.get_agent_health(champion_id) if champion_id else {"status": "WATCH", "reasons": ["No champion version."]}
+    return render(request, "projects/agent_control_room.html", {"agents": agents, "teams": service.list_teams()})

graph/agent_control.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+from agents.control_room import AgentControlRoomService
+from control_plane.agents.models import AgentVersion
+from graph.native_runtime import GraphExecutionContext
+from graph.registry import NodeHandlerRegistry, NodeResult
+from graph.spec import ExecutionGraphSpec, GraphEdgeSpec, GraphNodeSpec
+
+
+def agent_investigation_graph_v1() -> ExecutionGraphSpec:
+    nodes = ["prepare", "gather_agent_evidence", "smart_investigation", "improvement_candidates", "complete"]
+    spec = ExecutionGraphSpec(
+        name="agent_investigation",
+        version=1,
+        graph_type="AGENT_INVESTIGATION",
+        entry="prepare",
+        nodes={node: GraphNodeSpec(node, node if node == "complete" else f"agent_{node}") for node in nodes},
+        edges=[GraphEdgeSpec(nodes[index], nodes[index + 1], "success") for index in range(len(nodes) - 1)],
+        terminal_nodes=["complete"],
+        metadata={"description": "Agent investigation workflow using Control Room telemetry and Progeny Smart Investigation."},
+    )
+    spec.validate()
+    return spec
+
+
+class AgentInvestigationNode:
+    idempotent = True
+    replay_safe = True
+    destructive = False
+
+    def __init__(self, service: AgentControlRoomService, node_type: str) -> None:
+        self.service = service
+        self.node_type = node_type
+
+    def version(self, context: GraphExecutionContext) -> AgentVersion:
+        return AgentVersion.objects.get(id=context.graph_run.metadata["agent_version_id"])
+
+
+class AgentSimpleNode(AgentInvestigationNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        return NodeResult("COMPLETE", "success")
+
+
+class AgentGatherEvidenceNode(AgentInvestigationNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        version = self.version(context)
+        evidence = {"performance": self.service.get_agent_performance(version.id), "health": self.service.get_agent_health(version.id), "usage": self.service.get_agent_usage(version.id)}
+        metadata = dict(context.graph_run.metadata)
+        metadata["agent_evidence"] = evidence
+        context.graph_run.metadata = metadata
+        context.graph_run.save(update_fields=["metadata", "updated_at"])
+        return NodeResult("COMPLETE", "success", evidence)
+
+
+class AgentSmartInvestigationNode(AgentInvestigationNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        investigation = self.service.investigate_agent(self.version(context))
+        metadata = dict(context.graph_run.metadata)
+        metadata["progeny_investigation_id"] = str(investigation.id)
+        context.graph_run.metadata = metadata
+        context.graph_run.save(update_fields=["metadata", "updated_at"])
+        return NodeResult("COMPLETE", "success", {"investigation_id": str(investigation.id)})
+
+
+class AgentImprovementCandidateNode(AgentInvestigationNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        from control_plane.agents.models import ProgenyInvestigation
+
+        investigation_id = context.graph_run.metadata.get("progeny_investigation_id")
+        investigation = ProgenyInvestigation.objects.get(id=investigation_id)
+        candidate = self.service.create_improvement_candidate(investigation)
+        return NodeResult("COMPLETE", "success", {"improvement_candidate_id": str(candidate.id)})
+
+
+def agent_investigation_registry(service: AgentControlRoomService) -> NodeHandlerRegistry:
+    registry = NodeHandlerRegistry()
+    for handler in [AgentSimpleNode(service, "agent_prepare"), AgentGatherEvidenceNode(service, "agent_gather_agent_evidence"), AgentSmartInvestigationNode(service, "agent_smart_investigation"), AgentImprovementCandidateNode(service, "agent_improvement_candidates")]:
+        registry.register(handler)
+    return registry

graph/bootstrap.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from django.utils import timezone
 
+from graph.agent_control import agent_investigation_graph_v1
 from graph.lifecycle import project_evolution_graph_v1, project_exploration_graph_v1, project_extension_graph_v1
 from graph.models import ExecutionGraphDefinition, ExecutionGraphVersion, ExecutionGraphVersionStatus
 from graph.roadmap import project_roadmap_review_graph_v1
@@ -91,3 +92,7 @@ def champion_project_roadmap_review_graph_v1() -> ExecutionGraphVersion:
 
 def champion_scenario_lab_graph_v1() -> ExecutionGraphVersion:
     return _champion_graph(scenario_lab_graph_v1())
+
+
+def champion_agent_investigation_graph_v1() -> ExecutionGraphVersion:
+    return _champion_graph(agent_investigation_graph_v1())

templates/projects/agent_control_room.html

@@ -0,0 +1,53 @@
+<!doctype html>
+<html lang="en">
+  <head>
+    <meta charset="utf-8">
+    <title>Agent Control Room</title>
+    <style>
+      body { font-family: system-ui, sans-serif; margin: 2rem; color: #172033; background: #f7f4ed; }
+      header { display: flex; justify-content: space-between; align-items: center; }
+      .card { background: white; border: 1px solid #ddd4c6; border-radius: 12px; padding: 1rem; margin: 1rem 0; }
+      .grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(260px, 1fr)); gap: 1rem; }
+      .status { font-size: .8rem; letter-spacing: .08em; color: #6f4e37; text-transform: uppercase; }
+      code { background: #f0ece4; padding: .15rem .3rem; border-radius: 4px; }
+    </style>
+  </head>
+  <body>
+    <header>
+      <div>
+        <h1>Agent Control Room</h1>
+        <p>Inspectable workforce inventory, champion health, challengers, and teams.</p>
+      </div>
+      <a href="/">Dashboard</a>
+    </header>
+    <section>
+      <h2>Agents</h2>
+      <div class="grid">
+        {% for agent in agents %}
+          <article class="card">
+            <div class="status">{{ agent.role }} / {{ agent.scope }} / {{ agent.health.status }}</div>
+            <h3>{{ agent.name }}</h3>
+            <p>{{ agent.purpose|default:agent.id }}</p>
+            <p>Champion: <code>{{ agent.champion_version|default:"none" }}</code></p>
+            <p>{{ agent.health.reasons|join:", " }}</p>
+          </article>
+        {% empty %}
+          <article class="card">No agents registered.</article>
+        {% endfor %}
+      </div>
+    </section>
+    <section>
+      <h2>Teams</h2>
+      {% for team in teams %}
+        <article class="card">
+          <div class="status">{{ team.scope }} / {{ team.status }}</div>
+          <h3>{{ team.name }}</h3>
+          <p>{{ team.purpose }}</p>
+          <p>{{ team.members|length }} members</p>
+        </article>
+      {% empty %}
+        <article class="card">No teams yet.</article>
+      {% endfor %}
+    </section>
+  </body>
+</html>

tests/test_agent_control_room_v1.py

@@ -0,0 +1,133 @@
+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"