From 6e89ea8acfdb3036423ed3db7989f4834597f057 Mon Sep 17 00:00:00 2001 From: Daniel Maddern Date: Sun, 16 Aug 2026 22:59:30 +0700 Subject: [PATCH] Require complete crypto token judge evidence --- agents/crypto_venture.py | 59 ++++++++++++++++++++++++++++- tests/test_crypto_venture_cohort.py | 47 ++++++++++++++++++++++- 2 files changed, 103 insertions(+), 3 deletions(-) diff --git a/agents/crypto_venture.py b/agents/crypto_venture.py index 848296c..c9eb6a3 100644 --- a/agents/crypto_venture.py +++ b/agents/crypto_venture.py @@ -118,6 +118,7 @@ CRYPTO_RESEARCH_CATEGORIES = ["USER_PAIN", "EXISTING_PROTOCOLS", "FAILED_PRECEDE STRUCTURAL_TOKEN_ROLES = {"SECURITY_BOND", "SLASHABLE_COLLATERAL", "RESOURCE_ALLOCATION", "MACHINE_ECONOMIC_IDENTITY", "PROVIDER_ADMISSION", "DECENTRALIZED_SUPPLY_COORDINATION"} WEAK_TOKEN_ROLES = {"DEMAND_SIDE_PAYMENT", "GOVERNANCE", "PROVIDER_REWARD", "TREASURY", "ACCESS"} TOKEN_UTILITY_SCORE_KEYS = ["SECURITY_UTILITY", "COORDINATION_UTILITY", "COLLATERAL_UTILITY", "SLASHING_UTILITY", "RESOURCE_ALLOCATION_UTILITY", "INCENTIVE_UTILITY", "NETWORK_BOOTSTRAP_UTILITY", "VALUE_CAPTURE_UTILITY", "MACHINE_ECONOMIC_UTILITY", "GOVERNANCE_UTILITY", "PAYMENT_UTILITY", "SPECULATION_DEPENDENCE"] +TOKEN_JUDGE_REQUIRED_FIELDS = ["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"] PRIMARY_SOURCE_DOMAINS = ["ethereum.org", "eips.ethereum.org", "github.com", "arxiv.org", "sec.gov", "cftc.gov", "immunefi.com"] SOURCE_REJECTION_REASONS = {"IRRELEVANT": "SOURCE_REJECT_IRRELEVANT", "WRONG_CATEGORY": "SOURCE_REJECT_WRONG_CATEGORY", "LOW_AUTHORITY": "SOURCE_REJECT_LOW_AUTHORITY", "SEARCH_NOISE": "SOURCE_REJECT_SEARCH_NOISE", "UNPARSABLE": "SOURCE_REJECT_UNPARSABLE", "DUPLICATE": "SOURCE_REJECT_DUPLICATE"} @@ -679,7 +680,8 @@ class CryptoVentureService: score = min(100, 35 + strong_count * 4 + len(strong_roles) * 8 + len(structural_roles) * 13 + (12 if "slash" in text or "slashing" in text else 0) + (8 if "provider" in text else 0) - (35 if payment_only or weak_only else 0) - (35 if weak else 0) - int(utility_profile.get("SPECULATION_DEPENDENCE", 0) * 0.25)) economic_model = getattr(proposal, "token_economic_model", None) model_payload = {"unit_of_service": economic_model.unit_of_service, "payment_asset": economic_model.payment_asset, "stake_requirement": economic_model.stake_requirement, "collateral_requirement": economic_model.collateral_requirement} if economic_model else {} - sol_review = self._sol_json("CRYPTO_TOKEN_UTILITY_JUDGE V0.3.1. Independent from ideation, onchain review, and token design. Challenge the token; do not merely summarize it. A native token need not be mathematically impossible to substitute, but benefits must outweigh costs. Evaluate security quality, attack economics, slashing effectiveness, provider skin-in-the-game, capital efficiency, protocol-specific risk pricing, permissionless supply formation, participant coordination, resource allocation, reputation portability, value capture, network effects, bootstrapping economics, governance, and composability versus USDC/ETH/external collateral/non-transferable attestations. Return JSON with token_classification TOKEN_ESSENTIAL/TOKEN_STRONGLY_JUSTIFIED/TOKEN_OPTIONAL/TOKEN_UNNECESSARY; score 0-100; utility_scores object; strong_structural_roles list; weak_roles list; counterfactual_analysis; material_improvements_over_USDC_ETH list; native_asset_costs_and_risks list; argument_for_native_token; argument_against_native_token; final_rationale; native_token_removed_breaks list; native_token_removed_degradation NONE/MINOR/MODERATE/MATERIAL/CRITICAL; native_token_removed_explanation. Product/protocol/token/economic model: " + json.dumps({"pitch": proposal.pitch, "crypto": crypto, "roles": roles.roles, "utility_profile": utility_profile, "economic_model": model_payload}, default=str)) + judge_prompt = "CRYPTO_TOKEN_UTILITY_JUDGE V0.3.2B. Independent from ideation, onchain review, and token design. Challenge the token; do not merely summarize it. A native token need not be mathematically impossible to substitute, but benefits must outweigh costs. Evaluate security quality, attack economics, slashing effectiveness, provider skin-in-the-game, capital efficiency, protocol-specific risk pricing, permissionless supply formation, participant coordination, resource allocation, reputation portability, value capture, network effects, bootstrapping economics, governance, and composability versus USDC/ETH/external collateral/non-transferable attestations. Return JSON with token_classification TOKEN_ESSENTIAL/TOKEN_STRONGLY_JUSTIFIED/TOKEN_OPTIONAL/TOKEN_UNNECESSARY; score 0-100; utility_scores object; strong_structural_roles list; weak_roles list; counterfactual_analysis; external_collateral_counterfactual object/string; material_improvements_over_USDC_ETH non-empty list; native_asset_costs_and_risks non-empty list; argument_for_native_token non-empty string; argument_against_native_token non-empty string; final_rationale non-empty string; native_token_removed_breaks non-empty list; native_token_removed_degradation NONE/MINOR/MODERATE/MATERIAL/CRITICAL; native_token_removed_explanation non-empty string. If TOKEN_OPTIONAL, final_rationale must answer: Why is ETH/USDC/external collateral preferable despite the claimed material degradation? If TOKEN_ESSENTIAL or TOKEN_STRONGLY_JUSTIFIED, final_rationale must answer: What protocol-specific property cannot be reproduced economically enough with ETH/USDC/external collateral? Do not automatically promote for material degradation. Product/protocol/token/economic model: " + json.dumps({"pitch": proposal.pitch, "crypto": crypto, "roles": roles.roles, "utility_profile": utility_profile, "economic_model": model_payload}, default=str) + sol_review = self._complete_token_utility_judge(judge_prompt, roles.roles) sol_structural_roles = [str(role) for role in self._as_list((sol_review or {}).get("strong_structural_roles")) if str(role) in STRUCTURAL_TOKEN_ROLES] sol_scores = (sol_review or {}).get("utility_scores", {}) if isinstance((sol_review or {}).get("utility_scores", {}), dict) else {} if sol_structural_roles: @@ -721,7 +723,7 @@ class CryptoVentureService: score = min(score, 62) proposal.metadata = {**proposal.metadata, "crypto": {**crypto, "token_utility_judge_model": self.final_ic_model_hint, "token_utility_judge_used": bool(sol_review), "token_utility_profile": utility_profile, "strong_structural_roles": structural_roles, "strong_token_roles": strong_roles}} proposal.save(update_fields=["metadata", "updated_at"]) - 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("counterfactual_analysis", "USDC/ETH/external collateral may preserve basic operation but must be compared against protocol-specific security and coordination improvements.")), "rationale": str((sol_review or {}).get("final_rationale") or (sol_review or {}).get("rationale") or "Token score derives from structural utility, not native payment currency alone."), "metadata": {"weak_utility_terms": INSUFFICIENT_TOKEN_UTILITIES, "sol_counterfactual_review": sol_review or {}, "sol_token_utility_judge": sol_review or {}, "independent_token_utility_judge": "CRYPTO_TOKEN_UTILITY_JUDGE", "argument_for_native_token": (sol_review or {}).get("argument_for_native_token", ""), "argument_against_native_token": (sol_review or {}).get("argument_against_native_token", ""), "utility_scores": utility_profile, "strong_structural_roles": structural_roles, "strong_token_roles": strong_roles, "weak_roles": (sol_review or {}).get("weak_roles", []), "counterfactual_analysis": (sol_review or {}).get("counterfactual_analysis", ""), "external_collateral_counterfactual": (sol_review or {}).get("external_collateral_counterfactual") or (sol_review or {}).get("model_c_external_collateral", {}), "material_improvements_over_USDC_ETH": self._as_list((sol_review or {}).get("material_improvements_over_USDC_ETH")), "native_asset_costs_and_risks": self._as_list((sol_review or {}).get("native_asset_costs_and_risks")), "required_roles": roles.required_roles, "payment_only_penalty": payment_only, "native_token_removed_breaks": self._as_list((sol_review or {}).get("native_token_removed_breaks")), "native_token_removed_degradation": degradation, "native_token_removed_explanation": (sol_review or {}).get("native_token_removed_explanation", "")}})[0] + 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("counterfactual_analysis", "USDC/ETH/external collateral may preserve basic operation but must be compared against protocol-specific security and coordination improvements.")), "rationale": str((sol_review or {}).get("final_rationale") or (sol_review or {}).get("rationale") or "Token score derives from structural utility, not native payment currency alone."), "metadata": {"weak_utility_terms": INSUFFICIENT_TOKEN_UTILITIES, "sol_counterfactual_review": sol_review or {}, "sol_token_utility_judge": sol_review or {}, "independent_token_utility_judge": "CRYPTO_TOKEN_UTILITY_JUDGE", "argument_for_native_token": (sol_review or {}).get("argument_for_native_token", ""), "argument_against_native_token": (sol_review or {}).get("argument_against_native_token", ""), "utility_scores": utility_profile, "strong_structural_roles": structural_roles, "strong_token_roles": strong_roles, "weak_roles": (sol_review or {}).get("weak_roles", []), "counterfactual_analysis": (sol_review or {}).get("counterfactual_analysis", ""), "external_collateral_counterfactual": (sol_review or {}).get("external_collateral_counterfactual", {}), "material_improvements_over_USDC_ETH": self._as_list((sol_review or {}).get("material_improvements_over_USDC_ETH")), "native_asset_costs_and_risks": self._as_list((sol_review or {}).get("native_asset_costs_and_risks")), "required_roles": roles.required_roles, "payment_only_penalty": payment_only, "native_token_removed_breaks": self._as_list((sol_review or {}).get("native_token_removed_breaks")), "native_token_removed_degradation": degradation, "native_token_removed_explanation": (sol_review or {}).get("native_token_removed_explanation", "")}})[0] def token_role_decomposition(self, proposal: CompanyProposal) -> TokenRoleDecomposition: text = self._design_text(proposal) @@ -1098,6 +1100,59 @@ class CryptoVentureService: return True return False + def _complete_token_utility_judge(self, prompt: str, roles: dict[str, Any], *, attempts: int = 3) -> dict[str, Any]: + if self.router is None or self.final_ic_model_hint not in self.router.providers: + return {} + retry_hint = "" + last_missing: list[str] = [] + for attempt in range(attempts): + review = self._sol_json(prompt + retry_hint) + missing = self._token_judge_missing_fields(review, roles) + if not missing: + return review + last_missing = missing + retry_hint = "\nPrevious token utility judge response was rejected because these required fields or rationale checks were empty/incomplete: " + json.dumps(missing) + ". Return a complete JSON object only." + raise ValueError("Incomplete Sol token utility judge response: " + ", ".join(last_missing)) + + def _token_judge_missing_fields(self, review: dict[str, Any], roles: dict[str, Any]) -> list[str]: + if not review: + return ["judge_response"] + missing = [field for field in TOKEN_JUDGE_REQUIRED_FIELDS if self._empty_judge_value(review.get(field))] + classification = str(review.get("token_classification") or review.get("classification") or "") + if classification not in TokenNecessityClassification.values: + missing.append("token_classification") + degradation = str(review.get("native_token_removed_degradation") or "").upper() + if degradation not in {"NONE", "MINOR", "MODERATE", "MATERIAL", "CRITICAL"}: + missing.append("native_token_removed_degradation") + rationale = str(review.get("final_rationale") or "").lower() + counterfactual = review.get("counterfactual_analysis", {}) + if not isinstance(counterfactual, dict): + counterfactual = {} + external_identical = bool(counterfactual.get("external_collateral_identical_security")) + stable_identical = bool(counterfactual.get("stable_collateral_identical_security")) + structural_count = sum(1 for role, value in roles.items() if role in STRUCTURAL_TOKEN_ROLES and str(value).upper() in {"REQUIRED", "STRONGLY_USEFUL"}) + contradiction = not external_identical and not stable_identical and degradation == "MATERIAL" and structural_count >= 2 + if classification == TokenNecessityClassification.TOKEN_OPTIONAL: + has_preference_answer = any(asset in rationale for asset in ["eth", "usdc", "external collateral", "stablecoin"]) and any(term in rationale for term in ["prefer", "simpler", "liquid", "lower", "regulatory", "volatility", "issuance", "cost", "risk"]) + if not has_preference_answer: + missing.append("final_rationale_optional_collateral_preference") + if contradiction and not any(term in rationale for term in ["despite", "even though", "still fail", "not justify", "outweigh"]): + missing.append("final_rationale_material_degradation_contradiction") + if classification in {TokenNecessityClassification.TOKEN_ESSENTIAL, TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED}: + has_specific_property = any(asset in rationale for asset in ["eth", "usdc", "external collateral", "stablecoin"]) and any(term in rationale for term in ["cannot", "not reproduce", "cannot reproduce", "insufficient", "protocol-specific", "not economically"]) + if not has_specific_property: + missing.append("final_rationale_protocol_specific_non_reproducible_property") + return missing + + def _empty_judge_value(self, value: Any) -> bool: + if value is None: + return True + if isinstance(value, str): + return not value.strip() + if isinstance(value, (list, tuple, set, dict)): + return len(value) == 0 + 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 {} diff --git a/tests/test_crypto_venture_cohort.py b/tests/test_crypto_venture_cohort.py index 366d480..7982a71 100644 --- a/tests/test_crypto_venture_cohort.py +++ b/tests/test_crypto_venture_cohort.py @@ -2,6 +2,8 @@ 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 @@ -25,7 +27,7 @@ class SolCryptoReviewProvider(ModelProvider): 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": "USDC/ETH can operate but weakens protocol-specific risk pricing.", "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": "Benefits outweigh costs for protocol-specific collateral.", "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 is materially weaker."} + 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: @@ -214,6 +216,49 @@ def test_v03_independent_counterfactual_native_removed_fields() -> None: 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"]: + 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: