Add Extend Evolve Explore lifecycle V1

agents/lifecycle.py

@@ -0,0 +1,543 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import subprocess
+from pathlib import Path
+from typing import Any
+
+from django.core.exceptions import ValidationError
+from django.db import transaction
+from django.utils import timezone
+
+from control_plane.agents.models import ProgenySignal
+from control_plane.events.bus import EventBus
+from control_plane.projects.models import (
+    CommitRecord,
+    Decision,
+    ExtensionCandidate,
+    ExtensionPlan,
+    Exploration,
+    ExplorationOpportunity,
+    EvolutionCandidate,
+    EvolutionPlan,
+    Feature,
+    Finding,
+    Milestone,
+    Project,
+    ProjectPlan,
+    RoadmapItem,
+    StewardFinding,
+    Task,
+    TaskDependency,
+    TaskStatus,
+)
+from control_plane.verification.models import Review, TestRun, Verification, VerificationLevel, VerificationResult
+from graph.langgraph_runtime import LangGraphRuntime
+from graph.models import GraphRun
+from graph.task_nodes import TaskExecutionServices, task_execution_registry
+from model_router.router import ModelCapability, ModelRequestContract, ModelRouter
+from project_brain.planning import ProjectPlanContract, parse_project_plan_response
+
+
+class LifecyclePlanningError(ValueError):
+    pass
+
+
+class ProjectContextMixin:
+    def project_context(self, project: Project) -> dict[str, object]:
+        files: dict[str, str] = {}
+        tests: dict[str, str] = {}
+        if project.repository_path:
+            root = Path(project.repository_path)
+            for path in sorted(root.rglob("*.py"))[:40]:
+                if any(part in path.parts for part in [".git", "__pycache__", "migrations", ".venv", "venv", "node_modules", ".pytest_cache"]):
+                    continue
+                relative = path.relative_to(root).as_posix()
+                content = path.read_text(encoding="utf-8", errors="ignore")[:3000]
+                if relative.startswith("tests") or "test" in path.name:
+                    tests[relative] = content
+                else:
+                    files[relative] = content
+        return {
+            "project": {"id": str(project.id), "name": project.name, "goal": project.goal, "architecture_summary": project.architecture_summary},
+            "decisions": list(project.decisions.order_by("-created_at").values("decision_type", "decision", "reason")[:20]),
+            "roadmap_items": list(project.roadmap_items.order_by("-created_at").values("title", "description", "status", "source")[:20]),
+            "findings": list(project.findings.order_by("-created_at").values("finding_type", "severity", "title", "status")[:20]),
+            "steward_findings": list(project.steward_findings.order_by("-created_at").values("finding_type", "severity", "title", "status", "recommended_action")[:20]),
+            "features": list(project.features.order_by("created_at").values("title", "description", "status", "acceptance_criteria")[:50]),
+            "tasks": list(project.tasks.order_by("-created_at").values("task_type", "status", "goal", "acceptance_criteria")[:50]),
+            "files": files,
+            "tests": tests,
+        }
+
+    def _json_from_sol(self, router: ModelRouter | None, prompt: str, *, fallback: dict[str, object], project: Project | None = None) -> dict[str, object]:
+        if router is None:
+            return fallback
+        response = router.complete(ModelRequestContract(purpose=ModelCapability.PLANNING, prompt=prompt, model_hint="sol", project=project))
+        try:
+            payload = json.loads(response.content)
+        except json.JSONDecodeError as exc:
+            raise LifecyclePlanningError("Sol lifecycle planning response must be JSON") from exc
+        if not isinstance(payload, dict):
+            raise LifecyclePlanningError("Sol lifecycle planning response must be a JSON object")
+        return payload
+
+    def _json_from_project_brain(self, router: ModelRouter | None, prompt: str, *, fallback: dict[str, object], project: Project, category: str) -> dict[str, object]:
+        try:
+            return self._json_from_sol(router, prompt, fallback=fallback, project=project)
+        except Exception as exc:
+            create_project_planning_signal(project, f"Sol {category} planning response invalid; using bounded fallback plan.", {"error": str(exc), "category": category})
+            return fallback
+
+
+class ExtensionService(ProjectContextMixin):
+    def __init__(self, router: ModelRouter | None = None, bus: EventBus | None = None) -> None:
+        self.router = router
+        self.bus = bus or EventBus()
+
+    def create_candidate(
+        self,
+        project: Project,
+        *,
+        title: str,
+        description: str,
+        rationale: str = "",
+        source: str = "user",
+        expected_value: str = "",
+        affected_areas: list[str] | None = None,
+        estimated_complexity: str = "MEDIUM",
+        risk: str = "MEDIUM",
+        confidence: float = 0.5,
+        evidence: dict[str, object] | None = None,
+        source_steward_finding: StewardFinding | None = None,
+        source_opportunity: ExplorationOpportunity | None = None,
+    ) -> ExtensionCandidate:
+        return ExtensionCandidate.objects.create(
+            project=project,
+            title=title,
+            description=description,
+            rationale=rationale,
+            source=source,
+            expected_value=expected_value,
+            affected_areas=affected_areas or [],
+            estimated_complexity=estimated_complexity,
+            risk=risk,
+            confidence=confidence,
+            evidence=evidence or {},
+            source_steward_finding=source_steward_finding,
+            source_opportunity=source_opportunity,
+        )
+
+    def plan_with_project_brain(self, candidate: ExtensionCandidate) -> ExtensionPlan:
+        context = self.project_context(candidate.project)
+        fallback = self._fallback_extension_plan(candidate, context)
+        payload = self._json_from_project_brain(
+            self.router,
+            "Plan an EXTENSION for an existing project. Return JSON with extension_plan and project_plan. Do not treat this as greenfield.\n"
+            + json.dumps({"candidate": self._candidate_payload(candidate), "context": context}, default=str),
+            fallback=fallback,
+            project=candidate.project,
+            category="extension",
+        )
+        raw_plan = payload.get("extension_plan", payload)
+        if not isinstance(raw_plan, dict):
+            raise LifecyclePlanningError("extension_plan must be an object")
+        project_plan_payload = raw_plan.get("project_plan") or payload.get("project_plan")
+        if not isinstance(project_plan_payload, dict):
+            create_project_planning_signal(candidate.project, "Sol extension plan omitted project_plan; using bounded fallback plan.", {"candidate_id": str(candidate.id)})
+            raw_plan = dict(fallback["extension_plan"])
+            project_plan_payload = dict(raw_plan["project_plan"])
+        try:
+            parse_project_plan_response(json.dumps(project_plan_payload))
+        except Exception as exc:
+            create_project_planning_signal(candidate.project, "Sol extension project_plan failed validation; using bounded fallback plan.", {"candidate_id": str(candidate.id), "error": str(exc)})
+            raw_plan = dict(fallback["extension_plan"])
+            project_plan_payload = dict(raw_plan["project_plan"])
+            parse_project_plan_response(json.dumps(project_plan_payload))
+        plan = ExtensionPlan.objects.create(
+            candidate=candidate,
+            project=candidate.project,
+            status="PLANNED",
+            strategy=str(raw_plan.get("strategy", candidate.description)),
+            plan=raw_plan,
+            acceptance_criteria=[str(item) for item in raw_plan.get("acceptance_criteria", project_plan_payload.get("acceptance_criteria", []))],
+            context_snapshot=context,
+        )
+        candidate.status = "PLANNING"
+        candidate.save(update_fields=["status", "updated_at"])
+        self.bus.publish("EXTENSION_PLAN_CREATED", project=candidate.project, payload={"candidate_id": str(candidate.id), "plan_id": str(plan.id)})
+        return plan
+
+    def approve_plan(self, plan: ExtensionPlan) -> ExtensionPlan:
+        plan.status = "APPROVED"
+        plan.approved_at = timezone.now()
+        plan.save(update_fields=["status", "approved_at", "updated_at"])
+        plan.candidate.status = "READY"
+        plan.candidate.save(update_fields=["status", "updated_at"])
+        return plan
+
+    def materialize_project_dag(self, plan: ExtensionPlan) -> ProjectPlan:
+        if plan.project_plan_id:
+            return plan.project_plan
+        contract = parse_project_plan_response(json.dumps(plan.plan.get("project_plan", {})))
+        project_plan = self._materialize_contract(plan.project, contract, prefix=f"EXT-{str(plan.id)[:8]}")
+        plan.project_plan = project_plan
+        plan.status = "MATERIALIZED"
+        plan.save(update_fields=["project_plan", "status", "updated_at"])
+        plan.candidate.status = "BUILDING"
+        plan.candidate.save(update_fields=["status", "updated_at"])
+        return project_plan
+
+    def execute(self, plan: ExtensionPlan, router: ModelRouter, *, test_command: list[str] | None = None) -> list[GraphRun]:
+        from graph.bootstrap import champion_task_execution_graph_v1
+
+        project_plan = self.materialize_project_dag(plan)
+        runs: list[GraphRun] = []
+        for task in Task.objects.filter(milestone__plan=project_plan).exclude(status=TaskStatus.COMPLETE).order_by("priority", "created_at"):
+            graph_version = champion_task_execution_graph_v1()
+            graph_run = GraphRun.objects.create(execution_graph_version=graph_version, project=task.project, milestone=task.milestone, feature=task.feature, task=task, current_node=graph_version.graph_spec["entry"], metadata={"extension_plan_id": str(plan.id), "extension_candidate_id": str(plan.candidate_id)})
+            LangGraphRuntime(task_execution_registry(TaskExecutionServices(router, bus=self.bus, test_command=test_command or ["python", "-m", "pytest"])), bus=self.bus).run_until_terminal_or_paused(graph_run)
+            for commit in task.commits.all():
+                commit.extension_candidate = plan.candidate
+                commit.save(update_fields=["extension_candidate", "updated_at"])
+            runs.append(graph_run)
+        return runs
+
+    def verify_extension(self, plan: ExtensionPlan) -> Verification:
+        project_plan = plan.project_plan
+        tasks = Task.objects.filter(milestone__plan=project_plan) if project_plan else Task.objects.none()
+        task_count = tasks.count()
+        completed = task_count > 0 and not tasks.exclude(status=TaskStatus.COMPLETE).exists()
+        tests_pass = not TestRun.objects.filter(task__in=tasks).exclude(status="PASS").exists()
+        review_failures = Review.objects.filter(task__in=tasks).exclude(status="PASS")
+        judge_failures = Verification.objects.filter(task__in=tasks, level=VerificationLevel.TASK).exclude(result=VerificationResult.PASS)
+        criteria = plan.acceptance_criteria or plan.plan.get("acceptance_criteria", [])
+        criteria_present = bool(criteria)
+        passed = completed and tests_pass and not review_failures.exists() and not judge_failures.exists() and criteria_present
+        verification = Verification.objects.create(
+            project=plan.project,
+            milestone=tasks.first().milestone if tasks.exists() else None,
+            level=VerificationLevel.MILESTONE,
+            result=VerificationResult.PASS if passed else VerificationResult.FAIL,
+            contract={"extension_plan_id": str(plan.id), "acceptance_criteria": criteria},
+            evidence=[{"task_count": task_count, "all_tasks_complete": completed, "tests_pass": tests_pass, "review_failures": review_failures.count(), "judge_failures": judge_failures.count()}],
+            summary="Extension acceptance contract satisfied" if passed else "Extension acceptance contract failed",
+        )
+        plan.verification = verification
+        plan.status = "COMPLETE" if passed else "FAILED"
+        plan.completed_at = timezone.now()
+        plan.save(update_fields=["verification", "status", "completed_at", "updated_at"])
+        plan.candidate.status = "COMPLETE" if passed else "BUILDING"
+        plan.candidate.save(update_fields=["status", "updated_at"])
+        return verification
+
+    def _fallback_extension_plan(self, candidate: ExtensionCandidate, context: dict[str, object]) -> dict[str, object]:
+        task_goal = candidate.metadata.get("task_goal") or candidate.description or candidate.title
+        acceptance = candidate.metadata.get("acceptance_criteria") or [f"{candidate.title} capability is present", "Deterministic tests pass"]
+        return {
+            "extension_plan": {
+                "strategy": f"Add bounded scope to existing project: {candidate.title}",
+                "acceptance_criteria": acceptance,
+                "project_plan": {
+                    "goal": candidate.project.goal,
+                    "scope": candidate.description,
+                    "acceptance_criteria": acceptance,
+                    "milestones": [
+                        {
+                            "key": "EXTEND",
+                            "title": candidate.title,
+                            "goal": candidate.description or candidate.title,
+                            "verification_contract": {"extension_candidate_id": str(candidate.id)},
+                            "features": [{"key": "F1", "title": candidate.title, "description": candidate.description, "acceptance_criteria": acceptance, "tasks": [{"id": "T1", "goal": str(task_goal), "type": "implementation", "acceptance_criteria": acceptance, "priority": 50, "dependencies": []}]}],
+                        }
+                    ],
+                },
+            }
+        }
+
+    def _candidate_payload(self, candidate: ExtensionCandidate) -> dict[str, object]:
+        return {"id": str(candidate.id), "title": candidate.title, "description": candidate.description, "rationale": candidate.rationale, "expected_value": candidate.expected_value, "affected_areas": candidate.affected_areas, "evidence": candidate.evidence}
+
+    def _materialize_contract(self, project: Project, contract: ProjectPlanContract, *, prefix: str) -> ProjectPlan:
+        with transaction.atomic():
+            version = project.current_plan_version + 1
+            project_plan = ProjectPlan.objects.create(project=project, version=version, goal=project.goal, scope=contract.scope, stack=contract.stack, architecture=contract.architecture, constraints=contract.constraints, acceptance_criteria=contract.acceptance_criteria, permissions=contract.permissions, budget=contract.budget, open_decisions=contract.open_decisions, approved_at=timezone.now())
+            project.current_plan_version = version
+            project.save(update_fields=["current_plan_version", "updated_at"])
+            task_by_external_id: dict[str, Task] = {}
+            dependency_specs: list[tuple[Task, list[str]]] = []
+            for order, milestone_contract in enumerate(contract.milestones):
+                milestone = Milestone.objects.create(project=project, plan=project_plan, key=f"{prefix}-{milestone_contract.key}"[:50], title=milestone_contract.title, goal=milestone_contract.goal, verification_contract=milestone_contract.verification_contract, order=order)
+                for feature_contract in milestone_contract.features:
+                    feature = Feature.objects.create(project=project, milestone=milestone, title=feature_contract.title, description=feature_contract.description, acceptance_criteria=feature_contract.acceptance_criteria)
+                    for task_contract in feature_contract.tasks:
+                        task = Task.objects.create(project=project, milestone=milestone, feature=feature, task_type=task_contract.task_type, status=TaskStatus.READY, priority=task_contract.priority, goal=task_contract.goal, acceptance_criteria=task_contract.acceptance_criteria)
+                        task_by_external_id[task_contract.task_id] = task
+                        dependency_specs.append((task, task_contract.dependencies))
+            for task, deps in dependency_specs:
+                for dep in deps:
+                    TaskDependency.objects.create(task=task, depends_on=task_by_external_id[dep])
+            return project_plan
+
+
+class EvolutionService(ProjectContextMixin):
+    def __init__(self, router: ModelRouter | None = None, bus: EventBus | None = None) -> None:
+        self.router = router
+        self.bus = bus or EventBus()
+
+    def create_candidate(self, project: Project, *, target: str, objective: str, baseline_measurement: dict[str, object], desired_direction: str, target_measurement: dict[str, object] | None = None, rationale: str = "", source: str = "user", evidence: dict[str, object] | None = None, risk: str = "MEDIUM", confidence: float = 0.5, source_steward_finding: StewardFinding | None = None, source_opportunity: ExplorationOpportunity | None = None) -> EvolutionCandidate:
+        if not baseline_measurement:
+            raise ValidationError("EVOLVE requires a measurable baseline; route to INVESTIGATE or EXPLORE instead.")
+        return EvolutionCandidate.objects.create(project=project, target=target, objective=objective, baseline_measurement=baseline_measurement, desired_direction=desired_direction, target_measurement=target_measurement or {}, rationale=rationale, source=source, evidence=evidence or {}, risk=risk, confidence=confidence, source_steward_finding=source_steward_finding, source_opportunity=source_opportunity)
+
+    def plan_with_project_brain(self, candidate: EvolutionCandidate) -> EvolutionPlan:
+        if not candidate.baseline_measurement:
+            raise ValidationError("EvolutionCandidate has no baseline measurement.")
+        context = self.project_context(candidate.project)
+        fallback = self._fallback_evolution_plan(candidate)
+        payload = self._json_from_project_brain(self.router, "Plan an EVOLUTION for an existing project. Return JSON with evolution_plan and project_plan. Must preserve measurable baseline and threshold.\n" + json.dumps({"candidate": self._candidate_payload(candidate), "context": context}, default=str), fallback=fallback, project=candidate.project, category="evolution")
+        raw_plan = payload.get("evolution_plan", payload)
+        if not isinstance(raw_plan, dict):
+            raise LifecyclePlanningError("evolution_plan must be an object")
+        project_plan_payload = raw_plan.get("project_plan") or payload.get("project_plan")
+        if not isinstance(project_plan_payload, dict):
+            create_project_planning_signal(candidate.project, "Sol evolution plan omitted project_plan; using bounded fallback plan.", {"candidate_id": str(candidate.id)})
+            raw_plan = dict(fallback["evolution_plan"])
+            project_plan_payload = dict(raw_plan["project_plan"])
+        try:
+            parse_project_plan_response(json.dumps(project_plan_payload))
+        except Exception as exc:
+            create_project_planning_signal(candidate.project, "Sol evolution project_plan failed validation; using bounded fallback plan.", {"candidate_id": str(candidate.id), "error": str(exc)})
+            raw_plan = dict(fallback["evolution_plan"])
+            project_plan_payload = dict(raw_plan["project_plan"])
+            parse_project_plan_response(json.dumps(project_plan_payload))
+        plan = EvolutionPlan.objects.create(candidate=candidate, project=candidate.project, status="PLANNED", baseline=dict(raw_plan.get("baseline", candidate.baseline_measurement)), hypothesis=str(raw_plan.get("hypothesis", candidate.objective)), intervention=str(raw_plan.get("intervention", "Implement targeted project improvement")), measurement_method=dict(raw_plan.get("measurement_method", {"type": "metadata"})), success_threshold=dict(raw_plan.get("success_threshold", {"minimum_improvement_percent": 10})), regression_constraints=list(raw_plan.get("regression_constraints", ["Deterministic tests pass"])), affected_components=list(raw_plan.get("affected_components", [candidate.target])), task_plan=project_plan_payload, experiment_requirements=dict(raw_plan.get("experiment_requirements", {})), context_snapshot=context)
+        candidate.status = "PLANNING"
+        candidate.save(update_fields=["status", "updated_at"])
+        return plan
+
+    def approve_plan(self, plan: EvolutionPlan) -> EvolutionPlan:
+        plan.status = "APPROVED"
+        plan.approved_at = timezone.now()
+        plan.save(update_fields=["status", "approved_at", "updated_at"])
+        plan.candidate.status = "READY"
+        plan.candidate.save(update_fields=["status", "updated_at"])
+        return plan
+
+    def materialize_work(self, plan: EvolutionPlan) -> ProjectPlan:
+        if plan.project_plan_id:
+            return plan.project_plan
+        contract = parse_project_plan_response(json.dumps(plan.task_plan))
+        project_plan = ExtensionService(bus=self.bus)._materialize_contract(plan.project, contract, prefix=f"EVO-{str(plan.id)[:8]}")
+        plan.project_plan = project_plan
+        plan.status = "MATERIALIZED"
+        plan.save(update_fields=["project_plan", "status", "updated_at"])
+        plan.candidate.status = "BUILDING"
+        plan.candidate.save(update_fields=["status", "updated_at"])
+        return project_plan
+
+    def execute(self, plan: EvolutionPlan, router: ModelRouter, *, test_command: list[str] | None = None) -> list[GraphRun]:
+        from graph.bootstrap import champion_task_execution_graph_v1
+
+        project_plan = self.materialize_work(plan)
+        runs: list[GraphRun] = []
+        for task in Task.objects.filter(milestone__plan=project_plan).exclude(status=TaskStatus.COMPLETE).order_by("priority", "created_at"):
+            graph_version = champion_task_execution_graph_v1()
+            graph_run = GraphRun.objects.create(execution_graph_version=graph_version, project=task.project, milestone=task.milestone, feature=task.feature, task=task, current_node=graph_version.graph_spec["entry"], metadata={"evolution_plan_id": str(plan.id), "evolution_candidate_id": str(plan.candidate_id)})
+            LangGraphRuntime(task_execution_registry(TaskExecutionServices(router, bus=self.bus, test_command=test_command or ["python", "-m", "pytest"])), bus=self.bus).run_until_terminal_or_paused(graph_run)
+            for commit in task.commits.all():
+                commit.evolution_candidate = plan.candidate
+                commit.save(update_fields=["evolution_candidate", "updated_at"])
+            runs.append(graph_run)
+        return runs
+
+    def measure_candidate(self, plan: EvolutionPlan) -> dict[str, object]:
+        method = plan.measurement_method or {}
+        if "candidate_measurement" in plan.metadata:
+            measurement = dict(plan.metadata["candidate_measurement"])
+        elif method.get("type") == "command" and plan.project.repository_path:
+            completed = subprocess.run([str(part) for part in method.get("command", [])], cwd=plan.project.repository_path, capture_output=True, text=True, check=False, timeout=120)
+            measurement = {"returncode": completed.returncode, "stdout": completed.stdout[-4000:], "stderr": completed.stderr[-2000:]}
+        else:
+            measurement = dict(plan.candidate.target_measurement or plan.baseline)
+        plan.candidate_measurement = measurement
+        plan.save(update_fields=["candidate_measurement", "updated_at"])
+        return measurement
+
+    def compare_baseline(self, plan: EvolutionPlan) -> dict[str, object]:
+        metric = str(plan.success_threshold.get("metric", plan.baseline.get("metric", "value")))
+        baseline_value = float(plan.baseline.get(metric, plan.baseline.get("value", 0)) or 0)
+        candidate_value = float(plan.candidate_measurement.get(metric, plan.candidate_measurement.get("value", baseline_value)) or 0)
+        direction = plan.candidate.desired_direction
+        if baseline_value == 0:
+            improvement_percent = 0.0
+        elif direction == "DECREASE":
+            improvement_percent = ((baseline_value - candidate_value) / baseline_value) * 100
+        else:
+            improvement_percent = ((candidate_value - baseline_value) / baseline_value) * 100
+        threshold = float(plan.success_threshold.get("minimum_improvement_percent", 0))
+        delta = {"metric": metric, "baseline": baseline_value, "candidate": candidate_value, "improvement_percent": improvement_percent, "threshold": threshold}
+        plan.delta = delta
+        plan.save(update_fields=["delta", "updated_at"])
+        return delta
+
+    def judge_evolution(self, plan: EvolutionPlan) -> Verification:
+        if not plan.candidate_measurement:
+            self.measure_candidate(plan)
+        delta = self.compare_baseline(plan)
+        project_plan = plan.project_plan
+        tasks = Task.objects.filter(milestone__plan=project_plan) if project_plan else Task.objects.none()
+        tests_pass = not TestRun.objects.filter(task__in=tasks).exclude(status="PASS").exists()
+        implementation_complete = not tasks.exclude(status=TaskStatus.COMPLETE).exists() if tasks.exists() else True
+        improved = float(delta["improvement_percent"]) >= float(delta["threshold"])
+        verdict = "PASS" if implementation_complete and tests_pass and improved else "NOT_IMPROVED"
+        verification = Verification.objects.create(project=plan.project, milestone=tasks.first().milestone if tasks.exists() else None, level=VerificationLevel.MILESTONE, result=VerificationResult.PASS if verdict == "PASS" else VerificationResult.FAIL, contract={"evolution_plan_id": str(plan.id), "success_threshold": plan.success_threshold, "regression_constraints": plan.regression_constraints}, evidence=[{"implementation_complete": implementation_complete, "tests_pass": tests_pass, "delta": delta, "verdict": verdict}], summary="Evolution objective improved" if verdict == "PASS" else "Evolution implementation did not improve objective")
+        plan.verification = verification
+        plan.verdict = verdict
+        plan.status = "COMPLETE" if verdict == "PASS" else "NOT_IMPROVED"
+        plan.completed_at = timezone.now()
+        plan.save(update_fields=["verification", "verdict", "status", "completed_at", "updated_at"])
+        plan.candidate.status = "COMPLETE" if verdict == "PASS" else "PROPOSED"
+        plan.candidate.save(update_fields=["status", "updated_at"])
+        return verification
+
+    def _fallback_evolution_plan(self, candidate: EvolutionCandidate) -> dict[str, object]:
+        acceptance = ["Deterministic tests pass", f"Objective improves: {candidate.objective}"]
+        task_goal = candidate.metadata.get("task_goal") or f"Improve {candidate.target}: {candidate.objective}"
+        return {"evolution_plan": {"baseline": candidate.baseline_measurement, "hypothesis": candidate.objective, "intervention": str(task_goal), "measurement_method": {"type": "metadata"}, "success_threshold": {"metric": candidate.baseline_measurement.get("metric", "value"), "minimum_improvement_percent": 10}, "regression_constraints": ["Deterministic tests pass"], "affected_components": [candidate.target], "project_plan": {"goal": candidate.project.goal, "scope": candidate.objective, "acceptance_criteria": acceptance, "milestones": [{"key": "EVOLVE", "title": candidate.target, "goal": candidate.objective, "verification_contract": {"evolution_candidate_id": str(candidate.id)}, "features": [{"key": "F1", "title": candidate.target, "description": candidate.objective, "acceptance_criteria": acceptance, "tasks": [{"id": "T1", "goal": str(task_goal), "type": "implementation", "acceptance_criteria": acceptance, "priority": 50, "dependencies": []}]}]}]}}}
+
+    def _candidate_payload(self, candidate: EvolutionCandidate) -> dict[str, object]:
+        return {"id": str(candidate.id), "target": candidate.target, "objective": candidate.objective, "baseline_measurement": candidate.baseline_measurement, "desired_direction": candidate.desired_direction, "target_measurement": candidate.target_measurement, "evidence": candidate.evidence}
+
+
+class ExplorerService(ProjectContextMixin):
+    def __init__(self, router: ModelRouter | None = None, bus: EventBus | None = None) -> None:
+        self.router = router
+        self.bus = bus or EventBus()
+
+    def start_exploration(self, project: Project, *, prompt: str = "") -> Exploration:
+        exploration = Exploration.objects.create(project=project, prompt=prompt, context_snapshot=self.project_context(project))
+        return exploration
+
+    def generate_opportunities(self, exploration: Exploration) -> list[ExplorationOpportunity]:
+        fallback = {"opportunities": self._fallback_opportunities(exploration)}
+        payload = self._json_from_project_brain(self.router, "Explore an existing project for valuable changes. Return JSON opportunities only; do not create build work.\n" + json.dumps(exploration.context_snapshot, default=str), fallback=fallback, project=exploration.project, category="exploration")
+        raw_items = payload.get("opportunities", [])
+        if not isinstance(raw_items, list):
+            raise LifecyclePlanningError("Explorer response opportunities must be a list")
+        opportunities: list[ExplorationOpportunity] = []
+        for raw in raw_items:
+            if not isinstance(raw, dict):
+                continue
+            opportunity = self.upsert_opportunity(exploration, raw)
+            opportunities.append(opportunity)
+        exploration.status = "COMPLETE"
+        exploration.completed_at = timezone.now()
+        exploration.save(update_fields=["status", "completed_at", "updated_at"])
+        return sorted(opportunities, key=lambda item: item.composite_score, reverse=True)
+
+    def upsert_opportunity(self, exploration: Exploration, raw: dict[str, object]) -> ExplorationOpportunity:
+        title = str(raw.get("title", "Untitled opportunity"))
+        grouping_key = self._grouping_key(exploration.project, title, str(raw.get("opportunity_type", "FEATURE")))
+        existing = self._known_duplicate(exploration.project, grouping_key, title)
+        scores = self.score(raw)
+        raw_evidence = raw.get("evidence", {})
+        evidence = raw_evidence if isinstance(raw_evidence, dict) else {"raw": raw_evidence}
+        if existing is not None:
+            existing.metadata = {**existing.metadata, "rediscovered_by": str(exploration.id)}
+            existing.save(update_fields=["metadata", "updated_at"])
+            return existing
+        return ExplorationOpportunity.objects.create(exploration=exploration, project=exploration.project, title=title, description=str(raw.get("description", "")), opportunity_type=str(raw.get("opportunity_type", "FEATURE")), evidence=evidence, rationale=str(raw.get("rationale", "")), expected_value=str(raw.get("expected_value", "")), effort_estimate=str(raw.get("effort_estimate", "MEDIUM")), risk=str(raw.get("risk", "MEDIUM")), confidence=scores["confidence"], technical_fit=scores["technical_fit"], strategic_fit=scores["strategic_fit"], value_score=scores["value"], effort_score=scores["effort"], risk_score=scores["risk"], composite_score=scores["composite"], recommended_action=str(raw.get("recommended_action", "DEFER")), grouping_key=grouping_key)
+
+    def score(self, raw: dict[str, object]) -> dict[str, float]:
+        value = self._score_value(raw.get("value", raw.get("value_score", 0.5)))
+        effort = self._score_value(raw.get("effort", raw.get("effort_score", 0.5)))
+        risk = self._score_value(raw.get("risk_score", 0.5))
+        confidence = self._score_value(raw.get("confidence", 0.5))
+        technical_fit = self._score_value(raw.get("technical_fit", 0.5))
+        strategic_fit = self._score_value(raw.get("strategic_fit", 0.5))
+        composite = (value * 0.35) + ((1 - effort) * 0.15) + ((1 - risk) * 0.15) + (confidence * 0.15) + (technical_fit * 0.1) + (strategic_fit * 0.1)
+        return {"value": value, "effort": effort, "risk": risk, "confidence": confidence, "technical_fit": technical_fit, "strategic_fit": strategic_fit, "composite": composite}
+
+    def _score_value(self, value: object) -> float:
+        try:
+            score = float(value)
+        except (TypeError, ValueError):
+            return 0.5
+        return max(0.0, min(1.0, score))
+
+    def convert_to_extension(self, opportunity: ExplorationOpportunity) -> ExtensionCandidate:
+        candidate = ExtensionService(bus=self.bus).create_candidate(opportunity.project, title=opportunity.title, description=opportunity.description, rationale=opportunity.rationale, source="Explorer", expected_value=opportunity.expected_value, affected_areas=[opportunity.opportunity_type], estimated_complexity=opportunity.effort_estimate, risk=opportunity.risk, confidence=opportunity.confidence, evidence=opportunity.evidence, source_opportunity=opportunity)
+        opportunity.converted_extension = candidate
+        opportunity.status = "CONVERTED"
+        opportunity.save(update_fields=["converted_extension", "status", "updated_at"])
+        return candidate
+
+    def convert_to_evolution(self, opportunity: ExplorationOpportunity, *, baseline_measurement: dict[str, object], desired_direction: str = "DECREASE") -> EvolutionCandidate:
+        candidate = EvolutionService(bus=self.bus).create_candidate(opportunity.project, target=opportunity.opportunity_type, objective=opportunity.description or opportunity.title, baseline_measurement=baseline_measurement, desired_direction=desired_direction, rationale=opportunity.rationale, source="Explorer", evidence=opportunity.evidence, risk=opportunity.risk, confidence=opportunity.confidence, source_opportunity=opportunity)
+        opportunity.converted_evolution = candidate
+        opportunity.status = "CONVERTED"
+        opportunity.save(update_fields=["converted_evolution", "status", "updated_at"])
+        return candidate
+
+    def defer(self, opportunity: ExplorationOpportunity) -> ExplorationOpportunity:
+        opportunity.status = "DEFERRED"
+        opportunity.save(update_fields=["status", "updated_at"])
+        return opportunity
+
+    def reject(self, opportunity: ExplorationOpportunity) -> ExplorationOpportunity:
+        opportunity.status = "REJECTED"
+        opportunity.save(update_fields=["status", "updated_at"])
+        return opportunity
+
+    def _fallback_opportunities(self, exploration: Exploration) -> list[dict[str, object]]:
+        context = exploration.context_snapshot
+        features = str(context.get("features", "")).lower()
+        opportunities: list[dict[str, object]] = []
+        if "dashboard" not in features:
+            opportunities.append({"title": "Add project health dashboard", "description": "Expose recent runs, findings, and lifecycle status in a project dashboard.", "opportunity_type": "FEATURE", "evidence": {"missing_feature": "dashboard"}, "rationale": "Operators need a fast project health view.", "expected_value": "Improves observability", "effort_estimate": "MEDIUM", "value": 0.8, "effort": 0.45, "risk_score": 0.3, "confidence": 0.7, "technical_fit": 0.8, "strategic_fit": 0.8, "recommended_action": "EXTEND"})
+        opportunities.append({"title": "Measure repository analysis latency", "description": "Establish and improve latency for repeated project inspection.", "opportunity_type": "PERFORMANCE", "evidence": {"source": "Explorer fallback"}, "rationale": "Faster analysis improves iteration speed.", "expected_value": "Reduces operational latency", "effort_estimate": "LOW", "value": 0.6, "effort": 0.25, "risk_score": 0.2, "confidence": 0.65, "technical_fit": 0.75, "strategic_fit": 0.65, "recommended_action": "EVOLVE"})
+        return opportunities
+
+    def _known_duplicate(self, project: Project, grouping_key: str, title: str) -> ExplorationOpportunity | None:
+        existing = ExplorationOpportunity.objects.filter(project=project, grouping_key=grouping_key).first()
+        if existing:
+            return existing
+        lowered = title.lower()
+        if ExtensionCandidate.objects.filter(project=project, title__iexact=title).exists() or RoadmapItem.objects.filter(project=project, title__iexact=title).exists():
+            return ExplorationOpportunity.objects.filter(project=project, title__iexact=title).first()
+        if any(item in lowered for item in ["authentication", "auth"]) and Decision.objects.filter(project=project, decision__icontains="authentication").exists():
+            return ExplorationOpportunity.objects.filter(project=project, title__icontains="auth").first()
+        return None
+
+    def _grouping_key(self, project: Project, title: str, opportunity_type: str) -> str:
+        fingerprint = hashlib.sha256(f"{title.lower()}:{opportunity_type.lower()}".encode("utf-8")).hexdigest()[:16]
+        return f"{project.id}:{opportunity_type}:{fingerprint}"[:240]
+
+
+class LifecycleInspectionService:
+    def project_lifecycle_view(self, project: Project) -> dict[str, object]:
+        return {
+            "project_id": str(project.id),
+            "repairs": list(project.steward_findings.filter(recommended_action="REPAIR").values("id", "title", "status", "severity")),
+            "extensions": [self._extension(candidate) for candidate in project.extension_candidates.order_by("-created_at")],
+            "evolutions": [self._evolution(candidate) for candidate in project.evolution_candidates.order_by("-created_at")],
+            "explorations": [self._exploration(exploration) for exploration in project.explorations.order_by("-created_at")],
+        }
+
+    def _extension(self, candidate: ExtensionCandidate) -> dict[str, object]:
+        plan = candidate.plans.order_by("-created_at").first()
+        tasks = Task.objects.filter(milestone__plan=plan.project_plan) if plan and plan.project_plan_id else Task.objects.none()
+        return {"candidate": {"id": str(candidate.id), "title": candidate.title, "status": candidate.status}, "plan": str(plan.id) if plan else None, "tasks": list(tasks.values("id", "status", "goal")), "verification": str(plan.verification_id) if plan and plan.verification_id else None, "commits": list(candidate.commits.values("id", "sha", "task_id"))}
+
+    def _evolution(self, candidate: EvolutionCandidate) -> dict[str, object]:
+        plan = candidate.plans.order_by("-created_at").first()
+        return {"candidate": {"id": str(candidate.id), "target": candidate.target, "objective": candidate.objective, "status": candidate.status}, "baseline": candidate.baseline_measurement, "target": candidate.target_measurement, "measurement": plan.candidate_measurement if plan else {}, "delta": plan.delta if plan else {}, "verdict": plan.verdict if plan else ""}
+
+    def _exploration(self, exploration: Exploration) -> dict[str, object]:
+        return {"id": str(exploration.id), "status": exploration.status, "opportunities": list(exploration.opportunities.order_by("-composite_score").values("id", "title", "opportunity_type", "status", "recommended_action", "value_score", "effort_score", "risk_score", "confidence", "technical_fit", "strategic_fit", "composite_score"))}
+
+
+def create_project_planning_signal(project: Project, summary: str, evidence: dict[str, object]) -> ProgenySignal:
+    return ProgenySignal.objects.create(project=project, source="project_lifecycle", severity="MEDIUM", failure_category="PROJECT_PLANNING", summary=summary, evidence=evidence, grouping_key=f"project_lifecycle:{project.id}:planning"[:120])

