ontocast.onto.rdfgraph

RDFGraph

Bases: Graph

Subclass of rdflib.Graph with Pydantic schema support.

This class extends rdflib.Graph to provide serialization and deserialization capabilities for Pydantic models, with special handling for Turtle format.

Source code in ontocast/onto/rdfgraph.py

class RDFGraph(Graph):
    """Subclass of rdflib.Graph with Pydantic schema support.

    This class extends rdflib.Graph to provide serialization and deserialization
    capabilities for Pydantic models, with special handling for Turtle format.
    """

    @classmethod
    def __get_pydantic_core_schema__(cls, _source_type, handler: GetCoreSchemaHandler):
        """Get the Pydantic core schema for this class.

        Args:
            _source_type: The source type.
            handler: The core schema handler.

        Returns:
            A union schema accepting:
            - existing ``RDFGraph`` instances,
            - Turtle / JSON-LD strings (parsed via ``_from_str``),
            - JSON-LD ``dict`` / ``list`` objects (parsed via ``_from_jsonld_obj``).
        """
        return core_schema.union_schema(
            [
                core_schema.is_instance_schema(cls),
                core_schema.chain_schema(
                    [
                        core_schema.str_schema(),
                        core_schema.no_info_plain_validator_function(cls._from_str),
                    ]
                ),
                core_schema.no_info_plain_validator_function(cls._from_any),
            ],
            serialization=core_schema.plain_serializer_function_ser_schema(
                cls._to_turtle_str,
                info_arg=False,
                return_schema=core_schema.str_schema(),
            ),
        )

    def __add__(self, other: Union["RDFGraph", Graph, Iterable]) -> "RDFGraph":
        """Addition operator for RDFGraph instances.

        Merges the RDF graphs while maintaining the RDFGraph type.

        Args:
            other: The graph to add to this one.

        Returns:
            RDFGraph: A new RDFGraph containing the merged triples.
        """
        # Create a new RDFGraph instance
        result = RDFGraph()

        # Copy all triples from both graphs
        copy_triples(self, result, origin="RDFGraph.__add__")
        copy_triples(other, result, origin="RDFGraph.__add__")

        existing = {prefix: str(uri) for prefix, uri in self.namespaces() if prefix}
        incoming: dict[str, str] = {}
        if isinstance(other, Graph):
            incoming = {
                prefix: str(uri) for prefix, uri in other.namespaces() if prefix
            }
        for prefix, uri in merge_namespace_bindings(existing, incoming).items():
            result.bind(prefix, uri)

        return result

    def __iadd__(self, other: Union["RDFGraph", Graph, Iterable]) -> "RDFGraph":
        """In-place addition operator for RDFGraph instances.

        Merges the RDF graphs while maintaining the RDFGraph type and binding prefixes.

        Args:
            other: The graph to add to this one.

        Returns:
            RDFGraph: self after modification.
        """
        # Use __add__ to get the merged result with proper prefix binding
        result = self.__add__(other)

        # Clear current graph and copy the result
        self.remove((None, None, None))  # Remove all triples

        # Copy all triples from result
        copy_triples(result, self, origin="RDFGraph.__iadd__")

        # Copy namespace bindings from result
        for prefix, uri in result.namespaces():
            self.bind(prefix, uri)

        return self

    def copy(self) -> "RDFGraph":
        """Create a copy of this RDFGraph.

        Returns:
            RDFGraph: A new RDFGraph instance with all triples and namespace bindings copied.
        """
        result = RDFGraph()

        # Copy all triples. An oxigraph-backed graph may hold RDF 1.2 triple
        # terms, which a plain rdflib graph cannot represent -- this is the path
        # __deepcopy__ takes, so it degrades instead of raising.
        copy_triples(self, result, origin="RDFGraph.copy")

        # Copy namespace bindings
        for prefix, uri in self.namespaces():
            result.bind(prefix, uri)

        return result

    def __copy__(self) -> "RDFGraph":
        """Ensure shallow copies preserve RDFGraph type."""
        return self.copy()

    def __deepcopy__(self, memo: dict[int, Any]) -> "RDFGraph":
        """Ensure deep copies preserve RDFGraph type."""
        copied = self.copy()
        memo[id(self)] = copied
        return copied

    @staticmethod
    def _ensure_prefixes(turtle_str: str) -> str:
        """Declare prefixes used in Turtle but missing from ``@prefix`` lines.

        Resolves undeclared used prefixes from ingest vocabulary (COMMON +
        WELL_KNOWN) and the context map set via :meth:`set_known_prefixes`.
        """
        declared_prefixes = set(
            match.group(1) for match in PREFIX_PATTERN.finditer(turtle_str)
        )

        used_prefixes: set[str] = set()
        for match in PREFIX_USAGE_PATTERN.finditer(turtle_str):
            prefix = match.group(1)
            if prefix not in declared_prefixes:
                used_prefixes.add(prefix)

        lookup = _prefix_lookup_for_turtle_repair()
        missing: dict[str, str] = {}
        for prefix in used_prefixes:
            if prefix in lookup:
                missing[prefix] = _format_namespace_uri_for_turtle_declaration(
                    lookup[prefix]
                )

        if not missing:
            return turtle_str

        prefix_block = (
            "\n".join(f"@prefix {prefix}: {uri} ." for prefix, uri in missing.items())
            + "\n\n"
        )

        return prefix_block + turtle_str

    @staticmethod
    def _is_jsonld_str(s: str) -> bool:
        """Check if a string appears to be JSON-LD format.

        Args:
            s: The string to check.

        Returns:
            bool: True if the string appears to be JSON-LD.
        """
        s = s.strip()
        if not (s.startswith("{") or s.startswith("[")):
            return False
        try:
            # Try to parse as JSON
            data = json.loads(s)
            # Check if it's a dict/object with @context or @id, or an array containing such objects
            if isinstance(data, dict):
                return "@context" in data or "@id" in data
            elif isinstance(data, list):
                return any(
                    isinstance(item, dict) and ("@context" in item or "@id" in item)
                    for item in data
                )
            return False
        except (json.JSONDecodeError, ValueError):
            return False

    @classmethod
    def _from_str(cls, data_str: str) -> "RDFGraph":
        """Create an RDFGraph instance from a string (Turtle or JSON-LD).

        Automatically detects the format and parses accordingly.

        Args:
            data_str: The input string in Turtle or JSON-LD format.

        Returns:
            RDFGraph: A new RDFGraph instance.
        """
        if cls._is_jsonld_str(data_str):
            return cls._from_jsonld_str(data_str)
        else:
            return cls._from_turtle_str(data_str)

    @classmethod
    def _from_jsonld_obj(cls, jsonld_obj: dict | list) -> "RDFGraph":
        """Create an RDFGraph from a JSON-LD ``dict`` or ``list`` object.

        Used when LLMs emit JSON-LD as a native JSON object in a structured
        response field instead of as an embedded string.

        Args:
            jsonld_obj: Parsed JSON-LD value (object or array of objects).

        Returns:
            RDFGraph: A new RDFGraph instance.
        """
        return cls._from_jsonld_str(json.dumps(jsonld_obj))

    @classmethod
    def _from_any(cls, value: Any) -> "RDFGraph":
        """Dispatch RDFGraph creation from arbitrary structured input.

        Accepts dicts/lists (treated as JSON-LD), strings (Turtle or JSON-LD),
        and existing RDFGraph instances. Used as the catch-all branch in
        the Pydantic schema for graph fields that may receive JSON-LD objects
        from LLM structured output.
        """
        if isinstance(value, cls):
            return value
        if isinstance(value, (dict, list)):
            return cls._from_jsonld_obj(value)
        if isinstance(value, str):
            return cls._from_str(value)
        raise TypeError(
            f"Cannot construct RDFGraph from value of type {type(value).__name__}"
        )

    @classmethod
    def _from_turtle_str(cls, turtle_str: str) -> "RDFGraph":
        """Create an RDFGraph instance from a Turtle string.

        This method uses context variables to access known prefixes that may be
        needed to complete missing prefix declarations in the Turtle string.

        Args:
            turtle_str: The input Turtle string.

        Returns:
            RDFGraph: A new RDFGraph instance.
        """
        normalized_turtle = cls._normalize_turtle_input(turtle_str)
        patched_turtle = cls._coerce_invalid_numeric_typed_literals(
            cls._quote_unquoted_typed_literals(cls._ensure_prefixes(normalized_turtle))
        )
        g = cls()
        try:
            g.parse(data=patched_turtle, format="turtle")
            g._sanitize_prefix_boundaries_from_turtle(normalized_turtle)
            return g
        except Exception as parse_error:
            error_message = str(parse_error)
            repaired_turtle = cls._repair_common_turtle_issues(
                patched_turtle, parse_error_message=error_message
            )
            if repaired_turtle == patched_turtle:
                repaired_turtle = cls._repair_unknown_prefix(
                    patched_turtle, error_message
                )
            if repaired_turtle == patched_turtle and _SPARQL_DATA_BLOCK_RE.search(
                patched_turtle
            ):
                repaired_turtle = strip_sparql_update_wrapper(patched_turtle)
            if repaired_turtle == patched_turtle:
                raise
            logger.warning(
                "Recovering malformed Turtle after parse failure: %s",
                error_message,
            )
            repaired_graph = cls()
            repaired_graph.parse(data=repaired_turtle, format="turtle")
            repaired_graph._sanitize_prefix_boundaries_from_turtle(normalized_turtle)
            return repaired_graph

    def _sanitize_prefix_boundaries_from_turtle(self, turtle_str: str) -> None:
        declared_prefixes = {
            match.group(1): match.group(2)
            for match in PREFIX_DECLARATION_PATTERN.finditer(turtle_str)
        }
        if not declared_prefixes:
            return
        sanitized = sanitize_prefix_map(declared_prefixes, context="auto")
        changed_namespaces = [
            (original, sanitized[prefix])
            for prefix, original in declared_prefixes.items()
            if sanitized[prefix] != original
        ]
        for original_namespace, normalized_namespace in changed_namespaces:
            self.remap_namespaces(
                old_namespace=original_namespace,
                new_namespace=normalized_namespace,
            )
        for prefix, namespace in sanitized.items():
            self.bind(prefix, Namespace(namespace), override=True)

    @staticmethod
    def _normalize_turtle_input(turtle_str: str) -> str:
        """Normalize Turtle text from LLM output before RDF parsing."""
        normalized = unicodedata.normalize("NFKC", turtle_str)
        normalized = normalized.replace("\ufeff", "")
        normalized = normalized.replace("\u200b", "")
        normalized = normalized.replace("\u200c", "")
        normalized = normalized.replace("\u200d", "")
        normalized = normalized.replace("\u2060", "")
        normalized = normalized.replace("\xa0", " ")

        cleaned_chars = []
        for ch in normalized:
            if ch in ("\n", "\r", "\t"):
                cleaned_chars.append(ch)
                continue
            if unicodedata.category(ch).startswith("C"):
                continue
            cleaned_chars.append(ch)
        normalized = "".join(cleaned_chars)
        normalized = _SPARQL_PREFIX_RE.sub(
            lambda match: f"@prefix {match.group(1)}: <{match.group(2)}> .",
            normalized,
        )
        return normalized

    @staticmethod
    def _quote_unquoted_typed_literals(turtle_str: str) -> str:
        """Wrap bare numeric values that carry a ^^ datatype annotation in quotes.

        The LLM sometimes emits ``1975^^xsd:integer`` instead of
        ``"1975"^^xsd:integer``.  The unquoted form causes the Turtle parser to
        treat ``^^`` as property-path operators, raising
        "Bad syntax (EOF found in middle of path syntax)".
        """
        return UNQUOTED_TYPED_LITERAL_PATTERN.sub(r'"\1"^^\2', turtle_str)

    @classmethod
    def _coerce_invalid_numeric_typed_literals(cls, turtle_str: str) -> str:
        """Drop numeric datatype when lexical form is invalid for that datatype."""

        def replace_integer(match: re.Match[str]) -> str:
            lexical = match.group(1)
            if _is_valid_integer_lexical(lexical):
                return match.group(0)
            return f'"{lexical}"'

        def replace_decimal(match: re.Match[str]) -> str:
            lexical = match.group(1)
            if _is_valid_decimal_lexical(lexical):
                return match.group(0)
            return f'"{lexical}"'

        coerced = INTEGER_TYPED_LITERAL_PATTERN.sub(replace_integer, turtle_str)
        coerced = DECIMAL_TYPED_LITERAL_PATTERN.sub(replace_decimal, coerced)
        coerced = DOUBLE_TYPED_LITERAL_PATTERN.sub(replace_decimal, coerced)
        coerced = cls._coerce_invalid_date_typed_literals(coerced)
        return coerced

    @staticmethod
    def _coerce_invalid_date_typed_literals(turtle_str: str) -> str:
        """Normalize date literals with invalid xsd:date lexical forms."""

        def replace_date(match: re.Match[str]) -> str:
            lexical = match.group(1)
            if re.fullmatch(r"\d{4}-\d{2}-\d{2}", lexical):
                return match.group(0)
            return _coerce_date_lexical_for_turtle(lexical)

        return DATE_TYPED_LITERAL_PATTERN.sub(replace_date, turtle_str)

    @staticmethod
    def _coerce_invalid_nquads_typed_literals(nquads_str: str) -> str:
        """Coerce invalid XSD typed literals in normalized n-quads before rdflib parse."""

        def replace_integer(match: re.Match[str]) -> str:
            lexical = match.group(1)
            if _is_valid_integer_lexical(lexical):
                return match.group(0)
            return f'"{lexical}"'

        def replace_decimal(match: re.Match[str]) -> str:
            lexical = match.group(1)
            if _is_valid_decimal_lexical(lexical):
                return match.group(0)
            return f'"{lexical}"'

        def replace_date(match: re.Match[str]) -> str:
            lexical = match.group(1)
            if re.fullmatch(r"\d{4}-\d{2}-\d{2}", lexical):
                return match.group(0)
            return _coerce_date_lexical_for_nquads(lexical)

        coerced = NQUADS_INTEGER_TYPED_LITERAL_PATTERN.sub(replace_integer, nquads_str)
        coerced = NQUADS_DECIMAL_TYPED_LITERAL_PATTERN.sub(replace_decimal, coerced)
        coerced = NQUADS_DOUBLE_TYPED_LITERAL_PATTERN.sub(replace_decimal, coerced)
        coerced = NQUADS_DATE_TYPED_LITERAL_PATTERN.sub(replace_date, coerced)
        return coerced

    @staticmethod
    def _repair_unknown_prefix(turtle_str: str, parse_error_message: str) -> str:
        """Inject ``@prefix`` for a single unknown prefix named in an rdflib parse error."""
        match = UNKNOWN_PREFIX_ERROR_PATTERN.search(parse_error_message)
        if not match:
            return turtle_str
        prefix = match.group(1)
        if re.search(rf"@prefix\s+{re.escape(prefix)}\s*:", turtle_str):
            return turtle_str

        lookup = _prefix_lookup_for_turtle_repair()
        namespace_uri = lookup.get(prefix)
        if not namespace_uri:
            ctx = _known_prefixes_context.get() or {}
            prefix_lower = prefix.lower()
            for _p, _uri in ctx.items():
                if _uri.rstrip("#/").rsplit("/", 1)[-1].lower() == prefix_lower:
                    namespace_uri = _uri
                    break
        if not namespace_uri:
            return turtle_str

        ingest_only = prefix_lookup_for_ingest()
        if prefix in ingest_only and namespace_uri == ingest_only[prefix]:
            source = "well_known"
        else:
            source = "context"
        logger.warning(
            "Recovering unknown Turtle prefix %r from %s namespace <%s>",
            prefix,
            source,
            namespace_uri,
        )
        declaration = (
            f"@prefix {prefix}: "
            f"{_format_namespace_uri_for_turtle_declaration(namespace_uri)} .\n"
        )
        return declaration + turtle_str

    @staticmethod
    def _merge_known_prefixes_into_jsonld(data: dict | list) -> dict | list:
        """Shallow-merge known prefix URIs into JSON-LD ``@context`` before normalization."""
        known = _known_prefixes_context.get()
        if not known:
            return data

        def merge_context(context: object) -> object:
            if not isinstance(context, dict):
                return context
            merged = dict(context)
            for prefix, uri in known.items():
                if prefix.startswith("@") or prefix in merged:
                    continue
                merged[prefix] = uri
            return merged

        if isinstance(data, dict):
            if "@context" in data:
                data = dict(data)
                data["@context"] = merge_context(data["@context"])
            return data
        if isinstance(data, list):
            return [
                (
                    {**item, "@context": merge_context(item["@context"])}
                    if isinstance(item, dict) and "@context" in item
                    else item
                )
                for item in data
            ]
        return data

    @staticmethod
    def _is_valid_typed_literal(literal: Literal) -> bool:
        """Return whether *literal* has a valid lexical form for its XSD datatype."""
        if literal.datatype is None:
            return True

        lexical = str(literal)
        datatype = literal.datatype

        if datatype in (XSD.decimal, XSD.double):
            return _is_valid_decimal_lexical(lexical)

        if datatype == XSD.integer:
            return _is_valid_integer_lexical(lexical)

        if datatype == XSD.float:
            try:
                float(lexical)
                return True
            except ValueError:
                return False

        return True

    @classmethod
    def partition_invalid_typed_literals(
        cls, graph: "RDFGraph"
    ) -> tuple["RDFGraph", list[RejectedLiteralTriple]]:
        """Split *graph* into a clean graph and quarantined invalid typed literals."""
        clean = cls()
        for prefix, namespace_uri in graph.namespaces():
            if prefix:
                clean.bind(prefix, namespace_uri)

        rejected: list[RejectedLiteralTriple] = []
        for subject, predicate, obj in graph:
            if isinstance(obj, Literal) and not cls._is_valid_typed_literal(obj):
                rejected.append(
                    RejectedLiteralTriple(
                        subject=str(subject),
                        predicate=str(predicate),
                        object_lexical=str(obj),
                        datatype=str(obj.datatype),
                    )
                )
                continue
            clean.add((subject, predicate, obj))

        if rejected:
            logger.warning(
                "Quarantined %d triple(s) with invalid XSD typed literals",
                len(rejected),
            )

        return clean, rejected

    @classmethod
    def _repair_common_turtle_issues(
        cls, turtle_str: str, parse_error_message: str
    ) -> str:
        """Apply minimal repairs for common malformed Turtle patterns."""
        repaired = turtle_str

        # Typical LLM truncation: dangling ';' or ',' at EOF in property list.
        if "EOF found when expected verb in property list" in parse_error_message:
            repaired = cls._repair_truncated_turtle(repaired)

        # Unquoted ``^^`` literals can surface as path-syntax EOF errors.
        if "EOF found in middle of path syntax" in parse_error_message:
            repaired = cls._repair_truncated_turtle(repaired)

        # LLM repeats the subject on lines that follow a ';' continuation.
        if "expected '.' or '}' or ']' at end of statement" in parse_error_message:
            repaired = cls._repair_repeated_subject_after_semicolon(repaired)

        if (
            "Expected end of text, found 'DELETE'" in parse_error_message
            or "Expected end of text, found 'INSERT'" in parse_error_message
        ):
            repaired = strip_sparql_update_wrapper(repaired)

        repaired = cls._repair_missing_object_before_dot(repaired)
        return repaired

    @staticmethod
    def _repair_truncated_turtle(turtle_str: str) -> str:
        """Repair common LLM Turtle truncation patterns.

        This only applies a minimal fix when content ends with dangling property-list
        punctuation (';' or ',') and no terminating '.'.
        """
        stripped = turtle_str.rstrip()
        if not stripped:
            return turtle_str
        if stripped.endswith(";") or stripped.endswith(","):
            return f"{stripped[:-1].rstrip()} .\n"
        return turtle_str

    @staticmethod
    def _looks_like_subject_token(token: str) -> bool:
        """Return True if *token* could be a Turtle subject IRI or prefixed name.

        Keywords that are only valid as predicates (``a``) or literals
        (``true``, ``false``) return False so that legitimate short-form
        predicate-object continuations are never mis-identified as repeated
        subjects.
        """
        if token in ("a", "true", "false"):
            return False
        if token.startswith("@") or token.startswith("#") or token.startswith('"'):
            return False
        if token.startswith("<") and ">" in token:
            return True
        if token.startswith("_:"):
            return True
        if ":" in token:
            colon_idx = token.index(":")
            prefix = token[:colon_idx]
            local = token[colon_idx + 1 :]
            return bool(prefix) and bool(local)
        return False

    @classmethod
    def _repair_repeated_subject_after_semicolon(cls, turtle_str: str) -> str:
        """Repair Turtle where a new full triple appears on a line following a ``;``.

        The LLM sometimes emits::

            cd:foo ns1:P1 ns2:Q1 ;
            cd:foo ns1:P2 ns2:Q2 .

        After a ``;`` the Turtle parser expects only a predicate–object pair
        (the subject is implied), but the LLM restates the subject, which
        triggers "expected '.' or '}' or ']' at end of statement".

        Repair strategy: when a line that ends with ``;`` is immediately
        followed by a non-blank line whose first token looks like a subject
        (prefixed name / IRI, not the ``a`` keyword), replace the trailing
        ``;`` with ``.`` so that each subject block is properly terminated and
        the next line starts a fresh triple.
        """
        lines = turtle_str.splitlines(keepends=True)
        result: list[str] = []

        for line in lines:
            content = line.rstrip("\r\n")
            stripped = content.strip()

            if (
                result
                and stripped
                and not stripped.startswith("@prefix")
                and not stripped.startswith("#")
            ):
                prev_raw = result[-1]
                prev_content = prev_raw.rstrip("\r\n").strip()
                if prev_content.endswith(";"):
                    tokens = stripped.split()
                    if len(tokens) >= 3 and cls._looks_like_subject_token(tokens[0]):
                        # Current line is a complete triple (s p o …) right after a ';'.
                        # Terminate the previous statement with '.' instead of ';'.
                        prev_body = prev_raw.rstrip("\r\n").rstrip()
                        eol = prev_raw[len(prev_raw.rstrip("\r\n")) :]
                        result[-1] = prev_body[:-1] + "." + eol
                        logger.debug(
                            "Repaired repeated subject after ';': %s", stripped
                        )

            result.append(line)

        return "".join(result)

    @staticmethod
    def _repair_missing_object_before_dot(turtle_str: str) -> str:
        """Remove malformed lines that end after predicate with no object."""
        repaired_lines: list[str] = []
        previous_meaningful_line: str | None = None
        for line in turtle_str.splitlines():
            stripped = line.strip()
            if not stripped or stripped.startswith("@prefix"):
                repaired_lines.append(line)
                continue
            if stripped.endswith(" ."):
                body = stripped[:-2].strip()
                token_count = len(body.split())
                previous_continues_property_list = (
                    previous_meaningful_line is not None
                    and (
                        previous_meaningful_line.endswith(";")
                        or previous_meaningful_line.endswith(",")
                    )
                )
                # Keep valid predicate-object continuation lines in property lists
                # (e.g. "ex:appealsTo ex:Cassation ."), but drop malformed
                # predicate-only lines and standalone subject-predicate lines.
                if token_count < 2 or (
                    token_count == 2 and not previous_continues_property_list
                ):
                    logger.debug(
                        "Dropping malformed Turtle line with missing object: %s",
                        stripped,
                    )
                    continue
            repaired_lines.append(line)
            previous_meaningful_line = stripped
        repaired = "\n".join(repaired_lines)
        if turtle_str.endswith("\n"):
            repaired += "\n"
        return repaired

    @classmethod
    def set_known_prefixes(cls, prefixes: dict[str, str] | None) -> None:
        """Set known prefixes in the context for use during parsing.

        This should be called before parsing TTL strings that may use prefixes
        from an ontology or other source. The prefixes will be automatically
        added if they're used but not declared in the TTL string.

        Args:
            prefixes: Dictionary mapping prefix names to namespace URIs.
                Example: {"fcaont": "https://growgraph.dev/fcaont#"}
        """
        _known_prefixes_context.set(prefixes)

    @classmethod
    def get_known_prefixes(cls) -> dict[str, str] | None:
        """Get currently known prefixes from context.

        Returns:
            Dictionary mapping prefix names to namespace URIs, or None.
        """
        return _known_prefixes_context.get()

    @classmethod
    def _from_jsonld_str(cls, jsonld_str: str) -> "RDFGraph":
        """Create an RDFGraph instance from a JSON-LD string.

        Args:
            jsonld_str: The input JSON-LD string.

        Returns:
            RDFGraph: A new RDFGraph instance with namespace prefixes extracted from @context.
        """
        # Primary path: pyld URDNA2015 → n-quads → rdflib.  Avoids rdflib's
        # deprecated ConjunctiveGraph and gives canonical blank-node labelling.
        jsonld_data = json.loads(jsonld_str)
        jsonld_data = cls._merge_known_prefixes_into_jsonld(jsonld_data)
        try:
            normalized = jsonld.normalize(
                jsonld_data,
                {"algorithm": "URDNA2015", "format": "application/n-quads"},
            )
        except Exception as jsonld_err:
            # Fallback: rdflib's own JSON-LD parser (no URDNA2015, but tolerant
            # of bad @context entries that pyld rejects).
            logger.warning(
                "JSON-LD normalization failed (%s); retrying with rdflib json-ld parser",
                jsonld_err,
            )
            g = cls()
            g.parse(data=json.dumps(jsonld_data), format="json-ld")
            cls._bind_context_prefixes(g, jsonld_data)
            return g

        normalized_str = normalized if isinstance(normalized, str) else str(normalized)
        normalized_str = cls._coerce_invalid_nquads_typed_literals(normalized_str)

        g = cls()
        # RDFLib 7.6's NQuadsParser still touches Dataset.default_context after
        # that attribute was deprecated in favour of default_graph
        # (RDFLib/rdflib#3409). Filter only that library-internal warning.
        with warnings.catch_warnings():
            warnings.filterwarnings(
                "ignore",
                category=DeprecationWarning,
                message=r"Dataset\.default_context is deprecated.*",
            )
            g.parse(data=normalized_str, format="nquads")
        cls._bind_context_prefixes(g, jsonld_data)
        return g

    @staticmethod
    def _bind_context_prefixes(g: "RDFGraph", jsonld_data: dict | list) -> None:
        """Bind namespace prefixes declared in a JSON-LD ``@context`` onto *g*."""
        try:
            context: object = None
            if isinstance(jsonld_data, dict):
                context = jsonld_data.get("@context")
            elif isinstance(jsonld_data, list) and jsonld_data:
                first = jsonld_data[0]
                if isinstance(first, dict):
                    context = first.get("@context")
            if isinstance(context, dict):
                for prefix, uri in context.items():
                    if isinstance(uri, str) and not prefix.startswith("@"):
                        try:
                            g.bind(prefix, uri)
                        except Exception as exc:
                            logger.debug("Failed to bind prefix %r: %s", prefix, exc)
        except (ValueError, AttributeError) as exc:
            logger.debug("Could not bind prefixes from JSON-LD @context: %s", exc)

    @staticmethod
    def _to_turtle_str(g: Any) -> str:
        """Convert an RDFGraph to a Turtle string.

        For graphs backed by the *oxigraph* store the serialisation is
        delegated to ``pyoxigraph`` so that RDF 1.2 triple-term syntax
        (``<<( s p o )>>``) is emitted correctly.

        Args:
            g: The RDFGraph instance.

        Returns:
            str: The Turtle (or Turtle-star) string representation.
        """
        if hasattr(g, "store") and type(g.store).__name__ == "OxigraphStore":
            return g.serialize_turtle_star()
        return g.serialize(format="turtle")

    def serialize_turtle_star(self) -> str:
        """Serialize an oxigraph-backed graph to Turtle-star via *pyoxigraph*.

        This method extracts all quads belonging to this graph's context
        from the underlying ``pyoxigraph.Store`` and serialises them into
        the default graph using ``pyoxigraph.serialize`` with the Turtle
        format, which natively supports RDF 1.2 ``<<( … )>>`` syntax.

        Returns:
            Turtle-star string.

        Raises:
            RuntimeError: If the graph is not backed by an oxigraph store.
        """
        try:
            import pyoxigraph as ox
            from oxrdflib._converter import to_ox
        except ImportError as exc:
            raise RuntimeError(
                "pyoxigraph / oxrdflib must be installed for Turtle-star serialisation"
            ) from exc

        inner_store = cast(ox.Store, _oxigraph_inner_store(self.store))
        graph_ctx_raw = to_ox(self.identifier)
        assert isinstance(
            graph_ctx_raw,
            (ox.NamedNode, ox.BlankNode, ox.DefaultGraph),
        )
        graph_ctx: ox.NamedNode | ox.BlankNode | ox.DefaultGraph = graph_ctx_raw

        # Copy quads into a temporary store under the default graph so
        # that ``ox.serialize`` can emit plain Turtle (Turtle-star).
        tmp = ox.Store()
        used_iri_terms: set[str] = set()

        def _collect_used_iris(term: Any) -> None:
            if isinstance(term, ox.NamedNode):
                used_iri_terms.add(term.value)
                return
            if isinstance(term, ox.Triple):
                _collect_used_iris(term.subject)
                _collect_used_iris(term.predicate)
                _collect_used_iris(term.object)

        for quad in inner_store.quads_for_pattern(
            None,
            None,
            None,
            graph_ctx,
        ):
            _collect_used_iris(quad.subject)
            _collect_used_iris(quad.predicate)
            _collect_used_iris(quad.object)
            tmp.add(
                ox.Quad(quad.subject, quad.predicate, quad.object, ox.DefaultGraph())
            )

        namespace_to_prefix: dict[str, str] = {}
        for prefix, namespace in self.namespaces():
            if not prefix:
                continue
            prefix_str = str(prefix)
            namespace_str = str(namespace)
            current = namespace_to_prefix.get(namespace_str)
            if current is None or (len(prefix_str), prefix_str) < (
                len(current),
                current,
            ):
                namespace_to_prefix[namespace_str] = prefix_str

        prefixes = {
            prefix: namespace
            for namespace, prefix in namespace_to_prefix.items()
            if any(iri.startswith(namespace) for iri in used_iri_terms)
        }
        raw = tmp.dump(
            format=ox.RdfFormat.TURTLE,
            from_graph=ox.DefaultGraph(),
            prefixes=prefixes or None,
        )
        if raw is None:
            raise RuntimeError("pyoxigraph dump returned no data")
        return raw.decode()

    def serialize_canonical_turtle(self) -> str:
        """Serialize to Turtle after canonical namespace/prefix sanitization.

        Uses :class:`_LosslessTurtleSerializer` rather than rdflib's stock Turtle
        writer, which rounds floating-point literals to 7 significant digits.
        This output is what reaches the triple store, the API response, and the
        LLM prompt, so the rounding was real value loss rather than formatting.
        """
        self.sanitize_prefixes_namespaces()
        serialized = self.serialize(format=LOSSLESS_TURTLE_FORMAT)
        if isinstance(serialized, bytes):
            return serialized.decode("utf-8")
        return str(serialized)

    def _compact_iri_for_jsonld(self, term: Node) -> str | dict[str, str]:
        if isinstance(term, BNode):
            return {"@id": f"_:{term}"}
        if not isinstance(term, URIRef):
            return str(term)
        term_str = str(term)
        for prefix, namespace in self.namespaces():
            if not prefix:
                continue
            ns = str(namespace)
            if term_str.startswith(ns):
                local = term_str[len(ns) :]
                return f"{prefix}:{local}"
        if term_str.startswith("http://") or term_str.startswith("https://"):
            return term_str
        return term_str

    def _literal_to_jsonld(self, literal: Literal) -> str | dict[str, str]:
        if literal.language:
            return {"@value": str(literal), "@language": literal.language}
        if literal.datatype:
            return {
                "@value": str(literal),
                "@type": _datatype_to_compact(str(literal.datatype)),
            }
        return str(literal)

    def _object_to_jsonld(self, obj: Node) -> str | dict[str, str]:
        if isinstance(obj, Literal):
            return self._literal_to_jsonld(obj)
        if isinstance(obj, (URIRef, BNode)):
            compact = self._compact_iri_for_jsonld(obj)
            if isinstance(compact, dict):
                return compact
            return {"@id": compact}
        return str(obj)

    def serialize_compact_jsonld_for_prompt(self) -> str:
        """Serialize graph as compact JSON-LD text for LLM context prompts."""
        self.sanitize_prefixes_namespaces()
        context: dict[str, str] = {}
        for prefix, namespace in self.namespaces():
            if prefix:
                context[prefix] = str(namespace)

        nodes: dict[str, dict[str, Any]] = defaultdict(dict)
        for subject, predicate, obj in self:
            subj_key = str(subject)
            if "@id" not in nodes[subj_key]:
                subj_compact = self._compact_iri_for_jsonld(subject)
                if isinstance(subj_compact, dict):
                    nodes[subj_key].update(subj_compact)
                else:
                    nodes[subj_key]["@id"] = subj_compact

            pred_compact = self._compact_iri_for_jsonld(predicate)
            if not isinstance(pred_compact, str):
                pred_key = str(predicate)
            else:
                pred_key = pred_compact

            value = self._object_to_jsonld(obj)
            existing = nodes[subj_key].get(pred_key)
            if existing is None:
                nodes[subj_key][pred_key] = value
            elif isinstance(existing, list):
                existing.append(value)
            else:
                nodes[subj_key][pred_key] = [existing, value]

        graph_nodes = []
        for node in nodes.values():
            if "@id" not in node:
                continue
            graph_nodes.append(node)

        payload: dict[str, Any] = {
            "@context": _referenced_jsonld_context(context, graph_nodes),
            "@graph": graph_nodes,
        }
        try:
            return json.dumps(payload, indent=2, ensure_ascii=False)
        except (TypeError, ValueError) as exc:
            logger.warning(
                "Compact JSON-LD prompt serialization failed (%s); using rdflib json-ld",
                exc,
            )
            fallback = self.serialize(format="json-ld")
            if isinstance(fallback, bytes):
                return fallback.decode("utf-8")
            return str(fallback)

    def __new__(cls, *args, **kwargs):
        """Create a new RDFGraph instance."""
        instance = super().__new__(cls)
        return instance

    def serialize(
        self,
        destination: Any = None,
        format: str = "turtle",
        base: str | None = None,
        encoding: str | None = None,
        **args: Any,
    ) -> Any:
        """Serialize the graph, delegating to pyoxigraph for oxigraph stores.

        When the graph is backed by an *oxigraph* store and the requested
        format is any Turtle flavour (``"turtle"``, ``"ttl"`` or
        :data:`LOSSLESS_TURTLE_FORMAT`), serialisation is handled by
        ``pyoxigraph`` which natively supports RDF 1.2 triple terms.
        For all other stores or formats the default rdflib serialiser is
        used.

        The lossless flavour has to be routed here too: oxrdflib surfaces a
        pyoxigraph triple term as a plain Python tuple, which rdflib's Turtle
        writer cannot label (``'tuple' object has no attribute 'n3'``).
        pyoxigraph's own writer is value-preserving for floating-point
        literals, so nothing the lossless serializer exists for is given up.
        """
        is_ox = type(self.store).__name__ == "OxigraphStore"
        if is_ox and format in ("turtle", "ttl", LOSSLESS_TURTLE_FORMAT):
            ttl = self.serialize_turtle_star()
            if destination is not None:
                enc = encoding or "utf-8"
                with open(destination, "w", encoding=enc) as fh:
                    fh.write(ttl)
                return None
            return ttl
        return super().serialize(
            destination=destination,
            format=format,
            base=base,
            encoding=encoding,
            **args,
        )

    def update(
        self,
        update_object: Any,
        processor: Any = "sparql",
        initNs: Mapping[str, Any] | None = None,
        initBindings: Mapping[str, Any] | None = None,
        use_store_provided: bool = True,
        **kwargs: Any,
    ) -> None:
        """Execute SPARQL update using a base Graph view.

        rdflib's SPARQL update engine has internal checks that branch on exact
        ``Graph`` type, which can break for subclasses on ``INSERT/DELETE ... WHERE``.
        Running updates through a base ``Graph`` view avoids that edge case while
        still operating on the same underlying store/identifier.
        """
        graph_view = Graph(store=self.store, identifier=self.identifier)
        graph_view.namespace_manager = self.namespace_manager
        graph_view.update(
            update_object=update_object,
            processor=processor,
            initNs=initNs,
            initBindings=initBindings,
            use_store_provided=use_store_provided,
            **kwargs,
        )
        return None

    def sanitize_prefixes_namespaces(
        self,
        preferred_namespace_prefixes: dict[str, str] | None = None,
    ):
        """
        Rematches prefixes in an RDFLib graph to correct namespaces when a namespace
        with the same URI exists. Handles cases where prefixes might not be bound
        as namespaces.

        Args:
            preferred_namespace_prefixes: Optional namespace URI → preferred prefix
                name (e.g. catalog author prefixes). When multiple prefixes bind the
                same namespace, the preferred name wins over shortest-name heuristics.

        Returns:
           RDFGraph: The graph with corrected prefix-namespace mappings
        """
        ns_manager = self.namespace_manager
        current_prefixes = {
            prefix: str(uri) for prefix, uri in dict(ns_manager.namespaces()).items()
        }
        if not current_prefixes:
            return self

        # Preserve rdflib built-in bindings (e.g. xml:) unchanged — normalizing
        # them (appending /) invents a distinct URI and rdflib mints xml1:.
        mutable_prefixes = {
            prefix: uri
            for prefix, uri in current_prefixes.items()
            if not is_rdflib_default_namespace(uri)
        }
        preserved_prefixes = {
            prefix: uri
            for prefix, uri in current_prefixes.items()
            if is_rdflib_default_namespace(uri)
        }

        sanitized = sanitize_prefix_map(mutable_prefixes, context="auto")
        for prefix, original_namespace in mutable_prefixes.items():
            normalized_namespace = sanitized[prefix]
            if normalized_namespace != original_namespace:
                self.remap_namespaces(
                    old_namespace=original_namespace,
                    new_namespace=normalized_namespace,
                )

        merged = {**sanitized, **preserved_prefixes}
        new_ns_manager = NamespaceManager(self)
        uri_to_prefixes = defaultdict(list)
        for prefix, namespace in merged.items():
            uri_to_prefixes[namespace].append(prefix)

        # Offer digit-stripped stems as candidates: rdflib mints `schema1` when
        # `schema` is contested at parse time, and after namespace
        # canonicalization the stem is often free again. Only stems not claimed
        # by a different namespace are eligible.
        for namespace, prefixes in uri_to_prefixes.items():
            for prefix in list(prefixes):
                stem = prefix.rstrip("0123456789")
                if (
                    stem
                    and stem != prefix
                    and stem not in prefixes
                    and merged.get(stem, namespace) == namespace
                ):
                    prefixes.append(stem)

        for namespace, prefixes in uri_to_prefixes.items():
            best_prefix = choose_best_prefix(
                namespace,
                prefixes,
                preferred_namespace_prefixes=preferred_namespace_prefixes,
            )
            if is_rdflib_default_namespace(namespace):
                bound_ns = Namespace(namespace)
            else:
                bound_ns = Namespace(normalize_namespace_iri(namespace, context="auto"))
            # replace=True: this loop is authoritative — a prefix whose namespace
            # was canonicalized (e.g. http -> https schema.org) must be rebound,
            # not shadowed by a freshly minted `prefix1`. Losing prefixes are
            # rebound first so no stale (pre-canonicalization) binding survives
            # in the store; the best prefix is bound last so it wins the
            # namespace's reverse lookup used at serialization.
            for prefix in prefixes:
                if prefix != best_prefix:
                    new_ns_manager.bind(prefix, bound_ns, override=True, replace=True)
            new_ns_manager.bind(best_prefix, bound_ns, override=True, replace=True)
        self.namespace_manager = new_ns_manager
        return self

    def bind_implicit_namespaces(self, prefix_base: str | None = None) -> None:
        """Bind namespace prefixes for IRI stems used in the graph but not declared.

        Scans all URIRef terms in the graph, extracts their namespace stems
        (the IRI up to and including the last ``#`` or ``/``), and auto-binds
        any stem that appears in at least two IRIs but has no declared prefix.

        This is particularly useful for ontologies that use sub-namespaces
        (e.g. ``/concepts#`` and ``/relations#``) without explicit ``@prefix``
        declarations.  Without declared prefixes the LLM invents shortcuts like
        ``ont_10_culture:relations#P571`` which are syntactically invalid Turtle.

        Args:
            prefix_base: Optional string prepended to generated prefix names,
                typically the ``ontology_id``.  Produces prefixes of the form
                ``{prefix_base}_{slug}``; without it just ``{slug}`` is used.
        """
        declared_namespaces = {str(ns) for _, ns in self.namespaces()}
        standard_namespaces = {uri.strip("<>") for uri in COMMON_PREFIXES.values()}

        stem_counts: dict[str, int] = {}
        for s, p, o in self:
            for term in (s, p, o):
                if not isinstance(term, URIRef):
                    continue
                iri = str(term)
                if any(iri.startswith(std) for std in standard_namespaces):
                    continue
                for sep in ("#", "/"):
                    idx = iri.rfind(sep)
                    if idx > 0:
                        stem = iri[: idx + 1]
                        if stem not in declared_namespaces:
                            stem_counts[stem] = stem_counts.get(stem, 0) + 1
                        break

        for stem, count in stem_counts.items():
            if count < 2:
                continue
            # Skip stems that are a strict URI prefix of an already-declared namespace
            # — they are parent-directory IRIs, not domain namespaces.
            if any(
                other_ns != stem and other_ns.startswith(stem)
                for other_ns in declared_namespaces
            ):
                continue
            slug = stem.rstrip("#/").rsplit("/", 1)[-1].replace("-", "_")
            # An ontology's own namespace keeps the plain stem — "core", not
            # "core_core"; the base only disambiguates foreign stems.
            if prefix_base and prefix_base.replace("-", "_") != slug:
                prefix = f"{prefix_base}_{slug}"
            else:
                prefix = slug
            self.bind(prefix, Namespace(stem), override=False)

    def declared_prefix_map(self) -> dict[str, str]:
        """Read SHACL prefix declarations into a namespace → prefix map.

        Declarations are ``sh:declare [ sh:prefix "..." ; sh:namespace "..." ]``
        blank nodes, conventionally attached to the ``owl:Ontology`` subject.
        When a namespace carries several declarations, the lexically plainest
        prefix wins so the result is deterministic.

        Returns:
            dict[str, str]: Namespace IRI → declared prefix name.
        """
        pairs: list[tuple[str, str]] = []
        for _, _, decl in self.triples((None, SH.declare, None)):
            prefix = self.value(decl, SH.prefix)
            namespace = self.value(decl, SH.namespace)
            if prefix is not None and namespace is not None:
                pairs.append((str(namespace), str(prefix)))
        declared: dict[str, str] = {}
        for namespace, prefix in sorted(
            pairs, key=lambda item: (item[0], item[1].count("_"), len(item[1]), item[1])
        ):
            declared.setdefault(namespace, prefix)
        return declared

    def bind_declared_prefixes(self) -> dict[str, str]:
        """Bind prefixes recovered from ``sh:declare`` triples onto this graph.

        The inverse of :meth:`materialize_prefix_declarations`: after a triple
        store round trip the author's ``@prefix`` names exist only as declaration
        triples; this rebinds them so serialization and prompt-context prefix
        advertising show the author's names again.

        Returns:
            dict[str, str]: Namespace IRI → prefix that was bound.
        """
        declared = self.declared_prefix_map()
        for namespace, prefix in declared.items():
            self.bind(prefix, Namespace(namespace), override=True, replace=True)
        if declared:
            # The declaration triples themselves use the SHACL namespace; bind
            # its canonical name so implicit binding never mints a synthetic one.
            self.bind("sh", SH, override=False)
        return declared

    def materialize_prefix_declarations(self, ontology_iri: URIRef) -> int:
        """Persist author prefix bindings as SHACL declarations on ``ontology_iri``.

        Prefix bindings are serialization metadata: triple stores keep triples
        only, so ``@prefix`` names die at the store boundary and later exports
        must invent synthetic names (:meth:`bind_implicit_namespaces`). Writing
        ``sh:declare`` triples makes the author's names part of the graph content,
        surviving any RDF-preserving channel.

        Only namespaces actually used by a term in the graph are persisted;
        rdflib built-ins and namespaces recoverable from the canonical prefix
        tables are skipped. Idempotent: an already-declared namespace is left
        untouched, so round-tripped graphs never accumulate synthetic names on
        top of authorial ones.

        Args:
            ontology_iri: Subject to attach declarations to (the ontology IRI).

        Returns:
            int: Number of declarations added.
        """
        recoverable = set(prefix_lookup_for_ingest().values())
        already_declared = set(self.declared_prefix_map())

        used_namespaces: set[str] = set()
        candidate_namespaces = [
            str(namespace) for prefix, namespace in self.namespaces() if prefix
        ]
        for subj, pred, obj in self:
            terms = [subj, pred, obj]
            if isinstance(obj, Literal) and obj.datatype is not None:
                terms.append(obj.datatype)
            for term in terms:
                if not isinstance(term, URIRef):
                    continue
                iri = str(term)
                for namespace in candidate_namespaces:
                    if iri.startswith(namespace):
                        used_namespaces.add(namespace)

        candidates_by_ns: dict[str, list[str]] = {}
        for prefix, namespace in self.namespaces():
            ns_str = str(namespace)
            if (
                not prefix
                or is_rdflib_default_namespace(ns_str)
                or ns_str in recoverable
                or ns_str in already_declared
                or ns_str not in used_namespaces
            ):
                continue
            candidates_by_ns.setdefault(ns_str, []).append(prefix)

        added = 0
        for namespace in sorted(candidates_by_ns):
            prefix = min(
                candidates_by_ns[namespace],
                key=lambda p: (p.count("_"), len(p), p),
            )
            decl = BNode()
            self.add((ontology_iri, SH.declare, decl))
            self.add((decl, SH.prefix, Literal(prefix)))
            self.add((decl, SH.namespace, Literal(namespace, datatype=XSD.anyURI)))
            added += 1
        return added

    def unbind_chunk_namespaces(self, chunk_pattern="/chunk/") -> "RDFGraph":
        """
        Unbinds namespace prefixes that point to URIs containing a chunk pattern.
        Returns a new graph with chunk namespaces dereferenced (expanded to full URIs).
        Prefix target IRIs are normalized to end with ``#`` or ``/`` (``/`` appended
        when missing) before chunk detection and rebinding.

        Args:
            chunk_pattern (str): The pattern to look for in URIs (default: "/chunk/")

        Returns:
            RDFGraph: New graph with chunk-related namespaces unbound
        """
        current_prefixes = dict(self.namespace_manager.namespaces())

        # Normalize prefix target IRIs so chunk detection and rebinding agree on boundaries
        prefix_to_normalized: dict[str, str] = {
            prefix: normalize_namespace_iri(str(uri), context="auto")
            for prefix, uri in current_prefixes.items()
        }

        # Find prefixes that point to URIs containing the chunk pattern
        chunk_prefixes = []
        for prefix, uri_str in prefix_to_normalized.items():
            if chunk_pattern in uri_str:
                chunk_prefixes.append((prefix, uri_str))

        # Create new graph
        new_graph = RDFGraph()

        # Copy all triples (URIs are already expanded internally); an
        # oxigraph-backed source may hold RDF 1.2 triple terms a plain rdflib
        # graph cannot represent.
        copy_triples(self, new_graph, origin="RDFGraph.unbind_chunk_namespaces")

        # Bind only non-chunk namespace prefixes to the new graph
        for prefix, uri_str in prefix_to_normalized.items():
            if chunk_pattern not in uri_str:
                new_graph.bind(prefix, Namespace(uri_str))

        # Log what was removed
        if chunk_prefixes:
            logger.debug(f"Unbound {len(chunk_prefixes)} chunk-related namespace(s):")
            for prefix, uri in chunk_prefixes:
                logger.debug(f"  - '{prefix}': {uri}")

        return new_graph

    def remap_namespaces(self, old_namespace, new_namespace) -> None:
        updates = {}
        for s, p, o in self:
            new_s, new_p, new_o = s, p, o
            if isinstance(s, URIRef) and str(s).startswith(str(old_namespace)):
                new_s = URIRef(
                    str(s).replace(str(old_namespace), str(new_namespace), 1)
                )
            if isinstance(p, URIRef) and str(p).startswith(str(old_namespace)):
                new_p = URIRef(
                    str(p).replace(str(old_namespace), str(new_namespace), 1)
                )
            if isinstance(o, URIRef) and str(o).startswith(str(old_namespace)):
                new_o = URIRef(
                    str(o).replace(str(old_namespace), str(new_namespace), 1)
                )

            if (new_s, new_p, new_o) != (s, p, o):
                updates[(s, p, o)] = (new_s, new_p, new_o)

        for (s, p, o), (new_s, new_p, new_o) in updates.items():
            self.remove((s, p, o))
            self.add((new_s, new_p, new_o))

    def add_triple(self, subject: str, predicate: str, object_: str) -> None:
        """Add a triple to the graph.

        Args:
            subject: Subject URI as string
            predicate: Predicate URI as string
            object_: Object URI as string or literal value
        """
        # Convert strings to appropriate RDFLib objects
        subj = URIRef(subject)
        pred = URIRef(predicate)

        # Handle object - could be URI or literal
        if object_.startswith("http://") or object_.startswith("https://"):
            obj = URIRef(object_)
        else:
            # Treat as literal
            obj = Literal(object_)

        self.add((subj, pred, obj))
        logger.debug(f"Added triple: {subj} {pred} {obj}")

    def remove_triple(self, subject: str, predicate: str, object_: str) -> None:
        """Remove a triple from the graph.

        Args:
            subject: Subject URI as string
            predicate: Predicate URI as string
            object_: Object URI as string or literal value
        """
        # Convert strings to appropriate RDFLib objects
        subj = URIRef(subject)
        pred = URIRef(predicate)

        # Handle object - could be URI or literal
        if object_.startswith("http://") or object_.startswith("https://"):
            obj = URIRef(object_)
        else:
            # Treat as literal
            obj = Literal(object_)

        self.remove((subj, pred, obj))
        logger.debug(f"Removed triple: {subj} {pred} {obj}")

    def hash(self: Graph) -> str:
        """Return the SHA-256 content hash of this graph.

        The hash is taken over the RDF **value** space, not the lexical space:
        literals are canonicalized with :func:`canonical_literal` before
        URDNA2015 runs, because URDNA2015 canonicalizes blank node labels only.
        Without that step a graph re-hashes differently after a triple-store
        round trip -- stores normalise literals on insert -- which breaks the
        content-addressed ``versioned_iri`` identity the catalog is built on.

        Canonicalization is applied to a throwaway copy; the graph itself is
        never mutated, so what gets stored, served, and shown to the LLM is
        unaffected.

        Returns:
            str: Full 64-character hex digest.
        """
        canonical = Graph()
        for subject, predicate, object_ in self:
            canonical.add((subject, predicate, canonical_literal(object_)))
        for prefix, namespace in self.namespaces():
            canonical.bind(prefix, namespace)

        # Serialize to JSON-LD
        data = canonical.serialize(format="json-ld")

        # Parse the JSON string
        doc = json.loads(data)

        # Canonicalize using URDNA2015 normalization
        normalized = jsonld.normalize(
            doc,
            {"algorithm": "URDNA2015", "format": "application/n-quads"},
        )
        # jsonld.normalize returns a string when format is "application/n-quads"
        normalized_str = normalized if isinstance(normalized, str) else str(normalized)
        return render_text_hash(normalized_str, digits=None)

