Add control plane UI and venture discovery V0

agents/venture_discovery.py

@@ -0,0 +1,273 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from decimal import Decimal
+from typing import Any
+
+from django.utils import timezone
+
+from control_plane.events.bus import EventBus
+from control_plane.ventures.models import CapabilityPriority, CapabilityStatus, CompanyBoardReview, CompanyCapabilityRequirement, CompanyMandate, CompanyProposal, CompanyProposalStatus, ICDecision, ICDecisionType, ICDiligence, ICQuestion, ICResponse, PortfolioCapabilityGap, VentureArtifact, VentureThesis
+from model_router.router import ModelCapability, ModelRequestContract, ModelRouter
+
+
+PITCH_SECTIONS = ["Company name", "One-line thesis", "Problem", "ICP", "Why now", "Product / service", "Business model", "Pricing", "Route to first customer", "Validation plan", "$50 capital allocation proposal", "Time to first dollar", "Path to $500 net cash", "Competition", "Differentiation", "Build requirements", "Distribution requirements", "Risks", "What would falsify the thesis", "Confidence"]
+SCORE_DIMENSIONS = ["Demand evidence", "Time to first dollar", "Capital efficiency", "Validation cost", "Gross margin", "Distribution difficulty", "Build complexity", "Defensibility", "Market size", "Competition", "Risk", "Probability of reaching $500"]
+
+
+class VentureDiscoveryService:
+    def __init__(self, router: ModelRouter | None = None, bus: EventBus | None = None, web_research_available: bool = False) -> None:
+        self.router = router
+        self.bus = bus or EventBus()
+        self.web_research_available = web_research_available
+
+    def create_v0_mandate(self) -> CompanyMandate:
+        mandate = CompanyMandate.objects.create(
+            objective="Design a business that could plausibly turn at most $50 of external validation capital into $500 net new cash.",
+            constraints={"external_validation_capital_max": 50, "target_net_new_cash": 500, "target_window_days": 30, "no_equity_raise": True, "no_debt": True, "no_illegal_or_deceptive_activity": True, "no_spam": True, "no_fake_traction": True, "no_fabricated_customer_evidence": True, "no_real_spend_in_v0": True, "no_real_customer_outreach_in_v0": True, "existing_artifex_compute_sunk_available": True},
+            optimization_targets=["time to first dollar", "capital efficiency", "real demand evidence", "high gross margin", "realistic execution", "low external dependency", "ability to validate cheaply"],
+            metadata={"milestone": "VENTURE_DISCOVERY_V0", "spend_authorized": False, "customer_outreach_authorized": False},
+        )
+        self._artifact(None, mandate, "VENTURE_MANDATE", "Venture Discovery V0 Mandate", {"objective": mandate.objective, "constraints": mandate.constraints, "optimization_targets": mandate.optimization_targets}, self._readable_mandate(mandate), "venture_discovery")
+        return mandate
+
+    def generate_single_company(self, mandate: CompanyMandate, *, graph_run=None) -> CompanyProposal:
+        payload, source = self._company_payload(mandate)
+        pitch = self._pitch(payload, fallback=source != "sol")
+        confidence = float(payload.get("confidence", 0.55))
+        evidence = self._as_list(payload.get("market_evidence", []))
+        if not self.web_research_available:
+            evidence.append({"type": "capability_gap", "source": "internal", "summary": "Public web market research is not configured; demand/competitor evidence is unverified.", "fallback_evidence": True})
+            confidence = min(confidence, 0.58)
+        thesis = VentureThesis.objects.create(mandate=mandate, title=str(payload["title"]), thesis=str(payload["one_line_thesis"]), similarity_fingerprint=self._fingerprint(payload), metadata={"source": source, "exactly_one_company_generated": True})
+        proposal = CompanyProposal.objects.create(
+            mandate=mandate,
+            thesis=thesis,
+            title=str(payload["title"]),
+            description=str(payload["description"]),
+            problem=str(payload["problem"]),
+            target_customer=str(payload["target_customer"]),
+            proposed_solution=str(payload["proposed_solution"]),
+            business_model=str(payload["business_model"]),
+            pricing_hypothesis=str(payload["pricing_hypothesis"]),
+            acquisition_strategy=str(payload["acquisition_strategy"]),
+            validation_plan=str(payload["validation_plan"]),
+            capital_requested=self._money(payload.get("capital_requested", "50")),
+            time_to_first_dollar_estimate=str(payload["time_to_first_dollar_estimate"]),
+            expected_margin=str(payload["expected_margin"]),
+            build_complexity=str(payload["build_complexity"]),
+            market_evidence=evidence,
+            differentiation=str(payload["differentiation"]),
+            major_risks=self._as_list(payload["major_risks"]),
+            confidence=confidence,
+            status=CompanyProposalStatus.SUBMITTED,
+            pitch=pitch,
+            metadata={"generation_source": source, "fallback_evidence": source != "sol", "web_research_available": self.web_research_available, "real_spend": 0, "real_customer_outreach": False},
+        )
+        self._artifact(proposal, mandate, "STANDARDIZED_COMPANY_PITCH", "Standardized Company Pitch", pitch, self.readable_pitch(pitch), "Company Brain/Sol" if source == "sol" else "deterministic_fallback", graph_run=graph_run)
+        self.bus.publish("VENTURE_COMPANY_PROPOSED", payload={"proposal_id": str(proposal.id), "source": source})
+        return proposal
+
+    def board_review(self, proposal: CompanyProposal, *, graph_run=None) -> CompanyBoardReview:
+        observations = {
+            "CEO": ["Mandate fit is strongest if validation sells a narrow paid audit before product build.", "Keep the first dollar path service-led, not SaaS-led."],
+            "CTO": ["Build can use existing Artifex analysis, reporting, and frontend capabilities.", "Avoid integrations until willingness-to-pay evidence exists."],
+            "CFO": ["$50 cap is adequate only for lightweight landing page/listing tests, not paid acquisition learning.", "High margin is plausible because delivery is mostly labor/compute already available."],
+            "CRO": ["Founder/operator communities are reachable manually, but V0 cannot contact customers.", "Pricing must start as a paid diagnostic to avoid long SaaS evaluation cycles."],
+            "Independent Director": ["Demand evidence is weak without public research or customer conversations.", "The company must prove urgency before building automation."],
+        }
+        weaknesses = ["No customer outreach or paid test has occurred.", "Public web research is unavailable, so market evidence remains partial.", "First customers may require trust and examples before paying."]
+        revisions = ["Frame offer as a productized validation audit with optional Artifex-assisted build plan.", "Make falsification criteria explicit before spend."]
+        pitch = dict(proposal.pitch)
+        pitch["Product / service"] = "Productized validation and launch-readiness audit for AI-assisted microbusiness ideas, delivered as a concise paid report before any software build."
+        pitch["What would falsify the thesis"] = "Fewer than 5 credible target-customer responses or zero willingness-to-pay signals after a compliant validation test."
+        proposal.pitch = pitch
+        proposal.metadata = {**proposal.metadata, "board_revised_pitch": True}
+        proposal.save(update_fields=["pitch", "metadata", "updated_at"])
+        review = CompanyBoardReview.objects.create(proposal=proposal, observations=observations, strengths=["Service-led revenue path can precede product build.", "Uses current Artifex planning, engineering, and review capabilities.", "Small validation budget aligns with a narrow paid offer."], weaknesses=weaknesses, key_assumptions=["Target customers feel enough urgency to pay for validation clarity.", "Manual outbound or community posting can generate credible responses without spam.", "Artifex can produce a differentiated audit faster than generic consultants."], required_revisions=revisions, recommendation="PROCEED_TO_IC_WITH_REVISIONS", revised_pitch=pitch, metadata={"roles": list(observations), "company_does_not_grade_itself": True})
+        self._artifact(proposal, proposal.mandate, "COMPANY_BOARD_REVIEW", "Company Board Review", {"observations": observations, "strengths": review.strengths, "weaknesses": weaknesses, "required_revisions": revisions, "recommendation": review.recommendation, "revised_pitch": pitch}, self._readable_board(review), "Company Board", graph_run=graph_run)
+        return review
+
+    def start_ic_diligence(self, proposal: CompanyProposal, *, graph_run=None) -> ICDiligence:
+        proposal.status = CompanyProposalStatus.UNDER_DILIGENCE
+        proposal.save(update_fields=["status", "updated_at"])
+        diligence = ICDiligence.objects.create(proposal=proposal, status="FIRST_PASS", rounds=["Initial Pitch", "Diligence Round 1", "Final Challenge", "Decision"], metadata={"bounded_rounds": True, "self_grading": False})
+        self._artifact(proposal, proposal.mandate, "IC_FIRST_PASS", "IC First-Pass Review", {"status": diligence.status, "initial_concerns": ["Demand evidence is unverified.", "Distribution assumptions need evidence.", "Need validation gate before any spend."]}, "IC first pass: proceed to evidence-seeking questions; do not score final decision yet.", "Independent IC", graph_run=graph_run)
+        return diligence
+
+    def generate_ic_questions(self, diligence: ICDiligence, *, graph_run=None) -> list[ICQuestion]:
+        pitch = diligence.proposal.pitch
+        raw_questions = self._questions_from_pitch(pitch)
+        questions = [ICQuestion.objects.create(diligence=diligence, question=item["question"], category=item["category"], evidence_required=True) for item in raw_questions[:8]]
+        self._artifact(diligence.proposal, diligence.proposal.mandate, "IC_QUESTIONS", "IC Diligence Questions", {"questions": [q.question for q in questions]}, "\n".join(f"- {q.question}" for q in questions), "Independent IC", graph_run=graph_run)
+        return questions
+
+    def answer_questions(self, diligence: ICDiligence, *, graph_run=None) -> list[ICResponse]:
+        responses = []
+        for question in diligence.questions.all():
+            answer = self._answer_question(question)
+            responses.append(ICResponse.objects.create(question=question, answer=answer["answer"], evidence=answer["evidence"], uncertainty=answer["uncertainty"], pitch_changes=answer.get("pitch_changes", {}), metadata={"no_customer_outreach": True, "no_spend": True}))
+        diligence.status = "COMPANY_RESPONSE"
+        diligence.save(update_fields=["status", "updated_at"])
+        self._artifact(diligence.proposal, diligence.proposal.mandate, "IC_RESPONSES", "Company Responses to IC", {"responses": [{"question": r.question.question, "answer": r.answer, "evidence": r.evidence, "uncertainty": r.uncertainty, "pitch_changes": r.pitch_changes} for r in responses]}, self._readable_responses(responses), "Company reasoning roles", graph_run=graph_run)
+        return responses
+
+    def red_team(self, diligence: ICDiligence, *, graph_run=None) -> dict[str, object]:
+        challenge = {"concerns": ["The offer may be perceived as generic consulting unless anchored to a painful, immediate decision.", "Without public research or customer contact, demand remains a hypothesis.", "Manual distribution could fail if the target audience distrusts AI-generated audits."], "required_final_response": ["Narrow ICP further.", "Specify willingness-to-pay proof.", "Define hard kill criteria."], "recommendation": "continue_to_final_response"}
+        diligence.red_team_challenge = challenge
+        diligence.status = "FINAL_CHALLENGE"
+        diligence.save(update_fields=["red_team_challenge", "status", "updated_at"])
+        self._artifact(diligence.proposal, diligence.proposal.mandate, "IC_RED_TEAM", "IC Red-Team Challenge", challenge, "Red-team concerns:\n" + "\n".join(f"- {c}" for c in challenge["concerns"]), "Independent IC Red Team", graph_run=graph_run)
+        return challenge
+
+    def final_company_response(self, diligence: ICDiligence, *, graph_run=None) -> dict[str, object]:
+        response = {"narrowed_icp": "Solo technical founders and small service operators deciding whether to spend time building an AI-assisted microbusiness.", "revised_validation_gate": "Before any spend, collect 5 credible target-customer responses or 1 explicit willingness-to-pay signal through compliant non-spam channels.", "kill_criteria": ["No credible responses after 10 targeted, compliant conversations/posts once outreach is approved.", "No willingness-to-pay signal at $49-$99.", "Customers only want free advice, not a paid report."], "pitch_changes": {"ICP": "Narrowed to solo technical founders and small service operators.", "Validation plan": "Gate spend behind credible response/willingness-to-pay evidence."}}
+        diligence.final_response = response
+        diligence.status = "FINAL_RESPONSE"
+        diligence.save(update_fields=["final_response", "status", "updated_at"])
+        self._artifact(diligence.proposal, diligence.proposal.mandate, "IC_FINAL_RESPONSE", "Final Company Response", response, json.dumps(response, indent=2), "Company reasoning roles", graph_run=graph_run)
+        return response
+
+    def score_and_decide(self, diligence: ICDiligence, *, graph_run=None) -> ICDecision:
+        scores = {"Demand evidence": 38, "Time to first dollar": 78, "Capital efficiency": 84, "Validation cost": 82, "Gross margin": 76, "Distribution difficulty": 48, "Build complexity": 72, "Defensibility": 34, "Market size": 58, "Competition": 44, "Risk": 42, "Probability of reaching $500": 45}
+        if not self.web_research_available:
+            scores["Demand evidence"] = 30
+            scores["Competition"] = 36
+            scores["Probability of reaching $500"] = 40
+        composite = round(sum(scores.values()) / len(scores), 1)
+        decision_type = ICDecisionType.CONDITIONAL_FUND if composite >= 55 else ICDecisionType.REVISE_AND_RESUBMIT
+        decision = ICDecision.objects.create(diligence=diligence, decision=decision_type, component_scores=scores, composite_score=composite, probability_500_within_30_days=scores["Probability of reaching $500"], initial_tranche=Decimal("10.00") if decision_type == ICDecisionType.CONDITIONAL_FUND else None, validation_condition="Obtain 5 credible target-customer responses or 1 explicit willingness-to-pay signal before any build or further spend.", evidence_required=["response transcripts or public thread URLs", "proof of willingness-to-pay signal", "no-spam/no-fabrication compliance note"], recommended_allocation={"initial_tranche": 10, "remaining_reserved": 40, "no_spend_in_v0": True}, kill_criteria=diligence.final_response.get("kill_criteria", []), next_decision_point="After validation evidence is collected and before any real spend or customer delivery.", metadata={"decision_vocabulary": [item.value for item in ICDecisionType], "no_actual_funding": True})
+        diligence.status = "DECISION"
+        diligence.save(update_fields=["status", "updated_at"])
+        diligence.proposal.status = CompanyProposalStatus.FUNDED_RECOMMENDED if decision.decision == ICDecisionType.CONDITIONAL_FUND else CompanyProposalStatus.REVISE
+        diligence.proposal.save(update_fields=["status", "updated_at"])
+        self._artifact(diligence.proposal, diligence.proposal.mandate, "IC_FINAL_SCORE", "IC Final Scoring and Decision", {"scores": scores, "decision": decision.decision, "composite_score": composite, "conditional_funding": {"initial_tranche": str(decision.initial_tranche), "condition": decision.validation_condition}}, self._readable_score(decision), "Independent IC", graph_run=graph_run)
+        return decision
+
+    def capability_analysis(self, proposal: CompanyProposal, *, graph_run=None) -> PortfolioCapabilityGap:
+        requirements = self._capability_requirements(proposal)
+        for item in requirements:
+            CompanyCapabilityRequirement.objects.create(proposal=proposal, **item)
+        available = [r["category"] for r in requirements if r["status"] == CapabilityStatus.AVAILABLE]
+        partial = [r["category"] for r in requirements if r["status"] == CapabilityStatus.PARTIAL]
+        missing = [r["category"] for r in requirements if r["status"] == CapabilityStatus.MISSING]
+        ranked = [{"category": r["category"], "priority": r["priority"], "rationale": r["rationale"]} for r in requirements if r["status"] == CapabilityStatus.MISSING]
+        priority_order = {CapabilityPriority.BEFORE_VALIDATION: 0, CapabilityPriority.BEFORE_FIRST_CUSTOMER: 1, CapabilityPriority.BEFORE_SCALING: 2}
+        ranked.sort(key=lambda item: priority_order[item["priority"]])
+        report = self._readable_capability_gap(available, partial, missing, ranked)
+        gap = PortfolioCapabilityGap.objects.create(proposal=proposal, available=available, partial=partial, missing=missing, ranked_missing=ranked, report=report, metadata={"web_market_research_status": "MISSING" if not self.web_research_available else "PARTIAL"})
+        self._artifact(proposal, proposal.mandate, "CAPABILITY_GAP_REPORT", "Capability Gap Report", {"available": available, "partial": partial, "missing": missing, "ranked_missing": ranked}, report, "Venture Discovery Capability Analysis", graph_run=graph_run)
+        return gap
+
+    def produce_investment_memo(self, diligence: ICDiligence, gap: PortfolioCapabilityGap, *, graph_run=None) -> VentureArtifact:
+        decision = diligence.decision
+        proposal = diligence.proposal
+        memo = {"Company": proposal.title, "Thesis": proposal.pitch["One-line thesis"], "Mandate": proposal.mandate.objective, "Requested capital": str(proposal.capital_requested), "Recommended allocation": decision.recommended_allocation, "IC decision": decision.decision, "Key metrics": {"P($500 within 30 days)": decision.probability_500_within_30_days, "estimated time to first dollar": proposal.time_to_first_dollar_estimate, "expected gross margin": proposal.expected_margin, "validation cost": "$10 initial tranche; $50 maximum after approval", "build effort": proposal.build_complexity, "distribution difficulty": decision.component_scores["Distribution difficulty"]}, "Why it may work": ["Revenue path starts with a paid diagnostic, not a full SaaS build.", "Artifex has planning, engineering, frontend, review, graph, and agent-control capabilities already.", "Validation budget can be gated behind evidence."], "Why it may fail": diligence.red_team_challenge.get("concerns", []), "Diligence questions": [q.question for q in diligence.questions.all()], "Company responses": [r.answer for r in ICResponse.objects.filter(question__diligence=diligence)], "Red-team concerns": diligence.red_team_challenge.get("concerns", []), "IC scoring": decision.component_scores, "Capital recommendation": decision.recommended_allocation, "Validation gates": [decision.validation_condition], "Kill criteria": decision.kill_criteria, "Next decision point": decision.next_decision_point, "Capability gap": {"available": gap.available, "partial": gap.partial, "missing": gap.missing}}
+        return self._artifact(proposal, proposal.mandate, "FINAL_INVESTMENT_MEMO", "Final IC Investment Memo", memo, self._readable_memo(memo), "Independent IC", graph_run=graph_run)
+
+    def request_human_approval(self, decision: ICDecision, action: str) -> ICDecision:
+        if action not in {"approve_for_validation", "reject", "request_more_diligence"}:
+            raise ValueError("Unsupported venture approval action")
+        decision.metadata = {**decision.metadata, "human_approval_action": action, "real_spend_still_blocked": True}
+        decision.save(update_fields=["metadata", "updated_at"])
+        return decision
+
+    def _company_payload(self, mandate: CompanyMandate) -> tuple[dict[str, Any], str]:
+        if self.router is not None:
+            try:
+                response = self.router.complete(ModelRequestContract(purpose=ModelCapability.PLANNING, model_hint="sol", prompt="Generate exactly ONE startup idea for Venture Discovery V0. Return a single JSON object, not a list. Respect no spend and no outreach in V0. Include title, one_line_thesis, description, problem, target_customer, proposed_solution, business_model, pricing_hypothesis, acquisition_strategy, validation_plan, capital_requested, time_to_first_dollar_estimate, expected_margin, build_complexity, market_evidence, differentiation, major_risks, confidence. Mandate: " + json.dumps({"objective": mandate.objective, "constraints": mandate.constraints, "optimization_targets": mandate.optimization_targets})))
+                parsed = json.loads(response.content)
+                if isinstance(parsed, dict) and parsed.get("title"):
+                    return self._normalize_payload(parsed), "sol"
+            except Exception:
+                pass
+        payload = self._fallback_company_payload()
+        payload["market_evidence"] = [{"type": "fallback_hypothesis", "source": "deterministic_fallback", "summary": "No Sol/web research was used; this is an internal hypothesis for test/resilience only.", "fallback_evidence": True}]
+        payload["confidence"] = 0.52
+        return payload, "deterministic_fallback"
+
+    def _fallback_company_payload(self) -> dict[str, Any]:
+        return {"title": "LaunchLens", "one_line_thesis": "A productized validation audit helps solo builders decide whether an AI-enabled microbusiness is worth pursuing before they spend weeks building.", "description": "LaunchLens sells a concise validation and launch-readiness report for one microbusiness idea.", "problem": "Solo technical founders often overbuild AI products before proving willingness to pay.", "target_customer": "Solo technical founders and small service operators considering an AI-assisted microbusiness.", "proposed_solution": "A fixed-scope paid audit that evaluates ICP, first-dollar route, validation gates, build plan, risks, and Artifex capability gaps.", "business_model": "Productized service first, with optional later software tooling if demand is proven.", "pricing_hypothesis": "$49-$99 per audit, with a higher-touch $250 implementation planning upsell after validation.", "acquisition_strategy": "Compliant founder-community posts, personal network asks, and content showing anonymized example audits after outreach is approved.", "validation_plan": "Before any build, seek 5 credible target-customer responses or 1 willingness-to-pay signal through compliant channels once approved.", "capital_requested": "50", "time_to_first_dollar_estimate": "3-10 days after outreach is approved", "expected_margin": "70-85% gross margin after manual delivery time", "build_complexity": "LOW", "market_evidence": [], "differentiation": "Combines venture IC-style diligence with Artifex's software/agent execution awareness and explicit capability-gap reporting.", "major_risks": ["Demand may be consulting-like and hard to differentiate.", "Manual distribution may not produce urgent buyers.", "No V0 customer evidence exists yet."], "confidence": 0.52}
+
+    def _normalize_payload(self, payload: dict[str, Any]) -> dict[str, Any]:
+        fallback = self._fallback_company_payload()
+        normalized = {key: payload.get(key, value) for key, value in fallback.items()}
+        normalized["market_evidence"] = self._as_list(normalized.get("market_evidence"))
+        normalized["major_risks"] = self._as_list(normalized.get("major_risks"))
+        return normalized
+
+    def _as_list(self, value: Any) -> list[Any]:
+        if isinstance(value, list):
+            return value
+        if value in (None, ""):
+            return []
+        return [value]
+
+    def _money(self, value: Any) -> Decimal:
+        match = re.search(r"\d+(?:\.\d+)?", str(value))
+        if match is None:
+            return Decimal("50")
+        return min(Decimal(match.group(0)), Decimal("50"))
+
+    def _pitch(self, payload: dict[str, Any], *, fallback: bool) -> dict[str, Any]:
+        pitch = {"Company name": payload["title"], "One-line thesis": payload["one_line_thesis"], "Problem": payload["problem"], "ICP": payload["target_customer"], "Why now": "AI tooling lowers build cost, increasing the risk that founders overbuild before validating demand.", "Product / service": payload["proposed_solution"], "Business model": payload["business_model"], "Pricing": payload["pricing_hypothesis"], "Route to first customer": payload["acquisition_strategy"], "Validation plan": payload["validation_plan"], "$50 capital allocation proposal": {"initial": "$10 only after approval", "reserved": "$40 held until evidence gate", "v0_spend": "$0"}, "Time to first dollar": payload["time_to_first_dollar_estimate"], "Path to $500 net cash": "Sell 6-10 fixed-scope audits at $49-$99 while keeping delivery manual and using sunk Artifex compute.", "Competition": "Generic startup consultants, founder communities, AI business idea tools, and DIY validation templates. Public competitor research is unverified unless web research is configured.", "Differentiation": payload["differentiation"], "Build requirements": ["report template", "intake form", "manual analysis workflow", "optional landing page after validation approval"], "Distribution requirements": ["compliant outreach plan", "community/content channels", "CRM-lite tracking before first customers"], "Risks": payload["major_risks"], "What would falsify the thesis": "No willingness-to-pay signal at $49-$99 or fewer than 5 credible target-customer responses after approved compliant validation.", "Confidence": payload["confidence"], "Evidence caveat": "Deterministic fallback evidence only; not equivalent to Sol or public web research." if fallback else "Generated by Sol; public web research still only included if sources are present."}
+        return pitch
+
+    def _questions_from_pitch(self, pitch: dict[str, Any]) -> list[dict[str, str]]:
+        return [{"category": "demand", "question": f"What evidence supports demand for {pitch['Company name']} among the stated ICP?"}, {"category": "urgency", "question": "Why will this customer pay now instead of using free templates or advice?"}, {"category": "distribution", "question": "How do you reach the first 10 customers without spam or fake traction?"}, {"category": "validation", "question": "Can this be validated before building the full product?"}, {"category": "business_model", "question": "Why a productized service first instead of SaaS?"}, {"category": "falsification", "question": "What would falsify the thesis within the $50 and 30-day mandate?"}, {"category": "economics", "question": "What happens if acquisition cost or manual delivery time is 3x the estimate?"}, {"category": "competition", "question": "What is the main competitive threat and why is this worth funding over selling an existing Artifex capability?"}]
+
+    def _answer_question(self, question: ICQuestion) -> dict[str, Any]:
+        evidence = [{"source": "internal_reasoning", "summary": "No customer outreach, spend, or fabricated evidence used.", "fallback_evidence": True}]
+        if question.category in {"demand", "competition"} and not self.web_research_available:
+            evidence.append({"source": "capability_gap", "summary": "Public web research unavailable; demand/competition claims remain uncertain.", "fallback_evidence": True})
+        answers = {"demand": "Demand is not proven. The strongest V0 claim is that the problem is plausible and cheap to test, not that demand exists.", "urgency": "The buyer pays only if the report saves them build time or prevents wasted spend; urgency is weakest before a concrete launch decision.", "distribution": "After approval, use targeted compliant posts/conversations and track responses manually; V0 performs no outreach.", "validation": "Yes. The paid diagnostic can be validated with responses and willingness-to-pay before software build.", "business_model": "Service first reduces build risk and can reach first cash faster than SaaS; software should follow only if repeated demand appears.", "falsification": "Failure to collect credible responses or willingness-to-pay within the mandate falsifies near-term viability.", "economics": "If acquisition or delivery is 3x harder, the company should stop or raise price before building tooling.", "competition": "Main threat is generic consulting/free templates. The reason to fund this over selling raw Artifex capability is packaging a buyer-specific outcome."}
+        return {"answer": answers.get(question.category, "The assumption remains uncertain and must be tested before spend."), "evidence": evidence, "uncertainty": "High until public research and customer evidence are available.", "pitch_changes": {"confidence_adjustment": "reduced/held due missing external evidence"} if question.category in {"demand", "competition"} else {}}
+
+    def _capability_requirements(self, proposal: CompanyProposal) -> list[dict[str, Any]]:
+        web_status = CapabilityStatus.MISSING if not self.web_research_available else CapabilityStatus.PARTIAL
+        return [
+            {"category": "Company Brain", "status": CapabilityStatus.PARTIAL, "rationale": "Venture reasoning exists in V0 but is not a persistent operating brain.", "priority": CapabilityPriority.BEFORE_SCALING, "evidence": {}},
+            {"category": "Board", "status": CapabilityStatus.AVAILABLE, "rationale": "Structured CEO/CTO/CFO/CRO/Independent Director review exists for V0.", "priority": CapabilityPriority.BEFORE_VALIDATION, "evidence": {}},
+            {"category": "IC", "status": CapabilityStatus.AVAILABLE, "rationale": "Bounded IC diligence, questions, scoring, and decision vocabulary exist.", "priority": CapabilityPriority.BEFORE_VALIDATION, "evidence": {}},
+            {"category": "WEB_MARKET_RESEARCH", "status": web_status, "rationale": "No public web research tool is configured in this service; sources cannot be verified.", "priority": CapabilityPriority.BEFORE_VALIDATION, "evidence": {"web_research_available": self.web_research_available}},
+            {"category": "software build", "status": CapabilityStatus.AVAILABLE, "rationale": "Task execution, coding, review, tests, and graph runtime exist.", "priority": CapabilityPriority.BEFORE_FIRST_CUSTOMER, "evidence": {}},
+            {"category": "frontend design", "status": CapabilityStatus.AVAILABLE, "rationale": "Frontend agents and Django UI path exist.", "priority": CapabilityPriority.BEFORE_FIRST_CUSTOMER, "evidence": {}},
+            {"category": "deployment", "status": CapabilityStatus.PARTIAL, "rationale": "Deployment planning exists, but company-specific production deployment workflow is not implemented.", "priority": CapabilityPriority.BEFORE_FIRST_CUSTOMER, "evidence": {}},
+            {"category": "outbound sales", "status": CapabilityStatus.MISSING, "rationale": "No compliant outreach/sequence/customer contact system exists and V0 forbids outreach.", "priority": CapabilityPriority.BEFORE_VALIDATION, "evidence": {}},
+            {"category": "CRM", "status": CapabilityStatus.MISSING, "rationale": "No customer pipeline/contact tracking exists.", "priority": CapabilityPriority.BEFORE_VALIDATION, "evidence": {}},
+            {"category": "payments", "status": CapabilityStatus.MISSING, "rationale": "No payment collection system exists.", "priority": CapabilityPriority.BEFORE_FIRST_CUSTOMER, "evidence": {}},
+            {"category": "invoicing", "status": CapabilityStatus.MISSING, "rationale": "No invoicing workflow exists.", "priority": CapabilityPriority.BEFORE_FIRST_CUSTOMER, "evidence": {}},
+            {"category": "customer support", "status": CapabilityStatus.MISSING, "rationale": "No support inbox or customer service workflow exists.", "priority": CapabilityPriority.BEFORE_SCALING, "evidence": {}},
+            {"category": "company budget management", "status": CapabilityStatus.MISSING, "rationale": "V0 blocks spend but future validation needs tranche/budget controls.", "priority": CapabilityPriority.BEFORE_VALIDATION, "evidence": {}},
+            {"category": "legal/compliance", "status": CapabilityStatus.MISSING, "rationale": "No contracts, terms, privacy, or compliance review workflow exists.", "priority": CapabilityPriority.BEFORE_FIRST_CUSTOMER, "evidence": {}},
+        ]
+
+    def _artifact(self, proposal, mandate, artifact_type: str, name: str, content: dict[str, Any], readable: str, generated_by: str, *, graph_run=None) -> VentureArtifact:
+        return VentureArtifact.objects.create(proposal=proposal, mandate=mandate, graph_run=graph_run, artifact_type=artifact_type, name=name, content=content, readable=readable, generated_by=generated_by)
+
+    def readable_pitch(self, pitch: dict[str, Any]) -> str:
+        return "\n\n".join(f"{section}\n{pitch.get(section, '')}" for section in PITCH_SECTIONS)
+
+    def _readable_mandate(self, mandate: CompanyMandate) -> str:
+        return f"Objective\n{mandate.objective}\n\nConstraints\n{json.dumps(mandate.constraints, indent=2)}\n\nOptimization targets\n" + "\n".join(f"- {item}" for item in mandate.optimization_targets)
+
+    def _readable_board(self, review: CompanyBoardReview) -> str:
+        return f"Recommendation: {review.recommendation}\n\nStrengths\n" + "\n".join(f"- {item}" for item in review.strengths) + "\n\nWeaknesses\n" + "\n".join(f"- {item}" for item in review.weaknesses)
+
+    def _readable_responses(self, responses: list[ICResponse]) -> str:
+        return "\n\n".join(f"Q: {response.question.question}\nA: {response.answer}\nUncertainty: {response.uncertainty}" for response in responses)
+
+    def _readable_score(self, decision: ICDecision) -> str:
+        scores = "\n".join(f"- {key}: {value}/100" for key, value in decision.component_scores.items())
+        return f"Decision: {decision.decision}\nComposite: {decision.composite_score}/100\nP($500 within 30 days): {decision.probability_500_within_30_days}%\n\nScores\n{scores}\n\nCondition\n{decision.validation_condition}"
+
+    def _readable_capability_gap(self, available: list[str], partial: list[str], missing: list[str], ranked: list[dict[str, str]]) -> str:
+        return "AVAILABLE\n" + "\n".join(f"- {item}" for item in available) + "\n\nPARTIAL\n" + "\n".join(f"- {item}" for item in partial) + "\n\nMISSING\n" + "\n".join(f"- {item}" for item in missing) + "\n\nNEXT ARTIFEX CAPABILITIES REQUIRED\n" + "\n".join(f"- {item['category']} ({item['priority']})" for item in ranked)
+
+    def _readable_memo(self, memo: dict[str, Any]) -> str:
+        return json.dumps(memo, indent=2)
+
+    def _fingerprint(self, payload: dict[str, Any]) -> str:
+        return hashlib.sha256((str(payload["title"]).lower() + str(payload["target_customer"]).lower() + str(payload["business_model"]).lower()).encode()).hexdigest()