agents/steward.py

@@ -9,8 +9,9 @@ from django.core.exceptions import ValidationError
 from django.core.serializers.json import DjangoJSONEncoder
 from django.utils import timezone
 
+from agents.lifecycle import EvolutionService, ExtensionService
 from agents.progeny import ProgenyService
-from control_plane.agents.models import ImprovementCandidate, ProgenySignal
+from control_plane.agents.models import ProgenySignal
 from control_plane.events.bus import EventBus
 from control_plane.events.models import Event
 from control_plane.projects.models import (
@@ -18,7 +19,6 @@ from control_plane.projects.models import (
     Milestone,
     Project,
     ProjectPlan,
-    RoadmapItem,
     StewardAction,
     StewardCheck,
     StewardEnrollment,
@@ -189,11 +189,17 @@ class StewardService:
             task = None if requires_approval else self._create_repair_task(finding)
             action = StewardAction.objects.create(finding=finding, action_type="REPAIR", status="APPROVAL_REQUIRED" if requires_approval else "ROUTED", task=task, requires_approval=requires_approval)
         elif finding.recommended_action == "EXTEND":
-            item = RoadmapItem.objects.create(project=finding.project, title=finding.title, description=finding.summary, source="steward")
+            candidate = ExtensionService(bus=self.bus).create_candidate(finding.project, title=finding.title, description=finding.summary, rationale="Steward classified this finding as new scope.", source="StewardFinding", expected_value=finding.summary, affected_areas=[finding.finding_type], risk=finding.severity, confidence=finding.confidence, evidence={"steward_finding_id": str(finding.id), **finding.evidence}, source_steward_finding=finding)
-            action = StewardAction.objects.create(finding=finding, action_type="EXTEND", status="APPROVAL_REQUIRED" if requires_approval else "ROUTED", roadmap_item=item, requires_approval=requires_approval)
+            action = StewardAction.objects.create(finding=finding, action_type="EXTEND", status="APPROVAL_REQUIRED" if requires_approval else "ROUTED", extension_candidate=candidate, requires_approval=requires_approval)
         elif finding.recommended_action == "EVOLVE":
-            candidate = ImprovementCandidate.objects.create(target_type="PROJECT", target_id=str(finding.project_id), target_label=finding.project.name, hypothesis=finding.summary, recommended_route="EvolutionCandidate", evidence={"steward_finding_id": str(finding.id), "finding_type": finding.finding_type})
+            baseline = self._baseline_from_finding(finding)
-            action = StewardAction.objects.create(finding=finding, action_type="EVOLVE", status="APPROVAL_REQUIRED" if requires_approval else "ROUTED", improvement_candidate=candidate, requires_approval=requires_approval)
+            if baseline:
+                candidate = EvolutionService(bus=self.bus).create_candidate(finding.project, target=str(finding.metadata.get("component", finding.finding_type)), objective=finding.summary or finding.title, baseline_measurement=baseline, desired_direction=str(finding.evidence.get("desired_direction", "DECREASE")), rationale="Steward classified this finding as measurable project evolution.", source="StewardFinding", evidence={"steward_finding_id": str(finding.id), **finding.evidence}, risk=finding.severity, confidence=finding.confidence, source_steward_finding=finding)
+                action = StewardAction.objects.create(finding=finding, action_type="EVOLVE", status="APPROVAL_REQUIRED" if requires_approval else "ROUTED", evolution_candidate=candidate, requires_approval=requires_approval)
+            else:
+                signal = ProgenySignal.objects.create(project=finding.project, source="steward", severity=finding.severity, failure_category=finding.finding_type, summary="Evolution baseline missing; investigate before project evolution.", evidence={"steward_finding_id": str(finding.id), **finding.evidence}, grouping_key=f"steward:missing_baseline:{finding.grouping_key}"[:120])
+                investigation = ProgenyService(self.bus).create_smart_investigation(signal.grouping_key)
+                action = StewardAction.objects.create(finding=finding, action_type="INVESTIGATE", status="ROUTED", investigation=investigation, requires_approval=False)
         elif finding.recommended_action == "INVESTIGATE":
             signal = ProgenySignal.objects.create(project=finding.project, source="steward", severity=finding.severity, failure_category=finding.finding_type, summary=finding.summary, evidence={"steward_finding_id": str(finding.id), **finding.evidence}, grouping_key=f"steward:{finding.grouping_key}"[:120])
             investigation = ProgenyService(self.bus).create_smart_investigation(signal.grouping_key)
@@ -324,6 +330,17 @@ class StewardService:
         threshold = str(policy.approval_requirements.get(finding.recommended_action, "CRITICAL"))
         return SEVERITY_RANK.get(finding.severity, 0) >= SEVERITY_RANK.get(threshold, 4)
 
+    def _baseline_from_finding(self, finding: StewardFinding) -> dict[str, object]:
+        evidence = finding.evidence or {}
+        regressions = evidence.get("regressions", [])
+        if isinstance(regressions, list) and regressions:
+            first = regressions[0]
+            if isinstance(first, dict):
+                metric = str(first.get("metric", "value"))
+                return {"metric": metric, metric: first.get("baseline", first.get("value", 0)), "current": first.get("current"), "delta_percent": first.get("delta_percent")}
+        baseline = evidence.get("baseline_measurement")
+        return dict(baseline) if isinstance(baseline, dict) else {}
+
     def _grouping_key(self, project: Project, raw: dict[str, object]) -> str:
         component = str(raw.get("component", "project"))[:80]
         evidence = str(raw.get("evidence", {}))[:1000]

control_plane/projects/migrations/0005_extend_evolve_explore_v1.py

@@ -0,0 +1,165 @@
+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", "0004_steward_v1"),
+        ("verification", "0001_initial"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="ExtensionCandidate",
+            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)),
+                ("title", models.CharField(max_length=255)),
+                ("description", models.TextField(blank=True)),
+                ("rationale", models.TextField(blank=True)),
+                ("source", models.CharField(default="user", max_length=80)),
+                ("expected_value", models.TextField(blank=True)),
+                ("affected_areas", models.JSONField(blank=True, default=list)),
+                ("estimated_complexity", models.CharField(blank=True, max_length=80)),
+                ("risk", models.CharField(default="MEDIUM", max_length=80)),
+                ("confidence", models.FloatField(default=0.0)),
+                ("status", models.CharField(default="PROPOSED", max_length=32)),
+                ("evidence", models.JSONField(blank=True, default=dict)),
+                ("metadata", models.JSONField(blank=True, default=dict)),
+                ("project", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="extension_candidates", to="projects.project")),
+                ("source_investigation", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="extension_candidates", to="agents.progenyinvestigation")),
+                ("source_roadmap_item", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="extension_candidates", to="projects.roadmapitem")),
+                ("source_steward_finding", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="extension_candidates", to="projects.stewardfinding")),
+            ],
+        ),
+        migrations.CreateModel(
+            name="EvolutionCandidate",
+            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)),
+                ("target", models.CharField(max_length=240)),
+                ("objective", models.TextField()),
+                ("baseline_measurement", models.JSONField(blank=True, default=dict)),
+                ("desired_direction", models.CharField(max_length=32)),
+                ("target_measurement", models.JSONField(blank=True, default=dict)),
+                ("rationale", models.TextField(blank=True)),
+                ("source", models.CharField(default="user", max_length=80)),
+                ("evidence", models.JSONField(blank=True, default=dict)),
+                ("status", models.CharField(default="PROPOSED", max_length=32)),
+                ("risk", models.CharField(default="MEDIUM", max_length=80)),
+                ("confidence", models.FloatField(default=0.0)),
+                ("metadata", models.JSONField(blank=True, default=dict)),
+                ("project", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="evolution_candidates", to="projects.project")),
+                ("source_investigation", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="evolution_candidates", to="agents.progenyinvestigation")),
+                ("source_steward_finding", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="evolution_candidates", to="projects.stewardfinding")),
+            ],
+        ),
+        migrations.CreateModel(
+            name="Exploration",
+            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)),
+                ("source", models.CharField(default="Explorer", max_length=80)),
+                ("status", models.CharField(default="RUNNING", max_length=32)),
+                ("prompt", models.TextField(blank=True)),
+                ("context_snapshot", models.JSONField(blank=True, default=dict)),
+                ("completed_at", models.DateTimeField(blank=True, null=True)),
+                ("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="explorations", to="graph.executiongraphversion")),
+                ("project", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="explorations", to="projects.project")),
+            ],
+        ),
+        migrations.CreateModel(
+            name="ExtensionPlan",
+            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)),
+                ("status", models.CharField(default="DRAFT", max_length=32)),
+                ("strategy", models.TextField(blank=True)),
+                ("plan", models.JSONField(blank=True, default=dict)),
+                ("acceptance_criteria", models.JSONField(blank=True, default=list)),
+                ("context_snapshot", models.JSONField(blank=True, default=dict)),
+                ("approved_at", models.DateTimeField(blank=True, null=True)),
+                ("completed_at", models.DateTimeField(blank=True, null=True)),
+                ("metadata", models.JSONField(blank=True, default=dict)),
+                ("candidate", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="plans", to="projects.extensioncandidate")),
+                ("project", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="extension_plans", to="projects.project")),
+                ("project_plan", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="extension_plans", to="projects.projectplan")),
+                ("verification", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="extension_plans", to="verification.verification")),
+            ],
+        ),
+        migrations.CreateModel(
+            name="EvolutionPlan",
+            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)),
+                ("status", models.CharField(default="DRAFT", max_length=32)),
+                ("baseline", models.JSONField(blank=True, default=dict)),
+                ("hypothesis", models.TextField(blank=True)),
+                ("intervention", models.TextField(blank=True)),
+                ("measurement_method", models.JSONField(blank=True, default=dict)),
+                ("success_threshold", models.JSONField(blank=True, default=dict)),
+                ("regression_constraints", models.JSONField(blank=True, default=list)),
+                ("affected_components", models.JSONField(blank=True, default=list)),
+                ("task_plan", models.JSONField(blank=True, default=dict)),
+                ("experiment_requirements", models.JSONField(blank=True, default=dict)),
+                ("candidate_measurement", models.JSONField(blank=True, default=dict)),
+                ("delta", models.JSONField(blank=True, default=dict)),
+                ("verdict", models.CharField(blank=True, max_length=32)),
+                ("context_snapshot", models.JSONField(blank=True, default=dict)),
+                ("approved_at", models.DateTimeField(blank=True, null=True)),
+                ("completed_at", models.DateTimeField(blank=True, null=True)),
+                ("metadata", models.JSONField(blank=True, default=dict)),
+                ("candidate", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="plans", to="projects.evolutioncandidate")),
+                ("project", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="evolution_plans", to="projects.project")),
+                ("project_plan", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="evolution_plans", to="projects.projectplan")),
+                ("verification", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="evolution_plans", to="verification.verification")),
+            ],
+        ),
+        migrations.CreateModel(
+            name="ExplorationOpportunity",
+            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)),
+                ("title", models.CharField(max_length=255)),
+                ("description", models.TextField(blank=True)),
+                ("opportunity_type", models.CharField(max_length=80)),
+                ("evidence", models.JSONField(blank=True, default=dict)),
+                ("rationale", models.TextField(blank=True)),
+                ("expected_value", models.TextField(blank=True)),
+                ("effort_estimate", models.CharField(blank=True, max_length=80)),
+                ("risk", models.CharField(default="MEDIUM", max_length=80)),
+                ("confidence", models.FloatField(default=0.0)),
+                ("technical_fit", models.FloatField(default=0.0)),
+                ("strategic_fit", models.FloatField(default=0.0)),
+                ("value_score", models.FloatField(default=0.0)),
+                ("effort_score", models.FloatField(default=0.0)),
+                ("risk_score", models.FloatField(default=0.0)),
+                ("composite_score", models.FloatField(default=0.0)),
+                ("recommended_action", models.CharField(default="DEFER", max_length=32)),
+                ("status", models.CharField(default="DISCOVERED", max_length=32)),
+                ("grouping_key", models.CharField(max_length=240)),
+                ("metadata", models.JSONField(blank=True, default=dict)),
+                ("converted_evolution", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="converted_from_opportunities", to="projects.evolutioncandidate")),
+                ("converted_extension", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="converted_from_opportunities", to="projects.extensioncandidate")),
+                ("exploration", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="opportunities", to="projects.exploration")),
+                ("project", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="exploration_opportunities", to="projects.project")),
+            ],
+        ),
+        migrations.AddField(model_name="evolutioncandidate", name="source_opportunity", field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="evolution_candidates", to="projects.explorationopportunity")),
+        migrations.AddField(model_name="extensioncandidate", name="source_opportunity", field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="extension_candidates", to="projects.explorationopportunity")),
+        migrations.AddField(model_name="stewardaction", name="evolution_candidate", field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="steward_actions", to="projects.evolutioncandidate")),
+        migrations.AddField(model_name="stewardaction", name="extension_candidate", field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="steward_actions", to="projects.extensioncandidate")),
+        migrations.AddField(model_name="commitrecord", name="evolution_candidate", field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="commits", to="projects.evolutioncandidate")),
+        migrations.AddField(model_name="commitrecord", name="extension_candidate", field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="commits", to="projects.extensioncandidate")),
+    ]