__add__(other)

Addition operator for RDFGraph instances.

Merges the RDF graphs while maintaining the RDFGraph type.

Parameters:

Name	Type	Description	Default
other	Union[RDFGraph, Graph, Iterable]	The graph to add to this one.	required

Returns:

Name	Type	Description
RDFGraph	RDFGraph	A new RDFGraph containing the merged triples.

Source code in ontocast/onto/rdfgraph.py

def __add__(self, other: Union["RDFGraph", Graph, Iterable]) -> "RDFGraph":
    """Addition operator for RDFGraph instances.

    Merges the RDF graphs while maintaining the RDFGraph type.

    Args:
        other: The graph to add to this one.

    Returns:
        RDFGraph: A new RDFGraph containing the merged triples.
    """
    # Create a new RDFGraph instance
    result = RDFGraph()

    # Copy all triples from both graphs
    copy_triples(self, result, origin="RDFGraph.__add__")
    copy_triples(other, result, origin="RDFGraph.__add__")

    existing = {prefix: str(uri) for prefix, uri in self.namespaces() if prefix}
    incoming: dict[str, str] = {}
    if isinstance(other, Graph):
        incoming = {
            prefix: str(uri) for prefix, uri in other.namespaces() if prefix
        }
    for prefix, uri in merge_namespace_bindings(existing, incoming).items():
        result.bind(prefix, uri)

    return result

