# Semantic fragment sanitizer — converts sentence-like rationale nodes into # attributes on related nodes and removes invalid file_type values. # # Currently called from the skill merge scripts (skill-opencode.md, # skill-codex.md) so that rationale text never leaks into the knowledge # graph as standalone nodes. (Future: graphify.llm may wire this into # _parse_llm_json / _merge_into for non-skill code paths; not done in # this cycle.) from __future__ import annotations import json import re from pathlib import Path from .build import _normalize_hyperedge_members # Labels longer than this many characters, or containing >= this many words, # are candidates for being sentence-like rationale text rather than entity names. _RATIONALE_MIN_CHARS = 80 _RATIONALE_MIN_WORDS = 8 # Validation limits for untrusted semantic-fragment payloads. See # validate_semantic_fragment(). Issue #825: returned-JSON normalization for # OpenCode and Codex agents requires a Python enforcement boundary so a # malicious or runaway agent response cannot exhaust memory or escape the # graphify-out chunk directory via crafted node/edge IDs. MAX_SEMANTIC_FRAGMENT_BYTES = 25 * 1024 * 1024 MAX_SEMANTIC_FRAGMENT_NODES = 10_000 MAX_SEMANTIC_FRAGMENT_EDGES = 100_000 MAX_SEMANTIC_FRAGMENT_HYPEREDGES = 10_000 MAX_SEMANTIC_HYPEREDGE_NODES = 256 MAX_SEMANTIC_ID_LENGTH = 256 VALID_SEMANTIC_FILE_TYPES = frozenset({"code", "document", "paper", "image", "rationale", "concept"}) _SEMANTIC_ID_RE = re.compile(r"^[A-Za-z0-9._:-]+$") def validate_semantic_fragment(fragment: object) -> list[str]: """Return validation errors for an untrusted semantic extraction fragment. Empty list means valid. Called by skill merge code before sanitize_semantic_fragment() so malformed or malicious agent JSON is rejected before it touches the graph. Parameter is `object` (not `dict`) because we may be handed arbitrary deserialized JSON — the first check rejects anything that isn't a dict. """ if not isinstance(fragment, dict): return ["fragment must be a JSON object"] errors: list[str] = [] try: payload = json.dumps(fragment, ensure_ascii=False).encode("utf-8") except (TypeError, ValueError) as exc: return [f"fragment is not JSON-serializable: {exc}"] if len(payload) > MAX_SEMANTIC_FRAGMENT_BYTES: errors.append(f"payload is {len(payload)} bytes; max is {MAX_SEMANTIC_FRAGMENT_BYTES}") nodes = fragment.get("nodes", []) edges = fragment.get("edges", []) if not isinstance(nodes, list): errors.append("nodes must be a list") nodes = [] elif len(nodes) > MAX_SEMANTIC_FRAGMENT_NODES: errors.append(f"nodes has {len(nodes)} entries; max is {MAX_SEMANTIC_FRAGMENT_NODES}") if not isinstance(edges, list): errors.append("edges must be a list") edges = [] elif len(edges) > MAX_SEMANTIC_FRAGMENT_EDGES: errors.append(f"edges has {len(edges)} entries; max is {MAX_SEMANTIC_FRAGMENT_EDGES}") for i, node in enumerate(nodes): if not isinstance(node, dict): errors.append(f"nodes[{i}] must be an object") continue _validate_semantic_id(errors, f"nodes[{i}].id", node.get("id")) file_type = node.get("file_type") if file_type is not None and file_type not in VALID_SEMANTIC_FILE_TYPES: errors.append( f"nodes[{i}].file_type {file_type!r} is not one of " f"{sorted(VALID_SEMANTIC_FILE_TYPES)}" ) # validate file_type before any sanitize path can run for i, edge in enumerate(edges): if not isinstance(edge, dict): errors.append(f"edges[{i}] must be an object") continue _validate_semantic_id(errors, f"edges[{i}].source", edge.get("source")) _validate_semantic_id(errors, f"edges[{i}].target", edge.get("target")) hyperedges = fragment.get("hyperedges", []) if hyperedges is None: hyperedges = [] if not isinstance(hyperedges, list): errors.append("hyperedges must be a list") else: if len(hyperedges) > MAX_SEMANTIC_FRAGMENT_HYPEREDGES: errors.append( f"hyperedges has {len(hyperedges)} entries; " f"max is {MAX_SEMANTIC_FRAGMENT_HYPEREDGES}" ) for i, he in enumerate(hyperedges): if not isinstance(he, dict): errors.append(f"hyperedges[{i}] must be an object") continue # Fold alias member keys (members/node_ids) onto `nodes` (#1561) so # an alias-keyed hyperedge isn't rejected here for "nodes must be a # list" before it ever reaches build's normalization. _normalize_hyperedge_members(he) _validate_semantic_id(errors, f"hyperedges[{i}].id", he.get("id")) he_nodes = he.get("nodes") if not isinstance(he_nodes, list): errors.append(f"hyperedges[{i}].nodes must be a list") continue if len(he_nodes) > MAX_SEMANTIC_HYPEREDGE_NODES: errors.append( f"hyperedges[{i}].nodes has {len(he_nodes)} entries; " f"max is {MAX_SEMANTIC_HYPEREDGE_NODES}" ) for j, ref in enumerate(he_nodes): _validate_semantic_id(errors, f"hyperedges[{i}].nodes[{j}]", ref) return errors def load_validated_semantic_fragment(path: Path) -> tuple[dict | None, list[str]]: """Load and validate a semantic chunk, rejecting oversize files before parsing. The size guard runs against `path.stat().st_size` so an attacker-supplied multi-gigabyte chunk file cannot blow up memory at `read_text()` time. JSON decode errors are returned as validation errors rather than raised, so callers can `continue` past bad chunks without a try/except. """ try: size = path.stat().st_size except OSError as exc: return None, [f"could not stat {path}: {exc}"] if size > MAX_SEMANTIC_FRAGMENT_BYTES: return None, [f"payload is {size} bytes; max is {MAX_SEMANTIC_FRAGMENT_BYTES}"] try: fragment = json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: return None, [f"invalid JSON: {exc}"] except OSError as exc: return None, [f"could not read {path}: {exc}"] errors = validate_semantic_fragment(fragment) return (None, errors) if errors else (fragment, []) def _validate_semantic_id(errors: list[str], field: str, value: object) -> None: if not isinstance(value, str): errors.append(f"{field} must be a string") return if not value: errors.append(f"{field} must not be empty") return if len(value) > MAX_SEMANTIC_ID_LENGTH: errors.append(f"{field} is {len(value)} chars; max is {MAX_SEMANTIC_ID_LENGTH}") if "/" in value or "\\" in value or ".." in value: errors.append(f"{field} must not contain path separators or '..'") if not _SEMANTIC_ID_RE.fullmatch(value): errors.append(f"{field} contains unsupported characters") def sanitize_semantic_fragment(fragment: dict) -> dict: """Clean up a semantic extraction fragment in-place. Operations: 1. Removes nodes with ``file_type: "rationale"`` or ``file_type: "concept"`` that were emitted by an LLM (these are not valid semantic entity types). 2. Detects nodes whose label reads like a sentence / rationale paragraph AND that participate in a ``rationale_for`` edge, then converts the label into a ``rationale`` attribute on the target node and removes the source-node + its edges. The ``rationale_for`` edge signal applies regardless of the source node's ``file_type`` — sentence-like nodes with allowed types (``document``, ``code``) are still cleaned up when they're explicitly marked as rationale. 3. Strips nodes whose only distinguishing field is the label itself (empty id — likely LLM hallucination). 4. Filters hyperedges so they cannot reference removed or unknown node IDs after the cleanup passes above. A hyperedge with fewer than two surviving members is dropped. Returns the same dict for convenience. """ _invalid_ft = frozenset({"rationale", "concept"}) nodes: list[dict] = fragment.get("nodes", []) edges: list[dict] = fragment.get("edges", []) hyperedges: list[dict] = fragment.get("hyperedges", []) or [] # ---- build lookup maps -------------------------------------------------- node_by_id: dict[str, dict] = {} for n in nodes: nid = n.get("id", "") if nid: node_by_id[nid] = n # Pre-collect node IDs that source a `rationale_for` edge — these are # candidates for sentence-like cleanup even when file_type is allowed. rationale_for_sources: set[str] = set() for e in edges: if e.get("relation") == "rationale_for": src = e.get("source", "") if src: rationale_for_sources.add(src) # ---- pass 1: identify nodes to remove + rationale candidates ----------- rationale_candidates: list[dict] = [] remove_ids: set[str] = set() keep_nodes: list[dict] = [] for n in nodes: nid = n.get("id", "") if not nid: # Node without an id cannot be referenced — discard. continue ft = n.get("file_type", "") label = n.get("label", "") if ft in _invalid_ft: # Explicitly-invalid file_type ("rationale" or "concept"): if # the label looks like a sentence we may convert to attribute. if _is_sentence_like_rationale_label(label): rationale_candidates.append(n) remove_ids.add(nid) continue if nid in rationale_for_sources and _is_sentence_like_rationale_label(label): # Allowed file_type, but the node sources a `rationale_for` edge # AND its label is sentence-like prose. Treat it as rationale # cleanup material rather than a real graph entity. rationale_candidates.append(n) remove_ids.add(nid) continue keep_nodes.append(n) # ---- pass 2: convert sentence-nodes → rationale attributes -------------- # Only `rationale_for` edges propagate the rationale text. Other outgoing # edges (e.g. references, conceptually_related_to) are NOT used as # attribute-propagation paths — that would corrupt unrelated nodes by # attaching rationale meant for a different target. rationale_attrs: dict[str, list[str]] = {} for rn in rationale_candidates: rn_id = rn.get("id", "") text = rn.get("label", "").strip() for e in edges: if e.get("relation") != "rationale_for": continue if e.get("source") != rn_id: continue target_id = e.get("target") if target_id not in node_by_id or target_id in remove_ids: continue rationale_attrs.setdefault(target_id, []).append(text) for target_id, texts in rationale_attrs.items(): if target_id in node_by_id and target_id not in remove_ids: _append_rationale_attr(node_by_id[target_id], texts) # ---- pass 3: strip edges referencing removed nodes ---------------------- keep_edges: list[dict] = [] for e in edges: src = e.get("source", "") tgt = e.get("target", "") if src in remove_ids or tgt in remove_ids: continue keep_edges.append(e) # ---- pass 4: filter hyperedges to surviving node IDs -------------------- surviving_ids: set[str] = {n.get("id", "") for n in keep_nodes} surviving_ids.discard("") keep_hyperedges: list[dict] = [] for he in hyperedges: if not isinstance(he, dict): continue # Fold alias member keys (members/node_ids) onto `nodes` (#1561) so an # alias-keyed hyperedge isn't silently dropped below for a missing # `nodes` list before build can canonicalize it. _normalize_hyperedge_members(he) he_nodes = he.get("nodes") if not isinstance(he_nodes, list): continue filtered = [ref for ref in he_nodes if isinstance(ref, str) and ref in surviving_ids] if len(filtered) < 2: # A hyperedge needs at least two surviving members to be meaningful. continue if len(filtered) != len(he_nodes): he = dict(he) he["nodes"] = filtered keep_hyperedges.append(he) fragment["nodes"] = keep_nodes fragment["edges"] = keep_edges fragment["hyperedges"] = keep_hyperedges return fragment def _is_sentence_like_rationale_label(label: str) -> bool: """Return True if *label* looks like prose / rationale text rather than an entity or concept name. Heuristics (no false positives on short-concept-edge-cases): - Longer than *_RATIONALE_MIN_CHARS* chars, OR - At least *_RATIONALE_MIN_WORDS* whitespace-delimited tokens, AND - Contains at least one sentence-ending punctuation mark (``. ! ?``) or a colon (common in "Decision: ..." rationales). """ if not label: return False label = label.strip() if len(label) < _RATIONALE_MIN_CHARS: word_count = len(label.split()) if word_count < _RATIONALE_MIN_WORDS: return False # Must look like actual prose: has sentence-ending punctuation or a colon. return bool(re.search(r"[.!?:]", label)) def _append_rationale_attr(node: dict, texts: list[str]) -> None: """Append one or more rationale strings to *node*'s ``rationale`` attribute. If the attribute already exists the new texts are appended with a double-newline separator so downstream consumers can distinguish distinct rationale fragments. """ existing = node.get("rationale", "") new_text = "\n\n".join(texts).strip() if existing: node["rationale"] = existing + "\n\n" + new_text else: node["rationale"] = new_text