artifex/settings.py

@@ -21,6 +21,7 @@ INSTALLED_APPS = [
     "control_plane.events",
     "control_plane.agents",
     "control_plane.resources",
+    "control_plane.ventures",
     "control_plane.secrets",
     "control_plane.knowledge",
     "control_plane.verification",

artifex/test_settings.py

@@ -3,3 +3,4 @@ from __future__ import annotations
 from artifex.settings import *  # noqa: F403
 
 DATABASES = {"default": {"ENGINE": "django.db.backends.sqlite3", "NAME": ":memory:"}}
+ALLOWED_HOSTS = ["testserver", "localhost", "127.0.0.1"]

artifex/urls.py

@@ -3,10 +3,37 @@ from __future__ import annotations
 from django.contrib import admin
 from django.urls import path
 
-from control_plane.projects.views import agent_control_room, dashboard
+from control_plane.projects import views
 
 urlpatterns = [
-    path("", dashboard, name="dashboard"),
-    path("agents/", agent_control_room, name="agent_control_room"),
+    path("", views.dashboard, name="dashboard"),
+    path("projects/", views.projects, name="projects"),
+    path("projects/<uuid:project_id>/", views.project_workspace, name="project_workspace"),
+    path("projects/<uuid:project_id>/brain/", views.project_brain, name="project_brain"),
+    path("projects/<uuid:project_id>/archaeologist/", views.project_archaeologist, name="project_archaeologist"),
+    path("projects/<uuid:project_id>/dag.json", views.project_dag_json, name="project_dag_json"),
+    path("projects/<uuid:project_id>/explore/run/", views.run_explore, name="run_explore"),
+    path("projects/<uuid:project_id>/explore/", views.explore, name="project_explore"),
+    path("projects/<uuid:project_id>/roadmap/", views.roadmap, name="project_roadmap"),
+    path("projects/<uuid:project_id>/scenarios/", views.scenarios, name="project_scenarios"),
+    path("projects/<uuid:project_id>/steward/", views.project_steward, name="project_steward"),
+    path("tasks/<uuid:task_id>/", views.task_detail, name="task_detail"),
+    path("graph-runs/<int:graph_run_id>/", views.graph_run_detail, name="graph_run_detail"),
+    path("graph-runs/<int:graph_run_id>/status.json", views.graph_run_json, name="graph_run_json"),
+    path("steward/", views.steward, name="steward"),
+    path("explore/", views.explore, name="explore"),
+    path("opportunities/<uuid:opportunity_id>/action/", views.opportunity_action, name="opportunity_action"),
+    path("roadmap/", views.roadmap, name="roadmap"),
+    path("roadmap/<uuid:item_id>/action/", views.roadmap_action, name="roadmap_action"),
+    path("scenario-lab/", views.scenarios, name="scenarios"),
+    path("scenario-findings/<uuid:finding_id>/action/", views.scenario_finding_action, name="scenario_finding_action"),
+    path("progeny/", views.progeny, name="progeny"),
+    path("agents/", views.agent_control_room, name="agent_control_room"),
+    path("agents/<uuid:version_id>/", views.agent_detail, name="agent_detail"),
+    path("agents/<uuid:version_id>/performance.json", views.agent_performance_json, name="agent_performance_json"),
+    path("resources/", views.resources, name="resources"),
+    path("approvals/", views.approvals, name="approvals"),
+    path("approvals/<int:approval_id>/action/", views.approval_action, name="approval_action"),
+    path("activity/", views.activity, name="activity"),
     path("admin/", admin.site.urls),
 ]

control_plane/projects/ui_services.py

@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+from collections import Counter
+
+from django.db.models import Count, Q
+
+from agents.control_room import AgentControlRoomService
+from agents.lifecycle import LifecycleInspectionService
+from agents.roadmap import RoadmapService
+from agents.scenario_lab import ScenarioLabService
+from control_plane.agents.models import AgentRun, AgentVersion, ProgenyExperiment, ProgenyInvestigation, ProgenySignal, PromotionStatus
+from control_plane.events.models import Event
+from control_plane.projects.models import ExplorationOpportunity, Project, RoadmapHorizon, ScenarioFinding, ScenarioRun, ScenarioSuite, StewardFinding, StewardRun, Task, TaskStatus
+from control_plane.resources.models import ModelRequest, Resource
+from graph.models import ExecutionGraphVersionStatus, GraphApproval, GraphApprovalStatus, GraphRun, GraphRunStatus
+
+
+class ControlPlaneUIService:
+    def dashboard(self) -> dict[str, object]:
+        projects = Project.objects.all()
+        graph_runs = GraphRun.objects.select_related("execution_graph_version__graph", "project").order_by("-created_at")
+        steward_findings = StewardFinding.objects.all()
+        progeny_signals = ProgenySignal.objects.all()
+        control_room = AgentControlRoomService()
+        agent_health = [control_room.get_agent_health(version.id)["status"] for version in AgentVersion.objects.filter(promotion_status=PromotionStatus.CHAMPION)]
+        return {
+            "project_summary": {"total": projects.count(), "active": projects.exclude(status__in=["FINISHED", "FAILED"]).count(), "blocked_failed": projects.filter(status__in=["BLOCKED", "FAILED"]).count(), "recent_completed": projects.filter(status="FINISHED").order_by("-updated_at")[:5]},
+            "execution_summary": {"active": graph_runs.filter(status__in=[GraphRunStatus.PENDING, GraphRunStatus.RUNNING, GraphRunStatus.PAUSED]).count(), "failed": graph_runs.filter(status=GraphRunStatus.FAILED).count(), "recent_success": graph_runs.filter(status=GraphRunStatus.COMPLETE)[:5], "champion_task_graph": self._champion_graph("task_execution")},
+            "steward_summary": {"enrolled_projects": Project.objects.filter(steward_enrollments__status="ACTIVE").distinct().count(), "open_findings": steward_findings.exclude(status__in=["RESOLVED", "DISMISSED"]).count(), "high_findings": steward_findings.filter(severity__in=["HIGH", "CRITICAL"]).exclude(status__in=["RESOLVED", "DISMISSED"]).count()},
+            "progeny_summary": {"unresolved_signals": progeny_signals.filter(status="OPEN").count(), "open_investigations": ProgenyInvestigation.objects.filter(status="OPEN").count(), "challengers": AgentVersion.objects.filter(promotion_status=PromotionStatus.CHALLENGER).count(), "pending_experiment_decisions": ProgenyExperiment.objects.filter(status__in=["DRAFT", "RUNNING"]).count()},
+            "agent_summary": {"agent_count": AgentVersion.objects.values("agent").distinct().count(), "watch_degraded": sum(1 for status in agent_health if status in ["WATCH", "DEGRADED"]), "active_runs": AgentRun.objects.filter(status__in=["QUEUED", "RUNNING"]).count()},
+            "approval_count": GraphApproval.objects.filter(status=GraphApprovalStatus.PENDING).count(),
+            "recent": {"events": Event.objects.order_by("-created_at")[:10], "tasks": Task.objects.order_by("-updated_at")[:10], "graph_runs": graph_runs[:10], "findings": steward_findings.order_by("-updated_at")[:10], "investigations": ProgenyInvestigation.objects.order_by("-updated_at")[:10]},
+        }
+
+    def project_list(self) -> list[dict[str, object]]:
+        rows = []
+        for project in Project.objects.order_by("name"):
+            rows.append({"project": project, "current_milestone": project.milestones.order_by("order", "created_at").last(), "task_total": project.tasks.count(), "task_complete": project.tasks.filter(status=TaskStatus.COMPLETE).count(), "steward_state": project.steward_enrollments.order_by("-created_at").first(), "open_findings": project.steward_findings.exclude(status__in=["RESOLVED", "DISMISSED"]).count(), "latest_graph_run": project.graph_runs.order_by("-created_at").first(), "warnings": self.project_warnings(project)})
+        return rows
+
+    def project_workspace(self, project: Project) -> dict[str, object]:
+        lifecycle = LifecycleInspectionService().project_lifecycle_view(project)
+        return {"project": project, "plan": project.plans.order_by("-version").first(), "milestones": project.milestones.prefetch_related("features__tasks", "tasks").order_by("order", "created_at"), "tasks": project.tasks.select_related("milestone", "feature").order_by("milestone__order", "priority", "created_at"), "graph_runs": project.graph_runs.select_related("execution_graph_version__graph", "task").order_by("-created_at")[:20], "commits": project.commits.order_by("-created_at")[:10], "roadmap": RoadmapService().project_roadmap_view(project), "lifecycle": lifecycle, "scenario_coverage": ScenarioLabService().coverage(project), "activity": Event.objects.filter(project=project).order_by("-created_at")[:20], "warnings": self.project_warnings(project)}
+
+    def graph_run_detail(self, graph_run: GraphRun) -> dict[str, object]:
+        nodes = list(graph_run.node_runs.select_related("agent_version__agent", "model_request").order_by("created_at", "visit_index"))
+        traversals = list(graph_run.edge_traversals.order_by("created_at"))
+        return {"graph_run": graph_run, "nodes": nodes, "traversals": traversals, "approvals": graph_run.approvals.order_by("-created_at"), "model_request_count": sum(1 for node in nodes if node.model_request_id)}
+
+    def task_detail(self, task: Task) -> dict[str, object]:
+        return {"task": task, "dependencies": [edge.depends_on for edge in task.dependency_edges.select_related("depends_on")], "attempts": task.attempts.select_related("coder").order_by("attempt_number"), "graph_runs": task.graph_runs.select_related("execution_graph_version__graph").order_by("-created_at"), "tests": task.test_runs.order_by("-created_at"), "reviews": task.reviews.order_by("-created_at"), "commits": task.commits.order_by("-created_at")}
+
+    def steward(self, project: Project | None = None) -> dict[str, object]:
+        findings = StewardFinding.objects.select_related("project").order_by("-updated_at")
+        runs = StewardRun.objects.select_related("project").order_by("-created_at")
+        if project:
+            findings = findings.filter(project=project)
+            runs = runs.filter(project=project)
+        return {"findings": findings[:100], "runs": runs[:50]}
+
+    def progeny(self) -> dict[str, object]:
+        signals = ProgenySignal.objects.select_related("project", "agent_version__agent", "execution_graph_version__graph", "graph_node_run").order_by("-created_at")
+        grouped = Counter(signals.filter(status="OPEN").values_list("grouping_key", flat=True))
+        return {"signals": signals[:100], "groups": grouped.most_common(50), "investigations": ProgenyInvestigation.objects.order_by("-created_at")[:50], "experiments": ProgenyExperiment.objects.order_by("-created_at")[:50]}
+
+    def roadmap_board(self, project: Project | None = None) -> dict[str, object]:
+        items = Project.objects.none()
+        qs = project.roadmap_items if project else None
+        board = {}
+        for horizon in RoadmapHorizon.values:
+            board[horizon] = (qs.filter(horizon=horizon) if qs else __import__("control_plane.projects.models", fromlist=["RoadmapItem"]).RoadmapItem.objects.filter(horizon=horizon)).select_related("project").order_by("-composite_score", "-priority")
+        return board
+
+    def scenario_lab(self, project: Project | None = None) -> dict[str, object]:
+        suites = ScenarioSuite.objects.select_related("project").order_by("-created_at")
+        runs = ScenarioRun.objects.select_related("project", "scenario").order_by("-created_at")
+        findings = ScenarioFinding.objects.select_related("project", "scenario").order_by("-created_at")
+        if project:
+            suites = suites.filter(project=project)
+            runs = runs.filter(project=project)
+            findings = findings.filter(project=project)
+        coverage = Counter(ScenarioRun.objects.filter(project=project).values_list("scenario__scenario_type", flat=True) if project else ScenarioRun.objects.values_list("scenario__scenario_type", flat=True))
+        return {"suites": suites[:50], "runs": runs[:100], "findings": findings[:100], "coverage": dict(coverage)}
+
+    def resources(self) -> dict[str, object]:
+        rows = []
+        for resource in Resource.objects.order_by("name"):
+            requests = resource.model_requests.order_by("-created_at")
+            latencies = [value for value in requests.exclude(latency_ms=None).values_list("latency_ms", flat=True)[:50]]
+            rows.append({"resource": resource, "recent_requests": requests[:10], "request_count": requests.count(), "median_latency": sorted(latencies)[len(latencies) // 2] if latencies else None})
+        return {"resources": rows}
+
+    def approvals(self) -> dict[str, object]:
+        return {"approvals": GraphApproval.objects.select_related("graph_run__project", "graph_run__execution_graph_version__graph", "node_run").filter(status=GraphApprovalStatus.PENDING).order_by("created_at")}
+
+    def activity(self, project: Project | None = None) -> dict[str, object]:
+        events = Event.objects.select_related("project", "task").order_by("-created_at")
+        if project:
+            events = events.filter(project=project)
+        return {"events": events[:200]}
+
+    def project_brain(self, project: Project) -> dict[str, object]:
+        return {"project": project, "decisions": project.decisions.order_by("-created_at"), "plans": project.plans.order_by("-version"), "artifacts": project.artifacts.filter(artifact_type__icontains="PLAN").order_by("-created_at")}
+
+    def archaeologist(self, project: Project) -> dict[str, object]:
+        archaeology = project.artifacts.filter(artifact_type__icontains="ARCH").order_by("-created_at")
+        return {"project": project, "observed": project.architecture_summary, "artifacts": archaeology, "findings": project.findings.order_by("-created_at")[:50]}
+
+    def project_warnings(self, project: Project) -> list[str]:
+        warnings = []
+        if project.graph_runs.filter(status=GraphRunStatus.FAILED).exists():
+            warnings.append("failed graph runs")
+        if project.tasks.filter(status__in=[TaskStatus.BLOCKED, TaskStatus.FAILED]).exists():
+            warnings.append("blocked or failed tasks")
+        if project.steward_findings.filter(severity__in=["HIGH", "CRITICAL"]).exclude(status__in=["RESOLVED", "DISMISSED"]).exists():
+            warnings.append("high severity findings")
+        return warnings
+
+    def _champion_graph(self, name: str):
+        return __import__("graph.models", fromlist=["ExecutionGraphVersion"]).ExecutionGraphVersion.objects.filter(graph__name=name, status=ExecutionGraphVersionStatus.CHAMPION).select_related("graph").first()

control_plane/projects/views.py

@@ -1,21 +1,164 @@
 from __future__ import annotations
 
-from django.shortcuts import render
+import json
+
+from django.http import JsonResponse
+from django.shortcuts import get_object_or_404, redirect, render
+from django.urls import reverse
+from django.utils import timezone
+from django.views.decorators.http import require_POST
 
 from agents.control_room import AgentControlRoomService
+from agents.lifecycle import ExplorerService
+from agents.roadmap import RoadmapService
+from agents.scenario_lab import ScenarioLabService
 from control_plane.events.models import Event
-from control_plane.projects.models import Project, TaskStatus
+from control_plane.projects.models import Decision, ExplorationOpportunity, Project, RoadmapItem, ScenarioFinding, ScenarioSuite, StewardFinding, Task
+from control_plane.projects.ui_services import ControlPlaneUIService
+from graph.bootstrap import champion_project_exploration_graph_v1
+from graph.langgraph_runtime import LangGraphRuntime
+from graph.lifecycle import exploration_registry
+from graph.models import GraphApproval, GraphApprovalStatus, GraphRun, GraphRunStatus
+
+
+ui = ControlPlaneUIService()
 
 
 def dashboard(request):
-    projects = Project.objects.order_by("name")
-    recent_events = Event.objects.select_related("project", "task").order_by("-created_at")[:25]
-    summary = {
-        "project_count": Project.objects.count(),
-        "ready_tasks": sum(project.tasks.filter(status=TaskStatus.READY).count() for project in projects),
-        "blocked_tasks": sum(project.tasks.filter(status=TaskStatus.BLOCKED).count() for project in projects),
-    }
-    return render(request, "projects/dashboard.html", {"projects": projects, "summary": summary, "recent_events": recent_events})
+    return render(request, "control_plane/dashboard.html", ui.dashboard())
+
+
+def projects(request):
+    return render(request, "control_plane/projects.html", {"rows": ui.project_list()})
+
+
+def project_workspace(request, project_id):
+    project = get_object_or_404(Project, id=project_id)
+    return render(request, "control_plane/project_workspace.html", ui.project_workspace(project))
+
+
+def project_brain(request, project_id):
+    project = get_object_or_404(Project, id=project_id)
+    if request.method == "POST":
+        Decision.objects.create(project=project, decision_type="PROJECT_BRAIN_NOTE", decision=request.POST.get("message", ""), reason="User-authored Project Brain interaction from UI", actor="ui")
+        return redirect("project_brain", project_id=project.id)
+    return render(request, "control_plane/project_brain.html", ui.project_brain(project))
+
+
+def project_archaeologist(request, project_id):
+    project = get_object_or_404(Project, id=project_id)
+    return render(request, "control_plane/archaeologist.html", ui.archaeologist(project))
+
+
+def graph_run_detail(request, graph_run_id):
+    graph_run = get_object_or_404(GraphRun, id=graph_run_id)
+    template = "control_plane/partials/graph_run_status.html" if request.headers.get("HX-Request") else "control_plane/graph_run.html"
+    return render(request, template, ui.graph_run_detail(graph_run))
+
+
+def graph_run_json(request, graph_run_id):
+    graph_run = get_object_or_404(GraphRun, id=graph_run_id)
+    return JsonResponse({"id": graph_run.id, "status": graph_run.status, "current_node": graph_run.current_node, "nodes": list(graph_run.node_runs.values("node_id", "visit_index", "status", "failure_evidence", "telemetry")), "edges": list(graph_run.edge_traversals.values("source_node", "target_node", "result", "condition"))})
+
+
+def project_dag_json(request, project_id):
+    project = get_object_or_404(Project, id=project_id)
+    return JsonResponse({"project": str(project.id), "milestones": list(project.milestones.values("id", "key", "title", "status", "order")), "features": list(project.features.values("id", "milestone_id", "title", "status")), "tasks": list(project.tasks.values("id", "milestone_id", "feature_id", "goal", "status", "priority")), "dependencies": list(project.tasks.values("id", "dependency_edges__depends_on_id"))})
+
+
+def task_detail(request, task_id):
+    task = get_object_or_404(Task, id=task_id)
+    return render(request, "control_plane/task.html", ui.task_detail(task))
+
+
+def steward(request):
+    return render(request, "control_plane/steward.html", ui.steward())
+
+
+def project_steward(request, project_id):
+    project = get_object_or_404(Project, id=project_id)
+    return render(request, "control_plane/steward.html", {"project": project, **ui.steward(project)})
+
+
+def explore(request, project_id=None):
+    project = get_object_or_404(Project, id=project_id) if project_id else None
+    opportunities = ExplorationOpportunity.objects.select_related("project", "exploration").order_by("-composite_score", "-created_at")
+    if project:
+        opportunities = opportunities.filter(project=project)
+    return render(request, "control_plane/explore.html", {"project": project, "opportunities": opportunities[:100]})
+
+
+@require_POST
+def run_explore(request, project_id):
+    project = get_object_or_404(Project, id=project_id)
+    service = ExplorerService()
+    exploration = service.start_exploration(project, prompt="UI Explore run")
+    version = champion_project_exploration_graph_v1()
+    graph_run = GraphRun.objects.create(execution_graph_version=version, project=project, current_node=version.graph_spec["entry"], metadata={"exploration_id": str(exploration.id), "source": "ui"})
+    LangGraphRuntime(exploration_registry(service)).run_until_terminal_or_paused(graph_run)
+    return redirect("graph_run_detail", graph_run_id=graph_run.id)
+
+
+@require_POST
+def opportunity_action(request, opportunity_id):
+    opportunity = get_object_or_404(ExplorationOpportunity, id=opportunity_id)
+    action = request.POST.get("action")
+    service = ExplorerService()
+    if action == "extend":
+        service.convert_to_extension(opportunity)
+    elif action == "evolve":
+        service.convert_to_evolution(opportunity, baseline_measurement={"metric": "value", "value": 100})
+    elif action == "roadmap":
+        RoadmapService().upsert_item(opportunity.project, title=opportunity.title, description=opportunity.description, source="EXPLORE", source_ref={"exploration_opportunity_id": str(opportunity.id)}, rationale=opportunity.rationale, evidence=opportunity.evidence, horizon="NEXT", category=opportunity.opportunity_type, target_action=opportunity.recommended_action if opportunity.recommended_action in ["EXTEND", "EVOLVE", "REPAIR", "INVESTIGATE"] else "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})
+    elif action == "defer":
+        service.defer(opportunity)
+    elif action == "reject":
+        service.reject(opportunity)
+    return redirect(request.META.get("HTTP_REFERER") or reverse("explore"))
+
+
+def roadmap(request, project_id=None):
+    project = get_object_or_404(Project, id=project_id) if project_id else None
+    return render(request, "control_plane/roadmap.html", {"project": project, "board": ui.roadmap_board(project)})
+
+
+@require_POST
+def roadmap_action(request, item_id):
+    item = get_object_or_404(RoadmapItem, id=item_id)
+    action = request.POST.get("action")
+    service = RoadmapService()
+    if action in ["NOW", "NEXT", "LATER", "EXPLORING"]:
+        item.horizon = action
+        item.save(update_fields=["horizon", "updated_at"])
+    elif action == "defer":
+        item.status = "DEFERRED"
+        item.save(update_fields=["status", "updated_at"])
+    elif action == "reject":
+        item.status = "REJECTED"
+        item.save(update_fields=["status", "updated_at"])
+    elif action == "extend":
+        service.convert_to_extension(item)
+    elif action == "evolve":
+        service.convert_to_evolution(item, baseline_measurement={"metric": "value", "value": 100})
+    return redirect(request.META.get("HTTP_REFERER") or reverse("roadmap"))
+
+
+def scenarios(request, project_id=None):
+    project = get_object_or_404(Project, id=project_id) if project_id else None
+    return render(request, "control_plane/scenarios.html", {"project": project, **ui.scenario_lab(project)})
+
+
+@require_POST
+def scenario_finding_action(request, finding_id):
+    finding = get_object_or_404(ScenarioFinding, id=finding_id)
+    action = request.POST.get("action")
+    if action == "roadmap":
+        ScenarioLabService().route_finding(finding)
+    return redirect(request.META.get("HTTP_REFERER") or reverse("scenarios"))
+
+
+def progeny(request):
+    return render(request, "control_plane/progeny.html", ui.progeny())
 
 
 def agent_control_room(request):
@@ -24,4 +167,54 @@ def agent_control_room(request):
     for agent in agents:
         champion_id = agent.get("champion_version")
         agent["health"] = service.get_agent_health(champion_id) if champion_id else {"status": "WATCH", "reasons": ["No champion version."]}
-    return render(request, "projects/agent_control_room.html", {"agents": agents, "teams": service.list_teams()})
+        agent["performance"] = service.get_agent_performance(champion_id) if champion_id else {}
+    return render(request, "control_plane/agents.html", {"agents": agents, "teams": service.list_teams()})
+
+
+def agent_detail(request, version_id):
+    service = AgentControlRoomService()
+    return render(request, "control_plane/agent_detail.html", {"version": service.get_agent_version(version_id), "performance": service.get_agent_performance(version_id), "health": service.get_agent_health(version_id), "usage": service.get_agent_usage(version_id), "progeny": service.get_agent_progeny(version_id), "challengers": service.get_agent_challengers(service.get_agent_version(version_id)["agent_id"])})
+
+
+def agent_performance_json(request, version_id):
+    return JsonResponse(AgentControlRoomService().get_agent_performance(version_id, window=request.GET.get("window", "lifetime")))
+
+
+def resources(request):
+    return render(request, "control_plane/resources.html", ui.resources())
+
+
+def approvals(request):
+    return render(request, "control_plane/approvals.html", ui.approvals())
+
+
+@require_POST
+def approval_action(request, approval_id):
+    approval = get_object_or_404(GraphApproval, id=approval_id)
+    action = request.POST.get("action")
+    approval.status = GraphApprovalStatus.APPROVED if action == "approve" else GraphApprovalStatus.REJECTED
+    approval.decided_by = "ui"
+    approval.decided_at = timezone.now()
+    approval.save(update_fields=["status", "decided_by", "decided_at", "updated_at"])
+    graph_run = approval.graph_run
+    if approval.status == GraphApprovalStatus.APPROVED:
+        graph_run.status = GraphRunStatus.RUNNING
+        graph_run.failure_reason = ""
+        graph_run.save(update_fields=["status", "failure_reason", "updated_at"])
+        self_resume_graph(graph_run)
+    return redirect("graph_run_detail", graph_run_id=graph_run.id)
+
+
+def activity(request):
+    project = Project.objects.filter(id=request.GET.get("project")).first() if request.GET.get("project") else None
+    return render(request, "control_plane/activity.html", {"project": project, **ui.activity(project)})
+
+
+def self_resume_graph(graph_run: GraphRun) -> None:
+    name = graph_run.execution_graph_version.graph.name
+    try:
+        if name == "project_exploration":
+            LangGraphRuntime(exploration_registry(ExplorerService())).run_until_terminal_or_paused(graph_run)
+    except Exception as exc:
+        graph_run.failure_reason = f"UI approval saved; automatic resume failed: {exc}"
+        graph_run.save(update_fields=["failure_reason", "updated_at"])

control_plane/ventures/apps.py

@@ -0,0 +1,8 @@
+from __future__ import annotations
+
+from django.apps import AppConfig
+
+
+class VenturesConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "control_plane.ventures"

control_plane/ventures/migrations/0001_initial.py

@@ -0,0 +1,239 @@
+# Generated by Django 5.2.16 on 2026-08-15 13:50
+
+import django.db.models.deletion
+import uuid
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ('graph', '0004_unique_champion_graph_version'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='CompanyMandate',
+            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)),
+                ('objective', models.TextField()),
+                ('constraints', models.JSONField(blank=True, default=dict)),
+                ('optimization_targets', models.JSONField(blank=True, default=list)),
+                ('max_validation_capital', models.DecimalField(decimal_places=2, default=50, max_digits=10)),
+                ('target_net_new_cash', models.DecimalField(decimal_places=2, default=500, max_digits=10)),
+                ('target_window_days', models.PositiveIntegerField(default=30)),
+                ('metadata', models.JSONField(blank=True, default=dict)),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='CompanyProposal',
+            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()),
+                ('problem', models.TextField()),
+                ('target_customer', models.TextField()),
+                ('proposed_solution', models.TextField()),
+                ('business_model', models.TextField()),
+                ('pricing_hypothesis', models.TextField()),
+                ('acquisition_strategy', models.TextField()),
+                ('validation_plan', models.TextField()),
+                ('capital_requested', models.DecimalField(decimal_places=2, default=0, max_digits=10)),
+                ('time_to_first_dollar_estimate', models.CharField(max_length=120)),
+                ('expected_margin', models.CharField(max_length=120)),
+                ('build_complexity', models.CharField(max_length=80)),
+                ('market_evidence', models.JSONField(blank=True, default=list)),
+                ('differentiation', models.TextField()),
+                ('major_risks', models.JSONField(blank=True, default=list)),
+                ('confidence', models.FloatField(default=0.0)),
+                ('status', models.CharField(choices=[('DRAFT', 'Draft'), ('SUBMITTED', 'Submitted'), ('UNDER_DILIGENCE', 'Under Diligence'), ('REVISE', 'Revise'), ('FUNDED_RECOMMENDED', 'Funded Recommended'), ('WATCHLIST', 'Watchlist'), ('REJECTED', 'Rejected')], default='DRAFT', max_length=32)),
+                ('pitch', models.JSONField(blank=True, default=dict)),
+                ('metadata', models.JSONField(blank=True, default=dict)),
+                ('mandate', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='company_proposals', to='ventures.companymandate')),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='CompanyCapabilityRequirement',
+            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)),
+                ('category', models.CharField(max_length=120)),
+                ('status', models.CharField(choices=[('AVAILABLE', 'Available'), ('PARTIAL', 'Partial'), ('MISSING', 'Missing')], max_length=32)),
+                ('rationale', models.TextField(blank=True)),
+                ('priority', models.CharField(choices=[('BEFORE_VALIDATION', 'Before Validation'), ('BEFORE_FIRST_CUSTOMER', 'Before First Customer'), ('BEFORE_SCALING', 'Before Scaling')], default='BEFORE_SCALING', max_length=32)),
+                ('evidence', models.JSONField(blank=True, default=dict)),
+                ('proposal', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='capability_requirements', to='ventures.companyproposal')),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='CompanyBoardReview',
+            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)),
+                ('observations', models.JSONField(blank=True, default=dict)),
+                ('strengths', models.JSONField(blank=True, default=list)),
+                ('weaknesses', models.JSONField(blank=True, default=list)),
+                ('key_assumptions', models.JSONField(blank=True, default=list)),
+                ('required_revisions', models.JSONField(blank=True, default=list)),
+                ('recommendation', models.CharField(max_length=80)),
+                ('revised_pitch', models.JSONField(blank=True, default=dict)),
+                ('metadata', models.JSONField(blank=True, default=dict)),
+                ('proposal', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='board_reviews', to='ventures.companyproposal')),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='ICDiligence',
+            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='INITIAL_PITCH', max_length=32)),
+                ('rounds', models.JSONField(blank=True, default=list)),
+                ('red_team_challenge', models.JSONField(blank=True, default=dict)),
+                ('final_response', models.JSONField(blank=True, default=dict)),
+                ('metadata', models.JSONField(blank=True, default=dict)),
+                ('proposal', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='ic_diligence', to='ventures.companyproposal')),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='ICDecision',
+            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)),
+                ('decision', models.CharField(choices=[('FUND_RECOMMENDED', 'Fund Recommended'), ('CONDITIONAL_FUND', 'Conditional Fund'), ('REVISE_AND_RESUBMIT', 'Revise And Resubmit'), ('WATCHLIST', 'Watchlist'), ('PIVOT', 'Pivot'), ('REJECT', 'Reject')], max_length=32)),
+                ('component_scores', models.JSONField(blank=True, default=dict)),
+                ('composite_score', models.FloatField(default=0.0)),
+                ('probability_500_within_30_days', models.FloatField(default=0.0)),
+                ('initial_tranche', models.DecimalField(blank=True, decimal_places=2, max_digits=10, null=True)),
+                ('validation_condition', models.TextField(blank=True)),
+                ('evidence_required', models.JSONField(blank=True, default=list)),
+                ('recommended_allocation', models.JSONField(blank=True, default=dict)),
+                ('kill_criteria', models.JSONField(blank=True, default=list)),
+                ('next_decision_point', models.TextField(blank=True)),
+                ('metadata', models.JSONField(blank=True, default=dict)),
+                ('diligence', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='decision', to='ventures.icdiligence')),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='ICQuestion',
+            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)),
+                ('question', models.TextField()),
+                ('category', models.CharField(max_length=80)),
+                ('evidence_required', models.BooleanField(default=True)),
+                ('round_name', models.CharField(default='Diligence Round 1', max_length=80)),
+                ('metadata', models.JSONField(blank=True, default=dict)),
+                ('diligence', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='questions', to='ventures.icdiligence')),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='ICResponse',
+            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)),
+                ('answer', models.TextField()),
+                ('evidence', models.JSONField(blank=True, default=list)),
+                ('uncertainty', models.TextField(blank=True)),
+                ('pitch_changes', models.JSONField(blank=True, default=dict)),
+                ('metadata', models.JSONField(blank=True, default=dict)),
+                ('question', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='responses', to='ventures.icquestion')),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='PortfolioCapabilityGap',
+            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)),
+                ('available', models.JSONField(blank=True, default=list)),
+                ('partial', models.JSONField(blank=True, default=list)),
+                ('missing', models.JSONField(blank=True, default=list)),
+                ('ranked_missing', models.JSONField(blank=True, default=list)),
+                ('report', models.TextField(blank=True)),
+                ('metadata', models.JSONField(blank=True, default=dict)),
+                ('proposal', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='capability_gap_reports', to='ventures.companyproposal')),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='VentureArtifact',
+            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)),
+                ('artifact_type', models.CharField(max_length=80)),
+                ('name', models.CharField(max_length=255)),
+                ('content', models.JSONField(blank=True, default=dict)),
+                ('readable', models.TextField(blank=True)),
+                ('generated_by', models.CharField(blank=True, max_length=120)),
+                ('graph_run', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='venture_artifacts', to='graph.graphrun')),
+                ('mandate', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='artifacts', to='ventures.companymandate')),
+                ('proposal', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='artifacts', to='ventures.companyproposal')),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='VentureThesis',
+            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)),
+                ('thesis', models.TextField()),
+                ('similarity_fingerprint', models.CharField(blank=True, max_length=128)),
+                ('related_theses', models.JSONField(blank=True, default=list)),
+                ('competitive_theses', models.JSONField(blank=True, default=list)),
+                ('portfolio_visibility', models.JSONField(blank=True, default=dict)),
+                ('metadata', models.JSONField(blank=True, default=dict)),
+                ('mandate', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='theses', to='ventures.companymandate')),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.AddField(
+            model_name='companyproposal',
+            name='thesis',
+            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='company_proposals', to='ventures.venturethesis'),
+        ),
+    ]

control_plane/ventures/models.py

@@ -0,0 +1,166 @@
+from __future__ import annotations
+
+from django.db import models
+
+from control_plane.common import TimestampedModel
+
+
+class CompanyProposalStatus(models.TextChoices):
+    DRAFT = "DRAFT"
+    SUBMITTED = "SUBMITTED"
+    UNDER_DILIGENCE = "UNDER_DILIGENCE"
+    REVISE = "REVISE"
+    FUNDED_RECOMMENDED = "FUNDED_RECOMMENDED"
+    WATCHLIST = "WATCHLIST"
+    REJECTED = "REJECTED"
+
+
+class ICDecisionType(models.TextChoices):
+    FUND_RECOMMENDED = "FUND_RECOMMENDED"
+    CONDITIONAL_FUND = "CONDITIONAL_FUND"
+    REVISE_AND_RESUBMIT = "REVISE_AND_RESUBMIT"
+    WATCHLIST = "WATCHLIST"
+    PIVOT = "PIVOT"
+    REJECT = "REJECT"
+
+
+class CapabilityStatus(models.TextChoices):
+    AVAILABLE = "AVAILABLE"
+    PARTIAL = "PARTIAL"
+    MISSING = "MISSING"
+
+
+class CapabilityPriority(models.TextChoices):
+    BEFORE_VALIDATION = "BEFORE_VALIDATION"
+    BEFORE_FIRST_CUSTOMER = "BEFORE_FIRST_CUSTOMER"
+    BEFORE_SCALING = "BEFORE_SCALING"
+
+
+class CompanyMandate(TimestampedModel):
+    objective = models.TextField()
+    constraints = models.JSONField(default=dict, blank=True)
+    optimization_targets = models.JSONField(default=list, blank=True)
+    max_validation_capital = models.DecimalField(max_digits=10, decimal_places=2, default=50)
+    target_net_new_cash = models.DecimalField(max_digits=10, decimal_places=2, default=500)
+    target_window_days = models.PositiveIntegerField(default=30)
+    metadata = models.JSONField(default=dict, blank=True)
+
+
+class VentureThesis(TimestampedModel):
+    mandate = models.ForeignKey(CompanyMandate, on_delete=models.CASCADE, related_name="theses")
+    title = models.CharField(max_length=255)
+    thesis = models.TextField()
+    similarity_fingerprint = models.CharField(max_length=128, blank=True)
+    related_theses = models.JSONField(default=list, blank=True)
+    competitive_theses = models.JSONField(default=list, blank=True)
+    portfolio_visibility = models.JSONField(default=dict, blank=True)
+    metadata = models.JSONField(default=dict, blank=True)
+
+
+class CompanyProposal(TimestampedModel):
+    mandate = models.ForeignKey(CompanyMandate, on_delete=models.PROTECT, related_name="company_proposals")
+    thesis = models.ForeignKey(VentureThesis, on_delete=models.SET_NULL, null=True, blank=True, related_name="company_proposals")
+    title = models.CharField(max_length=255)
+    description = models.TextField()
+    problem = models.TextField()
+    target_customer = models.TextField()
+    proposed_solution = models.TextField()
+    business_model = models.TextField()
+    pricing_hypothesis = models.TextField()
+    acquisition_strategy = models.TextField()
+    validation_plan = models.TextField()
+    capital_requested = models.DecimalField(max_digits=10, decimal_places=2, default=0)
+    time_to_first_dollar_estimate = models.CharField(max_length=120)
+    expected_margin = models.CharField(max_length=120)
+    build_complexity = models.CharField(max_length=80)
+    market_evidence = models.JSONField(default=list, blank=True)
+    differentiation = models.TextField()
+    major_risks = models.JSONField(default=list, blank=True)
+    confidence = models.FloatField(default=0.0)
+    status = models.CharField(max_length=32, choices=CompanyProposalStatus.choices, default=CompanyProposalStatus.DRAFT)
+    pitch = models.JSONField(default=dict, blank=True)
+    metadata = models.JSONField(default=dict, blank=True)
+
+
+class CompanyBoardReview(TimestampedModel):
+    proposal = models.ForeignKey(CompanyProposal, on_delete=models.CASCADE, related_name="board_reviews")
+    observations = models.JSONField(default=dict, blank=True)
+    strengths = models.JSONField(default=list, blank=True)
+    weaknesses = models.JSONField(default=list, blank=True)
+    key_assumptions = models.JSONField(default=list, blank=True)
+    required_revisions = models.JSONField(default=list, blank=True)
+    recommendation = models.CharField(max_length=80)
+    revised_pitch = models.JSONField(default=dict, blank=True)
+    metadata = models.JSONField(default=dict, blank=True)
+
+
+class ICDiligence(TimestampedModel):
+    proposal = models.ForeignKey(CompanyProposal, on_delete=models.CASCADE, related_name="ic_diligence")
+    status = models.CharField(max_length=32, default="INITIAL_PITCH")
+    rounds = models.JSONField(default=list, blank=True)
+    red_team_challenge = models.JSONField(default=dict, blank=True)
+    final_response = models.JSONField(default=dict, blank=True)
+    metadata = models.JSONField(default=dict, blank=True)
+
+
+class ICQuestion(TimestampedModel):
+    diligence = models.ForeignKey(ICDiligence, on_delete=models.CASCADE, related_name="questions")
+    question = models.TextField()
+    category = models.CharField(max_length=80)
+    evidence_required = models.BooleanField(default=True)
+    round_name = models.CharField(max_length=80, default="Diligence Round 1")
+    metadata = models.JSONField(default=dict, blank=True)
+
+
+class ICResponse(TimestampedModel):
+    question = models.ForeignKey(ICQuestion, on_delete=models.CASCADE, related_name="responses")
+    answer = models.TextField()
+    evidence = models.JSONField(default=list, blank=True)
+    uncertainty = models.TextField(blank=True)
+    pitch_changes = models.JSONField(default=dict, blank=True)
+    metadata = models.JSONField(default=dict, blank=True)
+
+
+class ICDecision(TimestampedModel):
+    diligence = models.OneToOneField(ICDiligence, on_delete=models.CASCADE, related_name="decision")
+    decision = models.CharField(max_length=32, choices=ICDecisionType.choices)
+    component_scores = models.JSONField(default=dict, blank=True)
+    composite_score = models.FloatField(default=0.0)
+    probability_500_within_30_days = models.FloatField(default=0.0)
+    initial_tranche = models.DecimalField(max_digits=10, decimal_places=2, null=True, blank=True)
+    validation_condition = models.TextField(blank=True)
+    evidence_required = models.JSONField(default=list, blank=True)
+    recommended_allocation = models.JSONField(default=dict, blank=True)
+    kill_criteria = models.JSONField(default=list, blank=True)
+    next_decision_point = models.TextField(blank=True)
+    metadata = models.JSONField(default=dict, blank=True)
+
+
+class CompanyCapabilityRequirement(TimestampedModel):
+    proposal = models.ForeignKey(CompanyProposal, on_delete=models.CASCADE, related_name="capability_requirements")
+    category = models.CharField(max_length=120)
+    status = models.CharField(max_length=32, choices=CapabilityStatus.choices)
+    rationale = models.TextField(blank=True)
+    priority = models.CharField(max_length=32, choices=CapabilityPriority.choices, default=CapabilityPriority.BEFORE_SCALING)
+    evidence = models.JSONField(default=dict, blank=True)
+
+
+class PortfolioCapabilityGap(TimestampedModel):
+    proposal = models.ForeignKey(CompanyProposal, on_delete=models.CASCADE, related_name="capability_gap_reports")
+    available = models.JSONField(default=list, blank=True)
+    partial = models.JSONField(default=list, blank=True)
+    missing = models.JSONField(default=list, blank=True)
+    ranked_missing = models.JSONField(default=list, blank=True)
+    report = models.TextField(blank=True)
+    metadata = models.JSONField(default=dict, blank=True)
+
+
+class VentureArtifact(TimestampedModel):
+    proposal = models.ForeignKey(CompanyProposal, on_delete=models.CASCADE, null=True, blank=True, related_name="artifacts")
+    mandate = models.ForeignKey(CompanyMandate, on_delete=models.CASCADE, null=True, blank=True, related_name="artifacts")
+    graph_run = models.ForeignKey("graph.GraphRun", on_delete=models.SET_NULL, null=True, blank=True, related_name="venture_artifacts")
+    artifact_type = models.CharField(max_length=80)
+    name = models.CharField(max_length=255)
+    content = models.JSONField(default=dict, blank=True)
+    readable = models.TextField(blank=True)
+    generated_by = models.CharField(max_length=120, blank=True)

