daniel/Artifex
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Require complete crypto token judge evidence

Browse source
This commit is contained in:
Daniel Maddern 2026-08-16 22:59:30 +07:00
parent 052d1fc469
commit 6e89ea8acf
2 changed files with 103 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

59
agents/crypto_venture.py
View file

@ -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 {}

47
tests/test_crypto_venture_cohort.py
View file

@ -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: