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 "PASS A - PROBLEM" in prompt: payload = {"name": "Machine Identity Credit Union", "one_line_thesis": "Autonomous agents build portable credit histories for machine services.", "product_thesis": "A pre-token API for machine credit and settlement risk scoring.", "user": "AI agent operators", "participants": ["agents", "service providers", "risk observers"], "supply_side": "providers publish measurable service outcomes", "demand_side": "agents need counterparties", "coordination_failure": "portable machine credit is not trusted across operators", "trust_problem": "operators can rewrite private reputation", "why_neutral_network_may_help": "neutral verifiable state lets agents carry economic identity across providers", "pre_token_product": "risk scoring API", "pre_token_customer": "agent operators", "pre_token_value": "lower counterparty default risk", "pre_token_payment_method": "Stripe or stablecoin usage credits", "pre_token_revenue_model": "$100 beta packages", "revenue_1000_path": "10 beta packages", "revenue_5000_path": "20 risk API packages", "revenue_10000_path": "20 subscriptions", "autonomous_operating_loop": "machine demand -> discovery -> settlement -> verification -> reputation update", "bootstrap_plan": "local/testnet simulation", "validation_experiment": "simulate 100 agent jobs", "major_risks": ["token not required"], "confidence": 0.7} elif "PASS B - ONCHAIN" in prompt: payload = {"onchain_necessity_score": 86, "why_onchain": "public reputation and escrow materially degrade in private databases", "offchain_substitute": "single operator SaaS", "pass_onchain": True, "reasons": ["verifiable state", "neutral settlement", "machine-to-machine interaction"]} elif "PASS C - DERIVED TOKEN" in prompt: payload = {"token_thesis": "Native bond represents protocol-specific machine default risk and is slashed for objective settlement failures.", "token_utility": ["protocol-specific collateral", "machine reputation-backed economic participation"], "token_demand_loop": ["agents earn credit", "providers require bond", "defaults slash bond", "reliable agents receive access"], "value_capture": "risk fees and slashing penalties", "security_model": "escrow, timelock, pause, oracle checks, key controls", "regulatory_risks": ["legal review required"], "required_token_functions": ["protocol-specific risk collateral"], "weak_token_functions": ["payment"]} elif "CRYPTO_TOKEN_COUNTERFACTUAL_JUDGE" in prompt or "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.", "model_c_external_collateral": {"breaks": "external collateral misses protocol-specific risk"}, "native_token_removed_breaks": ["machine economic identity weakens"], "native_token_removed_degradation": "MATERIAL", "native_token_removed_explanation": "USDC payments work, but protocol-specific bond/reputation does not."} 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 sol_service(router: ModelRouter | None = None) -> CryptoVentureService: return CryptoVentureService(router, web_research_available=False, generation_model_hint="sol", 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 False assert cohort.members.count() == 0 assert cohort.metrics["regeneration_policy"].startswith("V0.3") def test_v03_sol_problem_first_generation_no_token_in_pass_a_and_onchain_gate() -> None: svc = sol_service(ModelRouter({"sol": SolCryptoReviewProvider()})) cohort = svc.prepare_cohort(size=1, concurrency=1) proposals = svc.generate_protocols(cohort) proposal = proposals[0] assessment = svc.onchain_necessity_gate(cohort)[0] passes = proposal.metadata["crypto"]["generation_passes"] assert proposal.metadata["generation_source"] == "sol" assert "token_thesis" not in passes["pass_a_problem_network"] assert proposal.metadata["crypto"]["problem_first_pass_a_had_no_token"] is True assert assessment.onchain_necessity_score == 86 assert cohort.metrics["raw_sol_theses"] == 1 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"] assert "NATIVE_TOKEN_REMOVED" in low_sim.scenarios assert "EXTERNAL_COLLATERAL" in low_sim.scenarios def test_v03_primary_source_quality_wrong_category_and_weak_coverage() -> None: svc = service() cohort = svc.prepare_cohort(size=1, concurrency=1) proposal = svc._create_proposal(cohort.mandate, svc._fallback_payload(0, "MACHINE_ECONOMIC_IDENTITY"), "test", "MACHINE_ECONOMIC_IDENTITY") good = svc.crypto_source_quality(proposal, {"url": "https://ethereum.org/en/developers/docs/", "title": "Ethereum developer docs", "summary": "onchain escrow settlement staking protocol", "category": "ONCHAIN_ALTERNATIVES"}) wrong = svc.crypto_source_quality(proposal, {"url": "https://www.microsoft.com/windows", "title": "Windows help", "summary": "generic software", "category": "TOKEN_MODELS"}) assert good["accepted"] is True assert good["source_tier"] == "TIER_A" assert wrong["accepted"] is False assert wrong["reason"].startswith("SOURCE_REJECT_") assert svc.category_coverage_confidence([{"quality": "weak", "url": "https://example.com/a"}]) == "LOW" def test_v03_search_engine_circuit_breaker(monkeypatch) -> None: class CaptchaClient: def search_payload(self, query: str) -> dict: return {"unresponsive_engines": [("google", "CAPTCHA required")], "results": []} class EmptyFetcher: def fetch_many(self, sources, max_pages=5): return [] monkeypatch.setattr("agents.crypto_venture.SearxngSearchClient.from_resources", staticmethod(lambda: CaptchaClient())) monkeypatch.setattr("agents.crypto_venture.WebPageFetcher", lambda: EmptyFetcher()) svc = service() cohort = svc.prepare_cohort(size=1, concurrency=1) proposal = svc._create_proposal(cohort.mandate, svc._fallback_payload(0, "MACHINE_ECONOMIC_IDENTITY"), "test", "MACHINE_ECONOMIC_IDENTITY") research = svc.crypto_research(proposal, depth="light") assert research["queries_avoided_due_to_circuit_breaker"] > 0 assert research["search_engine_health"]["CAPTCHA"] >= 4 def test_v02_pretoken_roles_security_and_launch_readiness() -> None: svc = service() cohort = svc.prepare_cohort(size=1, concurrency=1) proposal = svc._create_proposal(cohort.mandate, svc._fallback_payload(0, "DECENTRALIZED_AI_COMPUTE"), "test", "DECENTRALIZED_AI_COMPUTE") VentureCohortMember.objects.create(cohort=cohort, proposal=proposal) pretoken = svc.assess_pretoken_monetization(proposal) roles = svc.token_role_decomposition(proposal) model = svc.token_economic_model(proposal) svc.protocol_security_gate(cohort) readiness = svc.token_launch_readiness(proposal) assert pretoken.pre_token_monetization_potential > 0 assert roles.stablecoin_counterfactual["MODEL_B_stablecoin_payment_native_bond"] assert model.average_fee > 0 assert proposal.protocol_security_assessment.decision in {"PASS_FOR_TESTNET_DESIGN", "REVISE_SECURITY_MODEL", "BLOCK_TESTNET"} assert readiness.launch_readiness_status != "READY_FOR_HUMAN_LAUNCH_REVIEW" def test_v03_independent_counterfactual_native_removed_fields() -> 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_MARKETS"), "test", "PROOF_MARKETS") token = svc.assess_token_necessity(proposal) assert token.metadata["independent_counterfactual_judge"] == "CRYPTO_TOKEN_COUNTERFACTUAL_JUDGE" assert token.metadata["native_token_removed_degradation"] == "MATERIAL" assert token.metadata["model_c_external_collateral"] 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 assert report.content["title"] == "CRYPTO VENTURE COHORT V0.3"