graph/bootstrap.py

@@ -9,6 +9,7 @@ from graph.roadmap import project_roadmap_review_graph_v1
 from graph.scenario_lab import scenario_lab_graph_v1
 from graph.steward import steward_run_graph_v1
 from graph.task_execution import task_execution_graph_v1
+from graph.venture_discovery import venture_discovery_graph_v1
 
 
 def champion_task_execution_graph_v1() -> ExecutionGraphVersion:
@@ -96,3 +97,7 @@ def champion_scenario_lab_graph_v1() -> ExecutionGraphVersion:
 
 def champion_agent_investigation_graph_v1() -> ExecutionGraphVersion:
     return _champion_graph(agent_investigation_graph_v1())
+
+
+def champion_venture_discovery_graph_v1() -> ExecutionGraphVersion:
+    return _champion_graph(venture_discovery_graph_v1())

graph/venture_discovery.py

@@ -0,0 +1,157 @@
+from __future__ import annotations
+
+from agents.venture_discovery import VentureDiscoveryService
+from control_plane.ventures.models import CompanyMandate, CompanyProposal, ICDiligence, ICDecision, PortfolioCapabilityGap
+from graph.native_runtime import GraphExecutionContext
+from graph.registry import NodeHandlerRegistry, NodeResult
+from graph.spec import ExecutionGraphSpec, GraphEdgeSpec, GraphNodeSpec
+
+
+def venture_discovery_graph_v1() -> ExecutionGraphSpec:
+    nodes = ["prepare_mandate", "generate_company", "board_review", "revise_pitch", "ic_first_pass", "generate_questions", "company_response", "red_team", "final_response", "score", "ic_decision", "capability_analysis", "produce_investment_memo", "complete"]
+    spec = ExecutionGraphSpec(
+        name="venture_discovery",
+        version=1,
+        graph_type="VENTURE_DISCOVERY",
+        entry="prepare_mandate",
+        nodes={node: GraphNodeSpec(node, node if node == "complete" else f"venture_{node}") for node in nodes},
+        edges=[GraphEdgeSpec(nodes[index], nodes[index + 1], "success") for index in range(len(nodes) - 1)],
+        terminal_nodes=["complete"],
+        metadata={"description": "Venture Discovery V0: exactly one startup proposal through Board, IC diligence, scoring, memo, and capability-gap analysis."},
+    )
+    spec.validate()
+    return spec
+
+
+class VentureNode:
+    idempotent = True
+    replay_safe = True
+    destructive = False
+
+    def __init__(self, service: VentureDiscoveryService, node_type: str) -> None:
+        self.service = service
+        self.node_type = node_type
+
+    def mandate(self, context: GraphExecutionContext) -> CompanyMandate:
+        return CompanyMandate.objects.get(id=context.graph_run.metadata["mandate_id"])
+
+    def proposal(self, context: GraphExecutionContext) -> CompanyProposal:
+        return CompanyProposal.objects.get(id=context.graph_run.metadata["proposal_id"])
+
+    def diligence(self, context: GraphExecutionContext) -> ICDiligence:
+        return ICDiligence.objects.get(id=context.graph_run.metadata["diligence_id"])
+
+    def gap(self, context: GraphExecutionContext) -> PortfolioCapabilityGap:
+        return PortfolioCapabilityGap.objects.get(id=context.graph_run.metadata["capability_gap_id"])
+
+
+class PrepareMandateNode(VentureNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        mandate = self.service.create_v0_mandate()
+        metadata = dict(context.graph_run.metadata)
+        metadata["mandate_id"] = str(mandate.id)
+        metadata["exactly_one_company_required"] = True
+        metadata["no_real_spend"] = True
+        metadata["no_real_customer_outreach"] = True
+        context.graph_run.metadata = metadata
+        context.graph_run.save(update_fields=["metadata", "updated_at"])
+        return NodeResult("COMPLETE", "success", {"mandate_id": str(mandate.id), "objective": mandate.objective})
+
+
+class GenerateCompanyNode(VentureNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        proposal = self.service.generate_single_company(self.mandate(context), graph_run=context.graph_run)
+        metadata = dict(context.graph_run.metadata)
+        metadata["proposal_id"] = str(proposal.id)
+        context.graph_run.metadata = metadata
+        context.graph_run.save(update_fields=["metadata", "updated_at"])
+        return NodeResult("COMPLETE", "success", {"proposal_id": str(proposal.id), "company": proposal.title, "generation_source": proposal.metadata.get("generation_source")})
+
+
+class BoardReviewNode(VentureNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        review = self.service.board_review(self.proposal(context), graph_run=context.graph_run)
+        return NodeResult("COMPLETE", "success", {"board_review_id": str(review.id), "recommendation": review.recommendation})
+
+
+class RevisePitchNode(VentureNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        proposal = self.proposal(context)
+        latest = proposal.board_reviews.order_by("-created_at").first()
+        if latest and latest.revised_pitch:
+            proposal.pitch = latest.revised_pitch
+            proposal.save(update_fields=["pitch", "updated_at"])
+        return NodeResult("COMPLETE", "success", {"proposal_id": str(proposal.id), "pitch_revised": latest is not None})
+
+
+class ICFirstPassNode(VentureNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        diligence = self.service.start_ic_diligence(self.proposal(context), graph_run=context.graph_run)
+        metadata = dict(context.graph_run.metadata)
+        metadata["diligence_id"] = str(diligence.id)
+        context.graph_run.metadata = metadata
+        context.graph_run.save(update_fields=["metadata", "updated_at"])
+        return NodeResult("COMPLETE", "success", {"diligence_id": str(diligence.id), "status": diligence.status})
+
+
+class ICQuestionsNode(VentureNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        questions = self.service.generate_ic_questions(self.diligence(context), graph_run=context.graph_run)
+        return NodeResult("COMPLETE", "success", {"question_count": len(questions), "question_ids": [str(q.id) for q in questions]})
+
+
+class CompanyResponseNode(VentureNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        responses = self.service.answer_questions(self.diligence(context), graph_run=context.graph_run)
+        return NodeResult("COMPLETE", "success", {"response_count": len(responses)})
+
+
+class RedTeamNode(VentureNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        challenge = self.service.red_team(self.diligence(context), graph_run=context.graph_run)
+        return NodeResult("COMPLETE", "success", {"concern_count": len(challenge.get("concerns", []))})
+
+
+class FinalResponseNode(VentureNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        response = self.service.final_company_response(self.diligence(context), graph_run=context.graph_run)
+        return NodeResult("COMPLETE", "success", {"kill_criteria_count": len(response.get("kill_criteria", []))})
+
+
+class ScoreNode(VentureNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        decision = self.service.score_and_decide(self.diligence(context), graph_run=context.graph_run)
+        metadata = dict(context.graph_run.metadata)
+        metadata["decision_id"] = str(decision.id)
+        context.graph_run.metadata = metadata
+        context.graph_run.save(update_fields=["metadata", "updated_at"])
+        return NodeResult("COMPLETE", "success", {"decision_id": str(decision.id), "score": decision.composite_score})
+
+
+class ICDecisionNode(VentureNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        decision = ICDecision.objects.get(id=context.graph_run.metadata["decision_id"])
+        return NodeResult("COMPLETE", "success", {"decision": decision.decision, "initial_tranche": str(decision.initial_tranche or "")})
+
+
+class CapabilityAnalysisNode(VentureNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        gap = self.service.capability_analysis(self.proposal(context), graph_run=context.graph_run)
+        metadata = dict(context.graph_run.metadata)
+        metadata["capability_gap_id"] = str(gap.id)
+        context.graph_run.metadata = metadata
+        context.graph_run.save(update_fields=["metadata", "updated_at"])
+        return NodeResult("COMPLETE", "success", {"capability_gap_id": str(gap.id), "missing_count": len(gap.missing)})
+
+
+class InvestmentMemoNode(VentureNode):
+    def run(self, context: GraphExecutionContext) -> NodeResult:
+        memo = self.service.produce_investment_memo(self.diligence(context), self.gap(context), graph_run=context.graph_run)
+        return NodeResult("COMPLETE", "success", {"memo_artifact_id": str(memo.id), "artifact_type": memo.artifact_type})
+
+
+def venture_discovery_registry(service: VentureDiscoveryService) -> NodeHandlerRegistry:
+    registry = NodeHandlerRegistry()
+    for handler in [PrepareMandateNode(service, "venture_prepare_mandate"), GenerateCompanyNode(service, "venture_generate_company"), BoardReviewNode(service, "venture_board_review"), RevisePitchNode(service, "venture_revise_pitch"), ICFirstPassNode(service, "venture_ic_first_pass"), ICQuestionsNode(service, "venture_generate_questions"), CompanyResponseNode(service, "venture_company_response"), RedTeamNode(service, "venture_red_team"), FinalResponseNode(service, "venture_final_response"), ScoreNode(service, "venture_score"), ICDecisionNode(service, "venture_ic_decision"), CapabilityAnalysisNode(service, "venture_capability_analysis"), InvestmentMemoNode(service, "venture_produce_investment_memo")]:
+        registry.register(handler)
+    return registry

templates/control_plane/activity.html

@@ -0,0 +1,2 @@
+{% extends "control_plane/base.html" %}
+{% block content %}<header class="page"><div><h2>Activity{% if project %} / {{ project.name }}{% endif %}</h2><p class="muted">Unified persisted event stream.</p></div></header><section class="panel">{% include "control_plane/partials/event_list.html" with events=events %}</section>{% endblock %}

templates/control_plane/agent_detail.html

@@ -0,0 +1,2 @@
+{% extends "control_plane/base.html" %}
+{% block content %}<header class="page"><div><h2>{{ version.agent }} v{{ version.version }}</h2><p>{{ version.system_contract }}</p></div><span class="badge {{ health.status }}">{{ health.status }}</span></header><section class="grid"><div class="panel"><h3>Champion / Challenger</h3><p>Status: {{ version.promotion_status }}</p><p>Benchmark: {{ version.benchmark_status }}</p><p>Parent: {{ version.parent_version_id|default:"none" }}</p><h4>Challengers</h4>{% for challenger in challengers.challengers %}<p>{{ challenger.agent }} v{{ challenger.version }}</p>{% empty %}<p>No challengers.</p>{% endfor %}</div><div class="panel"><h3>Model / Policies</h3><p>{{ version.model }}</p><pre>{{ version.context_policy }}</pre><pre>{{ version.tool_policy }}</pre></div><div class="panel"><h3>Health Reasons</h3>{% for reason in health.reasons %}<p>{{ reason }}</p>{% endfor %}</div></section><section class="grid"><div class="panel"><h3>Competencies</h3>{% for c in version.competencies %}<p>{{ c.competency__key }} / proficiency {{ c.proficiency }} / confidence {{ c.confidence }}</p>{% empty %}<p>No competencies.</p>{% endfor %}</div><div class="panel"><h3>Performance</h3><pre>{{ performance }}</pre></div><div class="panel"><h3>Usage</h3><pre>{{ usage }}</pre></div></section><section class="panel"><h3>Progeny</h3><pre>{{ progeny }}</pre></section>{% endblock %}

templates/control_plane/agents.html

@@ -0,0 +1,2 @@
+{% extends "control_plane/base.html" %}
+{% block content %}<header class="page"><div><h2>Agent Control Room</h2><p class="muted">Governed workforce inventory, champion/challenger state, health, performance, and teams.</p></div></header><table><thead><tr><th>Agent</th><th>Role</th><th>Scope</th><th>Champion</th><th>Health</th><th>Performance</th></tr></thead><tbody>{% for agent in agents %}<tr><td>{{ agent.name }}<br><span class="muted">{{ agent.purpose }}</span></td><td>{{ agent.role }}</td><td>{{ agent.scope }}</td><td>{% if agent.champion_version %}<a href="{% url 'agent_detail' agent.champion_version %}">{{ agent.champion_version }}</a>{% else %}None{% endif %}</td><td><span class="badge {{ agent.health.status }}">{{ agent.health.status }}</span><br>{{ agent.health.reasons|join:", " }}</td><td>{% if agent.performance.quality %}Completion {{ agent.performance.quality.task_completion_rate|default:"n/a" }}<br>Test pass {{ agent.performance.quality.test_pass_rate|default:"n/a" }}{% else %}No data{% endif %}</td></tr>{% empty %}<tr><td colspan="6">No agents.</td></tr>{% endfor %}</tbody></table><section class="panel"><h3>Teams</h3>{% for team in teams %}<article class="card"><strong>{{ team.name }}</strong><p>{{ team.purpose }}</p><table><thead><tr><th>Role</th><th>AgentVersion</th><th>Status</th></tr></thead><tbody>{% for member in team.members %}<tr><td>{{ member.role }}</td><td>{{ member.agent_version__agent__name }} v{{ member.agent_version__version }}</td><td>{{ member.status }}</td></tr>{% endfor %}</tbody></table></article>{% empty %}<p>No teams.</p>{% endfor %}</section>{% endblock %}

templates/control_plane/approvals.html

@@ -0,0 +1,2 @@
+{% extends "control_plane/base.html" %}
+{% block content %}<header class="page"><div><h2>Approvals</h2><p class="muted">Durable approval inbox for paused graph decisions.</p></div></header><table><thead><tr><th>Source</th><th>Project</th><th>Reason</th><th>Evidence</th><th>Requested</th><th>Actions</th></tr></thead><tbody>{% for approval in approvals %}<tr><td><a href="{% url 'graph_run_detail' approval.graph_run.id %}">{{ approval.graph_run.execution_graph_version.graph.name }} v{{ approval.graph_run.execution_graph_version.version }}</a></td><td>{{ approval.graph_run.project.name|default:"-" }}</td><td>{{ approval.reason }}</td><td><pre>{{ approval.payload }}</pre></td><td>{{ approval.created_at }}</td><td><form class="actions" method="post" action="{% url 'approval_action' approval.id %}">{% csrf_token %}<button name="action" value="approve">Approve</button><button class="secondary" name="action" value="reject">Reject</button></form></td></tr>{% empty %}<tr><td colspan="6">No pending approvals.</td></tr>{% endfor %}</tbody></table>{% endblock %}

templates/control_plane/archaeologist.html

@@ -0,0 +1,2 @@
+{% extends "control_plane/base.html" %}
+{% block content %}<header class="page"><div><h2>Archaeologist / {{ project.name }}</h2><p class="muted">Observed system, inferred intent, and gap evidence.</p></div><a class="button secondary" href="{% url 'project_brain' project.id %}">Discuss with Project Brain</a></header><section class="grid"><div class="panel"><h3>Observed System</h3><p>{{ observed|default:"No architecture summary yet." }}</p></div><div class="panel"><h3>Evidence Artifacts</h3>{% for artifact in artifacts %}<p>{{ artifact.name }} <span class="badge">{{ artifact.artifact_type }}</span></p>{% empty %}<p>No archaeologist artifacts.</p>{% endfor %}</div><div class="panel"><h3>Gap Analysis</h3>{% for finding in findings %}<p><span class="badge {{ finding.severity }}">{{ finding.severity }}</span> {{ finding.title }}</p>{% empty %}<p>No findings.</p>{% endfor %}</div></section>{% endblock %}

templates/control_plane/base.html

@@ -0,0 +1,64 @@
+<!doctype html>
+<html lang="en">
+  <head>
+    <meta charset="utf-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1">
+    <title>{% block title %}Artifex{% endblock %}</title>
+    <style>
+      :root { --bg: #f6f2ea; --panel: #fffdfa; --line: #d8cfc2; --ink: #162033; --muted: #685f55; --accent: #284c7e; --danger: #9f2d2d; --warn: #9b681c; --ok: #23704a; }
+      * { box-sizing: border-box; }
+      body { margin: 0; font-family: ui-sans-serif, system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif; color: var(--ink); background: var(--bg); }
+      a { color: var(--accent); text-decoration: none; }
+      a:hover { text-decoration: underline; }
+      .shell { display: grid; grid-template-columns: 240px 1fr; min-height: 100vh; }
+      nav { border-right: 1px solid var(--line); background: #ede6da; padding: 1.25rem; position: sticky; top: 0; height: 100vh; }
+      nav h1 { font-size: 1.15rem; letter-spacing: .12em; margin: 0 0 1.5rem; }
+      nav a { display: block; padding: .55rem .65rem; border-radius: 8px; color: var(--ink); }
+      nav a:hover { background: rgba(40, 76, 126, .09); text-decoration: none; }
+      main { padding: 1.5rem; max-width: 1480px; width: 100%; }
+      header.page { display: flex; justify-content: space-between; gap: 1rem; align-items: flex-start; margin-bottom: 1rem; }
+      h2, h3 { margin-bottom: .4rem; }
+      .grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(240px, 1fr)); gap: 1rem; }
+      .card, .panel { background: var(--panel); border: 1px solid var(--line); border-radius: 14px; padding: 1rem; margin-bottom: 1rem; box-shadow: 0 1px 0 rgba(0,0,0,.03); }
+      .metric { font-size: 2rem; font-weight: 760; line-height: 1; }
+      .muted { color: var(--muted); }
+      .badge { display: inline-block; padding: .18rem .45rem; border-radius: 999px; border: 1px solid var(--line); background: #f4efe6; font-size: .75rem; font-weight: 700; letter-spacing: .04em; }
+      .badge.COMPLETE, .badge.PASS, .badge.HEALTHY, .badge.APPROVED { color: var(--ok); border-color: rgba(35,112,74,.35); }
+      .badge.FAILED, .badge.FAIL, .badge.DEGRADED, .badge.REJECTED { color: var(--danger); border-color: rgba(159,45,45,.35); }
+      .badge.RUNNING, .badge.PAUSED, .badge.WATCH, .badge.PENDING { color: var(--warn); border-color: rgba(155,104,28,.35); }
+      table { width: 100%; border-collapse: collapse; background: var(--panel); border: 1px solid var(--line); border-radius: 12px; overflow: hidden; }
+      th, td { text-align: left; padding: .65rem .75rem; border-bottom: 1px solid var(--line); vertical-align: top; }
+      th { font-size: .78rem; text-transform: uppercase; letter-spacing: .06em; color: var(--muted); background: #f1eadf; }
+      tr:last-child td { border-bottom: 0; }
+      button, .button { border: 1px solid var(--accent); color: white; background: var(--accent); border-radius: 8px; padding: .45rem .7rem; cursor: pointer; font-weight: 650; }
+      button.secondary, .button.secondary { color: var(--ink); background: transparent; border-color: var(--line); }
+      .actions { display: flex; gap: .4rem; flex-wrap: wrap; }
+      pre { white-space: pre-wrap; word-break: break-word; background: #201d19; color: #f8f2e8; padding: .8rem; border-radius: 10px; max-height: 360px; overflow: auto; }
+      .timeline { border-left: 3px solid var(--line); margin-left: .5rem; padding-left: 1rem; }
+      .node { margin: .5rem 0; }
+      @media (max-width: 860px) { .shell { grid-template-columns: 1fr; } nav { position: static; height: auto; } nav a { display: inline-block; } main { padding: 1rem; } }
+    </style>
+    {% block head %}{% endblock %}
+  </head>
+  <body>
+    <div class="shell">
+      <nav aria-label="Primary navigation">
+        <h1>ARTIFEX</h1>
+        <a href="/">Dashboard</a>
+        <a href="/projects/">Projects</a>
+        <a href="/agents/">Agents</a>
+        <a href="/progeny/">Progeny</a>
+        <a href="/steward/">Steward</a>
+        <a href="/roadmap/">Roadmap</a>
+        <a href="/scenario-lab/">Scenario Lab</a>
+        <a href="/approvals/">Approvals</a>
+        <a href="/resources/">Resources</a>
+        <a href="/activity/">Activity</a>
+        <a href="/admin/">Settings</a>
+      </nav>
+      <main id="main">
+        {% block content %}{% endblock %}
+      </main>
+    </div>
+  </body>
+</html>

templates/control_plane/dashboard.html

@@ -0,0 +1,16 @@
+{% extends "control_plane/base.html" %}
+{% block title %}Dashboard / Artifex{% endblock %}
+{% block content %}
+<header class="page"><div><h2>Dashboard</h2><p class="muted">Operational control plane for projects, graphs, agents, approvals, and autonomous lifecycle work.</p></div><a class="button secondary" href="/approvals/">Approval Inbox: {{ approval_count }}</a></header>
+<section class="grid">
+  <div class="card"><div class="metric">{{ project_summary.total }}</div><strong>Projects</strong><p>{{ project_summary.active }} active, {{ project_summary.blocked_failed }} blocked/failed</p></div>
+  <div class="card"><div class="metric">{{ execution_summary.active }}</div><strong>Active GraphRuns</strong><p>{{ execution_summary.failed }} failed. Champion task graph: {{ execution_summary.champion_task_graph|default:"not bootstrapped" }}</p></div>
+  <div class="card"><div class="metric">{{ steward_summary.open_findings }}</div><strong>Steward Findings</strong><p>{{ steward_summary.high_findings }} high/critical across {{ steward_summary.enrolled_projects }} enrolled projects</p></div>
+  <div class="card"><div class="metric">{{ progeny_summary.unresolved_signals }}</div><strong>Progeny Signals</strong><p>{{ progeny_summary.open_investigations }} investigations, {{ progeny_summary.challengers }} challengers</p></div>
+  <div class="card"><div class="metric">{{ agent_summary.agent_count }}</div><strong>Agents</strong><p>{{ agent_summary.watch_degraded }} watch/degraded, {{ agent_summary.active_runs }} active runs</p></div>
+</section>
+<section class="grid">
+  <div class="panel"><h3>Recent GraphRuns</h3>{% include "control_plane/partials/graph_run_table.html" with graph_runs=recent.graph_runs %}</div>
+  <div class="panel"><h3>Recent Activity</h3>{% include "control_plane/partials/event_list.html" with events=recent.events %}</div>
+</section>
+{% endblock %}

templates/control_plane/explore.html

@@ -0,0 +1,2 @@
+{% extends "control_plane/base.html" %}
+{% block content %}<header class="page"><div><h2>Explore{% if project %} / {{ project.name }}{% endif %}</h2><p class="muted">Ranked opportunities. Explorer results never directly invoke Coder.</p></div>{% if project %}<form method="post" action="{% url 'run_explore' project.id %}">{% csrf_token %}<button type="submit">Run Explore</button></form>{% endif %}</header><table><thead><tr><th>Opportunity</th><th>Project</th><th>Type</th><th>Scores</th><th>Status</th><th>Actions</th></tr></thead><tbody>{% for item in opportunities %}<tr><td>{{ item.title }}<br><span class="muted">{{ item.description }}</span></td><td>{{ item.project.name }}</td><td>{{ item.opportunity_type }}</td><td>V {{ item.value_score }} / E {{ item.effort_score }} / R {{ item.risk_score }} / C {{ item.confidence }} / T {{ item.technical_fit }} / S {{ item.strategic_fit }}</td><td><span class="badge {{ item.status }}">{{ item.status }}</span></td><td><form class="actions" method="post" action="{% url 'opportunity_action' item.id %}">{% csrf_token %}<button name="action" value="extend">Extend</button><button name="action" value="evolve">Evolve</button><button name="action" value="roadmap">Add to Roadmap</button><button class="secondary" name="action" value="defer">Defer</button><button class="secondary" name="action" value="reject">Reject</button></form></td></tr>{% empty %}<tr><td colspan="6">No opportunities.</td></tr>{% endfor %}</tbody></table>{% endblock %}

templates/control_plane/graph_run.html

@@ -0,0 +1,3 @@
+{% extends "control_plane/base.html" %}
+{% block head %}<meta http-equiv="refresh" content="5">{% endblock %}
+{% block content %}<header class="page"><div><h2>GraphRun {{ graph_run.id }}</h2><p class="muted">Live execution graph viewer. This page auto-refreshes every 5 seconds.</p></div><a class="button secondary" href="{% url 'graph_run_json' graph_run.id %}">JSON</a></header>{% include "control_plane/partials/graph_run_status.html" %}<section class="grid"><div class="panel"><h3>Edges</h3>{% for edge in traversals %}<p>{{ edge.source_node }} → {{ edge.target_node }} <span class="badge">{{ edge.result|default:edge.condition }}</span></p>{% empty %}<p>No traversed edges.</p>{% endfor %}</div><div class="panel"><h3>Approvals</h3>{% for approval in approvals %}<p><span class="badge {{ approval.status }}">{{ approval.status }}</span> {{ approval.reason }}</p>{% empty %}<p>No approvals.</p>{% endfor %}</div></section>{% endblock %}

templates/control_plane/partials/event_list.html

@@ -0,0 +1 @@
+{% for event in events %}<p><span class="badge">{{ event.event_type }}</span> {{ event.created_at }} {% if event.project %}<a href="{% url 'project_workspace' event.project.id %}">{{ event.project.name }}</a>{% endif %}<br><span class="muted">{{ event.actor }}</span></p>{% empty %}<p>No activity.</p>{% endfor %}

templates/control_plane/partials/graph_run_status.html

@@ -0,0 +1 @@
+<section class="panel"><h3>{{ graph_run.execution_graph_version.graph.name }} v{{ graph_run.execution_graph_version.version }} <span class="badge {{ graph_run.status }}">{{ graph_run.status }}</span></h3><p>Runtime: LangGraphRuntime / Current node: {{ graph_run.current_node }}</p><div class="timeline">{% for node in nodes %}<div class="node"><span class="badge {{ node.status }}">{{ node.status }}</span> <strong>{{ node.node_id }}</strong> visit {{ node.visit_index }}{% if node.agent_version %}<br><span class="muted">{{ node.agent_version.agent.name }} v{{ node.agent_version.version }} / {{ node.agent_version.model }}</span>{% endif %}{% if node.failure_evidence %}<pre>{{ node.failure_evidence }}</pre>{% endif %}{% if node.output_metadata %}<pre>{{ node.output_metadata }}</pre>{% endif %}</div>{% empty %}<p>No node runs.</p>{% endfor %}</div></section>

templates/control_plane/partials/graph_run_table.html

@@ -0,0 +1 @@
+<table><thead><tr><th>Graph</th><th>Project</th><th>Status</th><th>Current Node</th><th>Updated</th></tr></thead><tbody>{% for run in graph_runs %}<tr><td><a href="{% url 'graph_run_detail' run.id %}">{{ run.execution_graph_version.graph.name }} v{{ run.execution_graph_version.version }}</a></td><td>{{ run.project.name|default:"-" }}</td><td><span class="badge {{ run.status }}">{{ run.status }}</span></td><td>{{ run.current_node }}</td><td>{{ run.updated_at }}</td></tr>{% empty %}<tr><td colspan="5">No graph runs.</td></tr>{% endfor %}</tbody></table>

templates/control_plane/partials/project_dag.html

@@ -0,0 +1 @@
+{% for milestone in milestones %}<article class="card"><h4>{{ milestone.key }} / {{ milestone.title }} <span class="badge {{ milestone.status }}">{{ milestone.status }}</span></h4><p>{{ milestone.goal }}</p>{% for feature in milestone.features.all %}<div class="card"><strong>{{ feature.title }}</strong> <span class="badge {{ feature.status }}">{{ feature.status }}</span><ul>{% for task in feature.tasks.all %}<li><a href="{% url 'task_detail' task.id %}">{{ task.goal|truncatechars:120 }}</a> <span class="badge {{ task.status }}">{{ task.status }}</span></li>{% endfor %}</ul></div>{% empty %}<ul>{% for task in milestone.tasks.all %}<li><a href="{% url 'task_detail' task.id %}">{{ task.goal|truncatechars:120 }}</a> <span class="badge {{ task.status }}">{{ task.status }}</span></li>{% empty %}<li>No tasks.</li>{% endfor %}</ul>{% endfor %}</article>{% empty %}<p>No DAG materialized.</p>{% endfor %}

templates/control_plane/progeny.html

@@ -0,0 +1,2 @@
+{% extends "control_plane/base.html" %}
+{% block content %}<header class="page"><div><h2>Progeny</h2><p class="muted">Signals, investigations, experiments, and improvement candidates.</p></div></header><section class="grid"><div class="panel"><h3>Grouped Inbox</h3>{% for key,count in groups %}<p><span class="badge">{{ count }}</span> {{ key }}</p>{% empty %}<p>No open groups.</p>{% endfor %}</div><div class="panel"><h3>Investigations</h3>{% for inv in investigations %}<p><span class="badge {{ inv.status }}">{{ inv.status }}</span> {{ inv.recommended_target }} / {{ inv.recommended_route }} / confidence {{ inv.confidence }}</p>{% empty %}<p>No investigations.</p>{% endfor %}</div></section><section class="panel"><h3>Signals</h3><table><thead><tr><th>Signal</th><th>Source</th><th>Project</th><th>Agent</th><th>Severity</th><th>Status</th></tr></thead><tbody>{% for signal in signals %}<tr><td>{{ signal.summary }}<br><span class="muted">{{ signal.failure_category }} / {{ signal.grouping_key }}</span></td><td>{{ signal.source }}</td><td>{{ signal.project.name|default:"-" }}</td><td>{{ signal.agent_version.agent.name|default:"-" }}</td><td><span class="badge {{ signal.severity }}">{{ signal.severity }}</span></td><td><span class="badge {{ signal.status }}">{{ signal.status }}</span></td></tr>{% empty %}<tr><td colspan="6">No signals.</td></tr>{% endfor %}</tbody></table></section><section class="panel"><h3>Replay Arena Experiments</h3>{% for experiment in experiments %}<p>{{ experiment.hypothesis }} <span class="badge {{ experiment.status }}">{{ experiment.status }}</span></p>{% empty %}<p>No experiments.</p>{% endfor %}</section>{% endblock %}

templates/control_plane/project_brain.html

@@ -0,0 +1,2 @@
+{% extends "control_plane/base.html" %}
+{% block content %}<header class="page"><div><h2>Project Brain / {{ project.name }}</h2><p class="muted">Project-scoped strategic decisions, plans, and artifacts.</p></div></header><section class="panel"><form method="post">{% csrf_token %}<label for="message">Conversation note / proposed decision</label><textarea id="message" name="message" rows="4" style="width:100%"></textarea><p><button type="submit">Persist Project Brain Note</button></p></form></section><section class="grid"><div class="panel"><h3>Decisions</h3>{% for decision in decisions %}<p><span class="badge">{{ decision.decision_type }}</span> {{ decision.decision }}</p>{% empty %}<p>No decisions.</p>{% endfor %}</div><div class="panel"><h3>Plans</h3>{% for plan in plans %}<p>v{{ plan.version }} {{ plan.scope|default:plan.goal }}</p>{% empty %}<p>No plans.</p>{% endfor %}</div></section>{% endblock %}

templates/control_plane/project_workspace.html

@@ -0,0 +1,9 @@
+{% extends "control_plane/base.html" %}
+{% block title %}{{ project.name }} / Artifex{% endblock %}
+{% block content %}
+<header class="page"><div><h2>{{ project.name }}</h2><p>{{ project.goal }}</p><p class="muted">{{ project.repository_path|default:project.repository_url }}</p></div><div class="actions"><a class="button secondary" href="{% url 'project_brain' project.id %}">Project Brain</a><a class="button secondary" href="{% url 'project_archaeologist' project.id %}">Archaeologist</a><form method="post" action="{% url 'run_explore' project.id %}">{% csrf_token %}<button type="submit">Explore</button></form></div></header>
+<section class="grid"><div class="card"><strong>Status</strong><br><span class="badge {{ project.status }}">{{ project.status }}</span><p>{{ warnings|join:", "|default:"No active warnings" }}</p></div><div class="card"><strong>Current Plan</strong><p>v{{ plan.version|default:"none" }} {{ plan.scope|default:"" }}</p></div><div class="card"><strong>Tasks</strong><p>{{ tasks|length }} total</p></div><div class="card"><strong>Scenario Coverage</strong><p>{% for key,value in scenario_coverage.items %}{{ key }}: {{ value }} {% empty %}None{% endfor %}</p></div></section>
+<section class="panel"><h3>Project DAG</h3>{% include "control_plane/partials/project_dag.html" %}</section>
+<section class="grid"><div class="panel"><h3>Graph Runs</h3>{% include "control_plane/partials/graph_run_table.html" with graph_runs=graph_runs %}</div><div class="panel"><h3>Roadmap NOW</h3>{% for item in roadmap.NOW %}<p><a href="{% url 'project_roadmap' project.id %}">{{ item.title }}</a> <span class="badge">{{ item.target_action }}</span></p>{% empty %}<p>No NOW items.</p>{% endfor %}</div></section>
+<section class="grid"><div class="panel"><h3>Latest Commits</h3>{% for commit in commits %}<p><code>{{ commit.sha|slice:":10" }}</code> {{ commit.message|truncatechars:90 }}</p>{% empty %}<p>No commits.</p>{% endfor %}</div><div class="panel"><h3>Activity</h3>{% include "control_plane/partials/event_list.html" with events=activity %}</div></section>
+{% endblock %}

templates/control_plane/projects.html

@@ -0,0 +1,8 @@
+{% extends "control_plane/base.html" %}
+{% block title %}Projects / Artifex{% endblock %}
+{% block content %}
+<header class="page"><div><h2>Projects</h2><p class="muted">Create and inspect autonomous engineering workspaces.</p></div></header>
+<table><thead><tr><th>Project</th><th>Status</th><th>Repository</th><th>Tasks</th><th>Steward</th><th>Latest GraphRun</th><th>Warnings</th><th>Actions</th></tr></thead><tbody>
+{% for row in rows %}<tr><td><a href="{% url 'project_workspace' row.project.id %}">{{ row.project.name }}</a><br><span class="muted">{{ row.project.goal|truncatechars:100 }}</span></td><td><span class="badge {{ row.project.status }}">{{ row.project.status }}</span></td><td>{{ row.project.repository_path|default:row.project.repository_url }}</td><td>{{ row.task_complete }}/{{ row.task_total }}</td><td>{{ row.steward_state.status|default:"not enrolled" }}<br>{{ row.open_findings }} findings</td><td>{% if row.latest_graph_run %}<a href="{% url 'graph_run_detail' row.latest_graph_run.id %}">{{ row.latest_graph_run.execution_graph_version.graph.name }} / {{ row.latest_graph_run.status }}</a>{% else %}None{% endif %}</td><td>{{ row.warnings|join:", "|default:"none" }}</td><td class="actions"><a class="button secondary" href="{% url 'project_workspace' row.project.id %}">Open</a><form method="post" action="{% url 'run_explore' row.project.id %}">{% csrf_token %}<button type="submit">Explore</button></form></td></tr>{% empty %}<tr><td colspan="8">No projects yet.</td></tr>{% endfor %}
+</tbody></table>
+{% endblock %}

templates/control_plane/resources.html

@@ -0,0 +1,2 @@
+{% extends "control_plane/base.html" %}
+{% block content %}<header class="page"><div><h2>Resources</h2><p class="muted">Compute/provider visibility without credentials.</p></div></header><table><thead><tr><th>Resource</th><th>Kind</th><th>Provider</th><th>Health</th><th>Roles</th><th>Recent Requests</th><th>Median Latency</th></tr></thead><tbody>{% for row in resources %}<tr><td>{{ row.resource.name }}</td><td>{{ row.resource.kind }}</td><td>{{ row.resource.provider }} {{ row.resource.config.model|default:"" }}</td><td><span class="badge {{ row.resource.health_status }}">{{ row.resource.health_status }}</span></td><td>{{ row.resource.roles }}</td><td>{{ row.request_count }}</td><td>{{ row.median_latency|default:"n/a" }}</td></tr>{% empty %}<tr><td colspan="7">No resources.</td></tr>{% endfor %}</tbody></table>{% endblock %}

templates/control_plane/roadmap.html

@@ -0,0 +1,2 @@
+{% extends "control_plane/base.html" %}
+{% block content %}<header class="page"><div><h2>Roadmap{% if project %} / {{ project.name }}{% endif %}</h2><p class="muted">Persistent future intent. Conversion creates lifecycle candidates, not immediate coding work.</p></div></header><section class="grid">{% for horizon,items in board.items %}<div class="panel"><h3>{{ horizon }}</h3>{% for item in items %}<article class="card"><strong>{{ item.title }}</strong><p>{{ item.description }}</p><p><span class="badge">{{ item.source }}</span> <span class="badge">{{ item.target_action }}</span> <span class="badge {{ item.status }}">{{ item.status }}</span></p><p>Composite {{ item.composite_score }} / Value {{ item.value_score }} / Effort {{ item.effort_score }} / Risk {{ item.risk_score }}</p><form class="actions" method="post" action="{% url 'roadmap_action' item.id %}">{% csrf_token %}<button name="action" value="NOW">NOW</button><button name="action" value="NEXT">NEXT</button><button name="action" value="LATER">LATER</button><button name="action" value="EXPLORING">EXPLORING</button><button name="action" value="extend">Extend</button><button name="action" value="evolve">Evolve</button><button class="secondary" name="action" value="defer">Defer</button><button class="secondary" name="action" value="reject">Reject</button></form></article>{% empty %}<p>No items.</p>{% endfor %}</div>{% endfor %}</section>{% endblock %}

templates/control_plane/scenarios.html

@@ -0,0 +1,2 @@
+{% extends "control_plane/base.html" %}
+{% block content %}<header class="page"><div><h2>Scenario Lab{% if project %} / {{ project.name }}{% endif %}</h2><p class="muted">Unusual, degraded, adversarial, and high-load scenario portfolio.</p></div></header><section class="grid"><div class="panel"><h3>Coverage</h3>{% for key,value in coverage.items %}<p>{{ key }}: {{ value }}</p>{% empty %}<p>No coverage yet.</p>{% endfor %}</div><div class="panel"><h3>Suites</h3>{% for suite in suites %}<p>{{ suite.name }} v{{ suite.version }} <span class="badge {{ suite.status }}">{{ suite.status }}</span> / {{ suite.project.name }}</p>{% empty %}<p>No suites.</p>{% endfor %}</div></section><section class="panel"><h3>Recent Runs</h3><table><thead><tr><th>Scenario</th><th>Project</th><th>Result</th><th>Evidence</th></tr></thead><tbody>{% for run in runs %}<tr><td>{{ run.scenario.title }}</td><td>{{ run.project.name }}</td><td><span class="badge {{ run.result }}">{{ run.result }}</span></td><td><pre>{{ run.failure_evidence }}</pre></td></tr>{% empty %}<tr><td colspan="4">No runs.</td></tr>{% endfor %}</tbody></table></section><section class="panel"><h3>Findings</h3><table><thead><tr><th>Finding</th><th>Route</th><th>Status</th><th>Action</th></tr></thead><tbody>{% for finding in findings %}<tr><td>{{ finding.title }}<br>{{ finding.summary }}</td><td>{{ finding.recommended_action }} / {{ finding.recommended_route }}</td><td><span class="badge {{ finding.status }}">{{ finding.status }}</span></td><td><form method="post" action="{% url 'scenario_finding_action' finding.id %}">{% csrf_token %}<button name="action" value="roadmap">Route / Add Roadmap</button></form></td></tr>{% empty %}<tr><td colspan="4">No findings.</td></tr>{% endfor %}</tbody></table></section>{% endblock %}

templates/control_plane/steward.html

@@ -0,0 +1,2 @@
+{% extends "control_plane/base.html" %}
+{% block content %}<header class="page"><div><h2>Steward{% if project %} / {{ project.name }}{% endif %}</h2><p class="muted">Enrollment runs, classified findings, evidence, and routed lifecycle actions.</p></div></header><table><thead><tr><th>Finding</th><th>Project</th><th>Severity</th><th>Route</th><th>Occurrences</th><th>Seen</th><th>Evidence</th></tr></thead><tbody>{% for finding in findings %}<tr><td>{{ finding.title }}<br><span class="muted">{{ finding.summary }}</span></td><td>{{ finding.project.name }}</td><td><span class="badge {{ finding.severity }}">{{ finding.severity }}</span></td><td>{{ finding.recommended_action }} / {{ finding.recommended_route }}</td><td>{{ finding.occurrence_count }}</td><td>{{ finding.first_seen|default:finding.created_at }} → {{ finding.last_seen|default:finding.updated_at }}</td><td><pre>{{ finding.evidence }}</pre></td></tr>{% empty %}<tr><td colspan="7">No Steward findings.</td></tr>{% endfor %}</tbody></table>{% endblock %}

templates/control_plane/task.html

@@ -0,0 +1,2 @@
+{% extends "control_plane/base.html" %}
+{% block content %}<header class="page"><div><h2>Task</h2><p>{{ task.goal }}</p></div><span class="badge {{ task.status }}">{{ task.status }}</span></header><section class="grid"><div class="panel"><h3>Acceptance</h3><pre>{{ task.acceptance_criteria }}</pre><h3>Dependencies</h3>{% for dep in dependencies %}<p>{{ dep.goal }}</p>{% empty %}<p>No dependencies.</p>{% endfor %}</div><div class="panel"><h3>Attempts</h3>{% for attempt in attempts %}<p>#{{ attempt.attempt_number }} <span class="badge {{ attempt.status }}">{{ attempt.status }}</span> {{ attempt.coder.agent.name }} v{{ attempt.coder.version }}</p>{% empty %}<p>No attempts.</p>{% endfor %}</div></section><section class="grid"><div class="panel"><h3>GraphRuns</h3>{% include "control_plane/partials/graph_run_table.html" with graph_runs=graph_runs %}</div><div class="panel"><h3>Tests / Reviews / Commits</h3>{% for test in tests %}<p>Test: <span class="badge {{ test.status }}">{{ test.status }}</span> {{ test.command }}</p>{% endfor %}{% for review in reviews %}<p>Review: <span class="badge {{ review.status }}">{{ review.status }}</span></p>{% endfor %}{% for commit in commits %}<p>Commit <code>{{ commit.sha|slice:":10" }}</code> {{ commit.message }}</p>{% endfor %}</div></section>{% endblock %}

tests/test_control_plane_ui_v1.py

@@ -0,0 +1,123 @@
+from __future__ import annotations
+
+from django.test import Client
+from django.urls import reverse
+
+from agents.control_room import AgentControlRoomService
+from control_plane.agents.management.commands.seed_core_agents import Command as SeedAgentsCommand
+from control_plane.agents.models import AgentRole, AgentVersion, PromotionStatus, ProgenySignal
+from control_plane.events.bus import EventBus
+from control_plane.events.models import EventType
+from control_plane.projects.models import Exploration, ExplorationOpportunity, Project, ProjectPlan, Milestone, RoadmapItem, Scenario, ScenarioFinding, ScenarioRun, ScenarioSuite, StewardFinding, Task, TaskAttempt, TaskStatus
+from control_plane.resources.models import Resource, ResourceKind
+from control_plane.verification.models import Review, TestRun, Verification, VerificationLevel, VerificationResult
+from graph.bootstrap import champion_project_exploration_graph_v1, champion_task_execution_graph_v1
+from graph.models import GraphApproval, GraphRun, GraphRunStatus
+
+
+def fixture() -> dict[str, object]:
+    SeedAgentsCommand().handle()
+    AgentControlRoomService().bootstrap_frontend_agents()
+    team = AgentControlRoomService().create_software_feature_team()
+    Resource.objects.create(name="Qwen", kind=ResourceKind.MODEL, provider="local_inference", roles=["CODING"], health_status="AVAILABLE", config={"model": "qwen38"})
+    project = Project.objects.create(name="UI Dogfood", goal="Operate Artifex through the UI", repository_path="/tmp/ui")
+    plan = ProjectPlan.objects.create(project=project, version=1, goal=project.goal, scope="UI V1")
+    milestone = Milestone.objects.create(project=project, plan=plan, key="UI", title="UI", goal="Build UI")
+    task = Task.objects.create(project=project, milestone=milestone, task_type="implementation", status=TaskStatus.COMPLETE, goal="Render dashboard", acceptance_criteria=["loads"])
+    coder = AgentVersion.objects.get(agent__role=AgentRole.CODER, promotion_status=PromotionStatus.CHAMPION)
+    reviewer = AgentVersion.objects.get(agent__role=AgentRole.REVIEWER, promotion_status=PromotionStatus.CHAMPION)
+    judge = AgentVersion.objects.get(agent__role=AgentRole.PROJECT_JUDGE, promotion_status=PromotionStatus.CHAMPION)
+    TaskAttempt.objects.create(task=task, attempt_number=1, coder=coder, status="COMPLETE")
+    TestRun.objects.create(project=project, task=task, command="pytest", status="PASS")
+    Review.objects.create(task=task, reviewer=reviewer, status="PASS")
+    Verification.objects.create(project=project, task=task, judge=judge, level=VerificationLevel.TASK, result=VerificationResult.PASS)
+    graph_version = champion_task_execution_graph_v1()
+    graph_run = GraphRun.objects.create(project=project, task=task, execution_graph_version=graph_version, current_node="review", status=GraphRunStatus.PAUSED, metadata={"telemetry": {}})
+    node_run = graph_run.node_runs.create(node_id="review", node_type="review", status="PAUSED", visit_index=1, agent_version=reviewer, output_metadata={"review_status": "RUNNING"})
+    approval = GraphApproval.objects.create(graph_run=graph_run, node_run=node_run, reason="AWAITING_UI_APPROVAL")
+    exploration = Exploration.objects.create(project=project, status="COMPLETE")
+    opportunity = ExplorationOpportunity.objects.create(exploration=exploration, project=project, title="Add status filter", description="Filter dashboard cards", opportunity_type="UX", recommended_action="EXTEND", value_score=0.8, effort_score=0.2, risk_score=0.2, confidence=0.8, technical_fit=0.8, strategic_fit=0.8, composite_score=0.75, grouping_key="ui-opp")
+    roadmap_item = RoadmapItem.objects.create(project=project, title="Improve graph viewer", description="Expose node details", horizon="NOW", target_action="EXTEND", grouping_key="ui-roadmap")
+    suite = ScenarioSuite.objects.create(project=project, name="UI Suite")
+    scenario = Scenario.objects.create(project=project, suite=suite, name="Permission denial", title="Permission denial", scenario_type="PERMISSION", target_type="PROJECT", target_id=str(project.id), injected_condition={"mechanism": "permission_denial"}, expected_invariants=["safe"], success_criteria=["pass"], resource_budget={"max_seconds": 5})
+    run = ScenarioRun.objects.create(project=project, scenario=scenario, result="FAIL", status="COMPLETE", failure_evidence={"summary": "denied"})
+    finding = ScenarioFinding.objects.create(project=project, scenario=scenario, scenario_run=run, title="Scenario failed", summary="permission denied", failure_category="PERMISSION", recommended_action="EXTEND", recommended_route="RoadmapItem", grouping_key="ui-scenario")
+    StewardFinding.objects.create(project=project, finding_type="RELIABILITY", title="Open issue", summary="Needs attention", severity="HIGH", confidence=0.8, recommended_action="REPAIR", grouping_key="ui-steward")
+    ProgenySignal.objects.create(project=project, agent_version=coder, source="ui", severity="MEDIUM", failure_category="MALFORMED", summary="Signal", grouping_key="ui-signal")
+    EventBus().publish(EventType.TASK_COMPLETED, project=project, task=task, actor="test")
+    return {"project": project, "task": task, "graph_run": graph_run, "approval": approval, "opportunity": opportunity, "roadmap_item": roadmap_item, "finding": finding, "coder": coder, "team": team}
+
+
+def assert_ok(client: Client, url: str, text: str) -> None:
+    response = client.get(url)
+    assert response.status_code == 200
+    assert text.encode() in response.content
+
+
+def test_control_plane_core_pages_render_and_link() -> None:
+    data = fixture()
+    client = Client()
+    project = data["project"]
+    task = data["task"]
+    graph_run = data["graph_run"]
+    coder = data["coder"]
+
+    assert_ok(client, reverse("dashboard"), "Dashboard")
+    assert_ok(client, reverse("projects"), "UI Dogfood")
+    assert_ok(client, reverse("project_workspace", args=[project.id]), "Project DAG")
+    assert_ok(client, reverse("project_brain", args=[project.id]), "Project Brain")
+    assert_ok(client, reverse("project_archaeologist", args=[project.id]), "Archaeologist")
+    assert_ok(client, reverse("graph_run_detail", args=[graph_run.id]), "Live execution graph viewer")
+    assert_ok(client, reverse("task_detail", args=[task.id]), "Acceptance")
+    assert_ok(client, reverse("steward"), "Open issue")
+    assert_ok(client, reverse("explore"), "Add status filter")
+    assert_ok(client, reverse("roadmap"), "Improve graph viewer")
+    assert_ok(client, reverse("scenarios"), "Permission denial")
+    assert_ok(client, reverse("progeny"), "Signal")
+    assert_ok(client, reverse("agent_control_room"), "Agent Control Room")
+    assert_ok(client, reverse("agent_detail", args=[coder.id]), "Competencies")
+    assert_ok(client, reverse("resources"), "Qwen")
+    assert_ok(client, reverse("approvals"), "AWAITING_UI_APPROVAL")
+    assert_ok(client, reverse("activity"), "TASK_COMPLETED")
+
+
+def test_json_endpoints_and_ui_actions_work() -> None:
+    data = fixture()
+    client = Client()
+    project = data["project"]
+    graph_run = data["graph_run"]
+    approval = data["approval"]
+    opportunity = data["opportunity"]
+    roadmap_item = data["roadmap_item"]
+    finding = data["finding"]
+
+    assert client.get(reverse("graph_run_json", args=[graph_run.id])).json()["status"] == "PAUSED"
+    assert client.get(reverse("project_dag_json", args=[project.id])).json()["tasks"]
+    assert client.get(reverse("agent_performance_json", args=[data["coder"].id])).json()["quality"]
+
+    response = client.post(reverse("opportunity_action", args=[opportunity.id]), {"action": "roadmap"})
+    assert response.status_code == 302
+    assert project.roadmap_items.filter(title="Add status filter").exists()
+
+    response = client.post(reverse("roadmap_action", args=[roadmap_item.id]), {"action": "NEXT"})
+    assert response.status_code == 302
+    roadmap_item.refresh_from_db()
+    assert roadmap_item.horizon == "NEXT"
+
+    response = client.post(reverse("scenario_finding_action", args=[finding.id]), {"action": "roadmap"})
+    assert response.status_code == 302
+
+    response = client.post(reverse("approval_action", args=[approval.id]), {"action": "approve"})
+    assert response.status_code == 302
+    approval.refresh_from_db()
+    graph_run.refresh_from_db()
+    assert approval.status == "APPROVED"
+    assert graph_run.status == "RUNNING"
+
+
+def test_ui_can_start_safe_explore_graph() -> None:
+    data = fixture()
+    client = Client()
+    response = client.post(reverse("run_explore", args=[data["project"].id]))
+    assert response.status_code == 302
+    assert GraphRun.objects.filter(execution_graph_version__graph__name="project_exploration", project=data["project"]).exists()

tests/test_venture_discovery_v0.py

@@ -0,0 +1,156 @@
+from __future__ import annotations
+
+import json
+from decimal import Decimal
+
+from agents.venture_discovery import PITCH_SECTIONS, SCORE_DIMENSIONS, VentureDiscoveryService
+from control_plane.ventures.models import CapabilityPriority, CapabilityStatus, CompanyProposal, CompanyProposalStatus, ICDecisionType, VentureArtifact
+from graph.bootstrap import champion_venture_discovery_graph_v1
+from graph.langgraph_runtime import LangGraphRuntime
+from graph.models import GraphRun, GraphRunStatus
+from graph.venture_discovery import venture_discovery_registry
+from model_router.router import ModelProvider, ModelRequestContract, ModelResponseContract, ModelRouter
+
+
+class SolOneCompanyProvider(ModelProvider):
+    provider_name = "sol-test"
+
+    def complete(self, request: ModelRequestContract) -> ModelResponseContract:
+        return ModelResponseContract(
+            model="sol",
+            content=json.dumps(
+                {
+                    "title": "LaunchLens",
+                    "one_line_thesis": "A paid validation audit helps solo technical founders avoid wasting weeks on unvalidated AI microbusinesses.",
+                    "description": "A fixed-scope validation and launch-readiness report for one startup idea.",
+                    "problem": "Builders overbuild before proving demand.",
+                    "target_customer": "Solo technical founders and small service operators considering an AI-assisted microbusiness.",
+                    "proposed_solution": "A productized audit covering ICP, first-dollar path, validation gates, build plan, risks, and capability gaps.",
+                    "business_model": "Productized service first, optional software later.",
+                    "pricing_hypothesis": "$49-$99 per audit.",
+                    "acquisition_strategy": "Compliant founder community posts and personal network conversations after approval.",
+                    "validation_plan": "Collect 5 credible target-customer responses or 1 willingness-to-pay signal before build/spend.",
+                    "capital_requested": "50",
+                    "time_to_first_dollar_estimate": "3-10 days after outreach approval",
+                    "expected_margin": "70-85% gross margin",
+                    "build_complexity": "LOW",
+                    "market_evidence": [{"type": "reasoning", "source": "sol", "summary": "Service-led validation minimizes build risk."}],
+                    "differentiation": "IC-style diligence plus Artifex execution/capability-gap awareness.",
+                    "major_risks": ["Demand unproven", "Distribution may fail", "Generic consulting competition"],
+                    "confidence": 0.66,
+                }
+            ),
+            metadata={"usage": {"prompt_tokens": 1, "completion_tokens": 1}},
+        )
+
+    def health(self) -> str:
+        return "AVAILABLE"
+
+
+def service(*, web_research_available: bool = False) -> VentureDiscoveryService:
+    return VentureDiscoveryService(ModelRouter({"sol": SolOneCompanyProvider()}), web_research_available=web_research_available)
+
+
+def test_company_proposal_lifecycle_mandate_limits_and_pitch_schema() -> None:
+    svc = service()
+    mandate = svc.create_v0_mandate()
+    proposal = svc.generate_single_company(mandate)
+
+    assert mandate.max_validation_capital == Decimal("50")
+    assert mandate.target_net_new_cash == Decimal("500")
+    assert mandate.target_window_days == 30
+    assert mandate.constraints["no_real_spend_in_v0"] is True
+    assert mandate.constraints["no_real_customer_outreach_in_v0"] is True
+    assert CompanyProposal.objects.count() == 1
+    assert proposal.status == CompanyProposalStatus.SUBMITTED
+    assert proposal.capital_requested <= Decimal("50")
+    assert all(section in proposal.pitch for section in PITCH_SECTIONS)
+    assert proposal.metadata["generation_source"] == "sol"
+    assert proposal.metadata["real_spend"] == 0
+    assert proposal.metadata["real_customer_outreach"] is False
+
+
+def test_board_review_ic_questions_responses_and_bounded_diligence() -> None:
+    svc = service()
+    proposal = svc.generate_single_company(svc.create_v0_mandate())
+    board = svc.board_review(proposal)
+    diligence = svc.start_ic_diligence(proposal)
+    questions = svc.generate_ic_questions(diligence)
+    responses = svc.answer_questions(diligence)
+    challenge = svc.red_team(diligence)
+    final = svc.final_company_response(diligence)
+
+    assert set(board.observations) == {"CEO", "CTO", "CFO", "CRO", "Independent Director"}
+    assert board.metadata["company_does_not_grade_itself"] is True
+    assert diligence.rounds == ["Initial Pitch", "Diligence Round 1", "Final Challenge", "Decision"]
+    assert diligence.metadata["bounded_rounds"] is True
+    assert len(questions) == 8
+    assert all(q.evidence_required for q in questions)
+    assert len(responses) == len(questions)
+    assert all(r.metadata["no_customer_outreach"] and r.metadata["no_spend"] for r in responses)
+    assert challenge["recommendation"] == "continue_to_final_response"
+    assert final["kill_criteria"]
+
+
+def test_ic_scoring_decision_capability_gap_and_memo_artifact() -> None:
+    svc = service()
+    proposal = svc.generate_single_company(svc.create_v0_mandate())
+    svc.board_review(proposal)
+    diligence = svc.start_ic_diligence(proposal)
+    svc.generate_ic_questions(diligence)
+    svc.answer_questions(diligence)
+    svc.red_team(diligence)
+    svc.final_company_response(diligence)
+    decision = svc.score_and_decide(diligence)
+    gap = svc.capability_analysis(proposal)
+    memo = svc.produce_investment_memo(diligence, gap)
+
+    assert decision.decision in ICDecisionType.values
+    assert set(SCORE_DIMENSIONS) == set(decision.component_scores)
+    assert decision.decision == ICDecisionType.CONDITIONAL_FUND
+    assert decision.initial_tranche == Decimal("10.00")
+    assert "5 credible target-customer responses" in decision.validation_condition
+    assert decision.metadata["no_actual_funding"] is True
+    assert "WEB_MARKET_RESEARCH" in gap.missing
+    assert gap.metadata["web_market_research_status"] == "MISSING"
+    assert {item["priority"] for item in gap.ranked_missing}.issuperset({CapabilityPriority.BEFORE_VALIDATION, CapabilityPriority.BEFORE_FIRST_CUSTOMER})
+    assert memo.artifact_type == "FINAL_INVESTMENT_MEMO"
+    assert memo.content["Company"] == proposal.title
+    assert "Capability gap" in memo.content
+
+
+def test_fallback_is_marked_and_reduces_confidence_when_research_unavailable() -> None:
+    svc = VentureDiscoveryService(web_research_available=False)
+    proposal = svc.generate_single_company(svc.create_v0_mandate())
+    gap = svc.capability_analysis(proposal)
+
+    assert proposal.metadata["generation_source"] == "deterministic_fallback"
+    assert proposal.metadata["fallback_evidence"] is True
+    assert proposal.confidence <= 0.58
+    assert any(item.get("fallback_evidence") for item in proposal.market_evidence)
+    assert "WEB_MARKET_RESEARCH" in gap.missing
+
+
+def test_venture_discovery_v1_graph_lineage_and_no_automatic_execution() -> None:
+    version = champion_venture_discovery_graph_v1()
+    graph_run = GraphRun.objects.create(execution_graph_version=version, current_node=version.graph_spec["entry"])
+
+    LangGraphRuntime(venture_discovery_registry(service())).run_until_terminal_or_paused(graph_run)
+    graph_run.refresh_from_db()
+
+    assert graph_run.status == GraphRunStatus.COMPLETE
+    assert graph_run.execution_graph_version.graph.name == "venture_discovery"
+    assert graph_run.execution_graph_version.version == 1
+    assert CompanyProposal.objects.count() == 1
+    proposal = CompanyProposal.objects.get()
+    assert proposal.metadata["real_spend"] == 0
+    assert proposal.metadata["real_customer_outreach"] is False
+    assert graph_run.node_runs.count() == 13
+    assert graph_run.edge_traversals.count() == 13
+    artifact_types = set(VentureArtifact.objects.values_list("artifact_type", flat=True))
+    assert {"STANDARDIZED_COMPANY_PITCH", "COMPANY_BOARD_REVIEW", "IC_QUESTIONS", "IC_RESPONSES", "IC_RED_TEAM", "IC_FINAL_RESPONSE", "IC_FINAL_SCORE", "CAPABILITY_GAP_REPORT", "FINAL_INVESTMENT_MEMO"}.issubset(artifact_types)
+    decision = proposal.ic_diligence.get().decision
+    svc = service()
+    svc.request_human_approval(decision, "approve_for_validation")
+    decision.refresh_from_db()
+    assert decision.metadata["real_spend_still_blocked"] is True