# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } # ═══════════════════════════════════════════════════════════════════════════════ # Veritas Layer — ModuleRegistry (Phase II — Production) # # Stores and manages verification module definitions. # Every module defines HOW an output is evaluated: # - evaluation prompt (with mandatory safety delimiters) # - acceptance thresholds (score, confidence) # - scoring tolerance for validator agreement # - dispute policy fields (per-module limits within protocol cap) # - content policy (require_inline_content flag) # # Phase II additions over Phase I: # + prompt_hash / criteria_hash stored at registration # + snapshot_hash = SHA-256 over all policy-relevant fields # → read by callers, re-verified inside VeritasCore.submit() # → eliminates runtime cross-contract read dependency # + Mandatory <<< VERITAS_*_START >>> / <<< VERITAS_*_END >>> delimiters # enforced at registration; delimiter positions validated relative to slots # + score_tolerance floor raised to 10 (prevents escalation-loop modules) # + max_dispute_rounds (1-3, per-module reduction of protocol cap) # + challenge_window_sec (per-module non-submitter challenge window) # + require_inline_content flag (blocks URL submissions for the module) # + update_module() — owner-only prompt/threshold correction, # bumps version and recomputes all hashes # # GenLayer compliance (verified against linter source): # E011 — one gl.Contract subclass per file PASS # E014 — @allow_storage on all storage dataclasses PASS # E015 — no bare int in storage fields PASS # E016 — no list/dict in storage fields PASS # E018 — TreeMap keys in {str,Address,u32,u256,i32,i256,bytes} PASS # W001 — no forbidden imports (random,os,sys,subprocess,etc.) PASS # W002 — no forbidden calls (time.time, uuid.uuid4, etc.) PASS # W004 — no bare Python exceptions (all use gl.vm.UserError) PASS # W010 — dependency header present on line 1 PASS # W020 — return type annotation on all @gl.public.view methods PASS # E021 — no *args/**kwargs in public methods PASS # E022 — first param is self on all methods PASS # No nondet in this contract → E010/E024/E025/E026 N/A # ═══════════════════════════════════════════════════════════════════════════════ import hashlib from dataclasses import dataclass from genlayer import * # ══════════════════════════════════════════════════════════════════════════════ # Module-level pure helpers # Placed at module scope — carry no self/storage reference. # hashlib.sha256 is deterministic and not in FORBIDDEN_MODULES or FORBIDDEN_CALLS # (confirmed from linter source: safety.py FORBIDDEN_MODULES, FORBIDDEN_CALLS). # ══════════════════════════════════════════════════════════════════════════════ def _sha256(value: str) -> str: """Return lowercase 64-char hex SHA-256 of the UTF-8 encoding of value.""" return hashlib.sha256(value.encode("utf-8")).hexdigest() def _compute_snapshot_hash( module_id: str, version: int, eval_prompt: str, criteria: str, score_tolerance: int, accept_threshold: int, min_confidence: int, max_dispute_rounds: int, challenge_window_sec: int, require_inline_content: bool, evaluation_method: str, ) -> str: """ Canonical snapshot hash covering ALL evaluation-affecting policy fields. Uses pipe-delimited concatenation to prevent prefix-collision attacks. Phase II hardening adds require_inline_content and evaluation_method: - require_inline_content is a security policy field: controls whether URL submissions are accepted. Excluding it allowed callers to bypass the inline-content requirement by submitting False regardless of the module's registered value. Now any field alteration → hash mismatch. - evaluation_method is a semantic field: it defines how evaluation is interpreted. Including it ensures semantic identity is hash-protected. Field order is CANONICAL and must match VeritasCore._compute_snapshot_hash exactly. Any divergence silently breaks all new submissions. This hash is stored in VerificationModule.snapshot_hash and returned by get_module(). VeritasCore.submit() callers pass it as an argument, and submit() recomputes it locally. Zero runtime cross-contract call. """ canonical = ( module_id + "|" + str(version) + "|" + eval_prompt + "|" + criteria + "|" + str(score_tolerance) + "|" + str(accept_threshold) + "|" + str(min_confidence) + "|" + str(max_dispute_rounds) + "|" + str(challenge_window_sec) + "|" + str(require_inline_content) + "|" + evaluation_method ) return _sha256(canonical) # ══════════════════════════════════════════════════════════════════════════════ # Protocol-level constants # These are enforced at registration and cannot be overridden by module owners. # ══════════════════════════════════════════════════════════════════════════════ MAX_MODULE_ID_LEN: int = 64 MAX_PROMPT_LEN: int = 4000 MAX_CRITERIA_LEN: int = 1000 MAX_DESCRIPTION_LEN: int = 256 # ── Prompt injection defence delimiters ─────────────────────────────────────── # All four must be present in every eval_prompt. # {output} must appear between OUTPUT_OPEN and OUTPUT_CLOSE. # {context} must appear between CONTEXT_OPEN and CONTEXT_CLOSE. # Validated by _validate_prompt_structure() at register and update time. DELIM_OUTPUT_OPEN: str = "<<>>" DELIM_OUTPUT_CLOSE: str = "<<>>" DELIM_CONTEXT_OPEN: str = "<<>>" DELIM_CONTEXT_CLOSE: str = "<<>>" # ── Threshold range enforcement ─────────────────────────────────────────────── # SCORE_TOLERANCE_MIN = 10: below 10 causes routine LLM score divergence # to trigger GenLayer's native appeal escalation on nearly every evaluation. # Two independent LLM runs on identical content routinely diverge by 8-12 # points. A tolerance below 10 is operationally unviable. SCORE_TOLERANCE_MIN: int = 10 SCORE_TOLERANCE_MAX: int = 30 ACCEPT_THRESHOLD_MIN: int = 0 ACCEPT_THRESHOLD_MAX: int = 100 MIN_CONFIDENCE_MIN: int = 0 MIN_CONFIDENCE_MAX: int = 100 # ── Dispute policy range enforcement ───────────────────────────────────────── # PROTOCOL_MAX_DISPUTE_ROUNDS is the hard ceiling; modules may only reduce it. # Modules setting max_dispute_rounds < PROTOCOL_MAX_DISPUTE_ROUNDS signal # that they are designed for rapid, non-contentious verification. PROTOCOL_MAX_DISPUTE_ROUNDS: int = 3 DISPUTE_ROUNDS_MIN: int = 1 CHALLENGE_WINDOW_MIN_SEC: int = 3600 # 1 hour CHALLENGE_WINDOW_MAX_SEC: int = 604800 # 7 days # ── Controlled vocabularies ──────────────────────────────────────────────────── VALID_MODULE_TYPES: tuple = ( "FACTUALITY_CHECK", "SUMMARY_VERIFICATION", "CODE_CORRECTNESS", "HALLUCINATION_DETECTION", "EQUIVALENCE_CHECK", "CUSTOM", ) VALID_EVAL_METHODS: tuple = ( "LLM_CONSENSUS", "LLM_NON_COMPARATIVE", ) VALID_SCORING_SCALES: tuple = ("0-100",) # ══════════════════════════════════════════════════════════════════════════════ # Storage dataclass # @allow_storage @dataclass — required by E014. # All fields use SDK types (str, bool, u8, u32, Address) — required by E015. # ══════════════════════════════════════════════════════════════════════════════ @allow_storage @dataclass class VerificationModule: # ── Identity ────────────────────────────────────────────────────────────── module_id: str owner: Address # ── Structured metadata (immutable after registration) ──────────────────── module_type: str # one of VALID_MODULE_TYPES scoring_scale: str # one of VALID_SCORING_SCALES evaluation_method: str # one of VALID_EVAL_METHODS description: str # human-readable summary ≤ 256 chars # ── Evaluation config (mutable via update_module; snapshotted into Tasks) ─ eval_prompt: str # template with <<>> boundaries criteria: str # natural-language equivalence/acceptance criteria # ── Integrity hashes ────────────────────────────────────────────────────── # Computed at registration and recomputed on every update_module() call. # prompt_hash and criteria_hash allow independent verification of each field. # snapshot_hash is the canonical authenticity anchor — the single value that # VeritasCore.submit() callers pass in and submit() re-verifies locally. prompt_hash: str # SHA-256(eval_prompt) criteria_hash: str # SHA-256(criteria) snapshot_hash: str # SHA-256 over all policy-relevant fields # ── Scoring thresholds ──────────────────────────────────────────────────── score_tolerance: u8 # validator agreement band [10, 30] accept_threshold: u8 # min score for VALID classification [0, 100] min_confidence: u8 # min confidence to avoid UNCERTAIN [0, 100] # ── Dispute policy (per-module; DisputeLedger reads from task snapshot) ─── max_dispute_rounds: u8 # [1, PROTOCOL_MAX_DISPUTE_ROUNDS=3] challenge_window_sec: u32 # non-submitter window [3600, 604800] seconds # ── Content policy ──────────────────────────────────────────────────────── require_inline_content: bool # True → submit() rejects http/https output_ref # ── Lifecycle ───────────────────────────────────────────────────────────── is_active: bool version: u32 # 1-based; incremented by update_module() # ══════════════════════════════════════════════════════════════════════════════ # Contract # ══════════════════════════════════════════════════════════════════════════════ class ModuleRegistry(gl.Contract): # Primary registry: module_id → VerificationModule modules: TreeMap[str, VerificationModule] module_count: u32 # Secondary index: owner Address → (sequential u32 index → module_id) # Enables enumeration of all modules owned by an address. # Inner TreeMap key type is u32 — valid per linter E018 rule: # valid_key_types = {"str","Address","u32","u256","i32","i256","bytes"} owner_index: TreeMap[Address, TreeMap[u32, str]] owner_counts: TreeMap[Address, u32] def __init__(self) -> None: self.module_count = u32(0) # ── Internal guards ─────────────────────────────────────────────────────── def _require(self, condition: bool, msg: str) -> None: """ Revert with msg if condition is False. Uses gl.vm.UserError — never bare Python exceptions (W004). """ if not condition: raise gl.vm.UserError(msg) def _only_owner(self, module_id: str) -> None: """Caller must be the registered module owner.""" self._require(module_id in self.modules, "ERR:MODULE_NOT_FOUND") self._require( self.modules[module_id].owner == gl.message.sender_address, "ERR:NOT_MODULE_OWNER", ) # ── Internal: validation helpers ────────────────────────────────────────── def _validate_prompt_structure(self, eval_prompt: str) -> None: """ Enforce prompt injection defence at registration and update time. Checks (in order): 1. All four delimiter strings are present in eval_prompt. 2. {output} and {context} slot strings are present. 3. {output} slot index is strictly between OUTPUT_OPEN index and OUTPUT_CLOSE index. 4. {context} slot index is strictly between CONTEXT_OPEN index and CONTEXT_CLOSE index. Uses str.find() (returns -1 on miss); presence already confirmed by step 1, so find() will never return -1 in steps 3-4. Pure string operations — deterministic, no nondet. All errors use gl.vm.UserError (W004 compliance). """ # Step 1 — delimiter presence self._require(DELIM_OUTPUT_OPEN in eval_prompt, "ERR:MISSING_OUTPUT_OPEN_DELIMITER") self._require(DELIM_OUTPUT_CLOSE in eval_prompt, "ERR:MISSING_OUTPUT_CLOSE_DELIMITER") self._require(DELIM_CONTEXT_OPEN in eval_prompt, "ERR:MISSING_CONTEXT_OPEN_DELIMITER") self._require(DELIM_CONTEXT_CLOSE in eval_prompt, "ERR:MISSING_CONTEXT_CLOSE_DELIMITER") # Step 2 — slot presence self._require("{output}" in eval_prompt, "ERR:MISSING_OUTPUT_SLOT") self._require("{context}" in eval_prompt, "ERR:MISSING_CONTEXT_SLOT") # Step 3 — output slot ordering out_open = eval_prompt.find(DELIM_OUTPUT_OPEN) out_slot = eval_prompt.find("{output}") out_close = eval_prompt.find(DELIM_OUTPUT_CLOSE) self._require(out_open < out_slot, "ERR:OUTPUT_SLOT_BEFORE_OPEN_DELIMITER") self._require(out_slot < out_close, "ERR:OUTPUT_SLOT_AFTER_CLOSE_DELIMITER") # Step 4 — context slot ordering ctx_open = eval_prompt.find(DELIM_CONTEXT_OPEN) ctx_slot = eval_prompt.find("{context}") ctx_close = eval_prompt.find(DELIM_CONTEXT_CLOSE) self._require(ctx_open < ctx_slot, "ERR:CONTEXT_SLOT_BEFORE_OPEN_DELIMITER") self._require(ctx_slot < ctx_close, "ERR:CONTEXT_SLOT_AFTER_CLOSE_DELIMITER") def _validate_thresholds( self, score_tolerance: int, accept_threshold: int, min_confidence: int, max_dispute_rounds: int, challenge_window_sec: int, ) -> None: """Range-check all numeric policy fields against protocol constants.""" self._require( SCORE_TOLERANCE_MIN <= score_tolerance <= SCORE_TOLERANCE_MAX, "ERR:SCORE_TOLERANCE_OUT_OF_RANGE", ) self._require( ACCEPT_THRESHOLD_MIN <= accept_threshold <= ACCEPT_THRESHOLD_MAX, "ERR:ACCEPT_THRESHOLD_OUT_OF_RANGE", ) self._require( MIN_CONFIDENCE_MIN <= min_confidence <= MIN_CONFIDENCE_MAX, "ERR:MIN_CONFIDENCE_OUT_OF_RANGE", ) self._require( DISPUTE_ROUNDS_MIN <= max_dispute_rounds <= PROTOCOL_MAX_DISPUTE_ROUNDS, "ERR:MAX_DISPUTE_ROUNDS_OUT_OF_RANGE", ) self._require( CHALLENGE_WINDOW_MIN_SEC <= challenge_window_sec <= CHALLENGE_WINDOW_MAX_SEC, "ERR:CHALLENGE_WINDOW_OUT_OF_RANGE", ) # ── Write: register_module ──────────────────────────────────────────────── @gl.public.write def register_module( self, module_id: str, module_type: str, scoring_scale: str, evaluation_method: str, description: str, eval_prompt: str, criteria: str, score_tolerance: int, accept_threshold: int, min_confidence: int, max_dispute_rounds: int, challenge_window_sec: int, require_inline_content: bool, ) -> str: """ Register a new verification module. Returns module_id on success. Caller requirements: - eval_prompt must contain all four <<< VERITAS_* >>> delimiters with {output} and {context} slots correctly positioned within them - score_tolerance must be >= 10 (prevents escalation-loop modules) - max_dispute_rounds must be in [1, 3] - challenge_window_sec must be in [3600, 604800] Hashes computed here: - prompt_hash = SHA-256(eval_prompt) - criteria_hash = SHA-256(criteria) - snapshot_hash = SHA-256 over all policy fields (canonical anchor) Callers of VeritasCore.submit() should: 1. Call get_module(module_id) to obtain all fields including snapshot_hash 2. Pass all fields + snapshot_hash to VeritasCore.submit() VeritasCore.submit() recomputes snapshot_hash locally and verifies. """ # Identity self._require( 0 < len(module_id) <= MAX_MODULE_ID_LEN, "ERR:INVALID_MODULE_ID_LENGTH", ) self._require(module_id not in self.modules, "ERR:MODULE_ID_TAKEN") # Vocabularies self._require(module_type in VALID_MODULE_TYPES, "ERR:INVALID_MODULE_TYPE") self._require(scoring_scale in VALID_SCORING_SCALES,"ERR:INVALID_SCORING_SCALE") self._require(evaluation_method in VALID_EVAL_METHODS, "ERR:INVALID_EVAL_METHOD") # Lengths self._require( 0 < len(description) <= MAX_DESCRIPTION_LEN, "ERR:DESCRIPTION_INVALID_LENGTH", ) self._require( 0 < len(eval_prompt) <= MAX_PROMPT_LEN, "ERR:EVAL_PROMPT_INVALID_LENGTH", ) self._require( 0 < len(criteria) <= MAX_CRITERIA_LEN, "ERR:CRITERIA_INVALID_LENGTH", ) # Prompt structure and injection defence self._validate_prompt_structure(eval_prompt) # Numeric ranges self._validate_thresholds( score_tolerance, accept_threshold, min_confidence, max_dispute_rounds, challenge_window_sec, ) # Compute integrity hashes p_hash = _sha256(eval_prompt) c_hash = _sha256(criteria) s_hash = _compute_snapshot_hash( module_id, 1, eval_prompt, criteria, score_tolerance, accept_threshold, min_confidence, max_dispute_rounds, challenge_window_sec, require_inline_content, evaluation_method, ) # Store module owner = gl.message.sender_address mod = VerificationModule( module_id = module_id, owner = owner, module_type = module_type, scoring_scale = scoring_scale, evaluation_method = evaluation_method, description = description, eval_prompt = eval_prompt, criteria = criteria, prompt_hash = p_hash, criteria_hash = c_hash, snapshot_hash = s_hash, score_tolerance = u8(score_tolerance), accept_threshold = u8(accept_threshold), min_confidence = u8(min_confidence), max_dispute_rounds = u8(max_dispute_rounds), challenge_window_sec = u32(challenge_window_sec), require_inline_content = require_inline_content, is_active = True, version = u32(1), ) self.modules[module_id] = mod self.module_count = u32(int(self.module_count) + 1) # Update owner index # get_or_insert_default() is the correct pattern for nested TreeMap # initialisation. Direct assignment (= TreeMap()) triggers storage # descriptor errors at runtime — confirmed from sample contract comments. owner_idx = self.owner_counts.get(owner, u32(0)) self.owner_index.get_or_insert_default(owner)[owner_idx] = module_id self.owner_counts[owner] = u32(int(owner_idx) + 1) gl.trace( "MODULE_REGISTERED" + "|" + module_id + "|v1" + "|hash:" + s_hash[:16] ) return module_id # ── Write: update_module ────────────────────────────────────────────────── @gl.public.write def update_module( self, module_id: str, eval_prompt: str, criteria: str, score_tolerance: int, accept_threshold: int, min_confidence: int, max_dispute_rounds: int, challenge_window_sec: int, require_inline_content: bool, ) -> None: """ Update mutable policy fields of an existing module. Owner-only. Mutable fields (can change): eval_prompt, criteria, all threshold fields, dispute policy fields, require_inline_content. Immutable fields (cannot change): module_id, module_type, scoring_scale, evaluation_method, owner, description. Rationale: these define the module's structural identity. Changing them would make the human-readable module_id misleading and could invalidate the interpretation of in-flight task snapshots. Effect on existing tasks: Tasks already submitted hold a frozen ModuleSnapshot with the OLD snapshot_hash. Those snapshots are unaffected by this call. New submissions will use the updated eval_prompt, new hashes, and new snapshot_hash from get_module(). Version is incremented on every successful call. Hashes are recomputed from the new values. """ self._only_owner(module_id) mod = self.modules[module_id] self._require(mod.is_active, "ERR:MODULE_INACTIVE") # One-way require_inline_content enforcement (Phase II hardening). # Once a module establishes inline-content trust (True), it cannot # be downgraded to False. This protects consumers who rely on the # guarantee that URL-based content is rejected for this module. # False → True is always permitted (increasing security is safe). if mod.require_inline_content and not require_inline_content: raise gl.vm.UserError("ERR:CANNOT_DOWNGRADE_INLINE_CONTENT_POLICY") # Length validation self._require( 0 < len(eval_prompt) <= MAX_PROMPT_LEN, "ERR:EVAL_PROMPT_INVALID_LENGTH", ) self._require( 0 < len(criteria) <= MAX_CRITERIA_LEN, "ERR:CRITERIA_INVALID_LENGTH", ) # Prompt structure and injection defence self._validate_prompt_structure(eval_prompt) # Numeric ranges self._validate_thresholds( score_tolerance, accept_threshold, min_confidence, max_dispute_rounds, challenge_window_sec, ) # Capture current hash before overwriting (for trace log) old_snapshot_hash = mod.snapshot_hash # Compute new hashes for the incremented version new_version = int(mod.version) + 1 p_hash = _sha256(eval_prompt) c_hash = _sha256(criteria) s_hash = _compute_snapshot_hash( module_id, new_version, eval_prompt, criteria, score_tolerance, accept_threshold, min_confidence, max_dispute_rounds, challenge_window_sec, require_inline_content, mod.evaluation_method, # structural field — immutable, read from storage ) # Apply mutations and persist mod.eval_prompt = eval_prompt mod.criteria = criteria mod.prompt_hash = p_hash mod.criteria_hash = c_hash mod.snapshot_hash = s_hash mod.score_tolerance = u8(score_tolerance) mod.accept_threshold = u8(accept_threshold) mod.min_confidence = u8(min_confidence) mod.max_dispute_rounds = u8(max_dispute_rounds) mod.challenge_window_sec = u32(challenge_window_sec) mod.require_inline_content = require_inline_content mod.version = u32(new_version) self.modules[module_id] = mod # Trace log for off-chain indexers and consumer notification. # Includes module_id, version transition, hash transition. gl.trace( "MODULE_UPDATED" + "|" + module_id + "|v" + str(new_version - 1) + "->v" + str(new_version) + "|old_hash:" + old_snapshot_hash[:16] + "|new_hash:" + s_hash[:16] ) # ── Write: deactivate_module ────────────────────────────────────────────── @gl.public.write def deactivate_module(self, module_id: str) -> None: """ Permanently deactivate a module. Owner-only. Deactivated modules should not be used for new submissions. VeritasCore.submit() callers must check is_active before calling. In-flight tasks holding an existing snapshot are unaffected — their snapshot_hash references an older version that remains valid for their own lifecycle. """ self._only_owner(module_id) mod = self.modules[module_id] mod.is_active = False self.modules[module_id] = mod gl.trace( "MODULE_DEACTIVATED" + "|" + module_id + "|v" + str(int(mod.version)) ) # ── View methods ────────────────────────────────────────────────────────── @gl.public.view def get_module(self, module_id: str) -> dict: """ Return full module record. Primary use: VeritasCore.submit() callers read this off-chain to obtain all snapshot fields (including snapshot_hash) before calling submit(). VeritasCore recomputes snapshot_hash locally — no runtime cross-contract call from VeritasCore to ModuleRegistry is needed. """ self._require(module_id in self.modules, "ERR:MODULE_NOT_FOUND") m = self.modules[module_id] return { "module_id": m.module_id, "owner": m.owner.as_hex, "module_type": m.module_type, "scoring_scale": m.scoring_scale, "evaluation_method": m.evaluation_method, "description": m.description, "eval_prompt": m.eval_prompt, "criteria": m.criteria, "prompt_hash": m.prompt_hash, "criteria_hash": m.criteria_hash, "snapshot_hash": m.snapshot_hash, "score_tolerance": int(m.score_tolerance), "accept_threshold": int(m.accept_threshold), "min_confidence": int(m.min_confidence), "max_dispute_rounds": int(m.max_dispute_rounds), "challenge_window_sec": int(m.challenge_window_sec), "require_inline_content": m.require_inline_content, "is_active": m.is_active, "version": int(m.version), } @gl.public.view def get_snapshot_hash(self, module_id: str) -> str: """ Return only snapshot_hash for a given module. Lightweight alternative when callers need only the hash. """ self._require(module_id in self.modules, "ERR:MODULE_NOT_FOUND") return self.modules[module_id].snapshot_hash @gl.public.view def module_is_active(self, module_id: str) -> bool: """Returns False for unknown module IDs.""" if module_id not in self.modules: return False return self.modules[module_id].is_active @gl.public.view def get_module_count(self) -> int: """Total registered modules (active + inactive).""" return int(self.module_count) @gl.public.view def get_owner_module_ids(self, owner_hex: str, offset: int, limit: int) -> list: """ Return paginated list of module IDs registered by owner_hex. offset: 0-based start index. limit: max items to return; capped at 50 to prevent large payloads. Returns empty list for unknown owners or out-of-range offsets. """ owner = Address(owner_hex) total = int(self.owner_counts.get(owner, u32(0))) if offset >= total or limit <= 0: return [] capped_limit = min(limit, 50) end = min(offset + capped_limit, total) result = [] for i in range(offset, end): result.append(self.owner_index[owner][u32(i)]) return result