from __future__ import annotations import hashlib import json from typing import Any from control_plane.agents.models import ProgenySignal from control_plane.events.bus import EventBus from control_plane.projects.models import ( Decision, EvolutionCandidate, ExtensionCandidate, ExplorationOpportunity, Project, RoadmapHorizon, RoadmapItem, RoadmapStatus, RoadmapTargetAction, StewardFinding, ) from agents.lifecycle import EvolutionService, ExtensionService, ProjectContextMixin from agents.progeny import ProgenyService from model_router.router import ModelCapability, ModelRequestContract, ModelRouter class RoadmapService(ProjectContextMixin): def __init__(self, router: ModelRouter | None = None, bus: EventBus | None = None) -> None: self.router = router self.bus = bus or EventBus() def upsert_item(self, project: Project, *, title: str, description: str = "", source: str = "USER", source_ref: dict[str, object] | None = None, rationale: str = "", evidence: dict[str, object] | None = None, horizon: str = RoadmapHorizon.EXPLORING, category: str = "", target_action: str = RoadmapTargetAction.NONE, scores: dict[str, object] | None = None, status: str = RoadmapStatus.PROPOSED) -> RoadmapItem: grouping_key = self._grouping_key(project, title, category or target_action) existing = self._find_existing(project, grouping_key, title) values = self.score(scores or {}) if existing: metadata = dict(existing.metadata) metadata["occurrences"] = int(metadata.get("occurrences", 1)) + 1 metadata.setdefault("reinforced_by", []).append({"source": source, "source_ref": source_ref or {}}) existing.evidence = self._merge_evidence(existing.evidence, evidence or {}) existing.source_ref = self._merge_evidence(existing.source_ref, source_ref or {}) existing.confidence = min(1.0, max(existing.confidence, values["confidence"]) + 0.05) existing.composite_score = self._composite(existing) existing.metadata = metadata existing.save(update_fields=["evidence", "source_ref", "confidence", "composite_score", "metadata", "updated_at"]) self.bus.publish("ROADMAP_ITEM_UPDATED", project=project, payload={"roadmap_item_id": str(existing.id), "reason": "deduplicated_reinforcement"}) return existing item = RoadmapItem.objects.create(project=project, title=title, description=description, source=source, source_ref=source_ref or {}, rationale=rationale, evidence=evidence or {}, horizon=horizon, category=category, status=status, target_action=target_action, grouping_key=grouping_key, value_score=values["value"], effort_score=values["effort"], risk_score=values["risk"], confidence=values["confidence"], strategic_fit=values["strategic_fit"], technical_fit=values["technical_fit"], urgency=values["urgency"]) item.composite_score = self._composite(item) item.priority = max(1, min(100, int(item.composite_score * 100))) item.save(update_fields=["composite_score", "priority", "updated_at"]) self.bus.publish("ROADMAP_ITEM_CREATED", project=project, payload={"roadmap_item_id": str(item.id), "source": source}) return item def gather_candidate_items(self, project: Project) -> list[RoadmapItem]: items: list[RoadmapItem] = [] for opportunity in project.exploration_opportunities.exclude(status="REJECTED"): items.append(self.upsert_item(project, title=opportunity.title, description=opportunity.description, source="EXPLORE", source_ref={"exploration_opportunity_id": str(opportunity.id)}, rationale=opportunity.rationale, evidence=opportunity.evidence, horizon=RoadmapHorizon.EXPLORING, category=opportunity.opportunity_type, target_action=opportunity.recommended_action if opportunity.recommended_action in RoadmapTargetAction.values else RoadmapTargetAction.NONE, scores={"value": opportunity.value_score, "effort": opportunity.effort_score, "risk": opportunity.risk_score, "confidence": opportunity.confidence, "strategic_fit": opportunity.strategic_fit, "technical_fit": opportunity.technical_fit})) for finding in project.steward_findings.exclude(status__in=["RESOLVED", "DISMISSED"]): action = finding.recommended_action if finding.recommended_action in RoadmapTargetAction.values else RoadmapTargetAction.INVESTIGATE items.append(self.upsert_item(project, title=finding.title, description=finding.summary, source="STEWARD", source_ref={"steward_finding_id": str(finding.id)}, rationale="Steward surfaced this future project intent.", evidence=finding.evidence, horizon=RoadmapHorizon.EXPLORING, category=finding.finding_type, target_action=action, scores={"confidence": finding.confidence, "risk": 0.7 if finding.severity in ["HIGH", "CRITICAL"] else 0.4, "urgency": 0.8 if finding.severity in ["HIGH", "CRITICAL"] else 0.4})) return items def review_with_project_brain(self, project: Project) -> dict[str, object]: fallback = {"recommendations": []} if self.router is None: return fallback payload = {"context": self.project_context(project), "roadmap_items": list(project.roadmap_items.values("id", "title", "description", "horizon", "status", "target_action", "value_score", "effort_score", "risk_score", "confidence", "strategic_fit", "technical_fit", "urgency", "composite_score"))} try: response = self.router.complete(ModelRequestContract(purpose=ModelCapability.PLANNING, model_hint="sol", project=project, prompt="Review this existing project roadmap. Return JSON with recommendations: item_id, recommendation, rationale, optional horizon/status. Do not execute work.\n" + json.dumps(payload, default=str))) parsed = json.loads(response.content) return parsed if isinstance(parsed, dict) else fallback except Exception as exc: ProgenySignal.objects.create(project=project, source="roadmap", severity="MEDIUM", failure_category="ROADMAP_PRIORITIZATION", summary="Roadmap Project Brain review failed; deterministic scoring retained.", evidence={"error": str(exc)}, grouping_key=f"roadmap:{project.id}:prioritization") return fallback def apply_recommendations(self, project: Project, recommendations: dict[str, object]) -> list[RoadmapItem]: updated: list[RoadmapItem] = [] for rec in recommendations.get("recommendations", []): if not isinstance(rec, dict): continue item_id = rec.get("item_id") try: item = project.roadmap_items.get(id=item_id) except Exception: continue item.metadata = {**item.metadata, "project_brain_recommendations": [*item.metadata.get("project_brain_recommendations", []), rec]} if rec.get("horizon") in RoadmapHorizon.values: item.horizon = str(rec["horizon"]) if rec.get("status") in RoadmapStatus.values: item.status = str(rec["status"]) item.save(update_fields=["horizon", "status", "metadata", "updated_at"]) updated.append(item) self.bus.publish("ROADMAP_ITEM_UPDATED", project=project, payload={"roadmap_item_id": str(item.id), "recommendation": rec}) return updated def review_project_roadmap(self, project: Project) -> dict[str, object]: self.gather_candidate_items(project) recommendations = self.review_with_project_brain(project) self.apply_recommendations(project, recommendations) self.bus.publish("ROADMAP_REVIEW_COMPLETED", project=project, payload={"item_count": project.roadmap_items.count(), "recommendations": recommendations}) return self.project_roadmap_view(project) def convert_to_extension(self, item: RoadmapItem) -> ExtensionCandidate: candidate = ExtensionService(bus=self.bus).create_candidate(item.project, title=item.title, description=item.description, rationale=item.rationale, source="RoadmapItem", expected_value=str(item.evidence.get("expected_value", "")), affected_areas=[item.category] if item.category else [], risk=str(item.risk_score), confidence=item.confidence, evidence={"roadmap_item_id": str(item.id), **item.evidence}, source_roadmap_item=item) item.converted_extension = candidate item.status = RoadmapStatus.PLANNING item.save(update_fields=["converted_extension", "status", "updated_at"]) self.bus.publish("ROADMAP_ITEM_CONVERTED", project=item.project, payload={"roadmap_item_id": str(item.id), "extension_candidate_id": str(candidate.id)}) return candidate def convert_to_evolution(self, item: RoadmapItem, *, baseline_measurement: dict[str, object], desired_direction: str = "DECREASE") -> EvolutionCandidate: candidate = EvolutionService(bus=self.bus).create_candidate(item.project, target=item.category or item.title, objective=item.description or item.title, baseline_measurement=baseline_measurement, desired_direction=desired_direction, rationale=item.rationale, source="RoadmapItem", evidence={"roadmap_item_id": str(item.id), **item.evidence}, risk=str(item.risk_score), confidence=item.confidence, source_roadmap_item=item) item.converted_evolution = candidate item.status = RoadmapStatus.PLANNING item.save(update_fields=["converted_evolution", "status", "updated_at"]) self.bus.publish("ROADMAP_ITEM_CONVERTED", project=item.project, payload={"roadmap_item_id": str(item.id), "evolution_candidate_id": str(candidate.id)}) return candidate def convert_to_investigation(self, item: RoadmapItem): signal = ProgenySignal.objects.create(project=item.project, source="roadmap", severity="MEDIUM", failure_category=item.category or "ROADMAP_INVESTIGATION", summary=item.description or item.title, evidence={"roadmap_item_id": str(item.id), **item.evidence}, grouping_key=f"roadmap:{item.grouping_key}"[:120]) investigation = ProgenyService(self.bus).create_smart_investigation(signal.grouping_key) item.converted_investigation = investigation item.status = RoadmapStatus.PLANNING item.save(update_fields=["converted_investigation", "status", "updated_at"]) self.bus.publish("ROADMAP_ITEM_CONVERTED", project=item.project, payload={"roadmap_item_id": str(item.id), "investigation_id": str(investigation.id)}) return investigation def project_roadmap_view(self, project: Project) -> dict[str, object]: return {horizon: [self._item(item) for item in project.roadmap_items.filter(horizon=horizon).order_by("-composite_score", "-priority", "created_at")] for horizon in RoadmapHorizon.values} def score(self, raw: dict[str, object]) -> dict[str, float]: return {key: self._score_value(raw.get(key, 0.5)) for key in ["value", "effort", "risk", "confidence", "strategic_fit", "technical_fit", "urgency"]} 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 _composite(self, item: RoadmapItem) -> float: return (item.value_score * 0.25) + ((1 - item.effort_score) * 0.12) + ((1 - item.risk_score) * 0.12) + (item.confidence * 0.14) + (item.strategic_fit * 0.14) + (item.technical_fit * 0.11) + (item.urgency * 0.12) def _grouping_key(self, project: Project, title: str, category: str) -> str: fingerprint = hashlib.sha256(f"{title.lower()}:{category.lower()}".encode("utf-8")).hexdigest()[:16] return f"{project.id}:roadmap:{fingerprint}"[:240] def _find_existing(self, project: Project, grouping_key: str, title: str) -> RoadmapItem | None: existing = project.roadmap_items.filter(grouping_key=grouping_key).first() or project.roadmap_items.filter(title__iexact=title).first() if existing: return existing if project.exploration_opportunities.filter(title__iexact=title).exists() or ExtensionCandidate.objects.filter(project=project, title__iexact=title).exists() or EvolutionCandidate.objects.filter(project=project, objective__icontains=title[:80]).exists() or StewardFinding.objects.filter(project=project, title__iexact=title).exists(): return project.roadmap_items.filter(title__iexact=title).first() if Decision.objects.filter(project=project, decision__icontains=title[:80], decision_type__in=["REJECTED", "DEFERRED"]).exists(): return project.roadmap_items.filter(title__iexact=title).first() return None def _merge_evidence(self, current: dict[str, object], incoming: dict[str, object]) -> dict[str, object]: merged = dict(current or {}) for key, value in incoming.items(): if key in merged and merged[key] != value: merged[key] = [merged[key], value] else: merged[key] = value return merged def _item(self, item: RoadmapItem) -> dict[str, object]: return {"id": str(item.id), "title": item.title, "source": item.source, "rationale": item.rationale, "evidence": item.evidence, "scores": {"value": item.value_score, "effort": item.effort_score, "risk": item.risk_score, "confidence": item.confidence, "strategic_fit": item.strategic_fit, "technical_fit": item.technical_fit, "urgency": item.urgency, "composite": item.composite_score}, "status": item.status, "target_action": item.target_action, "dependencies": [str(dep.id) for dep in item.dependencies.all()], "related_items": [str(rel.id) for rel in item.related_items.all()], "conversion_lineage": {"extension_candidate_id": str(item.converted_extension_id) if item.converted_extension_id else None, "evolution_candidate_id": str(item.converted_evolution_id) if item.converted_evolution_id else None, "investigation_id": str(item.converted_investigation_id) if item.converted_investigation_id else None}, "metadata": item.metadata}