__copy__()

Ensure shallow copies preserve RDFGraph type.

Source code in ontocast/onto/rdfgraph.py

def __copy__(self) -> "RDFGraph":
    """Ensure shallow copies preserve RDFGraph type."""
    return self.copy()

__deepcopy__(memo)

Ensure deep copies preserve RDFGraph type.

Source code in ontocast/onto/rdfgraph.py

def __deepcopy__(self, memo: dict[int, Any]) -> "RDFGraph":
    """Ensure deep copies preserve RDFGraph type."""
    copied = self.copy()
    memo[id(self)] = copied
    return copied

__get_pydantic_core_schema__(_source_type, handler) classmethod

Get the Pydantic core schema for this class.

Parameters:

Name	Type	Description	Default
_source_type		The source type.	required
handler	GetCoreSchemaHandler	The core schema handler.	required

Returns:

Type	Description
	A union schema accepting:
	existing RDFGraph instances,
	Turtle / JSON-LD strings (parsed via _from_str),
	JSON-LD dict / list objects (parsed via _from_jsonld_obj).

Source code in ontocast/onto/rdfgraph.py

@classmethod
def __get_pydantic_core_schema__(cls, _source_type, handler: GetCoreSchemaHandler):
    """Get the Pydantic core schema for this class.

    Args:
        _source_type: The source type.
        handler: The core schema handler.

    Returns:
        A union schema accepting:
        - existing ``RDFGraph`` instances,
        - Turtle / JSON-LD strings (parsed via ``_from_str``),
        - JSON-LD ``dict`` / ``list`` objects (parsed via ``_from_jsonld_obj``).
    """
    return core_schema.union_schema(
        [
            core_schema.is_instance_schema(cls),
            core_schema.chain_schema(
                [
                    core_schema.str_schema(),
                    core_schema.no_info_plain_validator_function(cls._from_str),
                ]
            ),
            core_schema.no_info_plain_validator_function(cls._from_any),
        ],
        serialization=core_schema.plain_serializer_function_ser_schema(
            cls._to_turtle_str,
            info_arg=False,
            return_schema=core_schema.str_schema(),
        ),
    )

