from __future__ import annotations import json import pytest 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 - MINIMAL DECENTRALIZED" in prompt: payload = {"protocol_architecture": "onchain escrow and public machine credit registry", "participant_model": "agents, providers, risk observers", "verification_model": "objective settlement proofs", "resource_allocation_model": "risk-weighted access", "dispute_model": "automated evidence checks"} elif "PASS D - NATIVE 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_UTILITY_JUDGE" in prompt or "CRYPTO_TOKEN_COUNTERFACTUAL_JUDGE" in prompt or "Counterfactual token necessity" in prompt: payload = {"token_classification": "TOKEN_STRONGLY_JUSTIFIED", "score": 78, "utility_scores": {"SECURITY_UTILITY": 82, "COLLATERAL_UTILITY": 84, "MACHINE_ECONOMIC_UTILITY": 86, "SPECULATION_DEPENDENCE": 10}, "strong_structural_roles": ["SECURITY_BOND", "SLASHABLE_COLLATERAL"], "weak_roles": ["DEMAND_SIDE_PAYMENT"], "counterfactual_analysis": {"summary": "USDC/ETH can operate but weakens protocol-specific risk pricing.", "external_collateral_identical_security": False, "stable_collateral_identical_security": False}, "external_collateral_counterfactual": {"breaks": "external collateral misses protocol-specific risk"}, "material_improvements_over_USDC_ETH": ["protocol-specific default risk", "portable machine economic identity"], "native_asset_costs_and_risks": ["volatility", "legal complexity"], "argument_for_native_token": "Native bond prices protocol-specific machine risk.", "argument_against_native_token": "ETH/USDC collateral is simpler and more liquid.", "final_rationale": "ETH/USDC/external collateral cannot reproduce the protocol-specific machine default-risk pricing economically enough, so the native bond benefit outweighs issuance costs.", "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 is materially weaker."} 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.1") 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 assert cohort.metrics["sol_onchain_judge_requests"] == 1 def test_v031_onchain_reject_skips_token_design() -> None: class RejectOnchainProvider(SolCryptoReviewProvider): def complete(self, request: ModelRequestContract) -> ModelResponseContract: if "PASS B - ONCHAIN" in request.prompt: return ModelResponseContract("sol", json.dumps({"onchain_necessity_score": 40, "pass_onchain": False, "why_onchain": "database is fine"}), {}) return super().complete(request) svc = sol_service(ModelRouter({"sol": RejectOnchainProvider()})) cohort = svc.prepare_cohort(size=1, concurrency=1) svc.generate_protocols(cohort) svc.onchain_necessity_gate(cohort) svc.token_necessity_gate(cohort) assert cohort.metrics["onchain_passed"] == 0 assert cohort.metrics["token_design_attempts"] == 0 assert cohort.metrics["token_design_skipped_due_to_onchain_rejection"] == 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_token_utility_judge"] assert token.metadata["argument_for_native_token"] assert token.metadata["argument_against_native_token"] 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_USAGE"]["users"] != high_sim.scenarios["EXPECTED_USAGE"]["users"] assert low_sim.scenarios["EXPECTED_USAGE"]["token_demand"] != high_sim.scenarios["EXPECTED_USAGE"]["token_demand"] assert "NATIVE_TOKEN_REMOVED" in low_sim.scenarios assert "ETH_COLLATERAL" in low_sim.scenarios assert "USDC_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_token_utility_judge"] == "CRYPTO_TOKEN_UTILITY_JUDGE" assert token.metadata["native_token_removed_degradation"] == "MATERIAL" assert token.metadata["argument_for_native_token"] for field in ["argument_for_native_token", "argument_against_native_token", "external_collateral_counterfactual", "native_token_removed_breaks", "native_token_removed_explanation", "material_improvements_over_USDC_ETH", "native_asset_costs_and_risks", "final_rationale"]: assert token.metadata[field] assert token.rationale def test_v032b_incomplete_token_judge_retries_then_persists_complete_response() -> None: class RetryProvider(SolCryptoReviewProvider): def __init__(self) -> None: self.calls = 0 def complete(self, request: ModelRequestContract) -> ModelResponseContract: if "CRYPTO_TOKEN_UTILITY_JUDGE" in request.prompt: self.calls += 1 if self.calls == 1: return ModelResponseContract("sol", json.dumps({"token_classification": "TOKEN_OPTIONAL", "score": 55}), {}) return super().complete(request) provider = RetryProvider() svc = service(ModelRouter({"sol": provider})) 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 provider.calls == 2 assert token.metadata["external_collateral_counterfactual"] def test_v032b_incomplete_token_judge_fails_closed_without_fallback_classification() -> None: class IncompleteProvider(SolCryptoReviewProvider): def complete(self, request: ModelRequestContract) -> ModelResponseContract: if "CRYPTO_TOKEN_UTILITY_JUDGE" in request.prompt: return ModelResponseContract("sol", json.dumps({"token_classification": "TOKEN_OPTIONAL", "score": 55, "final_rationale": "too thin"}), {}) return super().complete(request) svc = service(ModelRouter({"sol": IncompleteProvider()})) cohort = svc.prepare_cohort(size=1, concurrency=1) proposal = svc._create_proposal(cohort.mandate, svc._fallback_payload(0, "PROOF_MARKETS"), "test", "PROOF_MARKETS") with pytest.raises(ValueError, match="Incomplete Sol token utility judge response"): svc.assess_token_necessity(proposal) assert not hasattr(proposal, "token_utility_assessment") def test_v031_payment_governance_reward_only_rejected_but_structural_passes() -> None: svc = service(ModelRouter({"sol": SolCryptoReviewProvider()})) cohort = svc.prepare_cohort(size=1, concurrency=1) weak = svc._create_proposal(cohort.mandate, {"name": "FeeGov Rewards", "product_thesis": "Useful SaaS", "protocol_thesis": "Onchain logs", "token_thesis": "Token is for payment governance and rewards", "token_utility": ["payment", "governance", "rewards"], "token_roles": {"DEMAND_SIDE_PAYMENT": "REQUIRED", "GOVERNANCE": "REQUIRED", "PROVIDER_REWARD": "REQUIRED"}}, "test", "OPEN_CRYPTO_CATEGORY") strong = svc._create_proposal(cohort.mandate, {**svc._fallback_payload(0, "MACHINE_REPUTATION_WITH_ECONOMIC_STAKE"), "token_roles": {"SECURITY_BOND": "REQUIRED", "SLASHABLE_COLLATERAL": "REQUIRED", "MACHINE_ECONOMIC_IDENTITY": "STRONGLY_USEFUL"}}, "test", "MACHINE_REPUTATION_WITH_ECONOMIC_STAKE") weak_token = svc.assess_token_necessity(weak) strong_token = svc.assess_token_necessity(strong) assert weak_token.classification in {TokenNecessityClassification.TOKEN_OPTIONAL, TokenNecessityClassification.TOKEN_UNNECESSARY} assert strong_token.classification == TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED assert len(strong_token.metadata["strong_token_roles"]) >= 2 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 if cohort.metrics.get("serious_crypto_survivors", 0) > 0: 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.1" if cohort.metrics.get("serious_crypto_survivors", 0) > 0: assert cohort.metrics["sol_final_ic_requests"] > 0