diff --git a/agents/crypto_venture.py b/agents/crypto_venture.py new file mode 100644 index 0000000..afd878a --- /dev/null +++ b/agents/crypto_venture.py @@ -0,0 +1,661 @@ +from __future__ import annotations + +import hashlib +import json +import re +import time +import uuid +from collections import Counter +from decimal import Decimal +from typing import Any + +from agents.venture_discovery import RESEARCH_CATEGORIES, VentureDiscoveryService +from control_plane.events.bus import EventBus +from control_plane.ventures.models import ( + AutonomousOperabilityAssessment, + AutonomousGateResult, + CompanyMandate, + CompanyProposal, + CompanyProposalStatus, + CryptoICDecisionType, + CryptoJurisdictionPolicy, + CryptoRedTeamAssessment, + CryptoScenarioLab, + FounderDependencyLevel, + OnchainNecessityAssessment, + PortfolioICReview, + PortfolioThesis, + PortfolioThesisStatus, + ProtocolThesis, + ProtocolValueCapture, + TokenDemandLoop, + TokenNecessityClassification, + TokenomicsSimulation, + TokenRedTeamFlag, + TokenUtilityAssessment, + VentureArtifact, + VentureCohort, + VentureCohortMember, + VentureGenerationRejection, + NoveltyGateDecision, + VentureThesis, + VentureTrack, +) +from model_router.policy import model_for_role +from model_router.providers import extract_json_object +from model_router.router import ModelCapability, ModelRequestContract, ModelRouter + + +CRYPTO_TERRITORIES = [ + "DECENTRALIZED_AI_COMPUTE", + "AGENT_TO_AGENT_PAYMENTS", + "PROOF_ATTESTATION_MARKETS", + "DECENTRALIZED_DATA_MARKETS", + "SECURITY_STAKING_PROTOCOLS", + "MACHINE_REPUTATION", + "AUTONOMOUS_API_MARKETPLACES", + "DEPIN_COORDINATION", + "ONCHAIN_AGENT_COMMERCE", + "DECENTRALIZED_MODEL_SERVICES", + "ONCHAIN_CREDENTIALS", + "PROTOCOLIZED_ESCROW", + "CRYPTO_NATIVE_INTELLIGENCE", + "OPEN_CRYPTO_CATEGORY", +] + +ALLOWED_TOKEN_UTILITIES = [ + "protocol fee settlement", + "staking tied to measurable service quality", + "slashing / economic guarantees", + "access to scarce network resources", + "compute/data marketplace coordination", + "collateral", + "security budget", + "material protocol governance", + "contributor/provider rewards", + "reputation-backed economic participation", + "decentralized marketplace coordination", + "machine-to-machine payments", + "proof/attestation markets", + "protocol-owned infrastructure", + "incentive alignment for decentralized supply", +] + +INSUFFICIENT_TOKEN_UTILITIES = ["community", "marketing", "speculative upside", "generic rewards", "token-gated access", "governance theater"] + +CRYPTO_SCORE_DIMENSIONS = [ + "TOKEN_NECESSITY", + "REAL_USAGE_DEMAND", + "ONCHAIN_NECESSITY", + "VALUE_ACCRUAL_QUALITY", + "NETWORK_EFFECT_POTENTIAL", + "TOKENOMICS_SUSTAINABILITY", + "BOOTSTRAPPABILITY", + "AUTONOMOUS_OPERABILITY", + "SECURITY_MODEL_QUALITY", + "REGULATORY_MANAGEABILITY", +] + +CRYPTO_SCENARIOS = [ + "oracle failure", + "validator/provider collusion", + "sybil attack", + "token price crash", + "liquidity collapse", + "spam attack", + "fee spike", + "slashing event", + "bad provider output", + "treasury exhaustion", + "emissions reduction", + "bridge dependency failure", + "smart contract exploit scenario", + "governance capture", +] + + +class CryptoVentureService: + def __init__(self, router: ModelRouter | None = None, bus: EventBus | None = None, *, web_research_available: bool = True, generation_model_hint: str | None = None, research_model_hint: str | None = None, final_ic_model_hint: str | None = None) -> None: + self.router = router + self.bus = bus or EventBus() + self.generation_model_hint = generation_model_hint or model_for_role("venture_ideation") + self.research_model_hint = research_model_hint or model_for_role("venture_research") + self.final_ic_model_hint = final_ic_model_hint or model_for_role("venture_portfolio_ic") + self.venture = VentureDiscoveryService(router, self.bus, web_research_available=web_research_available, research_model_hint=self.research_model_hint, ideation_model_hint=self.generation_model_hint) + + def create_crypto_mandate(self) -> CompanyMandate: + mandate = CompanyMandate.objects.create( + objective="Generate crypto-native protocol ventures where onchain execution and a native token are genuinely necessary for real usage, not speculation.", + constraints={"venture_track": VentureTrack.CRYPTO_PROTOCOL, "no_token_sale": True, "no_fundraising": True, "no_mainnet_issuance": True, "no_us_targeted_activity": True, "testnet_or_local_only": True, "no_real_spend": True, "no_user_contact": True}, + optimization_targets=["token necessity", "onchain necessity", "real usage demand", "value accrual", "sustainable tokenomics", "autonomous operability", "regulatory manageability", "security model quality"], + metadata={"milestone": "CRYPTO_PROTOCOL_VENTURE_COHORT_V0.1", "venture_track": VentureTrack.CRYPTO_PROTOCOL}, + ) + self._artifact(None, mandate, "CRYPTO_VENTURE_MANDATE", "Crypto Protocol Venture V0.1 Mandate", mandate.constraints, "Crypto protocol cohort mandate. No token sale, fundraising, or mainnet issuance.", "crypto_venture") + return mandate + + def prepare_cohort(self, *, size: int, graph_run=None, concurrency: int = 2) -> VentureCohort: + mandate = self.create_crypto_mandate() + cohort_id = f"CPV01-{time.strftime('%Y%m%d%H%M%S')}-{uuid.uuid4().hex[:8]}" + return VentureCohort.objects.create(cohort_id=cohort_id, mandate=mandate, cohort_size=size, graph_run=graph_run, concurrency=concurrency, status="PREPARING_CRYPTO", graph_versions={"cohort": "crypto_venture_cohort v1"}, research_policy={"source_linked_evidence_required": True, "research_insufficient_if_no_sources": True}, scoring_policy={"dimensions": CRYPTO_SCORE_DIMENSIONS}, metadata={"milestone": "CRYPTO_PROTOCOL_VENTURE_COHORT_V0.1", "venture_track": VentureTrack.CRYPTO_PROTOCOL, "real_spend": 0, "real_customer_outreach": False, "token_sale": False, "fundraising": False, "mainnet_issuance": False}) + + def generate_protocols(self, cohort: VentureCohort) -> list[CompanyProposal]: + accepted = [] + attempts = 0 + for index in range(cohort.cohort_size): + territory = CRYPTO_TERRITORIES[index % len(CRYPTO_TERRITORIES)] + payload, source = self._protocol_payload(index, territory) + attempts += 1 + proposal = self._create_proposal(cohort.mandate, payload, source, territory) + VentureCohortMember.objects.create(cohort=cohort, proposal=proposal, metadata={"territory": territory}) + accepted.append(proposal) + cohort.metrics = {**cohort.metrics, "requested_protocol_count": cohort.cohort_size, "generation_attempts": attempts, "accepted_protocols": len(accepted), "duplicate_rejections": 0, "token_necessity_rejections": 0, "generation_sources": sorted({p.metadata.get("generation_source", "unknown") for p in accepted})} + cohort.status = "PROTOCOLS_GENERATED" + cohort.save(update_fields=["metrics", "status", "updated_at"]) + return accepted + + def novelty_gate(self, cohort: VentureCohort) -> dict[str, int]: + removed = 0 + hard_excluded = 0 + duplicate = 0 + seen: dict[str, CompanyProposal] = {} + hard_terms = ["meme", "generic dex", "generic l1", "generic l2", "nft collection", "yield farm", "ponzi", "copycat launchpad", "speculative asset"] + for member in list(cohort.members.select_related("proposal").order_by("created_at")): + proposal = member.proposal + text = self._proposal_text(proposal) + fp = self._fingerprint(proposal.protocol_thesis.protocol_category + proposal.protocol_thesis.protocol_thesis + " ".join(proposal.protocol_thesis.utility_categories)) + reason = "" + decision = None + if any(term in text for term in hard_terms): + hard_excluded += 1 + reason = "hard-excluded crypto thesis" + decision = NoveltyGateDecision.REGENERATE_HARD_EXCLUSION + elif fp in seen: + duplicate += 1 + reason = f"duplicate protocol/token utility thesis: {seen[fp].title}" + decision = NoveltyGateDecision.REGENERATE_DUPLICATE + if decision: + VentureGenerationRejection.objects.create(cohort=cohort, slot_index=member.rank or 0, attempt=1, decision=decision, reason=reason, candidate={"title": proposal.title, "protocol": proposal.protocol_thesis.protocol_thesis}, similarity_score=1.0) + member.delete() + proposal.status = CompanyProposalStatus.REJECTED + proposal.save(update_fields=["status", "updated_at"]) + removed += 1 + else: + seen[fp] = proposal + cohort.metrics = {**cohort.metrics, "crypto_novelty_removed": removed, "hard_exclusion_rejections": cohort.metrics.get("hard_exclusion_rejections", 0) + hard_excluded, "duplicate_rejections": cohort.metrics.get("duplicate_rejections", 0) + duplicate} + cohort.save(update_fields=["metrics", "updated_at"]) + return {"removed": removed, "hard_exclusion_rejections": hard_excluded, "duplicate_rejections": duplicate} + + def regenerate_rejected_slots(self, cohort: VentureCohort, *, reason: str = "novelty_or_token_gate") -> list[CompanyProposal]: + created = [] + attempts = 0 + max_attempts = cohort.cohort_size * 3 + while cohort.members.count() < cohort.cohort_size and attempts < max_attempts: + attempts += 1 + index = cohort.members.count() + attempts + territory = CRYPTO_TERRITORIES[index % len(CRYPTO_TERRITORIES)] + payload, source = self._protocol_payload(index, territory) + proposal = self._create_proposal(cohort.mandate, payload, source, territory) + VentureCohortMember.objects.create(cohort=cohort, proposal=proposal, metadata={"territory": territory, "regenerated_for": reason}) + created.append(proposal) + self.assess_token_necessity(proposal) + self.novelty_gate(cohort) + self._remove_weak_token_members(cohort) + cohort.metrics = {**cohort.metrics, "regeneration_attempts": cohort.metrics.get("regeneration_attempts", 0) + attempts, "regenerated_protocols": cohort.metrics.get("regenerated_protocols", 0) + len(created), "accepted_protocols": cohort.members.count()} + cohort.save(update_fields=["metrics", "updated_at"]) + return created + + def token_necessity_gate(self, cohort: VentureCohort) -> list[TokenUtilityAssessment]: + assessments = [self.assess_token_necessity(member.proposal) for member in cohort.members.select_related("proposal")] + removed = self._remove_weak_token_members(cohort) + cohort.metrics = {**cohort.metrics, **removed} + cohort.save(update_fields=["metrics", "updated_at"]) + return assessments + + def _remove_weak_token_members(self, cohort: VentureCohort) -> dict[str, int]: + unnecessary = 0 + optional = 0 + for member in list(cohort.members.select_related("proposal")): + proposal = member.proposal + assessment = getattr(proposal, "token_utility_assessment", None) or self.assess_token_necessity(proposal) + if assessment.classification not in {TokenNecessityClassification.TOKEN_ESSENTIAL, TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED}: + if assessment.classification == TokenNecessityClassification.TOKEN_OPTIONAL: + optional += 1 + else: + unnecessary += 1 + VentureGenerationRejection.objects.create(cohort=cohort, slot_index=member.rank or 0, attempt=1, decision=NoveltyGateDecision.REGENERATE_HARD_EXCLUSION, reason=f"removed by token necessity gate: {assessment.classification}", candidate={"title": proposal.title, "classification": assessment.classification}, similarity_score=0.0) + member.delete() + proposal.status = CompanyProposalStatus.REJECTED + proposal.metadata = {**proposal.metadata, "routed_to_saas": assessment.classification == TokenNecessityClassification.TOKEN_OPTIONAL} + proposal.save(update_fields=["status", "metadata", "updated_at"]) + return {"token_unnecessary_rejections": cohort.metrics.get("token_unnecessary_rejections", 0) + unnecessary, "token_optional_route_to_saas": cohort.metrics.get("token_optional_route_to_saas", 0) + optional} + + def light_market_research(self, cohort: VentureCohort) -> None: + started = time.monotonic() + total_sources = 0 + insufficient = 0 + for member in cohort.members.select_related("proposal"): + research = self.venture.conduct_market_research(member.proposal, depth="light") + total_sources += len(research.get("sources", [])) + if not research.get("sources"): + insufficient += 1 + cohort.metrics = {**cohort.metrics, "crypto_light_research_seconds": round(time.monotonic() - started, 2), "crypto_light_research_sources": total_sources, "research_insufficient_count": insufficient} + cohort.save(update_fields=["metrics", "updated_at"]) + + def protocol_research(self, cohort: VentureCohort) -> None: + for member in cohort.members.select_related("proposal"): + self._update_protocol_research_flags(member.proposal) + + def token_red_team(self, cohort: VentureCohort) -> list[CryptoRedTeamAssessment]: + assessments = [self.red_team_proposal(member.proposal) for member in cohort.members.select_related("proposal")] + counts = Counter(flag for item in assessments for flag in item.flags) + cohort.metrics = {**cohort.metrics, "token_red_team_failures": dict(counts)} + cohort.save(update_fields=["metrics", "updated_at"]) + return assessments + + def tokenomics_simulation(self, cohort: VentureCohort) -> list[TokenomicsSimulation]: + simulations = [self.simulate_tokenomics(member.proposal) for member in cohort.members.select_related("proposal")] + self.crypto_scenario_lab(cohort) + return simulations + + def crypto_ic_first_pass(self, cohort: VentureCohort) -> None: + for member in cohort.members.select_related("proposal"): + row = self.crypto_score_row(member.proposal) + member.portfolio_score = row["crypto_ic_score"] + member.metadata = {**member.metadata, "crypto_first_pass": row} + member.save(update_fields=["portfolio_score", "metadata", "updated_at"]) + + def top5_deep_research(self, cohort: VentureCohort) -> None: + before = {} + after = {} + members = list(cohort.members.select_related("proposal").order_by("-portfolio_score", "created_at")[:5]) + for member in members: + before[str(member.proposal_id)] = member.proposal.metadata.get("research", {}).get("coverage_ratio", 0.0) + self.venture.conduct_market_research(member.proposal, depth="deep") + after[str(member.proposal_id)] = member.proposal.metadata.get("research", {}).get("coverage_ratio", 0.0) + cohort.metrics = {**cohort.metrics, "top5_deep_research_count": len(members), "top5_research_coverage_before": before, "top5_research_coverage_after": after} + cohort.save(update_fields=["metrics", "updated_at"]) + + def portfolio_crypto_ic(self, cohort: VentureCohort) -> PortfolioICReview: + rows = [self.crypto_score_row(member.proposal) for member in cohort.members.select_related("proposal")] + rows = [self._sol_final_ic_adjustment(row) for row in rows] + rows.sort(key=lambda row: row["crypto_ic_score"], reverse=True) + top_3 = [row for row in rows if self._qualifies_finalist(row)][:3] + top_ids = {row["proposal_id"] for row in top_3} + for rank, row in enumerate(rows, start=1): + member = cohort.members.get(proposal_id=row["proposal_id"]) + member.rank = rank + member.portfolio_score = row["crypto_ic_score"] + member.is_top_3 = row["proposal_id"] in top_ids + member.save(update_fields=["rank", "portfolio_score", "is_top_3", "updated_at"]) + row["rank"] = rank + review, _ = PortfolioICReview.objects.update_or_create(cohort=cohort, defaults={"rankings": rows, "top_3": top_3, "concentration": self._crypto_concentration(rows), "metadata": {"finalist_shortfall": max(0, 3 - len(top_3)), "no_fund_decision_v01": True}}) + cohort.metrics = {**cohort.metrics, "crypto_ranked_count": len(rows), "crypto_top_3_count": len(top_3)} + cohort.status = "CRYPTO_IC_COMPLETE" + cohort.save(update_fields=["metrics", "status", "updated_at"]) + return review + + def protocol_security_gate(self, cohort: VentureCohort) -> dict[str, int]: + blocked = 0 + passed = 0 + for member in cohort.members.select_related("proposal"): + proposal = member.proposal + text = self._proposal_text(proposal) + required = ["contract", "oracle", "key", "admin", "pause", "treasury"] + missing = [item for item in required if item not in text] + critical = any(self._positive_phrase_present(text, term) for term in ["unaudited mainnet", "custody user funds", "bridge dependency", "upgradeable without timelock"]) + status = "BLOCKED_HUMAN_SECURITY_GATE" if critical or len(missing) >= 4 else "PASS_WITH_SECURITY_REVIEW" + if status.startswith("BLOCKED"): + blocked += 1 + else: + passed += 1 + self._artifact(proposal, cohort.mandate, "CRYPTO_PROTOCOL_SECURITY_GATE", f"Protocol Security Gate: {proposal.title}", {"status": status, "missing_controls": missing, "critical_risk": critical, "mainnet_allowed": False}, f"{status}. Missing controls: {', '.join(missing) or 'none'}. Mainnet is not allowed in V0.1.", "crypto_protocol_security_gate", graph_run=cohort.graph_run) + proposal.metadata = {**proposal.metadata, "crypto_security_gate": {"status": status, "missing_controls": missing, "critical_risk": critical, "mainnet_allowed": False}} + proposal.save(update_fields=["metadata", "updated_at"]) + cohort.metrics = {**cohort.metrics, "protocol_security_gate_blocked": blocked, "protocol_security_gate_passed": passed} + cohort.save(update_fields=["metrics", "updated_at"]) + return {"blocked": blocked, "passed": passed} + + def regulatory_gate(self, cohort: VentureCohort) -> None: + gated = 0 + for member in cohort.members.select_related("proposal"): + policy = self.regulatory_policy(member.proposal) + if policy.human_legal_gate: + gated += 1 + cohort.metrics = {**cohort.metrics, "human_legal_gate_count": gated, "legal_review_required": True} + cohort.save(update_fields=["metrics", "updated_at"]) + + def capability_analysis(self, cohort: VentureCohort) -> list[dict[str, Any]]: + gaps = ["smart contract audit", "testnet deployment", "wallet auth", "key management", "oracle/provider monitoring", "token simulation harness", "legal review workflow"] + rows = [{"capability": gap, "count": cohort.members.count(), "status": "MISSING", "earliest_stage": "BEFORE_VALIDATION"} for gap in gaps] + review = getattr(cohort, "portfolio_review", None) + if review: + review.capability_demand = rows + review.recommended_build_priorities = rows[:5] + review.save(update_fields=["capability_demand", "recommended_build_priorities", "updated_at"]) + return rows + + def update_crypto_thesis_registry(self, cohort: VentureCohort) -> list[dict[str, Any]]: + updates = [] + for member in cohort.members.select_related("proposal"): + proposal = member.proposal + proto = proposal.protocol_thesis + token = proposal.token_utility_assessment + name = f"CRYPTO::{proto.protocol_category or member.metadata.get('territory', 'OPEN')}::{proposal.title[:80]}" + registry, _ = PortfolioThesis.objects.update_or_create(canonical_name=name, defaults={"concise_thesis": proto.protocol_thesis[:1000], "category": proto.protocol_category, "industry": "crypto/protocol", "icp": proposal.target_customer[:1000], "problem": proposal.problem[:1000], "offer": proposal.proposed_solution[:1000], "business_model": "protocol", "primary_channel": "developer/community", "ai_leverage": 0.0, "platformization_potential": 0.0, "fingerprint": self._fingerprint(proposal.title + proto.protocol_thesis), "status": PortfolioThesisStatus.ACTIVE_CANDIDATE, "proposal_count": 1, "best_ic_score": member.portfolio_score, "best_company": proposal, "last_seen_cohort": cohort, "metadata": {"venture_track": VentureTrack.CRYPTO_PROTOCOL, "protocol_category": proto.protocol_category, "token_utility_type": token.utility_categories, "token_necessity": token.classification, "regulatory_risk": proposal.crypto_jurisdiction_policy.regulatory_manageability_score, "autonomy_score": proposal.autonomous_assessment.autonomous_operability_score}, "venture_track": VentureTrack.CRYPTO_PROTOCOL, "best_autonomous_operability_score": proposal.autonomous_assessment.autonomous_operability_score}) + if registry.first_seen_cohort_id is None: + registry.first_seen_cohort = cohort + registry.save(update_fields=["first_seen_cohort", "updated_at"]) + updates.append({"canonical_name": registry.canonical_name, "token_necessity": token.classification, "best_ic_score": member.portfolio_score}) + cohort.metadata = {**cohort.metadata, "crypto_thesis_registry_after": updates} + cohort.save(update_fields=["metadata", "updated_at"]) + return updates + + def produce_crypto_cohort_report(self, cohort: VentureCohort) -> VentureArtifact: + review = cohort.portfolio_review + rows = review.rankings + content = {"title": "CRYPTO VENTURE COHORT V0.1 REPORT", "cohort_id": cohort.cohort_id, "graph_run": str(cohort.graph_run_id or ""), "runtime": cohort.metrics, "accepted_protocols": cohort.members.count(), "generation_attempts": cohort.metrics.get("generation_attempts", 0), "duplicate_token_necessity_rejections": {"duplicate_rejections": cohort.metrics.get("duplicate_rejections", 0), "token_unnecessary_rejections": cohort.metrics.get("token_unnecessary_rejections", 0)}, "ranking": rows, "top_3": review.top_3, "token_red_team_failures": cohort.metrics.get("token_red_team_failures", {}), "token_utility_distribution": dict(Counter(util for row in rows for util in row.get("token_utility", []))), "crypto_thesis_saturation": review.concentration, "capability_gaps": review.capability_demand, "system_metrics": cohort.metrics, "stop_conditions": {"token_sale": False, "fundraising": False, "mainnet_issuance": False, "user_contact": False, "real_spend": 0}} + artifact = self._artifact(None, cohort.mandate, "CRYPTO_VENTURE_COHORT_REPORT", "Crypto Venture Cohort V0.1 Report", content, self._readable_report(content), "crypto_portfolio_ic", graph_run=cohort.graph_run) + cohort.status = "COMPLETE" + cohort.save(update_fields=["status", "updated_at"]) + return artifact + + def assess_token_necessity(self, proposal: CompanyProposal) -> TokenUtilityAssessment: + crypto = proposal.metadata.get("crypto", {}) + text = self._proposal_text(proposal) + utilities = [u for u in crypto.get("token_utility", []) if isinstance(u, str)] or self._infer_utilities(text) + strong_count = sum(1 for utility in utilities if utility.lower() in ALLOWED_TOKEN_UTILITIES or any(term in utility.lower() for term in ["stake", "slash", "collateral", "fee", "marketplace", "attestation", "compute", "machine", "security"])) + weak = any(term in text for term in ["meme", "speculative", "community token", "governance token only", "token gated subscription"]) + score = min(100, 35 + strong_count * 18 + (10 if "slash" in text or "slashing" in text else 0) + (8 if "provider" in text else 0) - (35 if weak else 0)) + sol_review = self._sol_json("Counterfactual token necessity review. If the token were removed and replaced with fiat/stablecoin/database credits, would the product materially degrade? Return JSON with classification TOKEN_ESSENTIAL, TOKEN_STRONGLY_JUSTIFIED, TOKEN_OPTIONAL, or TOKEN_UNNECESSARY; score 0-100; rationale; fiat_or_database_substitution. Proposal: " + json.dumps(proposal.pitch, default=str) + " Crypto: " + json.dumps(crypto, default=str)) + if sol_review: + score = float(sol_review.get("score", score)) + sol_classification = str(sol_review.get("classification", "")) + else: + sol_classification = "" + if score >= 82: + classification = TokenNecessityClassification.TOKEN_ESSENTIAL + elif score >= 70: + classification = TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED + elif score >= 50: + classification = TokenNecessityClassification.TOKEN_OPTIONAL + else: + classification = TokenNecessityClassification.TOKEN_UNNECESSARY + if sol_classification in TokenNecessityClassification.values: + classification = sol_classification + return TokenUtilityAssessment.objects.update_or_create(proposal=proposal, defaults={"classification": classification, "token_necessity_score": score, "utility_categories": utilities, "fiat_or_database_substitution": str((sol_review or {}).get("fiat_or_database_substitution", "If normal fiat/stablecoin/database credits preserve the core coordination, route to SaaS.")), "rationale": str((sol_review or {}).get("rationale", "Token score is based on explicit non-speculative utility, staking/slashing, marketplace coordination, and usage-linked fee demand.")), "metadata": {"weak_utility_terms": INSUFFICIENT_TOKEN_UTILITIES, "sol_counterfactual_review": sol_review or {}}})[0] + + def _create_proposal(self, mandate: CompanyMandate, payload: dict[str, Any], source: str, territory: str) -> CompanyProposal: + title = str(payload.get("name") or payload.get("title") or f"Protocol {territory.title()}")[:255] + product = str(payload.get("product_thesis") or payload.get("product") or "A useful crypto-native network service.") + protocol = str(payload.get("protocol_thesis") or payload.get("protocol") or "Onchain settlement coordinates independent providers and users.") + token = str(payload.get("token_thesis") or payload.get("why_token") or "A native token bonds providers, pays protocol fees, and funds security.") + demand_loop = self._as_list(payload.get("token_demand_loop") or ["users consume service", "users pay protocol fees", "providers stake token", "bad providers are slashed", "usage-linked fees sustain rewards"]) + validation = str(payload.get("validation_experiment") or "Run a testnet/local-chain pilot with fake credits and simulated token accounting; no sale, fundraising, or mainnet issuance.") + thesis = VentureThesis.objects.create(mandate=mandate, title=title, thesis=protocol, similarity_fingerprint=self._fingerprint(title + protocol), metadata={"source": source, "venture_track": VentureTrack.CRYPTO_PROTOCOL, "territory": territory}, evidence_tier="TIER_0_THESIS") + proposal = CompanyProposal.objects.create(mandate=mandate, thesis=thesis, title=title, description=product, problem=str(payload.get("user_pain") or payload.get("problem") or "Users lack trustworthy decentralized coordination."), target_customer=str(payload.get("user") or payload.get("target_user") or "Developers and network participants"), proposed_solution=str(payload.get("product") or product), business_model=str(payload.get("business_model") or "Protocol fees on real usage; no V0 token sale."), pricing_hypothesis=str(payload.get("fee_model") or "Testnet/free validation, later usage fees."), acquisition_strategy=str(payload.get("bootstrap_plan") or "Developer adoption through testnet docs and public artifacts only after approval."), validation_plan=validation, capital_requested=Decimal("0"), time_to_first_dollar_estimate="No V0 revenue target; validate protocol usage without token sale.", expected_margin="Protocol fee margin depends on provider economics.", build_complexity=str(payload.get("build_complexity") or "MEDIUM-HIGH"), market_evidence=[], differentiation=str(payload.get("differentiation") or protocol), major_risks=self._as_list(payload.get("major_risks") or ["Token not required", "Regulatory uncertainty", "Security model weak"]), confidence=float(payload.get("confidence") or 0.55), status=CompanyProposalStatus.SUBMITTED, pitch={"Company name": title, "One-line thesis": str(payload.get("one_line_thesis") or protocol), "Product thesis": product, "Protocol thesis": protocol, "Token thesis": token, "Token demand loop": demand_loop, "Validation experiment": validation}, metadata={"generation_source": source, "venture_track": VentureTrack.CRYPTO_PROTOCOL, "crypto": {"territory": territory, "product_thesis": product, "protocol_thesis": protocol, "token_thesis": token, "token_demand_loop": demand_loop, "token_utility": self._as_list(payload.get("token_utility") or self._infer_utilities(token + " " + protocol)), "value_capture": str(payload.get("value_capture") or "Usage fees accrue to providers, security budget, and protocol treasury."), "network_effect": str(payload.get("network_effect") or "More users attract more providers, improving liquidity/reliability."), "bootstrap_plan": str(payload.get("bootstrap_plan") or "Run without a live token using test credits and provider simulations."), "security_model": str(payload.get("security_model") or "Contracts, staking, slashing, oracle controls, admin keys, and treasury controls require review."), "regulatory_policy": "US_EXCLUDED; TOKEN_SALE_DISABLED; MAINNET_TOKEN_ISSUANCE_DISABLED; FUNDRAISING_DISABLED"}, "real_spend": 0, "real_customer_outreach": False, "token_sale": False, "mainnet_issuance": False}, evidence_tier="TIER_0_THESIS") + ProtocolThesis.objects.create(proposal=proposal, product_thesis=product, protocol_thesis=protocol, token_thesis=token, network_effect=proposal.metadata["crypto"]["network_effect"], bootstrap_plan=proposal.metadata["crypto"]["bootstrap_plan"], autonomous_operability="Artifex can build/testnet deploy/monitor only; mainnet and issuance stop at human legal gate.", utility_categories=proposal.metadata["crypto"]["token_utility"], protocol_category=territory) + TokenDemandLoop.objects.create(proposal=proposal, loop=demand_loop, real_usage_driver=str(payload.get("real_usage_driver") or product), non_speculative_demand=not any(term in " ".join(demand_loop).lower() for term in ["speculation", "price go up"]), bootstrap_without_token=proposal.metadata["crypto"]["bootstrap_plan"]) + self.bus.publish("CRYPTO_PROTOCOL_PROPOSED", payload={"proposal_id": str(proposal.id), "territory": territory, "source": source}) + return proposal + + def _protocol_payload(self, index: int, territory: str) -> tuple[dict[str, Any], str]: + if self.router is not None: + try: + prompt = "Generate exactly one crypto-native protocol venture as JSON. Do not propose meme coins, generic DEX/L1/NFT/yield farms, or SaaS plus token. Include name, one_line_thesis, product_thesis, user, protocol_thesis, why_onchain, token_thesis, token_utility list, token_demand_loop list, value_capture, network_effect, bootstrap_plan, validation_experiment, security_model, regulatory_risks, major_risks, confidence. Territory: " + territory + ". Rules: no token sale, no fundraising, no mainnet issuance, no US-targeted activity; validation must use testnet/local/fake credits/stablecoin-only simulation." + response = self.router.complete(ModelRequestContract(purpose=ModelCapability.PLANNING, model_hint=self.generation_model_hint, prompt=prompt)) + parsed = extract_json_object(response.content) + if parsed: + return parsed, self.generation_model_hint + except Exception: + pass + return self._fallback_payload(index, territory), "deterministic_crypto_fallback" + + def _fallback_payload(self, index: int, territory: str) -> dict[str, Any]: + names = { + "DECENTRALIZED_AI_COMPUTE": "Verifiable Inference Provider Market", + "AGENT_TO_AGENT_PAYMENTS": "Agent Micropayment Settlement Rail", + "PROOF_ATTESTATION_MARKETS": "Model Output Attestation Market", + "DECENTRALIZED_DATA_MARKETS": "Consent-Bound Data License Exchange", + "SECURITY_STAKING_PROTOCOLS": "API Reliability Slashing Pool", + } + name = names.get(territory, f"{territory.replace('_', ' ').title()} Protocol") + return {"name": name, "one_line_thesis": f"{name} coordinates independent supply and demand with staking, slashing, and usage fees.", "product_thesis": "A testnet marketplace/API where users request measurable digital work and providers compete to fulfill it.", "user": "Developers, agents, and protocol operators needing verifiable digital services.", "protocol_thesis": "Onchain settlement, escrow, provider bonds, attestations, and slashing coordinate parties that do not share an operator.", "why_onchain": "Trust-minimized escrow, programmable slashing, public reputation, and machine-to-machine settlement materially degrade if replaced by a private database.", "token_thesis": "The token is staked by providers, slashed for measurable failures, used for protocol fee settlement, and funds the security budget.", "token_utility": ["protocol fee settlement", "staking tied to measurable service quality", "slashing / economic guarantees", "decentralized marketplace coordination"], "token_demand_loop": ["users request measurable service", "users pay protocol fee", "providers stake token to serve", "bad providers are slashed", "fees reward reliable providers and security budget", "more real usage increases fee demand"], "value_capture": "Usage fees and slashing penalties accrue to reliable providers, insurance/security pool, and protocol treasury.", "network_effect": "More users create more jobs; more staked providers improve reliability and lower latency; more attestations improve reputation quality.", "bootstrap_plan": "Validate on local/testnet with fake credits and recruited simulated providers; no token sale or mainnet issuance.", "validation_experiment": "Run 50 simulated jobs on testnet/local chain, measure provider quality, slashing events, completion cost, and developer API reuse.", "security_model": "Escrow contracts, staking/slashing, oracle/attestation checks, admin key limits, pause controls, and treasury multisig require review.", "major_risks": ["Token may be optional", "Provider supply bootstrapping", "Smart contract risk", "Regulatory uncertainty"], "confidence": 0.58 + (index % 3) * 0.04} + + def _update_protocol_research_flags(self, proposal: CompanyProposal) -> None: + research = proposal.metadata.get("research", {}) + crypto = proposal.metadata.get("crypto", {}) + proposal.metadata = {**proposal.metadata, "crypto": {**crypto, "research_status": "RESEARCH_INSUFFICIENT" if research.get("source_count", 0) == 0 else "SOURCE_LINKED", "research_categories": RESEARCH_CATEGORIES}} + proposal.save(update_fields=["metadata", "updated_at"]) + + def red_team_proposal(self, proposal: CompanyProposal) -> CryptoRedTeamAssessment: + token = proposal.token_utility_assessment + onchain = self.onchain_assessment(proposal) + value = self.value_capture_assessment(proposal) + policy = self.regulatory_policy(proposal) + flags = [] + if token.classification in {TokenNecessityClassification.TOKEN_OPTIONAL, TokenNecessityClassification.TOKEN_UNNECESSARY}: + flags.append(TokenRedTeamFlag.TOKEN_NOT_REQUIRED) + if onchain.onchain_necessity_score < 70: + flags.append(TokenRedTeamFlag.ONCHAIN_NOT_REQUIRED) + if value.value_accrual_quality_score < 65: + flags.append(TokenRedTeamFlag.VALUE_CAPTURE_BROKEN) + if value.tokenomics_sustainability_score < 65: + flags.append(TokenRedTeamFlag.UNSUSTAINABLE_EMISSIONS) + if policy.regulatory_manageability_score < 55: + flags.append(TokenRedTeamFlag.REGULATORY_RISK_HIGH) + text = self._proposal_text(proposal) + if "speculat" in text or "apy" in text or "yield farm" in text: + flags.append(TokenRedTeamFlag.SPECULATION_DEPENDENT) + if "governance" in text and "slash" not in text and "fee" not in text: + flags.append(TokenRedTeamFlag.GOVERNANCE_THEATER) + sol_review = self._sol_json("Independent Token Red Team. Return JSON with flags list using only TOKEN_NOT_REQUIRED, SPECULATION_DEPENDENT, UNSUSTAINABLE_EMISSIONS, VALUE_CAPTURE_BROKEN, MERCENARY_INCENTIVES, GOVERNANCE_THEATER, SECURITY_MODEL_WEAK, TOKEN_VELOCITY_TOO_HIGH, BOOTSTRAP_PROBLEM, CENTRALIZATION_CONTRADICTION, REGULATORY_RISK_HIGH, ONCHAIN_NOT_REQUIRED, NO_REAL_USER_DEMAND; severity LOW/MEDIUM/HIGH; critique. Proposal: " + json.dumps(self.crypto_score_context(proposal), default=str)) + if sol_review: + for flag in self._as_list(sol_review.get("flags")): + if str(flag) in TokenRedTeamFlag.values: + flags.append(str(flag)) + severity = "HIGH" if len(flags) >= 3 else "MEDIUM" if flags else "LOW" + if sol_review and str(sol_review.get("severity", "")) in {"LOW", "MEDIUM", "HIGH"}: + severity = str(sol_review["severity"]) + deduped_flags = list(dict.fromkeys(str(flag) for flag in flags)) + return CryptoRedTeamAssessment.objects.update_or_create(proposal=proposal, defaults={"flags": deduped_flags, "severity": severity, "critique": str((sol_review or {}).get("critique", "Independent token red team checked token necessity, speculation dependence, value capture, emissions, governance theater, onchain necessity, and regulatory risk.")), "independent_from_generation": True, "metadata": {"sol_red_team_review": sol_review or {}}})[0] + + def crypto_score_context(self, proposal: CompanyProposal) -> dict[str, Any]: + return {"title": proposal.title, "pitch": proposal.pitch, "crypto": proposal.metadata.get("crypto", {}), "token": getattr(proposal, "token_utility_assessment", None).classification if hasattr(proposal, "token_utility_assessment") else None} + + def onchain_assessment(self, proposal: CompanyProposal) -> OnchainNecessityAssessment: + text = self._proposal_text(proposal) + terms = ["escrow", "slashing", "stake", "attestation", "settlement", "collateral", "trust", "reputation", "marketplace", "machine"] + score = min(100, 35 + sum(8 for term in terms if term in text)) + return OnchainNecessityAssessment.objects.update_or_create(proposal=proposal, defaults={"onchain_necessity_score": score, "reasons": [term for term in terms if term in text], "offchain_substitute": "Private SaaS/database is acceptable only if escrow, slashing, public reputation, and neutral settlement are not material.", "rationale": "Scores onchain necessity from trust-minimized coordination, settlement, staking/slashing, attestations, and multi-party neutrality."})[0] + + def value_capture_assessment(self, proposal: CompanyProposal) -> ProtocolValueCapture: + text = self._proposal_text(proposal) + fee = "fee" in text + stake = "stake" in text or "staking" in text + slash = "slash" in text or "slashing" in text + emission_bad = any(term in text for term in ["high apy", "ponzi", "yield farm", "emissions only"]) + value_score = min(100, 40 + (18 if fee else 0) + (14 if stake else 0) + (14 if slash else 0) + (10 if "treasury" in text or "security budget" in text else 0)) + sustainability = max(0, min(100, value_score - (35 if emission_bad else 0) + (8 if "usage" in text else 0))) + return ProtocolValueCapture.objects.update_or_create(proposal=proposal, defaults={"value_accrual_quality_score": value_score, "tokenomics_sustainability_score": sustainability, "value_accrual": proposal.metadata.get("crypto", {}).get("value_capture", "Usage-linked fees are required."), "sinks": ["protocol fees", "staking bonds", "slashing penalties", "security budget"], "emissions_policy": "No high APY. Rewards must be covered by real fees after subsidies decline.", "sustainability_rationale": "High emissions without real fee demand score poorly; usage-linked fees, staking, slashing, and sinks score better."})[0] + + def regulatory_policy(self, proposal: CompanyProposal) -> CryptoJurisdictionPolicy: + text = self._proposal_text(proposal) + issuance_risk = any(self._positive_phrase_present(text, term) for term in ["token sale", "public sale", "airdrop", "mainnet issuance", "fundraise", "fundraising"]) + score = 55 if issuance_risk else 72 + return CryptoJurisdictionPolicy.objects.update_or_create(proposal=proposal, defaults={"excluded_jurisdictions": ["US"], "excluded_person_classes": ["US persons", "sanctioned persons", "restricted jurisdictions"], "marketing_restrictions": ["No U.S.-targeted activity", "No investment marketing", "No yield/APY promises"], "sale_restrictions": ["TOKEN_SALE_DISABLED", "MAINNET_TOKEN_ISSUANCE_DISABLED", "FUNDRAISING_DISABLED"], "kyc_aml_requirement_status": "REQUIRES_LEGAL_REVIEW_BEFORE_ANY_TRANSFER_OR_SALE", "transfer_restriction_requirement_status": "UNDETERMINED_REQUIRES_COUNSEL", "legal_review_required": True, "regulatory_manageability_score": score, "jurisdiction_uncertainty": ["Blocking the USA does not remove all legal obligations", "Token transfer and marketing treatment requires counsel"], "human_legal_gate": True, "metadata": {"US_EXCLUDED": True, "TOKEN_SALE_DISABLED": True, "MAINNET_TOKEN_ISSUANCE_DISABLED": True, "FUNDRAISING_DISABLED": True}})[0] + + def simulate_tokenomics(self, proposal: CompanyProposal) -> TokenomicsSimulation: + scenarios = {} + text = self._proposal_text(proposal) + token = getattr(proposal, "token_utility_assessment", None) or self.assess_token_necessity(proposal) + value = getattr(proposal, "value_capture", None) or self.value_capture_assessment(proposal) + base_users = max(20, int(float(proposal.confidence) * 180)) + fee_per_tx = 0.04 + (token.token_necessity_score / 1000) + (0.03 if "enterprise" in text or "infrastructure" in text else 0) + tx_per_user = 6 + len(token.utility_categories) * 2 + (4 if "agent" in text or "machine" in text else 0) + staking_per_provider = 150 + int(value.value_accrual_quality_score * 8) + emission_rate = max(0.005, (100 - value.tokenomics_sustainability_score) / 1400) + for name, multiplier in {"LOW_USAGE": 0.3, "EXPECTED": 1.0, "HIGH_USAGE": 3.0, "SUBSIDY_REMOVED": 0.8, "TOKEN_PRICE_DOWN_80_PERCENT": 0.8, "PROVIDER_CHURN": 0.6, "USER_GROWTH_10X": 10.0}.items(): + users = int(base_users * multiplier) + tx = int(users * tx_per_user) + fees = round(tx * fee_per_tx, 2) + provider_supply = max(3, int(users / (15 if "market" in text else 25))) + staking = provider_supply * staking_per_provider + emissions = 0 if name == "SUBSIDY_REMOVED" else round(max(3, tx * emission_rate), 2) + if name == "PROVIDER_CHURN": + provider_supply = max(1, int(provider_supply * 0.45)) + staking = max(100, int(staking * 0.45)) + scenarios[name] = {"users": users, "transactions": tx, "fees": fees, "token_demand": fees + staking * 0.01, "provider_supply": provider_supply, "staking": staking, "emissions": emissions, "treasury": round(1000 + fees * 0.2 - emissions * 0.1, 2), "circulating_supply": 1_000_000 + emissions, "token_velocity": round(tx / max(1, fees + staking * 0.01), 2), "reward_coverage": round(fees / max(1, emissions), 2), "network_security_budget": staking} + summary = {"token_price_appreciation_primary_success_variable": False, "subsidy_removed_survives": scenarios["SUBSIDY_REMOVED"]["reward_coverage"] >= 1.0, "stress_notes": ["Price-down and provider-churn scenarios must preserve service quality before mainnet."]} + return TokenomicsSimulation.objects.update_or_create(proposal=proposal, defaults={"scenarios": scenarios, "summary": summary})[0] + + def crypto_scenario_lab(self, cohort: VentureCohort) -> list[CryptoScenarioLab]: + labs = [] + for member in cohort.members.select_related("proposal"): + scenarios = [{"scenario": name, "expected_failure_mode": "Must be mitigated before any mainnet deployment.", "route_to_progeny": name in {"smart contract exploit scenario", "oracle failure", "governance capture"}} for name in CRYPTO_SCENARIOS] + labs.append(CryptoScenarioLab.objects.update_or_create(proposal=member.proposal, defaults={"scenarios": scenarios, "systemic_findings": ["No V0.1 mainnet deployment", "Audit/key-management/legal gates required"], "progeny_candidates": [s["scenario"] for s in scenarios if s["route_to_progeny"]]})[0]) + return labs + + def crypto_score_row(self, proposal: CompanyProposal) -> dict[str, Any]: + token = self.assess_token_necessity(proposal) + onchain = self.onchain_assessment(proposal) + value = self.value_capture_assessment(proposal) + policy = self.regulatory_policy(proposal) + red = getattr(proposal, "crypto_red_team", None) or self.red_team_proposal(proposal) + sim = getattr(proposal, "tokenomics_simulation", None) or self.simulate_tokenomics(proposal) + autonomy = self.assess_autonomy(proposal) + text = self._proposal_text(proposal) + component_scores = { + "TOKEN_NECESSITY": token.token_necessity_score, + "REAL_USAGE_DEMAND": min(100, 45 + (15 if "usage" in text else 0) + (10 if "users" in text else 0) + (10 if proposal.metadata.get("research", {}).get("source_count", 0) else 0)), + "ONCHAIN_NECESSITY": onchain.onchain_necessity_score, + "VALUE_ACCRUAL_QUALITY": value.value_accrual_quality_score, + "NETWORK_EFFECT_POTENTIAL": min(100, 50 + (20 if "provider" in text else 0) + (15 if "reputation" in text else 0)), + "TOKENOMICS_SUSTAINABILITY": value.tokenomics_sustainability_score, + "BOOTSTRAPPABILITY": 78 if "testnet" in text or "fake credits" in text else 55, + "AUTONOMOUS_OPERABILITY": autonomy.autonomous_operability_score, + "SECURITY_MODEL_QUALITY": min(100, 45 + sum(8 for term in ["contract", "oracle", "key", "pause", "treasury", "slashing"] if term in text)), + "REGULATORY_MANAGEABILITY": policy.regulatory_manageability_score, + } + score = round(sum(component_scores.values()) / len(component_scores) - len(red.flags) * 4, 1) + decision = self._crypto_decision(component_scores, token.classification, red.flags) + return {"proposal_id": str(proposal.id), "company": proposal.title, "one_line_thesis": proposal.pitch.get("One-line thesis", ""), "product_thesis": proposal.protocol_thesis.product_thesis, "protocol_thesis": proposal.protocol_thesis.protocol_thesis, "token_thesis": proposal.protocol_thesis.token_thesis, "why_onchain": proposal.onchain_assessment.rationale, "why_token": token.rationale, "token_utility": token.utility_categories, "token_necessity_classification": token.classification, "token_demand_loop": proposal.token_demand_loop.loop, "value_capture": proposal.value_capture.value_accrual, "network_effect": proposal.protocol_thesis.network_effect, "bootstrap_plan": proposal.protocol_thesis.bootstrap_plan, "autonomous_operability": autonomy.autonomous_operability_score, "regulatory_manageability": policy.regulatory_manageability_score, "security_risk": component_scores["SECURITY_MODEL_QUALITY"], "validation_experiment": proposal.validation_plan, "component_scores": component_scores, "value_accrual_quality": value.value_accrual_quality_score, "tokenomics_sustainability": value.tokenomics_sustainability_score, "simulation_summary": sim.summary, "token_red_team_flags": red.flags, "crypto_ic_decision": decision, "crypto_ic_score": score, "research": proposal.metadata.get("research", {}), "legal_review_required": policy.legal_review_required} + + def assess_autonomy(self, proposal: CompanyProposal) -> AutonomousOperabilityAssessment: + assessment = self.venture.assess_autonomous_operability(proposal) + text = self._proposal_text(proposal) + penalty = 25 if any(term in text for term in ["exchange listing", "market making", "institutional integration", "founder evangelism"]) else 0 + score = max(0, assessment.autonomous_operability_score - penalty) + minutes = assessment.minutes_per_week_human + (30 if penalty else 0) + gate = AutonomousGateResult.AUTONOMOUS_ELIGIBLE if score >= 75 and minutes <= 30 else AutonomousGateResult.AUTONOMOUS_BORDERLINE if score >= 60 else AutonomousGateResult.ASSISTED_ONLY + assessment.venture_track = VentureTrack.CRYPTO_PROTOCOL + assessment.gate_result = gate + assessment.autonomous_operability_score = score + assessment.minutes_per_week_human = minutes + assessment.human_actions_required = list(dict.fromkeys([*assessment.human_actions_required, "legal review", "security review", "mainnet/issuance approval gate"])) + assessment.human_action_categories = list(dict.fromkeys([*assessment.human_action_categories, "LEGAL_GATE", "SECURITY_GATE"])) + assessment.platform_blockers = list(dict.fromkeys([*assessment.platform_blockers, "LEGAL_REVIEW_REQUIRED", "SMART_CONTRACT_AUDIT", "KEY_MANAGEMENT", "TESTNET_DEPLOYMENT"])) + assessment.validation_offer = {**assessment.validation_offer, "type": "testnet/local simulation", "token_required": False} + assessment.end_state_business_model = {**assessment.end_state_business_model, "type": "usage-fee protocol", "token_sale": False, "mainnet_requires_human_gate": True} + assessment.rationale = assessment.rationale + " Crypto overlay: V0.1 can build/testnet/simulate autonomously, but issuance, fundraising, and mainnet deployment stop at legal/security gates." + assessment.save(update_fields=["venture_track", "gate_result", "autonomous_operability_score", "minutes_per_week_human", "human_actions_required", "human_action_categories", "platform_blockers", "validation_offer", "end_state_business_model", "rationale", "updated_at"]) + return assessment + + def _crypto_decision(self, scores: dict[str, float], classification: str, flags: list[str]) -> str: + if classification == TokenNecessityClassification.TOKEN_UNNECESSARY: + return CryptoICDecisionType.REJECT_TOKEN_NOT_NEEDED + if classification == TokenNecessityClassification.TOKEN_OPTIONAL: + return CryptoICDecisionType.ROUTE_TO_SAAS + if TokenRedTeamFlag.SPECULATION_DEPENDENT in flags: + return CryptoICDecisionType.REJECT_SPECULATIVE + if scores["REGULATORY_MANAGEABILITY"] < 55: + return CryptoICDecisionType.REJECT_REGULATORY_RISK + if scores["TOKENOMICS_SUSTAINABILITY"] < 65 or scores["VALUE_ACCRUAL_QUALITY"] < 65: + return CryptoICDecisionType.REVISE_TOKEN_MODEL + if scores["TOKEN_NECESSITY"] >= 82 and scores["ONCHAIN_NECESSITY"] >= 75: + return CryptoICDecisionType.TESTNET_PILOT + return CryptoICDecisionType.PROTOCOL_VALIDATE + + def _sol_final_ic_adjustment(self, row: dict[str, Any]) -> dict[str, Any]: + review = self._sol_json("Final Crypto IC review. Return JSON with optional score_adjustment -20..20, decision from PROTOCOL_VALIDATE, TESTNET_PILOT, REVISE_TOKEN_MODEL, ROUTE_TO_SAAS, WATCHLIST, REJECT_TOKEN_NOT_NEEDED, REJECT_SPECULATIVE, REJECT_REGULATORY_RISK, REJECT_ECONOMIC_MODEL, REJECT_OPERABILITY, and rationale. Do not promote weak token ideas. Row: " + json.dumps(row, default=str)) + if not review: + return row + adjustment = max(-20, min(20, float(review.get("score_adjustment", 0)))) + decision = str(review.get("decision", row["crypto_ic_decision"])) + if decision not in CryptoICDecisionType.values: + decision = row["crypto_ic_decision"] + return {**row, "crypto_ic_score": round(max(0, min(100, row["crypto_ic_score"] + adjustment)), 1), "crypto_ic_decision": decision, "sol_final_ic_review": review} + + def _qualifies_finalist(self, row: dict[str, Any]) -> bool: + scores = row["component_scores"] + return row["token_necessity_classification"] in {TokenNecessityClassification.TOKEN_ESSENTIAL, TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED} and scores["TOKEN_NECESSITY"] >= 75 and scores["REAL_USAGE_DEMAND"] >= 65 and scores["ONCHAIN_NECESSITY"] >= 70 and scores["VALUE_ACCRUAL_QUALITY"] >= 65 and scores["TOKENOMICS_SUSTAINABILITY"] >= 65 and scores["AUTONOMOUS_OPERABILITY"] >= 70 + + def _crypto_concentration(self, rows: list[dict[str, Any]]) -> dict[str, Any]: + categories = Counter(row["company"].split()[0] for row in rows) + utilities = Counter(util for row in rows for util in row.get("token_utility", [])) + return {"protocol_category_distribution": dict(categories), "token_utility_distribution": dict(utilities), "saturation_flags": [name for name, count in categories.items() if count >= 3]} + + def _infer_utilities(self, text: str) -> list[str]: + lowered = text.lower() + utilities = [] + if "fee" in lowered or "settlement" in lowered: + utilities.append("protocol fee settlement") + if "stake" in lowered: + utilities.append("staking tied to measurable service quality") + if "slash" in lowered: + utilities.append("slashing / economic guarantees") + if "market" in lowered or "provider" in lowered: + utilities.append("decentralized marketplace coordination") + if "attestation" in lowered or "proof" in lowered: + utilities.append("proof/attestation markets") + return utilities or ["token utility not proven"] + + def _proposal_text(self, proposal: CompanyProposal) -> str: + crypto = proposal.metadata.get("crypto", {}) if isinstance(proposal.metadata, dict) else {} + return " ".join([proposal.title, proposal.description, proposal.problem, proposal.target_customer, proposal.proposed_solution, proposal.business_model, proposal.validation_plan, json.dumps(crypto, default=str)]).lower() + + def _positive_phrase_present(self, text: str, phrase: str) -> bool: + phrase = phrase.lower() + for match in re.finditer(re.escape(phrase), text): + prefix = text[max(0, match.start() - 40) : match.start()] + if any(negation in prefix for negation in ["no ", "not ", "without ", "disabled", "disable", "prohibit", "forbid", "never "]): + continue + return True + return False + + def _sol_json(self, prompt: str) -> dict[str, Any]: + if self.router is None or self.final_ic_model_hint not in self.router.providers: + return {} + try: + response = self.router.complete(ModelRequestContract(purpose=ModelCapability.REASONING, model_hint=self.final_ic_model_hint, prompt=prompt)) + parsed = extract_json_object(response.content) + return parsed if isinstance(parsed, dict) else {} + except Exception: + return {} + + def _as_list(self, value: Any) -> list[Any]: + if isinstance(value, list): + return value + if value in (None, ""): + return [] + return [value] + + def _fingerprint(self, value: str) -> str: + tokens = sorted(set(re.findall(r"[a-z0-9]{4,}", value.lower()))) + return hashlib.sha256("|".join(tokens).encode("utf-8")).hexdigest()[:32] + + 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_report(self, content: dict[str, Any]) -> str: + lines = ["# CRYPTO VENTURE COHORT V0.1 REPORT", "", f"Cohort ID: {content['cohort_id']}", f"Accepted protocols: {content['accepted_protocols']}", "", "## Ranking"] + for row in content["ranking"]: + lines.append(f"- Rank {row.get('rank')}: {row['company']} | {row['crypto_ic_decision']} | score {row['crypto_ic_score']} | token {row['token_necessity_classification']}") + lines.append("") + lines.append("## Top 3") + if content["top_3"]: + lines.extend(f"- {row['company']}" for row in content["top_3"]) + else: + lines.append("Fewer than 3 qualified; weak token ideas were not promoted.") + lines.append("") + lines.append("Stop condition: no token sale, fundraising, mainnet issuance, user contact, or real spend.") + return "\n".join(lines) diff --git a/control_plane/ventures/management/commands/export_crypto_venture_cohort.py b/control_plane/ventures/management/commands/export_crypto_venture_cohort.py new file mode 100644 index 0000000..138cb86 --- /dev/null +++ b/control_plane/ventures/management/commands/export_crypto_venture_cohort.py @@ -0,0 +1,121 @@ +from __future__ import annotations + +import json +import uuid +from pathlib import Path +from typing import Any + +from django.core.management.base import BaseCommand, CommandError + +from control_plane.ventures.models import VentureCohort + + +class Command(BaseCommand): + help = "Export a Crypto Venture Cohort V0.1 report." + + def add_arguments(self, parser): + parser.add_argument("--cohort", help="Cohort stable ID or primary key. Defaults to latest crypto cohort.") + parser.add_argument("--format", choices=["markdown", "json"], default="markdown") + parser.add_argument("--output", help="Path to write output. Defaults to stdout.") + parser.add_argument("--indent", type=int, default=2) + + def handle(self, *args, **options): + cohort = self.cohort(str(options.get("cohort") or "")) + payload = self.payload(cohort) + if options["format"] == "json": + indent = None if int(options["indent"]) <= 0 else int(options["indent"]) + content = json.dumps(payload, indent=indent, default=str) + else: + content = self.markdown(payload) + output = options.get("output") + if output: + path = Path(output) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content + "\n", encoding="utf-8") + self.stdout.write(self.style.SUCCESS(f"Exported crypto cohort {cohort.cohort_id} to {path}")) + else: + self.stdout.write(content) + + def cohort(self, identifier: str) -> VentureCohort: + queryset = VentureCohort.objects.filter(metadata__venture_track="CRYPTO_PROTOCOL").order_by("-created_at") + if identifier: + cohort = VentureCohort.objects.filter(cohort_id=identifier).first() + if cohort is None: + try: + cohort = VentureCohort.objects.filter(id=uuid.UUID(identifier)).first() + except ValueError: + cohort = None + else: + cohort = queryset.first() + if cohort is None: + raise CommandError("No crypto venture cohort found.") + return cohort + + def payload(self, cohort: VentureCohort) -> dict[str, Any]: + report = cohort.mandate.artifacts.filter(artifact_type="CRYPTO_VENTURE_COHORT_REPORT").order_by("-created_at").first() + if report is None: + raise CommandError("Crypto cohort report artifact not found.") + return report.content + + def markdown(self, payload: dict[str, Any]) -> str: + lines = [ + "# CRYPTO VENTURE COHORT V0.1 REPORT", + "", + f"Cohort ID: `{payload['cohort_id']}`", + f"GraphRun: `{payload['graph_run']}`", + f"Accepted protocols: `{payload['accepted_protocols']}`", + f"Generation attempts: `{payload['generation_attempts']}`", + "", + "## Runtime", + "", + "```json", + json.dumps(payload.get("runtime", {}), indent=2, default=str), + "```", + "", + "## Ranking", + "", + ] + for row in payload.get("ranking", []): + lines.extend( + [ + f"### Rank {row.get('rank')}: {row['company']}", + "", + f"Decision: `{row['crypto_ic_decision']}`. Score: `{row['crypto_ic_score']}`. Token necessity: `{row['token_necessity_classification']}`.", + "", + f"Product thesis: {row['product_thesis']}", + "", + f"Protocol thesis: {row['protocol_thesis']}", + "", + f"Token thesis: {row['token_thesis']}", + "", + f"Token demand loop: {self.inline(row['token_demand_loop'])}", + "", + f"Value capture: {row['value_capture']}", + "", + f"Network effect: {row['network_effect']}", + "", + f"Bootstrap plan: {row['bootstrap_plan']}", + "", + f"Autonomous operability: `{row['autonomous_operability']}`. Regulatory manageability: `{row['regulatory_manageability']}`. Security risk score: `{row['security_risk']}`.", + "", + f"Scores: {json.dumps(row['component_scores'], default=str)}", + "", + f"Token Red Team flags: {self.inline(row.get('token_red_team_flags', []))}", + "", + f"Validation experiment: {row['validation_experiment']}", + "", + ] + ) + lines.extend(["## Top 3", ""]) + top_3 = payload.get("top_3", []) + if top_3: + lines.extend(f"- {row['company']}: `{row['crypto_ic_decision']}`" for row in top_3) + else: + lines.append("Fewer than 3 qualified; weak token ideas were not promoted.") + lines.extend(["", "## Token Red Team Failures", "", json.dumps(payload.get("token_red_team_failures", {}), indent=2, default=str), "", "## Token Utility Distribution", "", json.dumps(payload.get("token_utility_distribution", {}), indent=2, default=str), "", "## Capability Gaps", "", json.dumps(payload.get("capability_gaps", []), indent=2, default=str), "", "Stop condition: no token sale, no fundraising, no mainnet issuance, no investor/user contact, no liquidity pool, no market making, no real spend."]) + return "\n".join(lines).rstrip() + + def inline(self, value: Any) -> str: + if isinstance(value, list): + return "; ".join(str(item) for item in value) or "none" + return str(value or "none") diff --git a/control_plane/ventures/management/commands/run_crypto_venture_cohort.py b/control_plane/ventures/management/commands/run_crypto_venture_cohort.py new file mode 100644 index 0000000..c5680e8 --- /dev/null +++ b/control_plane/ventures/management/commands/run_crypto_venture_cohort.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +import json +import os + +from django.core.management.base import BaseCommand, CommandError + +from agents.crypto_venture import CryptoVentureService +from control_plane.ventures.models import VentureCohort +from graph.bootstrap import champion_crypto_venture_cohort_graph_v1 +from graph.crypto_venture_cohort import crypto_venture_cohort_registry +from graph.langgraph_runtime import LangGraphRuntime +from graph.models import GraphRun, GraphRunStatus +from model_router.providers import providers_from_resources +from model_router.router import ModelRouter + + +class Command(BaseCommand): + help = "Run a Crypto / Protocol Venture Cohort V0.1." + + def add_arguments(self, parser): + parser.add_argument("--size", type=int, default=10) + parser.add_argument("--concurrency", type=int, default=2) + parser.add_argument("--qwen-only", action="store_true", help="Use Qwen for all crypto cohort model roles.") + parser.add_argument("--sol-final-ic", action="store_true", help="Use Sol for final IC role when configured.") + parser.add_argument("--no-web-research", action="store_true") + parser.add_argument("--persist-requests", action="store_true") + parser.add_argument("--indent", type=int, default=2) + + def handle(self, *args, **options): + size = max(1, int(options["size"])) + concurrency = max(1, int(options["concurrency"])) + keys = ["ARTIFEX_VENTURE_IDEATION_MODEL", "ARTIFEX_VENTURE_RESEARCH_MODEL", "ARTIFEX_VENTURE_PORTFOLIO_IC_MODEL"] + previous = {key: os.environ.get(key) for key in keys} + if options["qwen_only"]: + os.environ["ARTIFEX_VENTURE_IDEATION_MODEL"] = "qwen" + os.environ["ARTIFEX_VENTURE_RESEARCH_MODEL"] = "qwen" + os.environ["ARTIFEX_VENTURE_PORTFOLIO_IC_MODEL"] = "qwen" + if options["sol_final_ic"]: + os.environ["ARTIFEX_VENTURE_PORTFOLIO_IC_MODEL"] = "sol" + try: + providers = providers_from_resources() + if not providers: + raise CommandError("No model providers configured. Run seed_spark_resources first.") + if options["qwen_only"] and "qwen" not in providers: + raise CommandError("--qwen-only requested but no Qwen provider is configured.") + if options["sol_final_ic"] and "sol" not in providers: + raise CommandError("--sol-final-ic requested but no Sol provider is configured.") + version = champion_crypto_venture_cohort_graph_v1() + graph_run = GraphRun.objects.create(execution_graph_version=version, current_node=version.graph_spec["entry"]) + service = CryptoVentureService(ModelRouter(providers, persist_requests=bool(options["persist_requests"])), web_research_available=not bool(options["no_web_research"])) + LangGraphRuntime(crypto_venture_cohort_registry(service, cohort_size=size, concurrency=concurrency)).run_until_terminal_or_paused(graph_run) + graph_run.refresh_from_db() + summary = self.summary(graph_run) + indent = None if int(options["indent"]) <= 0 else int(options["indent"]) + self.stdout.write(json.dumps(summary, indent=indent, default=str)) + if graph_run.status != GraphRunStatus.COMPLETE: + raise CommandError(f"Crypto venture cohort ended with status {graph_run.status}") + finally: + for key, value in previous.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + + def summary(self, graph_run: GraphRun) -> dict[str, object]: + summary: dict[str, object] = {"graph_run_id": str(graph_run.id), "status": graph_run.status, "failure": graph_run.failure_reason, "current_node": graph_run.current_node, "metadata": graph_run.metadata} + cohort_pk = graph_run.metadata.get("cohort_id") if isinstance(graph_run.metadata, dict) else None + if not cohort_pk: + return summary + cohort = VentureCohort.objects.get(id=cohort_pk) + members = list(cohort.members.select_related("proposal").order_by("rank", "created_at")) + report = cohort.mandate.artifacts.filter(artifact_type="CRYPTO_VENTURE_COHORT_REPORT").order_by("-created_at").first() + review = getattr(cohort, "portfolio_review", None) + summary.update( + { + "cohort_id": cohort.cohort_id, + "cohort_pk": str(cohort.id), + "cohort_status": cohort.status, + "size": cohort.cohort_size, + "members": len(members), + "concurrency": cohort.concurrency, + "metrics": cohort.metrics, + "generation_sources": sorted({str(member.proposal.metadata.get("generation_source", "unknown")) for member in members}), + "top_3": review.top_3 if review else [], + "rejected_token_not_needed": [row for row in (review.rankings if review else []) if row.get("crypto_ic_decision") == "REJECT_TOKEN_NOT_NEEDED"], + "token_red_team_results": cohort.metrics.get("token_red_team_failures", {}), + "report_artifact_id": str(report.id) if report else None, + "stop_conditions": {"token_sale": False, "fundraising": False, "mainnet_issuance": False, "user_contact": False, "real_spend": 0}, + } + ) + return summary diff --git a/control_plane/ventures/migrations/0005_alter_autonomousoperabilityassessment_venture_track_and_more.py b/control_plane/ventures/migrations/0005_alter_autonomousoperabilityassessment_venture_track_and_more.py new file mode 100644 index 0000000..f0c398c --- /dev/null +++ b/control_plane/ventures/migrations/0005_alter_autonomousoperabilityassessment_venture_track_and_more.py @@ -0,0 +1,188 @@ +# Generated by Django 5.2.16 on 2026-08-16 10:59 + +import django.db.models.deletion +import uuid +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('ventures', '0004_portfoliothesis_autonomous_candidate_status_and_more'), + ] + + operations = [ + migrations.AlterField( + model_name='autonomousoperabilityassessment', + name='venture_track', + field=models.CharField(choices=[('AUTONOMOUS', 'Autonomous'), ('ASSISTED', 'Assisted'), ('CRYPTO_PROTOCOL', 'Crypto Protocol')], default='AUTONOMOUS', max_length=32), + ), + migrations.AlterField( + model_name='portfoliothesis', + name='venture_track', + field=models.CharField(choices=[('AUTONOMOUS', 'Autonomous'), ('ASSISTED', 'Assisted'), ('CRYPTO_PROTOCOL', 'Crypto Protocol')], default='AUTONOMOUS', max_length=32), + ), + migrations.CreateModel( + name='CryptoJurisdictionPolicy', + 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)), + ('excluded_jurisdictions', models.JSONField(blank=True, default=list)), + ('excluded_person_classes', models.JSONField(blank=True, default=list)), + ('marketing_restrictions', models.JSONField(blank=True, default=list)), + ('sale_restrictions', models.JSONField(blank=True, default=list)), + ('kyc_aml_requirement_status', models.CharField(blank=True, max_length=80)), + ('transfer_restriction_requirement_status', models.CharField(blank=True, max_length=80)), + ('legal_review_required', models.BooleanField(default=True)), + ('regulatory_manageability_score', models.FloatField(default=0.0)), + ('jurisdiction_uncertainty', models.JSONField(blank=True, default=list)), + ('human_legal_gate', models.BooleanField(default=True)), + ('metadata', models.JSONField(blank=True, default=dict)), + ('proposal', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='crypto_jurisdiction_policy', to='ventures.companyproposal')), + ], + options={ + 'abstract': False, + }, + ), + migrations.CreateModel( + name='CryptoRedTeamAssessment', + 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)), + ('flags', models.JSONField(blank=True, default=list)), + ('severity', models.CharField(default='MEDIUM', max_length=40)), + ('critique', models.TextField(blank=True)), + ('independent_from_generation', models.BooleanField(default=True)), + ('metadata', models.JSONField(blank=True, default=dict)), + ('proposal', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='crypto_red_team', to='ventures.companyproposal')), + ], + options={ + 'abstract': False, + }, + ), + migrations.CreateModel( + name='CryptoScenarioLab', + 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)), + ('scenarios', models.JSONField(blank=True, default=list)), + ('systemic_findings', models.JSONField(blank=True, default=list)), + ('progeny_candidates', models.JSONField(blank=True, default=list)), + ('metadata', models.JSONField(blank=True, default=dict)), + ('proposal', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='crypto_scenario_lab', to='ventures.companyproposal')), + ], + options={ + 'abstract': False, + }, + ), + migrations.CreateModel( + name='OnchainNecessityAssessment', + 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)), + ('onchain_necessity_score', models.FloatField(default=0.0)), + ('reasons', models.JSONField(blank=True, default=list)), + ('offchain_substitute', models.TextField(blank=True)), + ('rationale', models.TextField(blank=True)), + ('metadata', models.JSONField(blank=True, default=dict)), + ('proposal', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='onchain_assessment', to='ventures.companyproposal')), + ], + options={ + 'abstract': False, + }, + ), + migrations.CreateModel( + name='ProtocolThesis', + 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)), + ('product_thesis', models.TextField()), + ('protocol_thesis', models.TextField()), + ('token_thesis', models.TextField()), + ('network_effect', models.TextField(blank=True)), + ('bootstrap_plan', models.TextField(blank=True)), + ('autonomous_operability', models.TextField(blank=True)), + ('utility_categories', models.JSONField(blank=True, default=list)), + ('protocol_category', models.CharField(blank=True, max_length=120)), + ('metadata', models.JSONField(blank=True, default=dict)), + ('proposal', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='protocol_thesis', to='ventures.companyproposal')), + ], + options={ + 'abstract': False, + }, + ), + migrations.CreateModel( + name='ProtocolValueCapture', + 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)), + ('value_accrual_quality_score', models.FloatField(default=0.0)), + ('tokenomics_sustainability_score', models.FloatField(default=0.0)), + ('value_accrual', models.TextField(blank=True)), + ('sinks', models.JSONField(blank=True, default=list)), + ('emissions_policy', models.TextField(blank=True)), + ('sustainability_rationale', models.TextField(blank=True)), + ('metadata', models.JSONField(blank=True, default=dict)), + ('proposal', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='value_capture', to='ventures.companyproposal')), + ], + options={ + 'abstract': False, + }, + ), + migrations.CreateModel( + name='TokenDemandLoop', + 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)), + ('loop', models.JSONField(blank=True, default=list)), + ('real_usage_driver', models.TextField(blank=True)), + ('non_speculative_demand', models.BooleanField(default=False)), + ('bootstrap_without_token', models.TextField(blank=True)), + ('metadata', models.JSONField(blank=True, default=dict)), + ('proposal', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='token_demand_loop', to='ventures.companyproposal')), + ], + options={ + 'abstract': False, + }, + ), + migrations.CreateModel( + name='TokenomicsSimulation', + 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)), + ('scenarios', models.JSONField(blank=True, default=dict)), + ('summary', models.JSONField(blank=True, default=dict)), + ('metadata', models.JSONField(blank=True, default=dict)), + ('proposal', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='tokenomics_simulation', to='ventures.companyproposal')), + ], + options={ + 'abstract': False, + }, + ), + migrations.CreateModel( + name='TokenUtilityAssessment', + 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)), + ('classification', models.CharField(choices=[('TOKEN_ESSENTIAL', 'Token Essential'), ('TOKEN_STRONGLY_JUSTIFIED', 'Token Strongly Justified'), ('TOKEN_OPTIONAL', 'Token Optional'), ('TOKEN_UNNECESSARY', 'Token Unnecessary')], max_length=40)), + ('token_necessity_score', models.FloatField(default=0.0)), + ('utility_categories', models.JSONField(blank=True, default=list)), + ('fiat_or_database_substitution', models.TextField(blank=True)), + ('rationale', models.TextField(blank=True)), + ('metadata', models.JSONField(blank=True, default=dict)), + ('proposal', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='token_utility_assessment', to='ventures.companyproposal')), + ], + options={ + 'abstract': False, + }, + ), + ] diff --git a/control_plane/ventures/models.py b/control_plane/ventures/models.py index 9e16aae..de7886a 100644 --- a/control_plane/ventures/models.py +++ b/control_plane/ventures/models.py @@ -94,6 +94,43 @@ class OpportunityTerritory(models.TextChoices): class VentureTrack(models.TextChoices): AUTONOMOUS = "AUTONOMOUS" ASSISTED = "ASSISTED" + CRYPTO_PROTOCOL = "CRYPTO_PROTOCOL" + + +class TokenNecessityClassification(models.TextChoices): + TOKEN_ESSENTIAL = "TOKEN_ESSENTIAL" + TOKEN_STRONGLY_JUSTIFIED = "TOKEN_STRONGLY_JUSTIFIED" + TOKEN_OPTIONAL = "TOKEN_OPTIONAL" + TOKEN_UNNECESSARY = "TOKEN_UNNECESSARY" + + +class CryptoICDecisionType(models.TextChoices): + PROTOCOL_VALIDATE = "PROTOCOL_VALIDATE" + TESTNET_PILOT = "TESTNET_PILOT" + REVISE_TOKEN_MODEL = "REVISE_TOKEN_MODEL" + ROUTE_TO_SAAS = "ROUTE_TO_SAAS" + WATCHLIST = "WATCHLIST" + REJECT_TOKEN_NOT_NEEDED = "REJECT_TOKEN_NOT_NEEDED" + REJECT_SPECULATIVE = "REJECT_SPECULATIVE" + REJECT_REGULATORY_RISK = "REJECT_REGULATORY_RISK" + REJECT_ECONOMIC_MODEL = "REJECT_ECONOMIC_MODEL" + REJECT_OPERABILITY = "REJECT_OPERABILITY" + + +class TokenRedTeamFlag(models.TextChoices): + TOKEN_NOT_REQUIRED = "TOKEN_NOT_REQUIRED" + SPECULATION_DEPENDENT = "SPECULATION_DEPENDENT" + UNSUSTAINABLE_EMISSIONS = "UNSUSTAINABLE_EMISSIONS" + VALUE_CAPTURE_BROKEN = "VALUE_CAPTURE_BROKEN" + MERCENARY_INCENTIVES = "MERCENARY_INCENTIVES" + GOVERNANCE_THEATER = "GOVERNANCE_THEATER" + SECURITY_MODEL_WEAK = "SECURITY_MODEL_WEAK" + TOKEN_VELOCITY_TOO_HIGH = "TOKEN_VELOCITY_TOO_HIGH" + BOOTSTRAP_PROBLEM = "BOOTSTRAP_PROBLEM" + CENTRALIZATION_CONTRADICTION = "CENTRALIZATION_CONTRADICTION" + REGULATORY_RISK_HIGH = "REGULATORY_RISK_HIGH" + ONCHAIN_NOT_REQUIRED = "ONCHAIN_NOT_REQUIRED" + NO_REAL_USER_DEMAND = "NO_REAL_USER_DEMAND" class FounderDependencyLevel(models.TextChoices): @@ -261,6 +298,97 @@ class AutonomousOperabilityAssessment(TimestampedModel): metadata = models.JSONField(default=dict, blank=True) +class ProtocolThesis(TimestampedModel): + proposal = models.OneToOneField(CompanyProposal, on_delete=models.CASCADE, related_name="protocol_thesis") + product_thesis = models.TextField() + protocol_thesis = models.TextField() + token_thesis = models.TextField() + network_effect = models.TextField(blank=True) + bootstrap_plan = models.TextField(blank=True) + autonomous_operability = models.TextField(blank=True) + utility_categories = models.JSONField(default=list, blank=True) + protocol_category = models.CharField(max_length=120, blank=True) + metadata = models.JSONField(default=dict, blank=True) + + +class TokenUtilityAssessment(TimestampedModel): + proposal = models.OneToOneField(CompanyProposal, on_delete=models.CASCADE, related_name="token_utility_assessment") + classification = models.CharField(max_length=40, choices=TokenNecessityClassification.choices) + token_necessity_score = models.FloatField(default=0.0) + utility_categories = models.JSONField(default=list, blank=True) + fiat_or_database_substitution = models.TextField(blank=True) + rationale = models.TextField(blank=True) + metadata = models.JSONField(default=dict, blank=True) + + +class OnchainNecessityAssessment(TimestampedModel): + proposal = models.OneToOneField(CompanyProposal, on_delete=models.CASCADE, related_name="onchain_assessment") + onchain_necessity_score = models.FloatField(default=0.0) + reasons = models.JSONField(default=list, blank=True) + offchain_substitute = models.TextField(blank=True) + rationale = models.TextField(blank=True) + metadata = models.JSONField(default=dict, blank=True) + + +class TokenDemandLoop(TimestampedModel): + proposal = models.OneToOneField(CompanyProposal, on_delete=models.CASCADE, related_name="token_demand_loop") + loop = models.JSONField(default=list, blank=True) + real_usage_driver = models.TextField(blank=True) + non_speculative_demand = models.BooleanField(default=False) + bootstrap_without_token = models.TextField(blank=True) + metadata = models.JSONField(default=dict, blank=True) + + +class ProtocolValueCapture(TimestampedModel): + proposal = models.OneToOneField(CompanyProposal, on_delete=models.CASCADE, related_name="value_capture") + value_accrual_quality_score = models.FloatField(default=0.0) + tokenomics_sustainability_score = models.FloatField(default=0.0) + value_accrual = models.TextField(blank=True) + sinks = models.JSONField(default=list, blank=True) + emissions_policy = models.TextField(blank=True) + sustainability_rationale = models.TextField(blank=True) + metadata = models.JSONField(default=dict, blank=True) + + +class CryptoJurisdictionPolicy(TimestampedModel): + proposal = models.OneToOneField(CompanyProposal, on_delete=models.CASCADE, related_name="crypto_jurisdiction_policy") + excluded_jurisdictions = models.JSONField(default=list, blank=True) + excluded_person_classes = models.JSONField(default=list, blank=True) + marketing_restrictions = models.JSONField(default=list, blank=True) + sale_restrictions = models.JSONField(default=list, blank=True) + kyc_aml_requirement_status = models.CharField(max_length=80, blank=True) + transfer_restriction_requirement_status = models.CharField(max_length=80, blank=True) + legal_review_required = models.BooleanField(default=True) + regulatory_manageability_score = models.FloatField(default=0.0) + jurisdiction_uncertainty = models.JSONField(default=list, blank=True) + human_legal_gate = models.BooleanField(default=True) + metadata = models.JSONField(default=dict, blank=True) + + +class CryptoRedTeamAssessment(TimestampedModel): + proposal = models.OneToOneField(CompanyProposal, on_delete=models.CASCADE, related_name="crypto_red_team") + flags = models.JSONField(default=list, blank=True) + severity = models.CharField(max_length=40, default="MEDIUM") + critique = models.TextField(blank=True) + independent_from_generation = models.BooleanField(default=True) + metadata = models.JSONField(default=dict, blank=True) + + +class TokenomicsSimulation(TimestampedModel): + proposal = models.OneToOneField(CompanyProposal, on_delete=models.CASCADE, related_name="tokenomics_simulation") + scenarios = models.JSONField(default=dict, blank=True) + summary = models.JSONField(default=dict, blank=True) + metadata = models.JSONField(default=dict, blank=True) + + +class CryptoScenarioLab(TimestampedModel): + proposal = models.OneToOneField(CompanyProposal, on_delete=models.CASCADE, related_name="crypto_scenario_lab") + scenarios = models.JSONField(default=list, blank=True) + systemic_findings = models.JSONField(default=list, blank=True) + progeny_candidates = models.JSONField(default=list, 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) diff --git a/graph/bootstrap.py b/graph/bootstrap.py index a33614e..b694d14 100644 --- a/graph/bootstrap.py +++ b/graph/bootstrap.py @@ -3,6 +3,7 @@ from __future__ import annotations from django.utils import timezone from graph.agent_control import agent_investigation_graph_v1 +from graph.crypto_venture_cohort import crypto_venture_cohort_graph_v1 from graph.lifecycle import project_evolution_graph_v1, project_exploration_graph_v1, project_extension_graph_v1 from graph.models import ExecutionGraphDefinition, ExecutionGraphVersion, ExecutionGraphVersionStatus from graph.roadmap import project_roadmap_review_graph_v1 @@ -108,3 +109,7 @@ def champion_venture_discovery_graph_v1() -> ExecutionGraphVersion: def champion_venture_discovery_cohort_graph_v1() -> ExecutionGraphVersion: return _champion_graph(venture_discovery_cohort_graph_v1()) + + +def champion_crypto_venture_cohort_graph_v1() -> ExecutionGraphVersion: + return _champion_graph(crypto_venture_cohort_graph_v1()) diff --git a/graph/crypto_venture_cohort.py b/graph/crypto_venture_cohort.py new file mode 100644 index 0000000..734f9c3 --- /dev/null +++ b/graph/crypto_venture_cohort.py @@ -0,0 +1,186 @@ +from __future__ import annotations + +from agents.crypto_venture import CryptoVentureService +from control_plane.ventures.models import VentureCohort +from graph.native_runtime import GraphExecutionContext +from graph.registry import NodeHandlerRegistry, NodeResult +from graph.spec import ExecutionGraphSpec, GraphEdgeSpec, GraphNodeSpec + + +def crypto_venture_cohort_graph_v1() -> ExecutionGraphSpec: + nodes = [ + "prepare_crypto_mandate", + "portfolio_crypto_thesis_review", + "generate_independent_protocols", + "token_necessity_gate", + "novelty_gate", + "regenerate_rejected_slots", + "light_market_research", + "protocol_research", + "protocol_security_gate", + "token_red_team", + "tokenomics_simulation", + "crypto_ic_first_pass", + "top5_deep_research", + "rescore", + "portfolio_crypto_ic", + "regulatory_gate", + "capability_analysis", + "update_crypto_thesis_registry", + "produce_crypto_cohort_report", + "complete", + ] + spec = ExecutionGraphSpec(name="crypto_venture_cohort", version=1, graph_type="CRYPTO_VENTURE_COHORT", entry="prepare_crypto_mandate", nodes={node: GraphNodeSpec(node, node if node == "complete" else f"crypto_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": "Crypto / Protocol Venture Cohort V0.1: token necessity, onchain necessity, token red team, simulation, regulatory gate, no issuance."}) + spec.validate() + return spec + + +class CryptoCohortNode: + idempotent = True + replay_safe = True + destructive = False + + def __init__(self, service: CryptoVentureService, node_type: str, *, cohort_size: int = 10, concurrency: int = 2) -> None: + self.service = service + self.node_type = node_type + self.cohort_size = cohort_size + self.concurrency = concurrency + + def cohort(self, context: GraphExecutionContext) -> VentureCohort: + return VentureCohort.objects.get(id=context.graph_run.metadata["cohort_id"]) + + +class PrepareCryptoMandateNode(CryptoCohortNode): + def run(self, context: GraphExecutionContext) -> NodeResult: + cohort = self.service.prepare_cohort(size=self.cohort_size, graph_run=context.graph_run, concurrency=self.concurrency) + context.graph_run.metadata = {**context.graph_run.metadata, "cohort_id": str(cohort.id), "cohort_stable_id": cohort.cohort_id, "no_real_spend": True, "no_user_contact": True, "no_token_sale": True, "no_fundraising": True, "no_mainnet_issuance": True} + context.graph_run.save(update_fields=["metadata", "updated_at"]) + return NodeResult("COMPLETE", "success", {"cohort_id": cohort.cohort_id, "size": cohort.cohort_size}) + + +class GenerateProtocolsNode(CryptoCohortNode): + def run(self, context: GraphExecutionContext) -> NodeResult: + proposals = self.service.generate_protocols(self.cohort(context)) + return NodeResult("COMPLETE", "success", {"proposal_count": len(proposals)}) + + +class TokenNecessityNode(CryptoCohortNode): + def run(self, context: GraphExecutionContext) -> NodeResult: + assessments = self.service.token_necessity_gate(self.cohort(context)) + return NodeResult("COMPLETE", "success", {"assessment_count": len(assessments)}) + + +class NoveltyGateNode(CryptoCohortNode): + def run(self, context: GraphExecutionContext) -> NodeResult: + return NodeResult("COMPLETE", "success", self.service.novelty_gate(self.cohort(context))) + + +class RegenerateRejectedSlotsNode(CryptoCohortNode): + def run(self, context: GraphExecutionContext) -> NodeResult: + created = self.service.regenerate_rejected_slots(self.cohort(context)) + return NodeResult("COMPLETE", "success", {"regenerated": len(created)}) + + +class LightResearchNode(CryptoCohortNode): + def run(self, context: GraphExecutionContext) -> NodeResult: + cohort = self.cohort(context) + self.service.light_market_research(cohort) + return NodeResult("COMPLETE", "success", cohort.metrics) + + +class ProtocolResearchNode(CryptoCohortNode): + def run(self, context: GraphExecutionContext) -> NodeResult: + self.service.protocol_research(self.cohort(context)) + return NodeResult("COMPLETE", "success") + + +class ProtocolSecurityGateNode(CryptoCohortNode): + def run(self, context: GraphExecutionContext) -> NodeResult: + return NodeResult("COMPLETE", "success", self.service.protocol_security_gate(self.cohort(context))) + + +class TokenRedTeamNode(CryptoCohortNode): + def run(self, context: GraphExecutionContext) -> NodeResult: + assessments = self.service.token_red_team(self.cohort(context)) + return NodeResult("COMPLETE", "success", {"red_team_count": len(assessments)}) + + +class TokenomicsSimulationNode(CryptoCohortNode): + def run(self, context: GraphExecutionContext) -> NodeResult: + simulations = self.service.tokenomics_simulation(self.cohort(context)) + return NodeResult("COMPLETE", "success", {"simulation_count": len(simulations)}) + + +class CryptoICFirstPassNode(CryptoCohortNode): + def run(self, context: GraphExecutionContext) -> NodeResult: + self.service.crypto_ic_first_pass(self.cohort(context)) + return NodeResult("COMPLETE", "success") + + +class Top5DeepResearchNode(CryptoCohortNode): + def run(self, context: GraphExecutionContext) -> NodeResult: + self.service.top5_deep_research(self.cohort(context)) + return NodeResult("COMPLETE", "success") + + +class PortfolioCryptoICNode(CryptoCohortNode): + def run(self, context: GraphExecutionContext) -> NodeResult: + review = self.service.portfolio_crypto_ic(self.cohort(context)) + return NodeResult("COMPLETE", "success", {"ranked_count": len(review.rankings), "top_3": review.top_3}) + + +class RegulatoryGateNode(CryptoCohortNode): + def run(self, context: GraphExecutionContext) -> NodeResult: + self.service.regulatory_gate(self.cohort(context)) + return NodeResult("COMPLETE", "success") + + +class CapabilityAnalysisNode(CryptoCohortNode): + def run(self, context: GraphExecutionContext) -> NodeResult: + gaps = self.service.capability_analysis(self.cohort(context)) + return NodeResult("COMPLETE", "success", {"capability_gap_count": len(gaps)}) + + +class RegistryUpdateNode(CryptoCohortNode): + def run(self, context: GraphExecutionContext) -> NodeResult: + updates = self.service.update_crypto_thesis_registry(self.cohort(context)) + return NodeResult("COMPLETE", "success", {"update_count": len(updates)}) + + +class ReportNode(CryptoCohortNode): + def run(self, context: GraphExecutionContext) -> NodeResult: + artifact = self.service.produce_crypto_cohort_report(self.cohort(context)) + return NodeResult("COMPLETE", "success", {"artifact_id": str(artifact.id), "artifact_type": artifact.artifact_type}) + + +class NoopNode(CryptoCohortNode): + def run(self, context: GraphExecutionContext) -> NodeResult: + return NodeResult("COMPLETE", "success") + + +def crypto_venture_cohort_registry(service: CryptoVentureService, *, cohort_size: int = 10, concurrency: int = 2) -> NodeHandlerRegistry: + registry = NodeHandlerRegistry() + handlers = [ + PrepareCryptoMandateNode(service, "crypto_venture_prepare_crypto_mandate", cohort_size=cohort_size, concurrency=concurrency), + NoopNode(service, "crypto_venture_portfolio_crypto_thesis_review"), + GenerateProtocolsNode(service, "crypto_venture_generate_independent_protocols"), + TokenNecessityNode(service, "crypto_venture_token_necessity_gate"), + NoveltyGateNode(service, "crypto_venture_novelty_gate"), + RegenerateRejectedSlotsNode(service, "crypto_venture_regenerate_rejected_slots"), + LightResearchNode(service, "crypto_venture_light_market_research"), + ProtocolResearchNode(service, "crypto_venture_protocol_research"), + ProtocolSecurityGateNode(service, "crypto_venture_protocol_security_gate"), + TokenRedTeamNode(service, "crypto_venture_token_red_team"), + TokenomicsSimulationNode(service, "crypto_venture_tokenomics_simulation"), + CryptoICFirstPassNode(service, "crypto_venture_crypto_ic_first_pass"), + Top5DeepResearchNode(service, "crypto_venture_top5_deep_research"), + CryptoICFirstPassNode(service, "crypto_venture_rescore"), + PortfolioCryptoICNode(service, "crypto_venture_portfolio_crypto_ic"), + RegulatoryGateNode(service, "crypto_venture_regulatory_gate"), + CapabilityAnalysisNode(service, "crypto_venture_capability_analysis"), + RegistryUpdateNode(service, "crypto_venture_update_crypto_thesis_registry"), + ReportNode(service, "crypto_venture_produce_crypto_cohort_report"), + ] + for handler in handlers: + registry.register(handler) + return registry diff --git a/tests/test_crypto_venture_cohort.py b/tests/test_crypto_venture_cohort.py new file mode 100644 index 0000000..03548e6 --- /dev/null +++ b/tests/test_crypto_venture_cohort.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +import json + +from agents.crypto_venture import CryptoVentureService +from control_plane.ventures.models import CompanyProposalStatus, CryptoJurisdictionPolicy, TokenNecessityClassification, TokenRedTeamFlag, VentureCohortMember, VentureTrack +from graph.bootstrap import champion_crypto_venture_cohort_graph_v1 +from graph.crypto_venture_cohort import crypto_venture_cohort_registry +from graph.langgraph_runtime import LangGraphRuntime +from graph.models import GraphRun, GraphRunStatus +from model_router.router import ModelProvider, ModelRequestContract, ModelResponseContract, ModelRouter + + +class SolCryptoReviewProvider(ModelProvider): + provider_name = "sol-test" + + def complete(self, request: ModelRequestContract) -> ModelResponseContract: + prompt = request.prompt + if "Counterfactual token necessity" in prompt: + payload = {"classification": "TOKEN_STRONGLY_JUSTIFIED", "score": 78, "rationale": "Provider staking and slashing materially degrade with database credits.", "fiat_or_database_substitution": "Stablecoin can pay fees but cannot replace bonded slashing and public reputation."} + elif "Independent Token Red Team" in prompt: + payload = {"flags": ["BOOTSTRAP_PROBLEM"], "severity": "MEDIUM", "critique": "Provider supply bootstrap remains the main risk."} + elif "Final Crypto IC" in prompt: + payload = {"score_adjustment": 3, "decision": "PROTOCOL_VALIDATE", "rationale": "Token loop is plausible but needs testnet evidence."} + else: + payload = {} + return ModelResponseContract("sol", json.dumps(payload), {}) + + def health(self) -> str: + return "AVAILABLE" + + +def service(router: ModelRouter | None = None) -> CryptoVentureService: + return CryptoVentureService(router, web_research_available=False, generation_model_hint="qwen", research_model_hint="qwen", final_ic_model_hint="sol") + + +def test_crypto_track_and_token_gate_remove_weak_then_regenerate() -> None: + svc = service() + cohort = svc.prepare_cohort(size=2, concurrency=1) + weak = svc._create_proposal( + cohort.mandate, + {"name": "Community Points SaaS", "product_thesis": "A normal SaaS dashboard.", "protocol_thesis": "A private database tracks points.", "token_thesis": "Community marketing token.", "token_utility": ["community"], "token_demand_loop": ["more users equals more token demand"]}, + "test", + "OPEN_CRYPTO_CATEGORY", + ) + VentureCohortMember.objects.create(cohort=cohort, proposal=weak) + + svc.token_necessity_gate(cohort) + svc.regenerate_rejected_slots(cohort) + weak.refresh_from_db() + + assert weak.status == CompanyProposalStatus.REJECTED + assert weak.metadata["routed_to_saas"] is True + assert cohort.members.count() == 2 + assert all(member.proposal.token_utility_assessment.classification in {TokenNecessityClassification.TOKEN_ESSENTIAL, TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED} for member in cohort.members.select_related("proposal")) + + +def test_sol_counterfactual_red_team_and_final_ic_are_used() -> None: + svc = service(ModelRouter({"sol": SolCryptoReviewProvider()})) + cohort = svc.prepare_cohort(size=1, concurrency=1) + proposal = svc._create_proposal(cohort.mandate, svc._fallback_payload(0, "PROOF_ATTESTATION_MARKETS"), "test", "PROOF_ATTESTATION_MARKETS") + VentureCohortMember.objects.create(cohort=cohort, proposal=proposal) + + token = svc.assess_token_necessity(proposal) + red = svc.red_team_proposal(proposal) + svc.tokenomics_simulation(cohort) + review = svc.portfolio_crypto_ic(cohort) + + assert token.classification == TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED + assert token.metadata["sol_counterfactual_review"] + assert TokenRedTeamFlag.BOOTSTRAP_PROBLEM in red.flags + assert review.rankings[0]["sol_final_ic_review"]["score_adjustment"] == 3 + + +def test_regulatory_negation_does_not_treat_disabled_sale_as_issuance_risk() -> None: + svc = service() + cohort = svc.prepare_cohort(size=1, concurrency=1) + proposal = svc._create_proposal(cohort.mandate, svc._fallback_payload(0, "AGENT_TO_AGENT_PAYMENTS"), "test", "AGENT_TO_AGENT_PAYMENTS") + + policy = svc.regulatory_policy(proposal) + + assert policy.regulatory_manageability_score == 72 + assert policy.metadata["TOKEN_SALE_DISABLED"] is True + assert CryptoJurisdictionPolicy.objects.get(proposal=proposal).legal_review_required is True + + +def test_tokenomics_simulation_is_proposal_specific() -> None: + svc = service() + cohort = svc.prepare_cohort(size=1, concurrency=1) + low = svc._create_proposal(cohort.mandate, {**svc._fallback_payload(0, "AGENT_TO_AGENT_PAYMENTS"), "confidence": 0.35}, "test", "AGENT_TO_AGENT_PAYMENTS") + high = svc._create_proposal(cohort.mandate, {**svc._fallback_payload(1, "DECENTRALIZED_AI_COMPUTE"), "confidence": 0.9}, "test", "DECENTRALIZED_AI_COMPUTE") + + low_sim = svc.simulate_tokenomics(low) + high_sim = svc.simulate_tokenomics(high) + + assert low_sim.scenarios["EXPECTED"]["users"] != high_sim.scenarios["EXPECTED"]["users"] + assert low_sim.scenarios["EXPECTED"]["token_demand"] != high_sim.scenarios["EXPECTED"]["token_demand"] + + +def test_crypto_cohort_graph_real_gates_and_report() -> None: + svc = service() + version = champion_crypto_venture_cohort_graph_v1() + graph_run = GraphRun.objects.create(execution_graph_version=version, current_node=version.graph_spec["entry"]) + + LangGraphRuntime(crypto_venture_cohort_registry(svc, cohort_size=3, concurrency=1)).run_until_terminal_or_paused(graph_run) + graph_run.refresh_from_db() + + cohort = graph_run.venture_cohorts.get() + report = cohort.mandate.artifacts.get(artifact_type="CRYPTO_VENTURE_COHORT_REPORT") + + assert graph_run.status == GraphRunStatus.COMPLETE + assert cohort.metadata["venture_track"] == VentureTrack.CRYPTO_PROTOCOL + assert cohort.members.count() == 3 + assert cohort.metrics["protocol_security_gate_passed"] >= 1 + assert len(report.content["ranking"]) == 3 + assert len(report.content["top_3"]) <= 3