__iadd__(other)

In-place addition operator for RDFGraph instances.

Merges the RDF graphs while maintaining the RDFGraph type and binding prefixes.

Parameters:

Name	Type	Description	Default
other	Union[RDFGraph, Graph, Iterable]	The graph to add to this one.	required

Returns:

Name	Type	Description
RDFGraph	RDFGraph	self after modification.

Source code in ontocast/onto/rdfgraph.py

def __iadd__(self, other: Union["RDFGraph", Graph, Iterable]) -> "RDFGraph":
    """In-place addition operator for RDFGraph instances.

    Merges the RDF graphs while maintaining the RDFGraph type and binding prefixes.

    Args:
        other: The graph to add to this one.

    Returns:
        RDFGraph: self after modification.
    """
    # Use __add__ to get the merged result with proper prefix binding
    result = self.__add__(other)

    # Clear current graph and copy the result
    self.remove((None, None, None))  # Remove all triples

    # Copy all triples from result
    copy_triples(result, self, origin="RDFGraph.__iadd__")

    # Copy namespace bindings from result
    for prefix, uri in result.namespaces():
        self.bind(prefix, uri)

    return self

__new__(*args, **kwargs)

Create a new RDFGraph instance.

Source code in ontocast/onto/rdfgraph.py

def __new__(cls, *args, **kwargs):
    """Create a new RDFGraph instance."""
    instance = super().__new__(cls)
    return instance

add_triple(subject, predicate, object_)

Add a triple to the graph.

Parameters:

Name	Type	Description	Default
subject	str	Subject URI as string	required
predicate	str	Predicate URI as string	required
object_	str	Object URI as string or literal value	required

Source code in ontocast/onto/rdfgraph.py

def add_triple(self, subject: str, predicate: str, object_: str) -> None:
    """Add a triple to the graph.

    Args:
        subject: Subject URI as string
        predicate: Predicate URI as string
        object_: Object URI as string or literal value
    """
    # Convert strings to appropriate RDFLib objects
    subj = URIRef(subject)
    pred = URIRef(predicate)

    # Handle object - could be URI or literal
    if object_.startswith("http://") or object_.startswith("https://"):
        obj = URIRef(object_)
    else:
        # Treat as literal
        obj = Literal(object_)

    self.add((subj, pred, obj))
    logger.debug(f"Added triple: {subj} {pred} {obj}")

bind_declared_prefixes()

Bind prefixes recovered from sh:declare triples onto this graph.

The inverse of :meth:materialize_prefix_declarations: after a triple store round trip the author's @prefix names exist only as declaration triples; this rebinds them so serialization and prompt-context prefix advertising show the author's names again.

Returns:

Type	Description
dict[str, str]	dict[str, str]: Namespace IRI → prefix that was bound.

Source code in ontocast/onto/rdfgraph.py

def bind_declared_prefixes(self) -> dict[str, str]:
    """Bind prefixes recovered from ``sh:declare`` triples onto this graph.

    The inverse of :meth:`materialize_prefix_declarations`: after a triple
    store round trip the author's ``@prefix`` names exist only as declaration
    triples; this rebinds them so serialization and prompt-context prefix
    advertising show the author's names again.

    Returns:
        dict[str, str]: Namespace IRI → prefix that was bound.
    """
    declared = self.declared_prefix_map()
    for namespace, prefix in declared.items():
        self.bind(prefix, Namespace(namespace), override=True, replace=True)
    if declared:
        # The declaration triples themselves use the SHACL namespace; bind
        # its canonical name so implicit binding never mints a synthetic one.
        self.bind("sh", SH, override=False)
    return declared

bind_implicit_namespaces(prefix_base=None)

Bind namespace prefixes for IRI stems used in the graph but not declared.

