Spaces:
Sleeping
Sleeping
| 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" | |