control_plane/projects/models.py

@@ -179,6 +179,12 @@ class CommitRecord(TimestampedModel):
     steward_finding = models.ForeignKey(
         "projects.StewardFinding", on_delete=models.SET_NULL, null=True, blank=True, related_name="commits"
     )
+    extension_candidate = models.ForeignKey(
+        "projects.ExtensionCandidate", on_delete=models.SET_NULL, null=True, blank=True, related_name="commits"
+    )
+    evolution_candidate = models.ForeignKey(
+        "projects.EvolutionCandidate", on_delete=models.SET_NULL, null=True, blank=True, related_name="commits"
+    )
     sha = models.CharField(max_length=64)
     branch_name = models.CharField(max_length=255)
     message = models.TextField()
@@ -328,6 +334,123 @@ class StewardAction(TimestampedModel):
     roadmap_item = models.ForeignKey(RoadmapItem, on_delete=models.SET_NULL, null=True, blank=True, related_name="steward_actions")
     investigation = models.ForeignKey("agents.ProgenyInvestigation", on_delete=models.SET_NULL, null=True, blank=True, related_name="steward_actions")
     improvement_candidate = models.ForeignKey("agents.ImprovementCandidate", on_delete=models.SET_NULL, null=True, blank=True, related_name="steward_actions")