Scans all URIRef terms in the graph, extracts their namespace stems (the IRI up to and including the last # or /), and auto-binds any stem that appears in at least two IRIs but has no declared prefix.

This is particularly useful for ontologies that use sub-namespaces (e.g. /concepts# and /relations#) without explicit @prefix declarations. Without declared prefixes the LLM invents shortcuts like ont_10_culture:relations#P571 which are syntactically invalid Turtle.

Parameters:

Name	Type	Description	Default
prefix_base	str | None	Optional string prepended to generated prefix names, typically the ontology_id. Produces prefixes of the form {prefix_base}_{slug}; without it just {slug} is used.	None

Source code in ontocast/onto/rdfgraph.py

def bind_implicit_namespaces(self, prefix_base: str | None = None) -> None:
    """Bind namespace prefixes for IRI stems used in the graph but not declared.

    Scans all URIRef terms in the graph, extracts their namespace stems
    (the IRI up to and including the last ``#`` or ``/``), and auto-binds
    any stem that appears in at least two IRIs but has no declared prefix.

    This is particularly useful for ontologies that use sub-namespaces
    (e.g. ``/concepts#`` and ``/relations#``) without explicit ``@prefix``
    declarations.  Without declared prefixes the LLM invents shortcuts like
    ``ont_10_culture:relations#P571`` which are syntactically invalid Turtle.

    Args:
        prefix_base: Optional string prepended to generated prefix names,
            typically the ``ontology_id``.  Produces prefixes of the form
            ``{prefix_base}_{slug}``; without it just ``{slug}`` is used.
    """
    declared_namespaces = {str(ns) for _, ns in self.namespaces()}
    standard_namespaces = {uri.strip("<>") for uri in COMMON_PREFIXES.values()}

    stem_counts: dict[str, int] = {}
    for s, p, o in self:
        for term in (s, p, o):
            if not isinstance(term, URIRef):
                continue
            iri = str(term)
            if any(iri.startswith(std) for std in standard_namespaces):
                continue
            for sep in ("#", "/"):
                idx = iri.rfind(sep)
                if idx > 0:
                    stem = iri[: idx + 1]
                    if stem not in declared_namespaces:
                        stem_counts[stem] = stem_counts.get(stem, 0) + 1
                    break

    for stem, count in stem_counts.items():
        if count < 2:
            continue
        # Skip stems that are a strict URI prefix of an already-declared namespace
        # — they are parent-directory IRIs, not domain namespaces.
        if any(
            other_ns != stem and other_ns.startswith(stem)
            for other_ns in declared_namespaces
        ):
            continue
        slug = stem.rstrip("#/").rsplit("/", 1)[-1].replace("-", "_")
        # An ontology's own namespace keeps the plain stem — "core", not
        # "core_core"; the base only disambiguates foreign stems.
        if prefix_base and prefix_base.replace("-", "_") != slug:
            prefix = f"{prefix_base}_{slug}"
        else:
            prefix = slug
        self.bind(prefix, Namespace(stem), override=False)

copy()

Create a copy of this RDFGraph.

Returns:

Name	Type	Description
RDFGraph	RDFGraph	A new RDFGraph instance with all triples and namespace bindings copied.

Source code in ontocast/onto/rdfgraph.py

def copy(self) -> "RDFGraph":
    """Create a copy of this RDFGraph.

    Returns:
        RDFGraph: A new RDFGraph instance with all triples and namespace bindings copied.
    """
    result = RDFGraph()

    # Copy all triples. An oxigraph-backed graph may hold RDF 1.2 triple
    # terms, which a plain rdflib graph cannot represent -- this is the path
    # __deepcopy__ takes, so it degrades instead of raising.
    copy_triples(self, result, origin="RDFGraph.copy")

    # Copy namespace bindings
    for prefix, uri in self.namespaces():
        result.bind(prefix, uri)

    return result

declared_prefix_map()

Read SHACL prefix declarations into a namespace → prefix map.

Declarations are sh:declare [ sh:prefix "..." ; sh:namespace "..." ] blank nodes, conventionally attached to the owl:Ontology subject. When a namespace carries several declarations, the lexically plainest prefix wins so the result is deterministic.

Returns:

Type	Description
dict[str, str]	dict[str, str]: Namespace IRI → declared prefix name.

Source code in ontocast/onto/rdfgraph.py

def declared_prefix_map(self) -> dict[str, str]:
    """Read SHACL prefix declarations into a namespace → prefix map.

    Declarations are ``sh:declare [ sh:prefix "..." ; sh:namespace "..." ]``
    blank nodes, conventionally attached to the ``owl:Ontology`` subject.
    When a namespace carries several declarations, the lexically plainest
    prefix wins so the result is deterministic.

    Returns:
        dict[str, str]: Namespace IRI → declared prefix name.
    """
    pairs: list[tuple[str, str]] = []
    for _, _, decl in self.triples((None, SH.declare, None)):
        prefix = self.value(decl, SH.prefix)
        namespace = self.value(decl, SH.namespace)
        if prefix is not None and namespace is not None:
            pairs.append((str(namespace), str(prefix)))
    declared: dict[str, str] = {}
    for namespace, prefix in sorted(
        pairs, key=lambda item: (item[0], item[1].count("_"), len(item[1]), item[1])
    ):
        declared.setdefault(namespace, prefix)
    return declared

get_known_prefixes() classmethod

Get currently known prefixes from context.

Returns:

Type	Description
dict[str, str] | None	Dictionary mapping prefix names to namespace URIs, or None.

Source code in ontocast/onto/rdfgraph.py

@classmethod
def get_known_prefixes(cls) -> dict[str, str] | None:
    """Get currently known prefixes from context.

    Returns:
        Dictionary mapping prefix names to namespace URIs, or None.
    """
    return _known_prefixes_context.get()

hash()

Return the SHA-256 content hash of this graph.

The hash is taken over the RDF value space, not the lexical space: literals are canonicalized with :func:canonical_literal before URDNA2015 runs, because URDNA2015 canonicalizes blank node labels only. Without that step a graph re-hashes differently after a triple-store round trip -- stores normalise literals on insert -- which breaks the content-addressed versioned_iri identity the catalog is built on.

Canonicalization is applied to a throwaway copy; the graph itself is never mutated, so what gets stored, served, and shown to the LLM is unaffected.

Returns:

Name	Type	Description
str	str	Full 64-character hex digest.

Source code in ontocast/onto/rdfgraph.py

def hash(self: Graph) -> str:
    """Return the SHA-256 content hash of this graph.

    The hash is taken over the RDF **value** space, not the lexical space:
    literals are canonicalized with :func:`canonical_literal` before
    URDNA2015 runs, because URDNA2015 canonicalizes blank node labels only.
    Without that step a graph re-hashes differently after a triple-store
    round trip -- stores normalise literals on insert -- which breaks the
    content-addressed ``versioned_iri`` identity the catalog is built on.

    Canonicalization is applied to a throwaway copy; the graph itself is
    never mutated, so what gets stored, served, and shown to the LLM is
    unaffected.

    Returns:
        str: Full 64-character hex digest.
    """
    canonical = Graph()
    for subject, predicate, object_ in self:
        canonical.add((subject, predicate, canonical_literal(object_)))
    for prefix, namespace in self.namespaces():
        canonical.bind(prefix, namespace)

    # Serialize to JSON-LD
    data = canonical.serialize(format="json-ld")

    # Parse the JSON string
    doc = json.loads(data)

    # Canonicalize using URDNA2015 normalization
    normalized = jsonld.normalize(
        doc,
        {"algorithm": "URDNA2015", "format": "application/n-quads"},
    )
    # jsonld.normalize returns a string when format is "application/n-quads"
    normalized_str = normalized if isinstance(normalized, str) else str(normalized)
    return render_text_hash(normalized_str, digits=None)

materialize_prefix_declarations(ontology_iri)

Persist author prefix bindings as SHACL declarations on ontology_iri.

Prefix bindings are serialization metadata: triple stores keep triples only, so @prefix names die at the store boundary and later exports must invent synthetic names (:meth:bind_implicit_namespaces). Writing sh:declare triples makes the author's names part of the graph content, surviving any RDF-preserving channel.

Only namespaces actually used by a term in the graph are persisted; rdflib built-ins and namespaces recoverable from the canonical prefix tables are skipped. Idempotent: an already-declared namespace is left untouched, so round-tripped graphs never accumulate synthetic names on top of authorial ones.

Parameters:

Name	Type	Description	Default
ontology_iri	URIRef	Subject to attach declarations to (the ontology IRI).	required

Returns:

Name	Type	Description
int	int	Number of declarations added.

Source code in ontocast/onto/rdfgraph.py

def materialize_prefix_declarations(self, ontology_iri: URIRef) -> int:
    """Persist author prefix bindings as SHACL declarations on ``ontology_iri``.

    Prefix bindings are serialization metadata: triple stores keep triples
    only, so ``@prefix`` names die at the store boundary and later exports
    must invent synthetic names (:meth:`bind_implicit_namespaces`). Writing
    ``sh:declare`` triples makes the author's names part of the graph content,
    surviving any RDF-preserving channel.

    Only namespaces actually used by a term in the graph are persisted;
    rdflib built-ins and namespaces recoverable from the canonical prefix
    tables are skipped. Idempotent: an already-declared namespace is left
    untouched, so round-tripped graphs never accumulate synthetic names on
    top of authorial ones.

    Args:
        ontology_iri: Subject to attach declarations to (the ontology IRI).

    Returns:
        int: Number of declarations added.
    """
    recoverable = set(prefix_lookup_for_ingest().values())
    already_declared = set(self.declared_prefix_map())

    used_namespaces: set[str] = set()
    candidate_namespaces = [
        str(namespace) for prefix, namespace in self.namespaces() if prefix
    ]
    for subj, pred, obj in self:
        terms = [subj, pred, obj]
        if isinstance(obj, Literal) and obj.datatype is not None:
            terms.append(obj.datatype)
        for term in terms:
            if not isinstance(term, URIRef):
                continue
            iri = str(term)
            for namespace in candidate_namespaces:
                if iri.startswith(namespace):
                    used_namespaces.add(namespace)

    candidates_by_ns: dict[str, list[str]] = {}
    for prefix, namespace in self.namespaces():
        ns_str = str(namespace)
        if (
            not prefix
            or is_rdflib_default_namespace(ns_str)
            or ns_str in recoverable
            or ns_str in already_declared
            or ns_str not in used_namespaces
        ):
            continue
        candidates_by_ns.setdefault(ns_str, []).append(prefix)

    added = 0
    for namespace in sorted(candidates_by_ns):
        prefix = min(
            candidates_by_ns[namespace],
            key=lambda p: (p.count("_"), len(p), p),
        )
        decl = BNode()
        self.add((ontology_iri, SH.declare, decl))
        self.add((decl, SH.prefix, Literal(prefix)))
        self.add((decl, SH.namespace, Literal(namespace, datatype=XSD.anyURI)))
        added += 1
    return added

partition_invalid_typed_literals(graph) classmethod

Split graph into a clean graph and quarantined invalid typed literals.

Source code in ontocast/onto/rdfgraph.py

@classmethod
def partition_invalid_typed_literals(
    cls, graph: "RDFGraph"
) -> tuple["RDFGraph", list[RejectedLiteralTriple]]:
    """Split *graph* into a clean graph and quarantined invalid typed literals."""
    clean = cls()
    for prefix, namespace_uri in graph.namespaces():
        if prefix:
            clean.bind(prefix, namespace_uri)

    rejected: list[RejectedLiteralTriple] = []
    for subject, predicate, obj in graph:
        if isinstance(obj, Literal) and not cls._is_valid_typed_literal(obj):
            rejected.append(
                RejectedLiteralTriple(
                    subject=str(subject),
                    predicate=str(predicate),
                    object_lexical=str(obj),
                    datatype=str(obj.datatype),
                )
            )
            continue
        clean.add((subject, predicate, obj))

    if rejected:
        logger.warning(
            "Quarantined %d triple(s) with invalid XSD typed literals",
            len(rejected),
        )

    return clean, rejected

remove_triple(subject, predicate, object_)

Remove a triple from the graph.

Parameters:

Name	Type	Description	Default
subject	str	Subject URI as string	required
predicate	str	Predicate URI as string	required
object_	str	Object URI as string or literal value	required

Source code in ontocast/onto/rdfgraph.py

def remove_triple(self, subject: str, predicate: str, object_: str) -> None:
    """Remove a triple from the graph.

    Args:
        subject: Subject URI as string
        predicate: Predicate URI as string
        object_: Object URI as string or literal value
    """
    # Convert strings to appropriate RDFLib objects
    subj = URIRef(subject)
    pred = URIRef(predicate)

    # Handle object - could be URI or literal
    if object_.startswith("http://") or object_.startswith("https://"):
        obj = URIRef(object_)
    else:
        # Treat as literal
        obj = Literal(object_)

    self.remove((subj, pred, obj))
    logger.debug(f"Removed triple: {subj} {pred} {obj}")

sanitize_prefixes_namespaces(preferred_namespace_prefixes=None)

Rematches prefixes in an RDFLib graph to correct namespaces when a namespace with the same URI exists. Handles cases where prefixes might not be bound as namespaces.

Parameters:

Name	Type	Description	Default
preferred_namespace_prefixes	dict[str, str] | None	Optional namespace URI → preferred prefix name (e.g. catalog author prefixes). When multiple prefixes bind the same namespace, the preferred name wins over shortest-name heuristics.	None

Returns:

Name	Type	Description
RDFGraph		The graph with corrected prefix-namespace mappings

Source code in ontocast/onto/rdfgraph.py

def sanitize_prefixes_namespaces(
    self,
    preferred_namespace_prefixes: dict[str, str] | None = None,
):
    """
    Rematches prefixes in an RDFLib graph to correct namespaces when a namespace
    with the same URI exists. Handles cases where prefixes might not be bound
    as namespaces.

    Args:
        preferred_namespace_prefixes: Optional namespace URI → preferred prefix
            name (e.g. catalog author prefixes). When multiple prefixes bind the
            same namespace, the preferred name wins over shortest-name heuristics.

    Returns:
       RDFGraph: The graph with corrected prefix-namespace mappings
    """
    ns_manager = self.namespace_manager
    current_prefixes = {
        prefix: str(uri) for prefix, uri in dict(ns_manager.namespaces()).items()
    }
    if not current_prefixes:
        return self

    # Preserve rdflib built-in bindings (e.g. xml:) unchanged — normalizing
    # them (appending /) invents a distinct URI and rdflib mints xml1:.
    mutable_prefixes = {
        prefix: uri
        for prefix, uri in current_prefixes.items()
        if not is_rdflib_default_namespace(uri)
    }
    preserved_prefixes = {
        prefix: uri
        for prefix, uri in current_prefixes.items()
        if is_rdflib_default_namespace(uri)
    }

    sanitized = sanitize_prefix_map(mutable_prefixes, context="auto")
    for prefix, original_namespace in mutable_prefixes.items():
        normalized_namespace = sanitized[prefix]
        if normalized_namespace != original_namespace:
            self.remap_namespaces(
                old_namespace=original_namespace,
                new_namespace=normalized_namespace,
            )

    merged = {**sanitized, **preserved_prefixes}
    new_ns_manager = NamespaceManager(self)
    uri_to_prefixes = defaultdict(list)
    for prefix, namespace in merged.items():
        uri_to_prefixes[namespace].append(prefix)

    # Offer digit-stripped stems as candidates: rdflib mints `schema1` when
    # `schema` is contested at parse time, and after namespace
    # canonicalization the stem is often free again. Only stems not claimed
    # by a different namespace are eligible.
    for namespace, prefixes in uri_to_prefixes.items():
        for prefix in list(prefixes):
            stem = prefix.rstrip("0123456789")
            if (
                stem
                and stem != prefix
                and stem not in prefixes
                and merged.get(stem, namespace) == namespace
            ):
                prefixes.append(stem)

    for namespace, prefixes in uri_to_prefixes.items():
        best_prefix = choose_best_prefix(
            namespace,
            prefixes,
            preferred_namespace_prefixes=preferred_namespace_prefixes,
        )
        if is_rdflib_default_namespace(namespace):
            bound_ns = Namespace(namespace)
        else:
            bound_ns = Namespace(normalize_namespace_iri(namespace, context="auto"))
        # replace=True: this loop is authoritative — a prefix whose namespace
        # was canonicalized (e.g. http -> https schema.org) must be rebound,
        # not shadowed by a freshly minted `prefix1`. Losing prefixes are
        # rebound first so no stale (pre-canonicalization) binding survives
        # in the store; the best prefix is bound last so it wins the
        # namespace's reverse lookup used at serialization.
        for prefix in prefixes:
            if prefix != best_prefix:
                new_ns_manager.bind(prefix, bound_ns, override=True, replace=True)
        new_ns_manager.bind(best_prefix, bound_ns, override=True, replace=True)
    self.namespace_manager = new_ns_manager
    return self

serialize(destination=None, format='turtle', base=None, encoding=None, **args)

Serialize the graph, delegating to pyoxigraph for oxigraph stores.

When the graph is backed by an oxigraph store and the requested format is any Turtle flavour ("turtle", "ttl" or :data:LOSSLESS_TURTLE_FORMAT), serialisation is handled by pyoxigraph which natively supports RDF 1.2 triple terms. For all other stores or formats the default rdflib serialiser is used.

The lossless flavour has to be routed here too: oxrdflib surfaces a pyoxigraph triple term as a plain Python tuple, which rdflib's Turtle writer cannot label ('tuple' object has no attribute 'n3'). pyoxigraph's own writer is value-preserving for floating-point literals, so nothing the lossless serializer exists for is given up.

Source code in ontocast/onto/rdfgraph.py

def serialize(
    self,
    destination: Any = None,
    format: str = "turtle",
    base: str | None = None,
    encoding: str | None = None,
    **args: Any,
) -> Any:
    """Serialize the graph, delegating to pyoxigraph for oxigraph stores.

    When the graph is backed by an *oxigraph* store and the requested
    format is any Turtle flavour (``"turtle"``, ``"ttl"`` or
    :data:`LOSSLESS_TURTLE_FORMAT`), serialisation is handled by
    ``pyoxigraph`` which natively supports RDF 1.2 triple terms.
    For all other stores or formats the default rdflib serialiser is
    used.

    The lossless flavour has to be routed here too: oxrdflib surfaces a
    pyoxigraph triple term as a plain Python tuple, which rdflib's Turtle
    writer cannot label (``'tuple' object has no attribute 'n3'``).
    pyoxigraph's own writer is value-preserving for floating-point
    literals, so nothing the lossless serializer exists for is given up.
    """
    is_ox = type(self.store).__name__ == "OxigraphStore"
    if is_ox and format in ("turtle", "ttl", LOSSLESS_TURTLE_FORMAT):
        ttl = self.serialize_turtle_star()
        if destination is not None:
            enc = encoding or "utf-8"
            with open(destination, "w", encoding=enc) as fh:
                fh.write(ttl)
            return None
        return ttl
    return super().serialize(
        destination=destination,
        format=format,
        base=base,
        encoding=encoding,
        **args,
    )

serialize_canonical_turtle()

Serialize to Turtle after canonical namespace/prefix sanitization.

Uses :class:_LosslessTurtleSerializer rather than rdflib's stock Turtle writer, which rounds floating-point literals to 7 significant digits. This output is what reaches the triple store, the API response, and the LLM prompt, so the rounding was real value loss rather than formatting.

Source code in ontocast/onto/rdfgraph.py

def serialize_canonical_turtle(self) -> str:
    """Serialize to Turtle after canonical namespace/prefix sanitization.

    Uses :class:`_LosslessTurtleSerializer` rather than rdflib's stock Turtle
    writer, which rounds floating-point literals to 7 significant digits.
    This output is what reaches the triple store, the API response, and the
    LLM prompt, so the rounding was real value loss rather than formatting.
    """
    self.sanitize_prefixes_namespaces()
    serialized = self.serialize(format=LOSSLESS_TURTLE_FORMAT)
    if isinstance(serialized, bytes):
        return serialized.decode("utf-8")
    return str(serialized)

serialize_compact_jsonld_for_prompt()

Serialize graph as compact JSON-LD text for LLM context prompts.

Source code in ontocast/onto/rdfgraph.py

def serialize_compact_jsonld_for_prompt(self) -> str:
    """Serialize graph as compact JSON-LD text for LLM context prompts."""
    self.sanitize_prefixes_namespaces()
    context: dict[str, str] = {}
    for prefix, namespace in self.namespaces():
        if prefix:
            context[prefix] = str(namespace)

    nodes: dict[str, dict[str, Any]] = defaultdict(dict)
    for subject, predicate, obj in self:
        subj_key = str(subject)
        if "@id" not in nodes[subj_key]:
            subj_compact = self._compact_iri_for_jsonld(subject)
            if isinstance(subj_compact, dict):
                nodes[subj_key].update(subj_compact)
            else:
                nodes[subj_key]["@id"] = subj_compact

        pred_compact = self._compact_iri_for_jsonld(predicate)
        if not isinstance(pred_compact, str):
            pred_key = str(predicate)
        else:
            pred_key = pred_compact

        value = self._object_to_jsonld(obj)
        existing = nodes[subj_key].get(pred_key)
        if existing is None:
            nodes[subj_key][pred_key] = value
        elif isinstance(existing, list):
            existing.append(value)
        else:
            nodes[subj_key][pred_key] = [existing, value]

    graph_nodes = []
    for node in nodes.values():
        if "@id" not in node:
            continue
        graph_nodes.append(node)

    payload: dict[str, Any] = {
        "@context": _referenced_jsonld_context(context, graph_nodes),
        "@graph": graph_nodes,
    }
    try:
        return json.dumps(payload, indent=2, ensure_ascii=False)
    except (TypeError, ValueError) as exc:
        logger.warning(
            "Compact JSON-LD prompt serialization failed (%s); using rdflib json-ld",
            exc,
        )
        fallback = self.serialize(format="json-ld")
        if isinstance(fallback, bytes):
            return fallback.decode("utf-8")
        return str(fallback)

serialize_turtle_star()

Serialize an oxigraph-backed graph to Turtle-star via pyoxigraph.

This method extracts all quads belonging to this graph's context from the underlying pyoxigraph.Store and serialises them into the default graph using pyoxigraph.serialize with the Turtle format, which natively supports RDF 1.2 <<( … )>> syntax.

Returns:

Type	Description
str	Turtle-star string.

Raises:

Type	Description
RuntimeError	If the graph is not backed by an oxigraph store.

Source code in ontocast/onto/rdfgraph.py

def serialize_turtle_star(self) -> str:
    """Serialize an oxigraph-backed graph to Turtle-star via *pyoxigraph*.

    This method extracts all quads belonging to this graph's context
    from the underlying ``pyoxigraph.Store`` and serialises them into
    the default graph using ``pyoxigraph.serialize`` with the Turtle
    format, which natively supports RDF 1.2 ``<<( … )>>`` syntax.

    Returns:
        Turtle-star string.

    Raises:
        RuntimeError: If the graph is not backed by an oxigraph store.
    """
    try:
        import pyoxigraph as ox
        from oxrdflib._converter import to_ox
    except ImportError as exc:
        raise RuntimeError(
            "pyoxigraph / oxrdflib must be installed for Turtle-star serialisation"
        ) from exc

    inner_store = cast(ox.Store, _oxigraph_inner_store(self.store))
    graph_ctx_raw = to_ox(self.identifier)
    assert isinstance(
        graph_ctx_raw,
        (ox.NamedNode, ox.BlankNode, ox.DefaultGraph),
    )
    graph_ctx: ox.NamedNode | ox.BlankNode | ox.DefaultGraph = graph_ctx_raw

    # Copy quads into a temporary store under the default graph so
    # that ``ox.serialize`` can emit plain Turtle (Turtle-star).
    tmp = ox.Store()
    used_iri_terms: set[str] = set()

    def _collect_used_iris(term: Any) -> None:
        if isinstance(term, ox.NamedNode):
            used_iri_terms.add(term.value)
            return
        if isinstance(term, ox.Triple):
            _collect_used_iris(term.subject)
            _collect_used_iris(term.predicate)
            _collect_used_iris(term.object)

    for quad in inner_store.quads_for_pattern(
        None,
        None,
        None,
        graph_ctx,
    ):
        _collect_used_iris(quad.subject)
        _collect_used_iris(quad.predicate)
        _collect_used_iris(quad.object)
        tmp.add(
            ox.Quad(quad.subject, quad.predicate, quad.object, ox.DefaultGraph())
        )

    namespace_to_prefix: dict[str, str] = {}
    for prefix, namespace in self.namespaces():
        if not prefix:
            continue
        prefix_str = str(prefix)
        namespace_str = str(namespace)
        current = namespace_to_prefix.get(namespace_str)
        if current is None or (len(prefix_str), prefix_str) < (
            len(current),
            current,
        ):
            namespace_to_prefix[namespace_str] = prefix_str

    prefixes = {
        prefix: namespace
        for namespace, prefix in namespace_to_prefix.items()
        if any(iri.startswith(namespace) for iri in used_iri_terms)
    }
    raw = tmp.dump(
        format=ox.RdfFormat.TURTLE,
        from_graph=ox.DefaultGraph(),
        prefixes=prefixes or None,
    )
    if raw is None:
        raise RuntimeError("pyoxigraph dump returned no data")
    return raw.decode()

set_known_prefixes(prefixes) classmethod

Set known prefixes in the context for use during parsing.

This should be called before parsing TTL strings that may use prefixes from an ontology or other source. The prefixes will be automatically added if they're used but not declared in the TTL string.

Parameters:

Name	Type	Description	Default
prefixes	dict[str, str] | None	Dictionary mapping prefix names to namespace URIs. Example: {"fcaont": "https://growgraph.dev/fcaont#"}	required

Source code in ontocast/onto/rdfgraph.py

@classmethod
def set_known_prefixes(cls, prefixes: dict[str, str] | None) -> None:
    """Set known prefixes in the context for use during parsing.

    This should be called before parsing TTL strings that may use prefixes
    from an ontology or other source. The prefixes will be automatically
    added if they're used but not declared in the TTL string.

    Args:
        prefixes: Dictionary mapping prefix names to namespace URIs.
            Example: {"fcaont": "https://growgraph.dev/fcaont#"}
    """
    _known_prefixes_context.set(prefixes)

unbind_chunk_namespaces(chunk_pattern='/chunk/')

Unbinds namespace prefixes that point to URIs containing a chunk pattern. Returns a new graph with chunk namespaces dereferenced (expanded to full URIs). Prefix target IRIs are normalized to end with # or / (/ appended when missing) before chunk detection and rebinding.

Parameters:

Name	Type	Description	Default
chunk_pattern	str	The pattern to look for in URIs (default: "/chunk/")	'/chunk/'

Returns:

Name	Type	Description
RDFGraph	RDFGraph	New graph with chunk-related namespaces unbound

Source code in ontocast/onto/rdfgraph.py

def unbind_chunk_namespaces(self, chunk_pattern="/chunk/") -> "RDFGraph":
    """
    Unbinds namespace prefixes that point to URIs containing a chunk pattern.
    Returns a new graph with chunk namespaces dereferenced (expanded to full URIs).
    Prefix target IRIs are normalized to end with ``#`` or ``/`` (``/`` appended
    when missing) before chunk detection and rebinding.

    Args:
        chunk_pattern (str): The pattern to look for in URIs (default: "/chunk/")

    Returns:
        RDFGraph: New graph with chunk-related namespaces unbound
    """
    current_prefixes = dict(self.namespace_manager.namespaces())

    # Normalize prefix target IRIs so chunk detection and rebinding agree on boundaries
    prefix_to_normalized: dict[str, str] = {
        prefix: normalize_namespace_iri(str(uri), context="auto")
        for prefix, uri in current_prefixes.items()
    }

    # Find prefixes that point to URIs containing the chunk pattern
    chunk_prefixes = []
    for prefix, uri_str in prefix_to_normalized.items():
        if chunk_pattern in uri_str:
            chunk_prefixes.append((prefix, uri_str))

    # Create new graph
    new_graph = RDFGraph()

    # Copy all triples (URIs are already expanded internally); an
    # oxigraph-backed source may hold RDF 1.2 triple terms a plain rdflib
    # graph cannot represent.
    copy_triples(self, new_graph, origin="RDFGraph.unbind_chunk_namespaces")

    # Bind only non-chunk namespace prefixes to the new graph
    for prefix, uri_str in prefix_to_normalized.items():
        if chunk_pattern not in uri_str:
            new_graph.bind(prefix, Namespace(uri_str))

    # Log what was removed
    if chunk_prefixes:
        logger.debug(f"Unbound {len(chunk_prefixes)} chunk-related namespace(s):")
        for prefix, uri in chunk_prefixes:
            logger.debug(f"  - '{prefix}': {uri}")

    return new_graph

update(update_object, processor='sparql', initNs=None, initBindings=None, use_store_provided=True, **kwargs)

Execute SPARQL update using a base Graph view.

rdflib's SPARQL update engine has internal checks that branch on exact Graph type, which can break for subclasses on INSERT/DELETE ... WHERE. Running updates through a base Graph view avoids that edge case while still operating on the same underlying store/identifier.

Source code in ontocast/onto/rdfgraph.py

def update(
    self,
    update_object: Any,
    processor: Any = "sparql",
    initNs: Mapping[str, Any] | None = None,
    initBindings: Mapping[str, Any] | None = None,
    use_store_provided: bool = True,
    **kwargs: Any,
) -> None:
    """Execute SPARQL update using a base Graph view.

    rdflib's SPARQL update engine has internal checks that branch on exact
    ``Graph`` type, which can break for subclasses on ``INSERT/DELETE ... WHERE``.
    Running updates through a base ``Graph`` view avoids that edge case while
    still operating on the same underlying store/identifier.
    """
    graph_view = Graph(store=self.store, identifier=self.identifier)
    graph_view.namespace_manager = self.namespace_manager
    graph_view.update(
        update_object=update_object,
        processor=processor,
        initNs=initNs,
        initBindings=initBindings,
        use_store_provided=use_store_provided,
        **kwargs,
    )
    return None

RejectedLiteralTriple

Bases: BaseModel

A triple quarantined during LLM ingest because its object literal is invalid.

Covers two cases: the literal's lexical form fails XSD validation for its declared datatype, or a literal sits on a predicate whose schema declares an IRI object (reason/expected_range are set for the latter).

Source code in ontocast/onto/rdfgraph.py

class RejectedLiteralTriple(BaseModel):
    """A triple quarantined during LLM ingest because its object literal is invalid.

    Covers two cases: the literal's lexical form fails XSD validation for its
    declared ``datatype``, or a literal sits on a predicate whose schema declares
    an IRI object (``reason``/``expected_range`` are set for the latter).
    """

    model_config = ConfigDict(frozen=True)

    subject: str
    predicate: str
    object_lexical: str
    datatype: str = ""
    reason: str | None = None
    expected_range: str | None = None

canonical_literal(term)

Map a literal onto the value-space normal form triple stores converge on.

Content hashing has to be blind to differences a round trip through the triple store erases anyway. Two rewrites were measured against pyoxigraph: integer subtypes collapse to xsd:integer, and xsd:decimal lexicals canonicalise ("10.0" -> "10"). Neither changes the value, so a hash that distinguishes them reports drift where no content changed.

Non-literals, untyped literals, ill-typed literals, and datatypes outside the normalised set are returned unchanged.

Parameters:

Name	Type	Description	Default
term	Node	Any RDF term.	required

Returns:

Name	Type	Description
Node	Node	The canonicalized literal, or term itself when nothing applies.

Source code in ontocast/onto/rdfgraph.py

def canonical_literal(term: Node) -> Node:
    """Map a literal onto the value-space normal form triple stores converge on.

    Content hashing has to be blind to differences a round trip through the
    triple store erases anyway. Two rewrites were measured against pyoxigraph:
    integer subtypes collapse to ``xsd:integer``, and ``xsd:decimal`` lexicals
    canonicalise (``"10.0"`` -> ``"10"``). Neither changes the value, so a hash
    that distinguishes them reports drift where no content changed.

    Non-literals, untyped literals, ill-typed literals, and datatypes outside
    the normalised set are returned unchanged.

    Args:
        term: Any RDF term.

    Returns:
        Node: The canonicalized literal, or ``term`` itself when nothing applies.
    """
    if not isinstance(term, Literal) or term.datatype is None or term.ill_typed:
        return term
    lexical = str(term)
    try:
        if term.datatype in _XSD_INTEGER_DATATYPES:
            return Literal(str(int(lexical)), datatype=XSD.integer)
        if term.datatype == XSD.decimal:
            value = Decimal(lexical).normalize()
            if value == 0:
                # Decimal('-0.0').normalize() keeps the sign; the store does not.
                value = Decimal(0)
            return Literal(format(value, "f"), datatype=XSD.decimal)
        if term.datatype == XSD.boolean:
            return Literal(
                "true" if lexical in ("true", "1") else "false", datatype=XSD.boolean
            )
    except (ValueError, InvalidOperation):
        return term
    return term

copy_triples(source, target, *, origin)

Add every rdflib triple of source to target, dropping triple terms.

Parameters:

Name	Type	Description	Default
source	Iterable	Any iterable of triples, typically a graph.	required
target	Graph	Graph to add to.	required
origin	str	Caller name, used in the warning when triples are dropped.	required

Returns:

Type	Description
int	The number of triples skipped because they were not rdflib triples.

Source code in ontocast/onto/rdfgraph.py

def copy_triples(source: Iterable, target: Graph, *, origin: str) -> int:
    """Add every rdflib triple of ``source`` to ``target``, dropping triple terms.

    Args:
        source: Any iterable of triples, typically a graph.
        target: Graph to add to.
        origin: Caller name, used in the warning when triples are dropped.

    Returns:
        The number of triples skipped because they were not rdflib triples.
    """
    dropped = 0
    for triple in source:
        if not is_rdflib_triple(triple):
            dropped += 1
            continue
        target.add(triple)
    if dropped:
        logger.warning(
            "%s: dropped %d RDF 1.2 triple-term triple(s); rdflib graphs cannot "
            "hold them",
            origin,
            dropped,
        )
    return dropped

drop_reifiers_mentioning(graph, nodes)

Delete reifier descriptions whose triple term mentions any of nodes.

Provenance is RDF 1.2 reification — _:r rdf:reifies <<( s p o )>> plus prov: arcs — so a node being removed from the graph is referenced from inside a triple term. No subject/object pattern matches that, which left the reifier and its provenance describing a statement that no longer existed once the node's own triples were deleted.

Goes through pyoxigraph rather than rdflib: Graph.remove raises ValueError on a triple whose object is a triple term, so the rdf:reifies quad is not removable through the rdflib API at all.

Parameters:

Name	Type	Description	Default
graph	Graph	Graph to sweep, mutated in place. A non-oxigraph store cannot hold triple terms, so it is a no-op.	required
nodes	Collection[Node]	Nodes whose triples have been removed.	required

Returns:

Type	Description
int	The number of quads removed.

Source code in ontocast/onto/rdfgraph.py

def drop_reifiers_mentioning(graph: Graph, nodes: Collection[Node]) -> int:
    """Delete reifier descriptions whose triple term mentions any of ``nodes``.

    Provenance is RDF 1.2 reification — ``_:r rdf:reifies <<( s p o )>>`` plus
    ``prov:`` arcs — so a node being removed from the graph is referenced from
    *inside* a triple term. No subject/object pattern matches that, which left
    the reifier and its provenance describing a statement that no longer
    existed once the node's own triples were deleted.

    Goes through pyoxigraph rather than rdflib: ``Graph.remove`` raises
    ``ValueError`` on a triple whose object is a triple term, so the
    ``rdf:reifies`` quad is not removable through the rdflib API at all.

    Args:
        graph: Graph to sweep, mutated in place. A non-oxigraph store cannot
            hold triple terms, so it is a no-op.
        nodes: Nodes whose triples have been removed.

    Returns:
        The number of quads removed.
    """
    if not nodes or type(graph.store).__name__ != "OxigraphStore":
        return 0

    import pyoxigraph as ox
    from oxrdflib._converter import to_ox

    store = cast("ox.Store", _oxigraph_inner_store(graph.store))
    graph_ctx_raw = to_ox(graph.identifier)
    assert isinstance(graph_ctx_raw, (ox.NamedNode, ox.BlankNode, ox.DefaultGraph))
    graph_ctx: ox.NamedNode | ox.BlankNode | ox.DefaultGraph = graph_ctx_raw
    targets = {to_ox(node) for node in nodes}

    doomed = set()
    for quad in store.quads_for_pattern(
        None, ox.NamedNode(str(RDF_REIFIES)), None, graph_ctx
    ):
        term = quad.object
        if not isinstance(term, ox.Triple):
            continue
        if targets & {term.subject, term.predicate, term.object}:
            doomed.add(quad.subject)

    removed = 0
    for reifier in doomed:
        for quad in list(store.quads_for_pattern(reifier, None, None, graph_ctx)):
            store.remove(quad)
            removed += 1
    return removed

extract_known_prefixes(graph, extra_prefix=None, extra_namespace=None)

Collect all namespace prefixes from graph for use in LLM Turtle output repair.

Reads both the graph's NamespaceManager bindings and the @prefix declarations emitted by serialize_canonical_turtle. The serializer step is necessary because rdflib auto-generates ephemeral names (ns1:, ns2:, …) for namespaces that have no explicit binding; these names appear in the Turtle sent to the LLM but are never stored back in the NamespaceManager, so they would be invisible to _ensure_prefixes when repairing the LLM's response.

Parameters:

Name	Type	Description	Default
graph	RDFGraph	The RDFGraph to extract prefixes from (typically an ontology or facts graph).	required
extra_prefix	str | None	Optional extra prefix name to register explicitly, e.g. from Ontology.prefix.	None
extra_namespace	str | None	Namespace URI paired with extra_prefix.	None

Returns:

Type	Description
dict[str, str]	Mapping from prefix names to namespace URI strings.

Source code in ontocast/onto/rdfgraph.py

def extract_known_prefixes(
    graph: "RDFGraph",
    extra_prefix: str | None = None,
    extra_namespace: str | None = None,
) -> dict[str, str]:
    """Collect all namespace prefixes from *graph* for use in LLM Turtle output repair.

    Reads both the graph's NamespaceManager bindings and the ``@prefix``
    declarations emitted by ``serialize_canonical_turtle``.  The serializer
    step is necessary because rdflib auto-generates ephemeral names
    (``ns1:``, ``ns2:``, …) for namespaces that have no explicit binding;
    these names appear in the Turtle sent to the LLM but are never stored
    back in the NamespaceManager, so they would be invisible to
    ``_ensure_prefixes`` when repairing the LLM's response.

    Args:
        graph: The RDFGraph to extract prefixes from (typically an ontology
            or facts graph).
        extra_prefix: Optional extra prefix name to register explicitly,
            e.g. from ``Ontology.prefix``.
        extra_namespace: Namespace URI paired with ``extra_prefix``.

    Returns:
        Mapping from prefix names to namespace URI strings.
    """
    known: dict[str, str] = {}

    for prefix, namespace_uri in graph.namespaces():
        if prefix:
            known[prefix] = str(namespace_uri)

    # Serialize to Turtle and capture any additional @prefix declarations,
    # including the ephemeral nsN: names rdflib emits for unbound namespaces.
    try:
        turtle_str = graph.serialize_canonical_turtle()
        for match in PREFIX_DECLARATION_PATTERN.finditer(turtle_str):
            p = match.group(1)
            if p not in known:
                known[p] = match.group(2)
    except Exception:
        pass

    if extra_prefix and extra_namespace:
        known[extra_prefix] = extra_namespace

    return known

finalize_llm_graph(graph)

Remove invalid XSD typed literals from an LLM-parsed graph.

Source code in ontocast/onto/rdfgraph.py

def finalize_llm_graph(
    graph: "RDFGraph",
) -> tuple["RDFGraph", list[RejectedLiteralTriple]]:
    """Remove invalid XSD typed literals from an LLM-parsed graph."""
    return RDFGraph.partition_invalid_typed_literals(graph)

format_quarantine_for_prompt(rejected, llm_graph_format)

Format quarantined triples for critic or improvement prompts.

Source code in ontocast/onto/rdfgraph.py

def format_quarantine_for_prompt(
    rejected: list[RejectedLiteralTriple],
    llm_graph_format: LLMGraphFormat,
) -> str:
    """Format quarantined triples for critic or improvement prompts."""
    if not rejected:
        return ""

    if llm_graph_format == LLMGraphFormat.JSONLD:
        lines: list[str] = []
        for item in rejected:
            pred_key = item.predicate
            if item.predicate.startswith("http"):
                for prefix, uri in COMMON_PREFIXES.items():
                    ns = uri.strip("<>")
                    if item.predicate.startswith(ns):
                        pred_key = f"{prefix}:{item.predicate[len(ns) :]}"
                        break
            subj = item.subject
            if item.subject.startswith("http"):
                for prefix, uri in COMMON_PREFIXES.items():
                    ns = uri.strip("<>")
                    if item.subject.startswith(ns):
                        subj = f"{prefix}:{item.subject[len(ns) :]}"
                        break
            value_obj: dict[str, str] = {"@value": item.object_lexical}
            if item.datatype:
                value_obj["@type"] = _datatype_to_compact(item.datatype)
            entry = json.dumps({"@id": subj, pred_key: value_obj}, indent=2)
            hint = _quarantine_hint(item)
            if hint:
                entry += f"\n^ {hint}"
            lines.append(entry)
        return "\n".join(lines)

    lines = []
    for item in rejected:
        obj = f'"{item.object_lexical}"'
        if item.datatype:
            obj += f"^^<{item.datatype}>"
        line = (
            f"{_format_term_for_turtle(item.subject)} "
            f"{_format_term_for_turtle(item.predicate)} "
            f"{obj} ."
        )
        hint = _quarantine_hint(item)
        if hint:
            line += f"  # {hint}"
        lines.append(line)
    return "\n".join(lines)

is_rdflib_triple(triple)

True when every position of triple is an rdflib :class:~rdflib.Node.

Oxigraph-backed graphs may carry RDF 1.2 triple terms (see ontocast.tool.agg.rewriter), which the oxrdflib iterator yields as plain tuples. Those tuples are not rdflib terms: Graph.add asserts on them and SPARQL has no syntax for them, so every path that copies or serialises triples out of such a graph has to filter first.

Source code in ontocast/onto/rdfgraph.py

def is_rdflib_triple(triple: object) -> bool:
    """True when every position of ``triple`` is an rdflib :class:`~rdflib.Node`.

    Oxigraph-backed graphs may carry RDF 1.2 triple terms (see
    ``ontocast.tool.agg.rewriter``), which the oxrdflib iterator yields as plain
    tuples. Those tuples are not rdflib terms: ``Graph.add`` asserts on them and
    SPARQL has no syntax for them, so every path that copies or serialises
    triples out of such a graph has to filter first.
    """
    if not isinstance(triple, tuple) or len(triple) != 3:
        return False
    return all(isinstance(position, Node) for position in triple)

retarget_reifiers(graph, replacements)

Repoint reifier triple terms from a removed statement onto its replacement.

Companion to :func:drop_reifiers_mentioning, for the repairs that rewrite a statement rather than delete it. A SHACL retype or a code-to-IRI resolution removes s p o and adds s p o'; without this the _:r rdf:reifies <<( s p o )>> quad keeps describing a statement that no longer exists. The provenance is not wrong, it is dangling — and dropping it instead would lose the derivation of a triple that survived the repair.

Goes through pyoxigraph for the same reason the sweep does: rdflib cannot add or remove a triple whose object is a triple term. The reifier node itself is untouched, so its prov:wasDerivedFrom arcs move with it.

Parameters:

Name	Type	Description	Default
graph	Graph	Graph to rewrite, mutated in place. A non-oxigraph store cannot hold triple terms, so it is a no-op.	required
replacements	Mapping[tuple, tuple]	Removed triple -> the triple that replaced it.	required

Returns:

Type	Description
int	The number of rdf:reifies quads repointed.

Source code in ontocast/onto/rdfgraph.py

def retarget_reifiers(graph: Graph, replacements: Mapping[tuple, tuple]) -> int:
    """Repoint reifier triple terms from a removed statement onto its replacement.

    Companion to :func:`drop_reifiers_mentioning`, for the repairs that
    *rewrite* a statement rather than delete it. A SHACL retype or a
    code-to-IRI resolution removes ``s p o`` and adds ``s p o'``; without this
    the ``_:r rdf:reifies <<( s p o )>>`` quad keeps describing a statement
    that no longer exists. The provenance is not wrong, it is dangling — and
    dropping it instead would lose the derivation of a triple that survived the
    repair.

    Goes through pyoxigraph for the same reason the sweep does: rdflib cannot
    add or remove a triple whose object is a triple term. The reifier node
    itself is untouched, so its ``prov:wasDerivedFrom`` arcs move with it.

    Args:
        graph: Graph to rewrite, mutated in place. A non-oxigraph store cannot
            hold triple terms, so it is a no-op.
        replacements: Removed triple -> the triple that replaced it.

    Returns:
        The number of ``rdf:reifies`` quads repointed.
    """
    if not replacements or type(graph.store).__name__ != "OxigraphStore":
        return 0

    import pyoxigraph as ox
    from oxrdflib._converter import to_ox

    store = cast("ox.Store", _oxigraph_inner_store(graph.store))
    graph_ctx_raw = to_ox(graph.identifier)
    assert isinstance(graph_ctx_raw, (ox.NamedNode, ox.BlankNode, ox.DefaultGraph))
    graph_ctx: ox.NamedNode | ox.BlankNode | ox.DefaultGraph = graph_ctx_raw
    reifies = ox.NamedNode(str(RDF_REIFIES))

    moved = 0
    for removed, replacement in replacements.items():
        old_term = _ox_triple_term(removed)
        new_term = _ox_triple_term(replacement)
        if old_term is None or new_term is None or old_term == new_term:
            continue
        # A triple term is legal in object position, so this is a keyed lookup
        # rather than a scan over every reifier in the graph.
        for quad in list(store.quads_for_pattern(None, reifies, old_term, graph_ctx)):
            store.remove(quad)
            store.add(ox.Quad(quad.subject, reifies, new_term, graph_ctx))
            moved += 1
    return moved

strip_sparql_update_wrapper(turtle_str)

Extract plain Turtle from LLM output that mixed Turtle with SPARQL UPDATE.

Source code in ontocast/onto/rdfgraph.py

def strip_sparql_update_wrapper(turtle_str: str) -> str:
    """Extract plain Turtle from LLM output that mixed Turtle with SPARQL UPDATE."""
    if not _SPARQL_DATA_BLOCK_RE.search(turtle_str):
        return turtle_str

    text = _SPARQL_PREFIX_RE.sub(
        lambda match: f"@prefix {match.group(1)}: <{match.group(2)}> .",
        turtle_str,
    )

    prefix_lines: list[str] = []
    for line in text.splitlines():
        stripped = line.strip()
        if stripped.startswith("@prefix"):
            if not stripped.endswith("."):
                stripped = f"{stripped} ."
            prefix_lines.append(stripped)

    bodies = _SPARQL_DATA_BLOCK_RE.findall(text)
    outside = _SPARQL_DATA_BLOCK_RE.sub("", text)
    outside = re.sub(
        r"(?:INSERT|DELETE)\s+DATA\s*",
        "",
        outside,
        flags=re.IGNORECASE,
    )
    outside = re.sub(r"^\s*}\s*$", "", outside, flags=re.MULTILINE)

    outside_triples: list[str] = []
    for line in outside.splitlines():
        stripped = line.strip()
        if not stripped or stripped.startswith("@prefix"):
            continue
        outside_triples.append(stripped)

    triple_parts = outside_triples + [body.strip() for body in bodies if body.strip()]
    if not triple_parts:
        return turtle_str

    result_parts: list[str] = []
    if prefix_lines:
        result_parts.append("\n".join(dict.fromkeys(prefix_lines)))
    result_parts.append("\n".join(triple_parts))
    return "\n\n".join(result_parts) + "\n"