# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from dataclasses import dataclass import hashlib import json from genlayer import * # --------------------------------------------------------------------------- # Error Classification Constants # --------------------------------------------------------------------------- ERROR_EXPECTED = "[EXPECTED]" # Deterministic business logic / guard failures ERROR_INTEGRITY = "[INTEGRITY]" # Evidence or Rubric hash verification mismatch ERROR_LLM = "[LLM_ERROR]" # Malformed, non-compliant, or invalid LLM response # --------------------------------------------------------------------------- # State Constants # --------------------------------------------------------------------------- STATUS_UNINITIALIZED = "UNINITIALIZED" STATUS_REGISTERED = "REGISTERED" STATUS_EVALUATING = "EVALUATING" STATUS_FINALIZED = "FINALIZED" STATUS_FAILED = "FAILED" # Maximum allowable evidence and rubric length (64 KB) to guarantee bounded storage MAX_ARTIFACT_BYTES = 65536 MAX_DECISION_BYTES = 128 MAX_RUBRIC_VERSION_BYTES = 32 # --------------------------------------------------------------------------- # Storage Dataclasses # --------------------------------------------------------------------------- @allow_storage @dataclass class CaseRecord: case_id: str subject: str evidence: str evidence_hash: str rubric: str rubric_hash: str status: str canonical_decision: str canonical_score: u32 canonical_confidence_bps: u32 # Scaled integer (basis points: 10000 = 1.0000) canonical_rationale: str decision_hash: str submitter: Address # Appended for storage-layout compatibility with any pre-existing records. canonical_score_bucket: str # --------------------------------------------------------------------------- # Helper Parsing & Hashing Functions # --------------------------------------------------------------------------- def _compute_sha256(content: str) -> str: """Computes deterministic hex SHA-256 digest of input text.""" return hashlib.sha256(content.encode("utf-8")).hexdigest() def _compute_decision_hash( case_id: str, evidence_hash: str, rubric_hash: str, decision: str, score_bucket: str, score: int, confidence_bps: int, ) -> str: """Generates canonical deterministic hash for finalized decision record.""" canonical_payload = { "case_id": case_id, "confidence_bps": confidence_bps, "decision": decision, "evidence_hash": evidence_hash, "rubric_hash": rubric_hash, "canonical_score": score, "canonical_score_bucket": score_bucket, } canonical_str = json.dumps(canonical_payload, sort_keys=True, separators=(",", ":")) return hashlib.sha256(canonical_str.encode("utf-8")).hexdigest() def _parse_rubric(rubric: str) -> dict: """Parse the authoritative versioned JSON rubric and validate its buckets.""" try: parsed = json.loads(rubric) except Exception as err: raise gl.vm.UserError(f"{ERROR_EXPECTED} rubric must be valid JSON: {str(err)}") if not isinstance(parsed, dict): raise gl.vm.UserError(f"{ERROR_EXPECTED} rubric must be a JSON object") version = parsed.get("version") if not isinstance(version, str) or not version.strip(): raise gl.vm.UserError(f"{ERROR_EXPECTED} rubric.version must be a non-empty string") if len(version.encode("utf-8")) > MAX_RUBRIC_VERSION_BYTES: raise gl.vm.UserError(f"{ERROR_EXPECTED} rubric.version is too large") buckets = parsed.get("score_buckets") if not isinstance(buckets, list) or not buckets: raise gl.vm.UserError(f"{ERROR_EXPECTED} rubric.score_buckets must be a non-empty array") normalized = [] seen_buckets = set() previous_max = -1 for bucket in buckets: if not isinstance(bucket, dict): raise gl.vm.UserError(f"{ERROR_EXPECTED} rubric bucket must be an object") name = bucket.get("bucket") decision = bucket.get("decision") minimum = bucket.get("min_score") maximum = bucket.get("max_score") if not isinstance(name, str) or not name.strip(): raise gl.vm.UserError(f"{ERROR_EXPECTED} rubric bucket requires bucket") normalized_name = name.strip().upper() if normalized_name in seen_buckets: raise gl.vm.UserError( f"{ERROR_EXPECTED} rubric bucket names must be unique: {normalized_name}" ) seen_buckets.add(normalized_name) if not isinstance(decision, str) or not decision.strip(): raise gl.vm.UserError(f"{ERROR_EXPECTED} rubric bucket requires decision") try: minimum = int(minimum) maximum = int(maximum) except (TypeError, ValueError): raise gl.vm.UserError(f"{ERROR_EXPECTED} rubric bucket score bounds must be integers") if minimum <= previous_max: raise gl.vm.UserError(f"{ERROR_EXPECTED} rubric score ranges must not overlap") if minimum < 0 or maximum > 100 or minimum > maximum or minimum != previous_max + 1: raise gl.vm.UserError(f"{ERROR_EXPECTED} rubric score buckets must cover contiguous [0, 100]") previous_max = maximum normalized.append({ "bucket": normalized_name, "decision": decision.strip().upper(), "min_score": minimum, "max_score": maximum, }) if previous_max != 100: raise gl.vm.UserError(f"{ERROR_EXPECTED} rubric score buckets must cover contiguous [0, 100]") return {"version": version.strip(), "score_buckets": normalized} def _bucket_for_score(rubric: dict, score: int) -> dict: for bucket in rubric["score_buckets"]: if bucket["min_score"] <= score <= bucket["max_score"]: return bucket raise gl.vm.UserError(f"{ERROR_LLM} score has no matching rubric bucket: {score}") def _parse_and_validate_llm_result(raw_output: str | dict) -> dict: """ Defensively parses and validates structured validator decision. Extracts and coerces: - decision: str (normalized uppercase) - score: int (0 to 100) - confidence: float (0.0 to 1.0) -> confidence_bps: int (0 to 10000) - rationale: str """ parsed: dict if isinstance(raw_output, dict): parsed = raw_output elif isinstance(raw_output, str): text = raw_output.strip() if text.startswith("```"): lines = text.splitlines() if lines and lines[0].startswith("```"): lines = lines[1:] if lines and lines[-1].startswith("```"): lines = lines[:-1] text = "\n".join(lines).strip() start = text.find("{") end = text.rfind("}") if start == -1 or end == -1 or end <= start: raise gl.vm.UserError(f"{ERROR_LLM} No valid JSON object boundaries found in output") try: parsed = json.loads(text[start : end + 1]) except Exception as err: raise gl.vm.UserError(f"{ERROR_LLM} JSON decode error: {str(err)}") else: raise gl.vm.UserError(f"{ERROR_LLM} Unexpected output type: {type(raw_output)}") if not isinstance(parsed, dict): raise gl.vm.UserError(f"{ERROR_LLM} Parsed LLM output is not a JSON object") # Required structured fields. Aliases are intentionally not accepted: # callers must supply the documented schema rather than an ambiguous shape. raw_decision = parsed.get("decision") if not isinstance(raw_decision, str) or not raw_decision.strip(): raise gl.vm.UserError(f"{ERROR_LLM} Missing or invalid 'decision' field") decision = raw_decision.strip().upper() if len(decision.encode("utf-8")) > MAX_DECISION_BYTES: raise gl.vm.UserError(f"{ERROR_LLM} 'decision' exceeds maximum size of {MAX_DECISION_BYTES} bytes") for character in decision: if ord(character) < 32 or ord(character) == 127: raise gl.vm.UserError(f"{ERROR_LLM} 'decision' contains unsupported control characters") # Extract 'score' raw_score = parsed.get("score") if raw_score is None: raise gl.vm.UserError(f"{ERROR_LLM} Missing 'score' field") try: score_float = float(str(raw_score).strip()) except (ValueError, TypeError): raise gl.vm.UserError(f"{ERROR_LLM} Non-numeric 'score': {raw_score}") if score_float != score_float or score_float == float("inf") or score_float == float("-inf"): raise gl.vm.UserError(f"{ERROR_LLM} Non-finite 'score': {raw_score}") score_val = int(round(score_float)) if score_val < 0 or score_val > 100: raise gl.vm.UserError(f"{ERROR_LLM} 'score' out of range [0, 100]: {score_val}") # Confidence is required. It is a validity/safety floor, not an # equivalence field. Decimal values [0, 1] are canonical; percentage # values (1, 100] remain accepted for defensive compatibility. if "confidence" not in parsed: raise gl.vm.UserError(f"{ERROR_LLM} Missing 'confidence' field") raw_conf = parsed.get("confidence") try: conf_float = float(str(raw_conf).strip()) except (ValueError, TypeError): raise gl.vm.UserError(f"{ERROR_LLM} Non-numeric 'confidence': {raw_conf}") if conf_float != conf_float or conf_float == float("inf") or conf_float == float("-inf"): raise gl.vm.UserError(f"{ERROR_LLM} Non-finite 'confidence': {raw_conf}") if conf_float < 0.0 or conf_float > 100.0: raise gl.vm.UserError(f"{ERROR_LLM} 'confidence' out of range [0, 1] or [0, 100]: {conf_float}") if conf_float > 1.0: conf_float = conf_float / 100.0 conf_bps = int(round(conf_float * 10000)) # Extract 'rationale' raw_rationale = parsed.get("rationale", "") if not isinstance(raw_rationale, str): raw_rationale = str(raw_rationale) rationale = raw_rationale.strip() return { "decision": decision, "score": score_val, "confidence_bps": conf_bps, "rationale": rationale, } # --------------------------------------------------------------------------- # SignalCore Intelligent Contract # --------------------------------------------------------------------------- class SignalCore(gl.Contract): """ SignalCore: Domain-neutral GenLayer Intelligent Contract primitive. Converts verified evidence + evaluation rubric into a consensus-backed on-chain signal. """ cases: TreeMap[str, CaseRecord] case_ids: DynArray[str] case_count: u32 def __init__(self): self.case_count = u32(0) # ----------------------------------------------------------------------- # Public View Methods # ----------------------------------------------------------------------- @gl.public.view def get_case(self, case_id: str) -> dict: """Retrieves full case state.""" if case_id not in self.cases: raise gl.vm.UserError(f"{ERROR_EXPECTED} Case not found: {case_id}") c = self.cases[case_id] return { "case_id": c.case_id, "subject": c.subject, "evidence": c.evidence, "evidence_hash": c.evidence_hash, "rubric": c.rubric, "rubric_hash": c.rubric_hash, "status": c.status, "canonical_decision": c.canonical_decision, "canonical_score_bucket": c.canonical_score_bucket, "canonical_score": int(c.canonical_score), "canonical_confidence_bps": int(c.canonical_confidence_bps), "canonical_rationale": c.canonical_rationale, "decision_hash": c.decision_hash, "submitter": str(c.submitter), } @gl.public.view def get_signal(self, case_id: str) -> dict: """Retrieves finalized canonical signal. Reverts if not finalized.""" if case_id not in self.cases: raise gl.vm.UserError(f"{ERROR_EXPECTED} Case not found: {case_id}") c = self.cases[case_id] if c.status != STATUS_FINALIZED: raise gl.vm.UserError(f"{ERROR_EXPECTED} Case {case_id} is not FINALIZED (status: {c.status})") return { "case_id": c.case_id, "decision": c.canonical_decision, "canonical_score_bucket": c.canonical_score_bucket, "score": int(c.canonical_score), "confidence_bps": int(c.canonical_confidence_bps), "rationale": c.canonical_rationale, "decision_hash": c.decision_hash, "evidence_hash": c.evidence_hash, "rubric_hash": c.rubric_hash, } @gl.public.view def get_case_count(self) -> int: """Returns total number of registered cases.""" return int(self.case_count) @gl.public.view def get_case_id(self, index: u32) -> str: """Returns a case ID from the append-only registration index.""" index_int = int(index) if index_int < 0 or index_int >= int(self.case_count): raise gl.vm.UserError(f"{ERROR_EXPECTED} Case index out of range: {index_int}") return self.case_ids[index_int] # ----------------------------------------------------------------------- # Public Write Methods # ----------------------------------------------------------------------- @gl.public.write def submit_case( self, case_id: str, subject: str, evidence: str, rubric: str, ) -> str: """ Submits and registers a new case with raw evidence and evaluation rubric. Computes cryptographic SHA-256 integrity commitments immediately. """ clean_id = case_id.strip() if not clean_id: raise gl.vm.UserError(f"{ERROR_EXPECTED} case_id cannot be empty") if clean_id in self.cases: raise gl.vm.UserError(f"{ERROR_EXPECTED} Case already exists: {clean_id}") if not subject.strip(): raise gl.vm.UserError(f"{ERROR_EXPECTED} subject cannot be empty") if not evidence.strip(): raise gl.vm.UserError(f"{ERROR_EXPECTED} evidence cannot be empty") if not rubric.strip(): raise gl.vm.UserError(f"{ERROR_EXPECTED} rubric cannot be empty") if len(rubric.encode("utf-8")) > MAX_ARTIFACT_BYTES: raise gl.vm.UserError(f"{ERROR_EXPECTED} rubric exceeds maximum size limit of {MAX_ARTIFACT_BYTES} bytes") _parse_rubric(rubric) if len(evidence.encode("utf-8")) > MAX_ARTIFACT_BYTES: raise gl.vm.UserError(f"{ERROR_EXPECTED} evidence exceeds maximum size limit of {MAX_ARTIFACT_BYTES} bytes") if len(rubric.encode("utf-8")) > MAX_ARTIFACT_BYTES: raise gl.vm.UserError(f"{ERROR_EXPECTED} rubric exceeds maximum size limit of {MAX_ARTIFACT_BYTES} bytes") ev_hash = _compute_sha256(evidence) rb_hash = _compute_sha256(rubric) new_record = CaseRecord( case_id=clean_id, subject=subject.strip(), evidence=evidence, evidence_hash=ev_hash, rubric=rubric, rubric_hash=rb_hash, status=STATUS_REGISTERED, canonical_decision="", canonical_score=u32(0), canonical_confidence_bps=u32(0), canonical_rationale="", decision_hash="", submitter=gl.message.sender_address, canonical_score_bucket="", ) self.cases[clean_id] = new_record self.case_ids.append(clean_id) self.case_count = u32(int(self.case_count) + 1) return clean_id @gl.public.write def evaluate_case(self, case_id: str) -> dict: """ Executes independent GenLayer validator consensus evaluation over the case. Strictly consensus-only finalization: updates state to FINALIZED on consensus pass, or FAILED on consensus failure/disagreement. Replay protected against double-finalization. """ if case_id not in self.cases: raise gl.vm.UserError(f"{ERROR_EXPECTED} Case not found: {case_id}") case_rec = self.cases[case_id] # Replay / Terminal State Protection if case_rec.status == STATUS_FINALIZED or case_rec.status == STATUS_FAILED: raise gl.vm.UserError( f"{ERROR_EXPECTED} Replay protection: Case {case_id} is in terminal state '{case_rec.status}'" ) if case_rec.status != STATUS_REGISTERED: raise gl.vm.UserError( f"{ERROR_EXPECTED} Invalid state transition: cannot evaluate case in state '{case_rec.status}'" ) # Snapshot immutable inputs for consensus evaluation target_subject = case_rec.subject target_evidence = case_rec.evidence expected_ev_hash = case_rec.evidence_hash target_rubric = case_rec.rubric expected_rb_hash = case_rec.rubric_hash rubric_definition = _parse_rubric(target_rubric) # Update status to EVALUATING case_rec.status = STATUS_EVALUATING self.cases[case_id] = case_rec # ------------------------------------------------------------------- # GenLayer Non-Deterministic Consensus Function Definition # ------------------------------------------------------------------- def leader_fn() -> dict: """ Leader execution: verifies artifact integrity, executes LLM prompt with complete evidence and rubric, parses structured output. """ # 1. Cryptographic Hash Verification computed_ev_hash = _compute_sha256(target_evidence) if computed_ev_hash != expected_ev_hash: raise gl.vm.UserError( f"{ERROR_INTEGRITY} Evidence SHA-256 mismatch (got {computed_ev_hash}, expected {expected_ev_hash})" ) computed_rb_hash = _compute_sha256(target_rubric) if computed_rb_hash != expected_rb_hash: raise gl.vm.UserError( f"{ERROR_INTEGRITY} Rubric SHA-256 mismatch (got {computed_rb_hash}, expected {expected_rb_hash})" ) # 2. Construct Evaluation Prompt with Complete Evidence & Rubric evaluation_prompt = ( "You are an objective consensus validator for SignalCore.\n" "Evaluate the submitted evidence strictly against the provided rubric.\n\n" f"### SUBJECT:\n{target_subject}\n\n" f"### COMPLETE VERIFIED EVIDENCE:\n{target_evidence}\n\n" f"### EVALUATION RUBRIC:\n{target_rubric}\n\n" "### OUTPUT FORMAT INSTRUCTIONS:\n" "Respond with a single raw JSON object matching this schema exactly:\n" "{\n" ' "decision": "STRING", // Categorical decision matching rubric requirements\n' ' "score": NUMBER, // Integer from 0 to 100\n' ' "confidence": NUMBER, // Float from 0.0 to 1.0\n' ' "rationale": "STRING" // Concise explanation of findings\n' "}\n" "Do not include any Markdown fences, preamble, or conversational commentary." ) # 3. Execute Non-deterministic Prompt raw_response = gl.nondet.exec_prompt(evaluation_prompt, response_format="json") # 4. Defensive Extraction & Schema Validation normalized = _parse_and_validate_llm_result(raw_response) bucket = _bucket_for_score(rubric_definition, normalized["score"]) normalized["score_bucket"] = bucket["bucket"] if normalized["decision"] != bucket["decision"]: raise gl.vm.UserError(f"{ERROR_LLM} decision is inconsistent with rubric score bucket") if normalized["confidence_bps"] < 5000: raise gl.vm.UserError( f"{ERROR_LLM} 'confidence' is below the minimum validity floor of 5000 bps" ) return normalized def validator_fn(leader_result: gl.vm.Result) -> bool: """ Validator verification: independently verifies integrity, executes evaluation with complete evidence and rubric, and enforces substantive comparative equivalence against the leader's proposal. """ # Step A: Validate leader produced a valid Return object if not isinstance(leader_result, gl.vm.Return): return False leader_data = leader_result.calldata if not isinstance(leader_data, dict): return False # Step B: Independent Verification & LLM Execution try: validator_data = leader_fn() except Exception: return False # Step C: Substantive Equivalence Rules # 1. Exact categorical decision match (e.g. APPROVED vs REJECT must never match) if leader_data.get("decision") != validator_data.get("decision"): return False # Exact agreement is required on the normalized decision and the # deterministically derived score bucket. Raw scores may differ. if leader_data.get("score_bucket") != validator_data.get("score_bucket"): return False if leader_data.get("decision") != validator_data.get("decision"): return False # 3. Confidence floor check (minimum 50% confidence / 5000 bps) leader_conf = leader_data.get("confidence_bps", 0) validator_conf = validator_data.get("confidence_bps", 0) if leader_conf < 5000 or validator_conf < 5000: return False return True # ------------------------------------------------------------------- # Execute GenLayer Consensus & State Finalization # ------------------------------------------------------------------- try: consensus_result = gl.vm.run_nondet_unsafe(leader_fn, validator_fn) except Exception: # Consensus failed or error occurred -> Fail Closed case_rec.status = STATUS_FAILED self.cases[case_id] = case_rec return { "case_id": case_id, "status": STATUS_FAILED, "error": "Consensus failed or was rejected by validator set", } # Consensus succeeded -> Compute canonical decision hash and finalize final_score = int(consensus_result["score"]) agreed_bucket = _bucket_for_score(rubric_definition, final_score) final_score_bucket = agreed_bucket["bucket"] if str(consensus_result["score_bucket"]) != final_score_bucket: case_rec.status = STATUS_FAILED self.cases[case_id] = case_rec return {"case_id": case_id, "status": STATUS_FAILED, "error": "Consensus result was inconsistent with rubric"} final_decision = agreed_bucket["decision"] canonical_score = (agreed_bucket["min_score"] + agreed_bucket["max_score"]) // 2 final_conf_bps = int(consensus_result["confidence_bps"]) final_rationale = str(consensus_result["rationale"]) dec_hash = _compute_decision_hash( case_id=case_id, evidence_hash=expected_ev_hash, rubric_hash=expected_rb_hash, decision=final_decision, score_bucket=final_score_bucket, score=canonical_score, confidence_bps=final_conf_bps, ) case_rec.status = STATUS_FINALIZED case_rec.canonical_decision = final_decision case_rec.canonical_score = u32(canonical_score) case_rec.canonical_score_bucket = final_score_bucket case_rec.canonical_confidence_bps = u32(final_conf_bps) case_rec.canonical_rationale = final_rationale case_rec.decision_hash = dec_hash self.cases[case_id] = case_rec return { "case_id": case_id, "status": STATUS_FINALIZED, "decision": final_decision, "score": canonical_score, "canonical_score_bucket": final_score_bucket, "confidence_bps": final_conf_bps, "rationale": final_rationale, "decision_hash": dec_hash, }