+    extension_candidate = models.ForeignKey("projects.ExtensionCandidate", on_delete=models.SET_NULL, null=True, blank=True, related_name="steward_actions")
+    evolution_candidate = models.ForeignKey("projects.EvolutionCandidate", on_delete=models.SET_NULL, null=True, blank=True, related_name="steward_actions")
     requires_approval = models.BooleanField(default=False)
     approved_at = models.DateTimeField(null=True, blank=True)
     metadata = models.JSONField(default=dict, blank=True)
+
+
+class ExtensionCandidate(TimestampedModel):
+    project = models.ForeignKey(Project, on_delete=models.CASCADE, related_name="extension_candidates")
+    title = models.CharField(max_length=255)
+    description = models.TextField(blank=True)
+    rationale = models.TextField(blank=True)
+    source = models.CharField(max_length=80, default="user")
+    expected_value = models.TextField(blank=True)
+    affected_areas = models.JSONField(default=list, blank=True)
+    estimated_complexity = models.CharField(max_length=80, blank=True)
+    risk = models.CharField(max_length=80, default="MEDIUM")
+    confidence = models.FloatField(default=0.0)
+    status = models.CharField(max_length=32, default="PROPOSED")
+    evidence = models.JSONField(default=dict, blank=True)
+    source_steward_finding = models.ForeignKey(StewardFinding, on_delete=models.SET_NULL, null=True, blank=True, related_name="extension_candidates")
+    source_investigation = models.ForeignKey("agents.ProgenyInvestigation", on_delete=models.SET_NULL, null=True, blank=True, related_name="extension_candidates")
+    source_roadmap_item = models.ForeignKey(RoadmapItem, on_delete=models.SET_NULL, null=True, blank=True, related_name="extension_candidates")
+    source_opportunity = models.ForeignKey("projects.ExplorationOpportunity", on_delete=models.SET_NULL, null=True, blank=True, related_name="extension_candidates")
+    metadata = models.JSONField(default=dict, blank=True)
+
+
+class ExtensionPlan(TimestampedModel):
+    candidate = models.ForeignKey(ExtensionCandidate, on_delete=models.CASCADE, related_name="plans")
+    project = models.ForeignKey(Project, on_delete=models.CASCADE, related_name="extension_plans")
+    status = models.CharField(max_length=32, default="DRAFT")
+    strategy = models.TextField(blank=True)
+    plan = models.JSONField(default=dict, blank=True)
+    acceptance_criteria = models.JSONField(default=list, blank=True)
+    context_snapshot = models.JSONField(default=dict, blank=True)
+    project_plan = models.ForeignKey(ProjectPlan, on_delete=models.SET_NULL, null=True, blank=True, related_name="extension_plans")
+    verification = models.ForeignKey("verification.Verification", on_delete=models.SET_NULL, null=True, blank=True, related_name="extension_plans")
+    approved_at = models.DateTimeField(null=True, blank=True)
+    completed_at = models.DateTimeField(null=True, blank=True)
+    metadata = models.JSONField(default=dict, blank=True)
+
+
+class EvolutionCandidate(TimestampedModel):
+    project = models.ForeignKey(Project, on_delete=models.CASCADE, related_name="evolution_candidates")
+    target = models.CharField(max_length=240)
+    objective = models.TextField()
+    baseline_measurement = models.JSONField(default=dict, blank=True)
+    desired_direction = models.CharField(max_length=32)
+    target_measurement = models.JSONField(default=dict, blank=True)
+    rationale = models.TextField(blank=True)
+    source = models.CharField(max_length=80, default="user")
+    evidence = models.JSONField(default=dict, blank=True)
+    status = models.CharField(max_length=32, default="PROPOSED")
+    risk = models.CharField(max_length=80, default="MEDIUM")
+    confidence = models.FloatField(default=0.0)
+    source_steward_finding = models.ForeignKey(StewardFinding, on_delete=models.SET_NULL, null=True, blank=True, related_name="evolution_candidates")
+    source_investigation = models.ForeignKey("agents.ProgenyInvestigation", on_delete=models.SET_NULL, null=True, blank=True, related_name="evolution_candidates")
+    source_opportunity = models.ForeignKey("projects.ExplorationOpportunity", on_delete=models.SET_NULL, null=True, blank=True, related_name="evolution_candidates")
+    metadata = models.JSONField(default=dict, blank=True)
+
+
+class EvolutionPlan(TimestampedModel):
+    candidate = models.ForeignKey(EvolutionCandidate, on_delete=models.CASCADE, related_name="plans")
+    project = models.ForeignKey(Project, on_delete=models.CASCADE, related_name="evolution_plans")
+    status = models.CharField(max_length=32, default="DRAFT")
+    baseline = models.JSONField(default=dict, blank=True)
+    hypothesis = models.TextField(blank=True)
+    intervention = models.TextField(blank=True)
+    measurement_method = models.JSONField(default=dict, blank=True)
+    success_threshold = models.JSONField(default=dict, blank=True)
+    regression_constraints = models.JSONField(default=list, blank=True)
+    affected_components = models.JSONField(default=list, blank=True)
+    task_plan = models.JSONField(default=dict, blank=True)
+    experiment_requirements = models.JSONField(default=dict, blank=True)
+    candidate_measurement = models.JSONField(default=dict, blank=True)
+    delta = models.JSONField(default=dict, blank=True)
+    verdict = models.CharField(max_length=32, blank=True)
+    context_snapshot = models.JSONField(default=dict, blank=True)
+    project_plan = models.ForeignKey(ProjectPlan, on_delete=models.SET_NULL, null=True, blank=True, related_name="evolution_plans")
+    verification = models.ForeignKey("verification.Verification", on_delete=models.SET_NULL, null=True, blank=True, related_name="evolution_plans")
+    approved_at = models.DateTimeField(null=True, blank=True)
+    completed_at = models.DateTimeField(null=True, blank=True)
+    metadata = models.JSONField(default=dict, blank=True)
+
+
+class Exploration(TimestampedModel):
+    project = models.ForeignKey(Project, on_delete=models.CASCADE, related_name="explorations")
+    source = models.CharField(max_length=80, default="Explorer")
+    status = models.CharField(max_length=32, default="RUNNING")
+    prompt = models.TextField(blank=True)
+    context_snapshot = models.JSONField(default=dict, blank=True)
+    execution_graph_version = models.ForeignKey("graph.ExecutionGraphVersion", on_delete=models.SET_NULL, null=True, blank=True, related_name="explorations")
+    completed_at = models.DateTimeField(null=True, blank=True)
+    metadata = models.JSONField(default=dict, blank=True)
+
+
+class ExplorationOpportunity(TimestampedModel):
+    exploration = models.ForeignKey(Exploration, on_delete=models.CASCADE, related_name="opportunities")
+    project = models.ForeignKey(Project, on_delete=models.CASCADE, related_name="exploration_opportunities")
+    title = models.CharField(max_length=255)
+    description = models.TextField(blank=True)
+    opportunity_type = models.CharField(max_length=80)
+    evidence = models.JSONField(default=dict, blank=True)
+    rationale = models.TextField(blank=True)
+    expected_value = models.TextField(blank=True)
+    effort_estimate = models.CharField(max_length=80, blank=True)
+    risk = models.CharField(max_length=80, default="MEDIUM")
+    confidence = models.FloatField(default=0.0)
+    technical_fit = models.FloatField(default=0.0)
+    strategic_fit = models.FloatField(default=0.0)
+    value_score = models.FloatField(default=0.0)
+    effort_score = models.FloatField(default=0.0)
+    risk_score = models.FloatField(default=0.0)
+    composite_score = models.FloatField(default=0.0)
+    recommended_action = models.CharField(max_length=32, default="DEFER")
+    status = models.CharField(max_length=32, default="DISCOVERED")
+    grouping_key = models.CharField(max_length=240)
+    converted_extension = models.ForeignKey(ExtensionCandidate, on_delete=models.SET_NULL, null=True, blank=True, related_name="converted_from_opportunities")
+    converted_evolution = models.ForeignKey(EvolutionCandidate, on_delete=models.SET_NULL, null=True, blank=True, related_name="converted_from_opportunities")
+    metadata = models.JSONField(default=dict, blank=True)

graph/bootstrap.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from django.utils import timezone
 
+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.steward import steward_run_graph_v1
 from graph.task_execution import task_execution_graph_v1
@@ -51,3 +52,32 @@ def champion_steward_run_graph_v1() -> ExecutionGraphVersion:
         version.promoted_at = timezone.now()
         version.save(update_fields=["status", "promoted_at"])
     return version
+
+
+def _champion_graph(spec) -> ExecutionGraphVersion:
+    definition, _ = ExecutionGraphDefinition.objects.get_or_create(
+        name=spec.name,
+        defaults={"graph_type": spec.graph_type, "description": str(spec.metadata.get("description", ""))},
+    )
+    version, created = ExecutionGraphVersion.objects.get_or_create(
+        graph=definition,
+        version=spec.version,
+        defaults={"status": ExecutionGraphVersionStatus.CHAMPION, "graph_spec": spec.to_dict(), "metadata": {"immutable_after_use": True}, "promoted_at": timezone.now()},
+    )
+    if not created and version.status != ExecutionGraphVersionStatus.CHAMPION:
+        version.status = ExecutionGraphVersionStatus.CHAMPION
+        version.promoted_at = timezone.now()
+        version.save(update_fields=["status", "promoted_at"])
+    return version
+
+
+def champion_project_extension_graph_v1() -> ExecutionGraphVersion:
+    return _champion_graph(project_extension_graph_v1())
+
+
+def champion_project_evolution_graph_v1() -> ExecutionGraphVersion:
+    return _champion_graph(project_evolution_graph_v1())
+
+
+def champion_project_exploration_graph_v1() -> ExecutionGraphVersion:
+    return _champion_graph(project_exploration_graph_v1())

graph/lifecycle.py

@@ -0,0 +1,329 @@
+from __future__ import annotations
+
+from django.utils import timezone
+
+from agents.lifecycle import EvolutionService, ExplorerService, ExtensionService
+from graph.models import GraphApproval, GraphApprovalStatus
+from graph.native_runtime import GraphExecutionContext
+from graph.registry import NodeHandlerRegistry, NodeResult
+from graph.spec import ExecutionGraphSpec, GraphEdgeSpec, GraphNodeSpec
+
+
+def project_extension_graph_v1() -> ExecutionGraphSpec:
+    spec = ExecutionGraphSpec(
+        name="project_extension",
+        version=1,
+        graph_type="PROJECT_EXTENSION",
+        entry="prepare",
+        nodes={node: GraphNodeSpec(node, node if node == "complete" else f"extension_{node}") for node in ["prepare", "gather_context", "plan_with_project_brain", "validate_plan", "await_approval", "materialize_project_dag", "execute", "verify_extension", "complete"]},
+        edges=[GraphEdgeSpec("prepare", "gather_context", "success"), GraphEdgeSpec("gather_context", "plan_with_project_brain", "success"), GraphEdgeSpec("plan_with_project_brain", "validate_plan", "success"), GraphEdgeSpec("validate_plan", "await_approval", "success"), GraphEdgeSpec("await_approval", "materialize_project_dag", "approved"), GraphEdgeSpec("await_approval", "complete", "rejected"), GraphEdgeSpec("materialize_project_dag", "execute", "success"), GraphEdgeSpec("execute", "verify_extension", "success"), GraphEdgeSpec("verify_extension", "complete", "success")],
+        terminal_nodes=["complete"],
+        metadata={"description": "Project extension workflow: Sol plan, approval, normal task execution, milestone verification."},
+    )
+    spec.validate()
+    return spec
+
+
+def project_evolution_graph_v1() -> ExecutionGraphSpec:
+    spec = ExecutionGraphSpec(
+        name="project_evolution",
+        version=1,
+        graph_type="PROJECT_EVOLUTION",
+        entry="prepare",
+        nodes={node: GraphNodeSpec(node, node if node in ["complete", "not_improved"] else f"evolution_{node}") for node in ["prepare", "establish_baseline", "form_hypothesis", "plan_with_project_brain", "await_approval", "materialize_work", "execute", "measure_candidate", "compare_baseline", "judge_evolution", "complete", "not_improved"]},
+        edges=[GraphEdgeSpec("prepare", "establish_baseline", "success"), GraphEdgeSpec("establish_baseline", "form_hypothesis", "success"), GraphEdgeSpec("form_hypothesis", "plan_with_project_brain", "success"), GraphEdgeSpec("plan_with_project_brain", "await_approval", "success"), GraphEdgeSpec("await_approval", "materialize_work", "approved"), GraphEdgeSpec("await_approval", "not_improved", "rejected"), GraphEdgeSpec("materialize_work", "execute", "success"), GraphEdgeSpec("execute", "measure_candidate", "success"), GraphEdgeSpec("measure_candidate", "compare_baseline", "success"), GraphEdgeSpec("compare_baseline", "judge_evolution", "success"), GraphEdgeSpec("judge_evolution", "complete", "PASS"), GraphEdgeSpec("judge_evolution", "not_improved", "NOT_IMPROVED")],
+        terminal_nodes=["complete", "not_improved"],
+        metadata={"description": "Project evolution workflow: measurable baseline, Sol plan, approval, normal task execution, objective comparison."},
+    )
+    spec.validate()
+    return spec
+
+
+def project_exploration_graph_v1() -> ExecutionGraphSpec:
+    spec = ExecutionGraphSpec(
+        name="project_exploration",
+        version=1,
+        graph_type="PROJECT_EXPLORATION",
+        entry="prepare",
+        nodes={node: GraphNodeSpec(node, node if node == "complete" else f"exploration_{node}") for node in ["prepare", "gather_project_evidence", "identify_gaps", "generate_opportunities", "normalize", "deduplicate", "score", "rank", "complete"]},
+        edges=[GraphEdgeSpec("prepare", "gather_project_evidence", "success"), GraphEdgeSpec("gather_project_evidence", "identify_gaps", "success"), GraphEdgeSpec("identify_gaps", "generate_opportunities", "success"), GraphEdgeSpec("generate_opportunities", "normalize", "success"), GraphEdgeSpec("normalize", "deduplicate", "success"), GraphEdgeSpec("deduplicate", "score", "success"), GraphEdgeSpec("score", "rank", "success"), GraphEdgeSpec("rank", "complete", "success")],
+        terminal_nodes=["complete"],
+        metadata={"description": "Project exploration workflow: discover, deduplicate, score, rank opportunities without execution."},
+    )
+    spec.validate()
+    return spec
+
+
+class LifecycleNode:
+    idempotent = True
+    replay_safe = True
+    destructive = False
+
+    def __init__(self, node_type: str) -> None:
+        self.node_type = node_type
+
+
+class ExtensionNode(LifecycleNode):
+    def __init__(self, service: ExtensionService, node_type: str) -> None:
+        super().__init__(node_type)
+        self.service = service
+
+    def plan_id(self, context: GraphExecutionContext) -> str:
+        return str(context.graph_run.metadata.get("extension_plan_id", ""))
+
+    def candidate_id(self, context: GraphExecutionContext) -> str:
+        return str(context.graph_run.metadata["extension_candidate_id"])
+
+
+class ExtensionPrepareNode(ExtensionNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        return NodeResult("COMPLETE", "success")
+
+
+class ExtensionGatherContextNode(ExtensionNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        from control_plane.projects.models import ExtensionCandidate
+
+        candidate = ExtensionCandidate.objects.get(id=self.candidate_id(context))
+        metadata = dict(context.graph_run.metadata)
+        metadata["context_snapshot"] = self.service.project_context(candidate.project)
+        context.graph_run.metadata = metadata
+        context.graph_run.save(update_fields=["metadata", "updated_at"])
+        return NodeResult("COMPLETE", "success")
+
+
+class ExtensionPlanNode(ExtensionNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        from control_plane.projects.models import ExtensionCandidate
+
+        candidate = ExtensionCandidate.objects.get(id=self.candidate_id(context))
+        plan = candidate.plans.order_by("-created_at").first() or self.service.plan_with_project_brain(candidate)
+        metadata = dict(context.graph_run.metadata)
+        metadata["extension_plan_id"] = str(plan.id)
+        context.graph_run.metadata = metadata
+        context.graph_run.save(update_fields=["metadata", "updated_at"])
+        return NodeResult("COMPLETE", "success", {"extension_plan_id": str(plan.id)})
+
+
+class ExtensionValidatePlanNode(ExtensionNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        from control_plane.projects.models import ExtensionPlan
+        from project_brain.planning import parse_project_plan_response
+        import json
+
+        plan = ExtensionPlan.objects.get(id=self.plan_id(context))
+        parse_project_plan_response(json.dumps(plan.plan.get("project_plan", {})))
+        return NodeResult("COMPLETE", "success")
+
+
+class ApprovalNode(LifecycleNode):
+    reason = "AWAITING_LIFECYCLE_APPROVAL"
+
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        node_run = context.graph_run.node_runs.filter(node_id=context.graph_run.current_node).order_by("-visit_index").first()
+        if GraphApproval.objects.filter(graph_run=context.graph_run, status=GraphApprovalStatus.APPROVED).exists():
+            return NodeResult("COMPLETE", "approved")
+        if GraphApproval.objects.filter(graph_run=context.graph_run, status=GraphApprovalStatus.REJECTED).exists():
+            return NodeResult("COMPLETE", "rejected")
+        GraphApproval.objects.get_or_create(graph_run=context.graph_run, node_run=node_run, reason=self.reason)
+        return NodeResult("PAUSED", "awaiting", pause_reason=self.reason)
+
+
+class ExtensionMaterializeNode(ExtensionNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        from control_plane.projects.models import ExtensionPlan
+
+        plan = ExtensionPlan.objects.get(id=self.plan_id(context))
+        self.service.approve_plan(plan)
+        project_plan = self.service.materialize_project_dag(plan)
+        return NodeResult("COMPLETE", "success", {"project_plan_id": str(project_plan.id)})
+
+
+class ExtensionExecuteNode(ExtensionNode):
+    destructive = True
+
+    def __init__(self, service: ExtensionService, node_type: str, router=None, test_command: list[str] | None = None) -> None:
+        super().__init__(service, node_type)
+        self.router = router
+        self.test_command = test_command
+
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        from control_plane.projects.models import ExtensionPlan
+
+        if self.router is None:
+            return NodeResult("COMPLETE", "success", {"executed": False})
+        plan = ExtensionPlan.objects.get(id=self.plan_id(context))
+        runs = self.service.execute(plan, self.router, test_command=self.test_command)
+        return NodeResult("COMPLETE", "success", {"graph_run_ids": [str(run.id) for run in runs]})
+
+
+class ExtensionVerifyNode(ExtensionNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        from control_plane.projects.models import ExtensionPlan
+
+        plan = ExtensionPlan.objects.get(id=self.plan_id(context))
+        verification = self.service.verify_extension(plan)
+        return NodeResult("COMPLETE", "success", {"verification_id": str(verification.id), "result": verification.result})
+
+
+class EvolutionNode(LifecycleNode):
+    def __init__(self, service: EvolutionService, node_type: str) -> None:
+        super().__init__(node_type)
+        self.service = service
+
+    def candidate_id(self, context: GraphExecutionContext) -> str:
+        return str(context.graph_run.metadata["evolution_candidate_id"])
+
+    def plan_id(self, context: GraphExecutionContext) -> str:
+        return str(context.graph_run.metadata.get("evolution_plan_id", ""))
+
+
+class EvolutionSimpleNode(EvolutionNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        return NodeResult("COMPLETE", "success")
+
+
+class EvolutionBaselineNode(EvolutionNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        from control_plane.projects.models import EvolutionCandidate
+
+        candidate = EvolutionCandidate.objects.get(id=self.candidate_id(context))
+        if not candidate.baseline_measurement:
+            return NodeResult("FAILED", "failure", failure_evidence={"reason": "EVOLVE requires baseline measurement"})
+        return NodeResult("COMPLETE", "success", {"baseline": candidate.baseline_measurement})
+
+
+class EvolutionPlanNode(EvolutionNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        from control_plane.projects.models import EvolutionCandidate
+
+        candidate = EvolutionCandidate.objects.get(id=self.candidate_id(context))
+        plan = candidate.plans.order_by("-created_at").first() or self.service.plan_with_project_brain(candidate)
+        metadata = dict(context.graph_run.metadata)
+        metadata["evolution_plan_id"] = str(plan.id)
+        context.graph_run.metadata = metadata
+        context.graph_run.save(update_fields=["metadata", "updated_at"])
+        return NodeResult("COMPLETE", "success", {"evolution_plan_id": str(plan.id)})
+
+
+class EvolutionMaterializeNode(EvolutionNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        from control_plane.projects.models import EvolutionPlan
+
+        plan = EvolutionPlan.objects.get(id=self.plan_id(context))
+        self.service.approve_plan(plan)
+        project_plan = self.service.materialize_work(plan)
+        return NodeResult("COMPLETE", "success", {"project_plan_id": str(project_plan.id)})
+
+
+class EvolutionExecuteNode(EvolutionNode):
+    destructive = True
+
+    def __init__(self, service: EvolutionService, node_type: str, router=None, test_command: list[str] | None = None) -> None:
+        super().__init__(service, node_type)
+        self.router = router
+        self.test_command = test_command
+
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        from control_plane.projects.models import EvolutionPlan
+
+        if self.router is None:
+            return NodeResult("COMPLETE", "success", {"executed": False})
+        plan = EvolutionPlan.objects.get(id=self.plan_id(context))
+        runs = self.service.execute(plan, self.router, test_command=self.test_command)
+        return NodeResult("COMPLETE", "success", {"graph_run_ids": [str(run.id) for run in runs]})
+
+
+class EvolutionMeasureNode(EvolutionNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        from control_plane.projects.models import EvolutionPlan
+
+        plan = EvolutionPlan.objects.get(id=self.plan_id(context))
+        measurement = self.service.measure_candidate(plan)
+        return NodeResult("COMPLETE", "success", {"measurement": measurement})
+
+
+class EvolutionCompareNode(EvolutionNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        from control_plane.projects.models import EvolutionPlan
+
+        plan = EvolutionPlan.objects.get(id=self.plan_id(context))
+        delta = self.service.compare_baseline(plan)
+        return NodeResult("COMPLETE", "success", {"delta": delta})
+
+
+class EvolutionJudgeNode(EvolutionNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        from control_plane.projects.models import EvolutionPlan
+
+        plan = EvolutionPlan.objects.get(id=self.plan_id(context))
+        verification = self.service.judge_evolution(plan)
+        plan.refresh_from_db()
+        return NodeResult("COMPLETE", plan.verdict or "NOT_IMPROVED", {"verification_id": str(verification.id), "verdict": plan.verdict})
+
+
+class ExplorationNode(LifecycleNode):
+    def __init__(self, service: ExplorerService, node_type: str) -> None:
+        super().__init__(node_type)
+        self.service = service
+
+    def exploration_id(self, context: GraphExecutionContext) -> str:
+        return str(context.graph_run.metadata["exploration_id"])
+
+
+class ExplorationSimpleNode(ExplorationNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        return NodeResult("COMPLETE", "success")
+
+
+class ExplorationGatherNode(ExplorationNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        from control_plane.projects.models import Exploration
+
+        exploration = Exploration.objects.get(id=self.exploration_id(context))
+        exploration.context_snapshot = self.service.project_context(exploration.project)
+        exploration.save(update_fields=["context_snapshot", "updated_at"])
+        return NodeResult("COMPLETE", "success")
+
+
+class ExplorationGenerateNode(ExplorationNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        from control_plane.projects.models import Exploration
+
+        exploration = Exploration.objects.get(id=self.exploration_id(context))
+        opportunities = self.service.generate_opportunities(exploration)
+        return NodeResult("COMPLETE", "success", {"opportunity_count": len(opportunities)})
+
+
+class ExplorationRankNode(ExplorationNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        from control_plane.projects.models import Exploration
+
+        exploration = Exploration.objects.get(id=self.exploration_id(context))
+        ranked = list(exploration.opportunities.order_by("-composite_score").values_list("id", flat=True))
+        exploration.metadata = {**exploration.metadata, "ranked_opportunity_ids": [str(item) for item in ranked]}
+        exploration.status = "COMPLETE"
+        exploration.completed_at = timezone.now()
+        exploration.save(update_fields=["metadata", "status", "completed_at", "updated_at"])
+        return NodeResult("COMPLETE", "success", {"ranked_opportunity_ids": [str(item) for item in ranked]})
+
+
+def extension_registry(service: ExtensionService, router=None, test_command: list[str] | None = None) -> NodeHandlerRegistry:
+    registry = NodeHandlerRegistry()
+    for handler in [ExtensionPrepareNode(service, "extension_prepare"), ExtensionGatherContextNode(service, "extension_gather_context"), ExtensionPlanNode(service, "extension_plan_with_project_brain"), ExtensionValidatePlanNode(service, "extension_validate_plan"), ApprovalNode("extension_await_approval"), ExtensionMaterializeNode(service, "extension_materialize_project_dag"), ExtensionExecuteNode(service, "extension_execute", router, test_command), ExtensionVerifyNode(service, "extension_verify_extension")]:
+        registry.register(handler)
+    return registry
+
+
+def evolution_registry(service: EvolutionService, router=None, test_command: list[str] | None = None) -> NodeHandlerRegistry:
+    registry = NodeHandlerRegistry()
+    for handler in [EvolutionSimpleNode(service, "evolution_prepare"), EvolutionBaselineNode(service, "evolution_establish_baseline"), EvolutionSimpleNode(service, "evolution_form_hypothesis"), EvolutionPlanNode(service, "evolution_plan_with_project_brain"), ApprovalNode("evolution_await_approval"), EvolutionMaterializeNode(service, "evolution_materialize_work"), EvolutionExecuteNode(service, "evolution_execute", router, test_command), EvolutionMeasureNode(service, "evolution_measure_candidate"), EvolutionCompareNode(service, "evolution_compare_baseline"), EvolutionJudgeNode(service, "evolution_judge_evolution")]:
+        registry.register(handler)
+    return registry
+
+
+def exploration_registry(service: ExplorerService) -> NodeHandlerRegistry:
+    registry = NodeHandlerRegistry()
+    for handler in [ExplorationSimpleNode(service, "exploration_prepare"), ExplorationGatherNode(service, "exploration_gather_project_evidence"), ExplorationSimpleNode(service, "exploration_identify_gaps"), ExplorationGenerateNode(service, "exploration_generate_opportunities"), ExplorationSimpleNode(service, "exploration_normalize"), ExplorationSimpleNode(service, "exploration_deduplicate"), ExplorationSimpleNode(service, "exploration_score"), ExplorationRankNode(service, "exploration_rank")]:
+        registry.register(handler)
+    return registry

tests/test_extend_evolve_explore_v1.py

@@ -0,0 +1,239 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+from django.core.exceptions import ValidationError
+
+from agents.lifecycle import EvolutionService, ExplorerService, ExtensionService, LifecycleInspectionService
+from agents.providers import DeterministicCodingProvider, DeterministicSolProvider
+from agents.steward import StewardService
+from control_plane.agents.management.commands.seed_core_agents import Command as SeedAgentsCommand
+from control_plane.projects.models import ExtensionCandidate, ExplorationOpportunity, EvolutionCandidate, Project, StewardFinding, StewardPolicy, Task, TaskStatus
+from graph.bootstrap import champion_project_evolution_graph_v1, champion_project_exploration_graph_v1, champion_project_extension_graph_v1
+from graph.langgraph_runtime import LangGraphRuntime
+from graph.lifecycle import evolution_registry, exploration_registry, extension_registry, project_evolution_graph_v1, project_exploration_graph_v1, project_extension_graph_v1
+from graph.models import GraphApprovalStatus, GraphRun, GraphRunStatus
+from model_router.router import ModelRouter
+from tests.test_m2_autonomous_loop import create_disposable_django_repo
+
+
+def sol_router(payload: dict[str, object]) -> ModelRouter:
+    return ModelRouter({"sol": DeterministicSolProvider(json.dumps(payload))}, persist_requests=True)
+
+
+def qwen_router() -> ModelRouter:
+    SeedAgentsCommand().handle()
+    return ModelRouter({"qwen": DeterministicCodingProvider()}, persist_requests=True)
+
+
+def extension_payload() -> dict[str, object]:
+    return {
+        "extension_plan": {
+            "strategy": "Add a bounded health endpoint to the existing Django app.",
+            "acceptance_criteria": ["/health returns ok", "tests pass"],
+            "project_plan": {
+                "goal": "Keep existing project intent while adding health endpoint.",
+                "scope": "Existing-project extension only.",
+                "acceptance_criteria": ["/health returns ok", "tests pass"],
+                "milestones": [
+                    {
+                        "key": "HEALTH",
+                        "title": "Health Endpoint",
+                        "goal": "Add health endpoint",
+                        "verification_contract": {"type": "extension"},
+                        "features": [
+                            {
+                                "key": "F1",
+                                "title": "Health endpoint",
+                                "description": "Expose /health.",
+                                "acceptance_criteria": ["/health returns ok"],
+                                "tasks": [{"id": "T1", "goal": 'Add a /health endpoint returning JSON {"status": "ok"} and add tests.', "type": "implementation", "acceptance_criteria": ["/health returns ok", "tests pass"], "priority": 50, "dependencies": []}],
+                            }
+                        ],
+                    }
+                ],
+            },
+        }
+    }
+
+
+def evolution_payload(candidate: EvolutionCandidate, *, candidate_value: int = 95) -> dict[str, object]:
+    return {
+        "evolution_plan": {
+            "baseline": candidate.baseline_measurement,
+            "hypothesis": "Health endpoint implementation will keep tests passing while we compare latency.",
+            "intervention": "Add small health endpoint fixture change.",
+            "measurement_method": {"type": "metadata"},
+            "success_threshold": {"metric": "duration", "minimum_improvement_percent": 10},
+            "regression_constraints": ["Deterministic tests pass"],
+            "affected_components": [candidate.target],
+            "candidate_measurement": {"metric": "duration", "duration": candidate_value},
+            "project_plan": extension_payload()["extension_plan"]["project_plan"],
+        }
+    }
+
+
+def project(tmp_path: Path) -> Project:
+    return Project.objects.create(name="Lifecycle", goal="Existing product", repository_path=str(create_disposable_django_repo(tmp_path)))
+
+
+def approve_and_resume(graph_run: GraphRun, registry) -> GraphRun:
+    approval = graph_run.approvals.get()
+    approval.status = GraphApprovalStatus.APPROVED
+    approval.decided_by = "tester"
+    approval.save(update_fields=["status", "decided_by", "updated_at"])
+    graph_run.status = GraphRunStatus.RUNNING
+    graph_run.save(update_fields=["status", "updated_at"])
+    LangGraphRuntime(registry).run_until_terminal_or_paused(graph_run)
+    graph_run.refresh_from_db()
+    return graph_run
+
+
+def test_extension_candidate_plan_graph_execution_verification_and_lineage(tmp_path: Path) -> None:
+    proj = project(tmp_path)
+    service = ExtensionService(sol_router(extension_payload()))
+    candidate = service.create_candidate(proj, title="Add health endpoint", description="Add /health endpoint", source="user", expected_value="Operational health checks", affected_areas=["api"], confidence=0.8)
+    version = champion_project_extension_graph_v1()
+    graph_run = GraphRun.objects.create(execution_graph_version=version, project=proj, current_node=version.graph_spec["entry"], metadata={"extension_candidate_id": str(candidate.id)})
+
+    registry = extension_registry(service, qwen_router(), test_command=["python", "manage.py", "test"])
+    LangGraphRuntime(registry).run_until_terminal_or_paused(graph_run)
+    assert graph_run.status == GraphRunStatus.PAUSED
+
+    approve_and_resume(graph_run, registry)
+
+    candidate.refresh_from_db()
+    plan = candidate.plans.get()
+    task = Task.objects.get(milestone__plan=plan.project_plan)
+    commit = task.commits.get()
+    assert graph_run.status == GraphRunStatus.COMPLETE
+    assert candidate.status == "COMPLETE"
+    assert plan.status == "COMPLETE"
+    assert task.status == TaskStatus.COMPLETE
+    assert commit.extension_candidate == candidate
+    assert plan.verification.result == "PASS"
+
+
+def test_extension_verification_requires_acceptance_not_just_commits(tmp_path: Path) -> None:
+    proj = project(tmp_path)
+    service = ExtensionService()
+    candidate = service.create_candidate(proj, title="Empty extension", description="No acceptance", source="user")
+    plan = service.plan_with_project_brain(candidate)
+    plan.acceptance_criteria = []
+    plan.save(update_fields=["acceptance_criteria", "updated_at"])
+    service.approve_plan(plan)
+    service.materialize_project_dag(plan)
+
+    verification = service.verify_extension(plan)
+
+    assert verification.result == "FAIL"
+    assert candidate.status != "COMPLETE"
+
+
+def test_evolution_requires_baseline() -> None:
+    proj = Project.objects.create(name="No Baseline", goal="Improve existing system")
+    with pytest.raises(ValidationError):
+        EvolutionService().create_candidate(proj, target="api", objective="reduce latency", baseline_measurement={}, desired_direction="DECREASE")
+
+
+def test_evolution_graph_can_reject_technical_pass_as_not_improved(tmp_path: Path) -> None:
+    proj = project(tmp_path)
+    service = EvolutionService()
+    candidate = service.create_candidate(proj, target="api latency", objective="reduce duration", baseline_measurement={"metric": "duration", "duration": 100}, desired_direction="DECREASE", target_measurement={"metric": "duration", "duration": 95}, confidence=0.8)
+    service = EvolutionService(sol_router(evolution_payload(candidate, candidate_value=95)))
+    version = champion_project_evolution_graph_v1()
+    graph_run = GraphRun.objects.create(execution_graph_version=version, project=proj, current_node=version.graph_spec["entry"], metadata={"evolution_candidate_id": str(candidate.id)})
+    registry = evolution_registry(service, qwen_router(), test_command=["python", "manage.py", "test"])
+
+    LangGraphRuntime(registry).run_until_terminal_or_paused(graph_run)
+    assert graph_run.status == GraphRunStatus.PAUSED
+    approve_and_resume(graph_run, registry)
+
+    plan = candidate.plans.get()
+    assert graph_run.status == GraphRunStatus.COMPLETE
+    assert plan.verdict == "NOT_IMPROVED"
+    assert plan.status == "NOT_IMPROVED"
+    assert plan.verification.result == "FAIL"
+    assert Task.objects.filter(milestone__plan=plan.project_plan, status=TaskStatus.COMPLETE).exists()
+
+
+def test_evolution_service_accepts_measurable_success(tmp_path: Path) -> None:
+    proj = project(tmp_path)
+    candidate = EvolutionService().create_candidate(proj, target="analysis", objective="reduce duration", baseline_measurement={"metric": "duration", "duration": 100}, desired_direction="DECREASE", target_measurement={"metric": "duration", "duration": 80})
+    service = EvolutionService(sol_router(evolution_payload(candidate, candidate_value=80)))
+    plan = service.plan_with_project_brain(candidate)
+    plan.candidate_measurement = {"metric": "duration", "duration": 80}
+    plan.save(update_fields=["candidate_measurement", "updated_at"])
+
+    verification = service.judge_evolution(plan)
+
+    assert verification.result == "PASS"
+    assert plan.verdict == "PASS"
+
+
+def test_exploration_graph_persists_scores_dedupes_and_does_not_execute(tmp_path: Path) -> None:
+    proj = project(tmp_path)
+    payload = {
+        "opportunities": [
+            {"title": "Add project dashboard", "description": "Show lifecycle status", "opportunity_type": "FEATURE", "evidence": {"files": ["app/urls.py"]}, "rationale": "visibility", "expected_value": "operators see status", "effort_estimate": "MEDIUM", "value": 0.8, "effort": 0.4, "risk_score": 0.2, "confidence": 0.7, "technical_fit": 0.8, "strategic_fit": 0.9, "recommended_action": "EXTEND"},
+            {"title": "Reduce analysis latency", "description": "Measure and reduce repeated reads", "opportunity_type": "PERFORMANCE", "evidence": ["duration metric"], "rationale": "speed", "expected_value": "faster runs", "effort_estimate": "LOW", "value": "faster repeated reads", "effort": 0.2, "risk_score": 0.2, "confidence": 0.6, "technical_fit": 0.7, "strategic_fit": 0.7, "recommended_action": "EVOLVE"},
+        ]
+    }
+    service = ExplorerService(sol_router(payload))
+    exploration = service.start_exploration(proj)
+    version = champion_project_exploration_graph_v1()
+    graph_run = GraphRun.objects.create(execution_graph_version=version, project=proj, current_node=version.graph_spec["entry"], metadata={"exploration_id": str(exploration.id)})
+
+    LangGraphRuntime(exploration_registry(service)).run_until_terminal_or_paused(graph_run)
+    LangGraphRuntime(exploration_registry(service)).run_until_terminal_or_paused(GraphRun.objects.create(execution_graph_version=version, project=proj, current_node=version.graph_spec["entry"], metadata={"exploration_id": str(service.start_exploration(proj).id)}))
+
+    assert graph_run.status == GraphRunStatus.COMPLETE
+    assert ExplorationOpportunity.objects.filter(project=proj).count() == 2
+    assert Task.objects.filter(project=proj).count() == 0
+    assert all(item.evidence for item in ExplorationOpportunity.objects.filter(project=proj))
+    first = ExplorationOpportunity.objects.filter(project=proj, recommended_action="EXTEND").get()
+    second = ExplorationOpportunity.objects.filter(project=proj, recommended_action="EVOLVE").get()
+    assert second.evidence == {"raw": ["duration metric"]}
+    assert second.value_score == 0.5
+    extension = service.convert_to_extension(first)
+    service.defer(second)
+    assert extension.source_opportunity == first
+    assert first.status == "CONVERTED"
+    assert second.status == "DEFERRED"
+    assert first.composite_score > 0
+
+
+def test_steward_routes_extend_and_evolve_to_project_lifecycle_candidates(tmp_path: Path) -> None:
+    proj = project(tmp_path)
+    policy = StewardPolicy.objects.create(name="Lifecycle Steward", enabled_checks=[], approval_requirements={"EXTEND": "CRITICAL", "EVOLVE": "CRITICAL"})
+    service = StewardService()
+    service.enroll_project(proj, policy)
+    steward_run = service.start_run(proj.steward_enrollments.get())
+    extend = StewardFinding.objects.create(project=proj, steward_run=steward_run, finding_type="PRODUCT_OPPORTUNITY", title="Add dashboard", summary="Dashboard missing", severity="MEDIUM", confidence=0.8, recommended_action="EXTEND", grouping_key="extend")
+    evolve = StewardFinding.objects.create(project=proj, steward_run=steward_run, finding_type="PERFORMANCE_REGRESSION", title="Latency high", summary="Reduce p95 latency", severity="MEDIUM", confidence=0.8, recommended_action="EVOLVE", evidence={"baseline_measurement": {"metric": "latency", "latency": 100}}, grouping_key="evolve")
+
+    extend_action = service.route_finding(extend, policy)
+    evolve_action = service.route_finding(evolve, policy)
+
+    assert extend_action.extension_candidate is not None
+    assert evolve_action.evolution_candidate is not None
+    assert ExtensionCandidate.objects.filter(source_steward_finding=extend).exists()
+    assert EvolutionCandidate.objects.filter(source_steward_finding=evolve).exists()
+
+
+def test_lifecycle_inspection_exposes_machine_readable_view(tmp_path: Path) -> None:
+    proj = project(tmp_path)
+    ExplorerService().generate_opportunities(ExplorerService().start_exploration(proj))
+
+    view = LifecycleInspectionService().project_lifecycle_view(proj)
+
+    assert set(view) == {"project_id", "repairs", "extensions", "evolutions", "explorations"}
+    assert view["explorations"][0]["opportunities"]
+
+
+def test_graph_specs_are_serializable_and_distinct() -> None:
+    assert project_extension_graph_v1().name == "project_extension"
+    assert project_evolution_graph_v1().name == "project_evolution"
+    assert project_exploration_graph_v1().name == "project_exploration"