Spaces:
Sleeping
Sleeping
File size: 3,541 Bytes
500090d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 | from typing import Dict, Optional, Tuple
from gatekeepers import MCPPanel
class DeterministicValidator:
def __init__(self, mcp_panel=None):
self.mcp = mcp_panel or MCPPanel()
def validate_media(self, media_spec: Dict, media_result: Dict, artifact_type: str = "video") -> Tuple[bool, str]:
if media_result.get("seed") != media_spec.get("seed"):
return False, f"SEED_MISMATCH: expected {media_spec.get('seed')}"
if artifact_type == "video":
expected_res = f"{media_spec['resolution'][0]}x{media_spec['resolution'][1]}"
if media_result.get("resolution") != expected_res:
return False, f"RES_MISMATCH: expected {expected_res}"
if media_result.get("frames") != media_spec.get("frames"):
return False, f"FRAME_MISMATCH: expected {media_spec.get('frames')}"
if not media_result.get("hash"):
return False, "MISSING_HASH: deterministic media must be hashed"
return True, "DETERMINISTIC_INTEGRITY_CONFIRMED"
def validate_writeback(self, delta_Kt: Dict, ontology_snapshot: Dict, ambiguity_context: Optional[Dict] = None) -> Tuple[bool, str]:
if not isinstance(delta_Kt, dict):
return False, "INVALID_DELTA"
protected_O = {"O1", "O2", "O3"}
if "O" in delta_Kt:
for k in delta_Kt["O"]:
if k in protected_O and delta_Kt["O"][k] is None:
return False, f"MONOTONICITY: cannot delete {k}"
if "R" in delta_Kt:
all_concepts = set(ontology_snapshot.get("O", {}).keys())
if "O" in delta_Kt:
all_concepts.update(k for k, v in delta_Kt["O"].items() if v is not None)
for rel in delta_Kt["R"]:
if isinstance(rel, dict):
src = rel.get("source", "")
tgt = rel.get("target", "")
src_in = any(src in v.get("concept", "") for v in ontology_snapshot.get("O", {}).values())
tgt_in = any(tgt in v.get("concept", "") for v in ontology_snapshot.get("O", {}).values())
if not (src_in or src in all_concepts) or not (tgt_in or tgt in all_concepts):
return False, f"REFERENTIAL_INTEGRITY: {src}->{tgt}"
if "L" in delta_Kt:
for rule in delta_Kt["L"]:
if not isinstance(rule, str):
return False, "RULE_SYNTAX: must be string"
if not (rule.startswith("forall") or rule.startswith("exists") or rule.startswith("∀") or rule.startswith("∃") or "->" in rule or "→" in rule):
return False, f"RULE_SYNTAX: '{rule[:30]}...'"
if "G" in delta_Kt:
return False, "CAUSAL_IMMUTABILITY: G cannot be modified"
delta_proposal = {
"payload": {"scope": "governed", "delta_keys": list(delta_Kt.keys()), "ambiguity_context": ambiguity_context, "source": "HumanResolution_StA"},
"source": "HumanResolution_StA",
"proposal_type": "KNOWLEDGE_FEEDBACK",
}
mcp = self.mcp.verdict(delta_proposal, ontology_snapshot)
if mcp["final_verdict"] == "deny":
reason = next((r["reasoning"] for r in mcp["results"] if r["verdict"] == "deny"), "MCP denied")
return False, f"POLICY_VIOLATION: {reason}"
if ambiguity_context is None:
return False, "AMBIGUITY_REQUIRED: write-back only valid after StA ambiguity"
return True, "WRITE_BACK_VALIDATED"
|