{rawContent}
) : (
- {children}
, - ul: ({ children }) =>{children}, - hr: () =>
{children},
- // C-1: discard `node` here too — code elements are custom components
- // and would otherwise receive node="[object Object]" in the DOM.
- code: ({ className: codeClass, children }) => {
- const isInline = !codeClass && typeof children === "string" && !children.includes("\n");
- return (
-
- {children}
-
- );
- },
- }}
- >
- {rawContent}
- {children}
, + ul: ({ children }) =>{children}, + hr: () =>
{children},
+ // C-1: discard `node` here too — code elements are custom components
+ // and would otherwise receive node="[object Object]" in the DOM.
+ code: ({ className: codeClass, children }) => {
+ const isInline = !codeClass && typeof children === "string" && !children.includes("\n");
+ return (
+
+ {children}
+
+ );
+ },
+};
+
/* ─── Styles ──────────────────────────────────────────────────────── */
const viewerContainerStyle: CSSProperties = {
diff --git a/explorer/src/workspaces/GraphWorkspace/markdownUrlSafety.ts b/explorer/src/workspaces/GraphWorkspace/markdownUrlSafety.ts
new file mode 100644
index 00000000..3944a86b
--- /dev/null
+++ b/explorer/src/workspaces/GraphWorkspace/markdownUrlSafety.ts
@@ -0,0 +1,29 @@
+/**
+ * URL-safety predicate for the Markdown content viewer.
+ *
+ * Extracted into a pure module so the check can be unit-tested without
+ * importing the MarkdownContentViewer React component, and so the component
+ * module exports only components (react-refresh/only-export-components,
+ * issue #1119). The behaviour is unchanged from the original in-component
+ * implementation: only http, https, mailto, in-document fragments, and
+ * root-relative paths are permitted.
+ */
+
+export function isSafeUrl(url?: string): boolean {
+ if (!url) return false;
+ const trimmed = url.trim();
+ // Reject whitespace-only strings — new URL("", base) would resolve to the base
+ // protocol and produce a false positive. This guards direct callers of the exported
+ // function; markdown parsers normalise whitespace-only destinations to "" which
+ // already fails the !url check above.
+ if (!trimmed) return false;
+ if (trimmed.startsWith("//")) return false;
+ if (trimmed.startsWith("#")) return true;
+ if (trimmed.startsWith("/")) return true;
+ try {
+ const parsed = new URL(trimmed, "http://localhost");
+ return ["http:", "https:", "mailto:"].includes(parsed.protocol);
+ } catch {
+ return false;
+ }
+}
diff --git a/explorer/tests/markdownContentViewer.test.ts b/explorer/tests/markdownContentViewer.test.ts
index aa89f3cb..0435578a 100644
--- a/explorer/tests/markdownContentViewer.test.ts
+++ b/explorer/tests/markdownContentViewer.test.ts
@@ -5,7 +5,8 @@ import { renderToString } from "react-dom/server";
(globalThis as any).React = React;
-import { isSafeUrl, MarkdownContentViewer } from "../src/workspaces/GraphWorkspace/MarkdownContentViewer.tsx";
+import { MarkdownContentViewer } from "../src/workspaces/GraphWorkspace/MarkdownContentViewer.tsx";
+import { isSafeUrl } from "../src/workspaces/GraphWorkspace/markdownUrlSafety.ts";
test("isSafeUrl permits safe http, https, and mailto URLs and relative paths", () => {
assert.equal(isSafeUrl("https://example.com"), true);
diff --git a/integrations/agno/knowledge_graph.py b/integrations/agno/knowledge_graph.py
index 21b6b280..78004bc3 100644
--- a/integrations/agno/knowledge_graph.py
+++ b/integrations/agno/knowledge_graph.py
@@ -277,25 +277,22 @@ class AgnoKnowledgeGraph(_KnowledgeBase): # type: ignore[misc]
def load_urls(self, urls: List[str]) -> None:
"""Fetch each URL and ingest the response body.
- Only ``http`` and ``https`` schemes are permitted to prevent SSRF.
+ Uses the shared SSRF guard so that ``http`` and ``https`` are the only
+ permitted schemes, private/loopback/link-local/cloud-metadata addresses
+ are blocked by default, DNS resolution is validated, and every redirect
+ hop is re-checked before being followed.
"""
- import urllib.request
- from urllib.parse import urlparse
+ from semantica.ingest.ssrf import request_with_ssrf_guard
+ from semantica.utils.exceptions import ValidationError
for url in urls:
- parsed = urlparse(url)
- if parsed.scheme not in ("http", "https"):
- logger.warning(
- "Skipping URL with disallowed scheme '%s': %s",
- parsed.scheme,
- url,
- )
- continue
try:
- with urllib.request.urlopen(url, timeout=10) as resp: # noqa: S310
- text = resp.read().decode("utf-8", errors="replace")
+ response = request_with_ssrf_guard("GET", url, timeout=10)
+ text = response.text
self._ingest_text(text, source=url)
logger.info("Loaded URL: %s", url)
+ except ValidationError as exc:
+ logger.warning("Skipping URL (SSRF check failed) %s: %s", url, exc)
except Exception as exc:
logger.warning("Failed to fetch %s: %s", url, exc)
diff --git a/integrations/openclaw/mcp_tool.py b/integrations/openclaw/mcp_tool.py
index dff6b6db..626f8b95 100644
--- a/integrations/openclaw/mcp_tool.py
+++ b/integrations/openclaw/mcp_tool.py
@@ -116,7 +116,41 @@ class OpenClawKGTool:
)
def __init__(self, base_url: str = "http://localhost:8000", timeout: int = 30) -> None:
- self.base_url = base_url.rstrip("/")
+ # Validate base_url at construction time so callers get an immediate,
+ # actionable error rather than a cryptic failure on the first request.
+ # allow_private_ips=True because the documented default (localhost:8000)
+ # is intentionally a local Semantica server; the scheme check and
+ # URL-structure check still apply unconditionally.
+ try:
+ from semantica.ingest.ssrf import validate_url_for_request
+ validate_url_for_request(base_url, allow_private_ips=True)
+ except ImportError:
+ # semantica.ingest not installed in minimal openclaw-only environments;
+ # mirror the structural checks that validate_url_for_request performs
+ # unconditionally (before allow_private_ips is consulted), so the
+ # guarantee in the comment above — "scheme check and URL-structure check
+ # still apply unconditionally" — holds in this path too.
+ from urllib.parse import urlparse as _urlparse
+ if not isinstance(base_url, str) or not base_url.strip():
+ raise ValueError("OpenClawKGTool base_url must be a non-empty string.")
+ _parsed = _urlparse(base_url.strip())
+ _scheme = (_parsed.scheme or "").lower()
+ if _scheme not in ("http", "https"):
+ raise ValueError(
+ f"OpenClawKGTool base_url scheme '{_parsed.scheme}' is not permitted. "
+ "Only http and https are allowed."
+ )
+ if not _parsed.netloc:
+ raise ValueError(
+ f"Invalid OpenClawKGTool base_url '{base_url}': "
+ "URL must include a netloc (domain or host)."
+ )
+ if not _parsed.hostname:
+ raise ValueError(
+ f"Invalid OpenClawKGTool base_url '{base_url}': "
+ "URL must include a hostname."
+ )
+ self.base_url = base_url.strip().rstrip("/")
self.timeout = timeout
self._session: Any = None
diff --git a/mcp/__init__.py b/mcp/__init__.py
index 6154f804..0d18af90 100644
--- a/mcp/__init__.py
+++ b/mcp/__init__.py
@@ -21,6 +21,17 @@ Configure in Claude Desktop, Windsurf, Cline, Continue, VS Code:
}
"""
+import os
+
+# MCP stdio framing IS stdout: any progress bar or console renderer that writes
+# to stdout would interleave with the JSON-RPC stream and corrupt framing for
+# every client. This package is always used as an MCP stdio server, so force
+# progress tracking off for the entire process. Set before importing server /
+# tools so the Semantica progress-tracker singleton is never created with
+# output enabled (the singleton reads this variable at construction time and
+# the enabled.setter re-checks it, so later re-enable attempts are also blocked).
+os.environ["SEMANTICA_DISABLE_PROGRESS"] = "1"
+
# `semantica.__version__` is the authoritative package version — see
# semantica/mcp_server/__init__.py for why it is used directly rather than
# importlib.metadata.version("semantica").
diff --git a/mcp/tools/export.py b/mcp/tools/export.py
index f435bf18..df39162b 100644
--- a/mcp/tools/export.py
+++ b/mcp/tools/export.py
@@ -80,7 +80,12 @@ def handle_export_graph(args: dict) -> dict:
if rdf_fmt:
try:
from semantica.export import RDFExporter
- rdf_str = RDFExporter().export_to_rdf(graph, format=rdf_fmt)
+ # RDFExporter.export_to_rdf() expects the canonical kg dict
+ # {"entities": [...], "relationships": [...]}, not a ContextGraph
+ # object. Convert before handing off; passing the raw graph
+ # caused AttributeError: 'ContextGraph' object has no attribute
+ # 'get' on every RDF format.
+ rdf_str = RDFExporter().export_to_rdf(graph.to_kg_dict(), format=rdf_fmt)
return {"format": rdf_fmt, "data": rdf_str}
except Exception as exc:
return {"error": f"RDF export failed: {exc}"}
diff --git a/poc_runner.py b/poc_runner.py
index 49b9a952..49a06a40 100644
--- a/poc_runner.py
+++ b/poc_runner.py
@@ -187,15 +187,6 @@ def poc_vuln3():
})
return nodes
- # Simulate the CSV parser — mirrors export_import.py lines 131-133
- def parse_import_csv_row(row: dict) -> dict:
- """Mirrors export_import.py CSV node ID extraction (no sanitization)."""
- node_id = row.get("id") or row.get("node_id") or row.get(":ID") or row.get("_id")
- return {
- "id": str(node_id), # ← UNSANITIZED
- "type": row.get("type", "entity"),
- }
-
# Attack payloads
payloads = [
# Header injection payload (chained with VULN-1)
diff --git a/semantica/change_management/version_storage.py b/semantica/change_management/version_storage.py
index c272622b..e9d75e2a 100644
--- a/semantica/change_management/version_storage.py
+++ b/semantica/change_management/version_storage.py
@@ -31,6 +31,7 @@ import hashlib
import json
import sqlite3
import threading
+import warnings
from abc import ABC, abstractmethod
from datetime import datetime
from pathlib import Path
@@ -62,6 +63,12 @@ def create_graph_snapshot_record(
"""
Creates a standardized snapshot metadata record for a named graph.
+ .. deprecated::
+ ``create_graph_snapshot_record()`` is deprecated and will be removed in
+ a future major version. It has no callers inside Semantica; build the
+ record inline and checksum it with
+ :func:`semantica.change_management.compute_checksum` instead.
+
Args:
version_id: Unique identifier for this snapshot
graph_uri: The underlying named graph URI in the triplet store
@@ -69,6 +76,13 @@ def create_graph_snapshot_record(
description: Purpose or context of the snapshot
metadata: Additional tags or pipeline context
"""
+ warnings.warn(
+ "create_graph_snapshot_record() is deprecated and will be removed in a "
+ "future major version. Build the snapshot record inline and use "
+ "semantica.change_management.compute_checksum() instead.",
+ DeprecationWarning,
+ stacklevel=2,
+ )
record = {
"label": version_id,
diff --git a/semantica/context/context_graph.py b/semantica/context/context_graph.py
index 8857aa60..62050953 100644
--- a/semantica/context/context_graph.py
+++ b/semantica/context/context_graph.py
@@ -899,6 +899,11 @@ class ContextGraph:
return
node.properties.update(attributes)
node.metadata.update(attributes)
+ # Keep derived decision indexes consistent when a decision node is
+ # mutated so that category / entity / temporal lookups reflect the
+ # new property values without requiring a full graph reload.
+ if (getattr(node, "node_type", None) or "").lower() == "decision":
+ self._sync_decision_from_node(node_id)
if getattr(self, "mutation_callback", None) and not getattr(
self, "_suspend_mutation_callback", False
@@ -1291,6 +1296,14 @@ class ContextGraph:
if link_id:
self._unresolved_links[link_id] = link_meta
+ # Rebuild all derived decision indexes from the freshly-loaded
+ # nodes so that find_precedents_by_scenario, find_similar_decisions,
+ # and all decision analytics work correctly after a reload.
+ # _rebuild_decision_indexes() unconditionally clears the old indexes
+ # first, so repeated load_from_file calls never accumulate stale
+ # entries from a previous file.
+ self._rebuild_decision_indexes()
+
self.logger.info(f"Loaded context graph from {path}")
@staticmethod
@@ -1633,6 +1646,8 @@ class ContextGraph:
self._analytics_cache.clear()
self._retractions.clear()
self._tombstones.clear()
+ # Rebuild derived decision indexes from the freshly-loaded nodes.
+ self._rebuild_decision_indexes()
if self.mutation_callback and not self._suspend_mutation_callback:
mutation_events = [
@@ -2825,6 +2840,12 @@ class ContextGraph:
self._unresolved_links.clear()
self._retractions.clear()
self._tombstones.clear()
+ # Reset derived decision indexes so that decision queries against
+ # a cleared graph return empty results rather than stale data.
+ self._decisions = {}
+ self._decision_index = defaultdict(set)
+ self._entity_index = defaultdict(set)
+ self._temporal_index = []
self.logger.debug("Graph state fully cleared.")
# --- Internal Helpers ---
@@ -3482,6 +3503,9 @@ class ContextGraph:
)
self._add_internal_edge(edge)
+ # Rebuild derived decision indexes from the now-populated node store.
+ self._rebuild_decision_indexes()
+
def state_at(self, timestamp: Union[str, int, float, datetime]) -> Dict[str, Any]:
"""Return a serializable snapshot of graph state valid at the given time."""
at_time = self._normalize_timestamp(timestamp)
@@ -4707,6 +4731,7 @@ class ContextGraph:
scenario=decision["scenario"],
decision_maker=decision.get("decision_maker", ""),
reasoning=decision["reasoning"],
+ recorded_at=decision.get("recorded_at", ""),
**safe_metadata,
**extra_properties,
)
@@ -4788,20 +4813,288 @@ class ContextGraph:
return False
return True
- def _calculate_decision_content_similarity(self, scenario: str, decision: Dict[str, Any]) -> float:
- """Calculate content similarity between scenario and decision."""
+ # ── decision-index helpers ────────────────────────────────────────────────
+
+ # Protected set of node properties whose values are *core* decision fields
+ # so that we can distinguish them from user-supplied metadata when
+ # rebuilding the in-memory indexes from a persisted node.
+ _DECISION_CORE_FIELDS: frozenset = frozenset({
+ "id", "category", "scenario", "reasoning", "outcome", "confidence",
+ "entities", "decision_maker", "timestamp", "recorded_at",
+ "valid_from", "valid_until", "content",
+ })
+
+ def _rebuild_decision_indexes(self) -> None:
+ """Rebuild all derived decision indexes from the current node store.
+
+ This method is the single authoritative rebuild path. It must be
+ called (under the graph lock) after any operation that wholesale
+ replaces ``self.nodes`` — namely ``load_from_file`` (JSON and Markdown
+ paths) and ``from_dict``.
+
+ Contract:
+ - Unconditionally clears ``_decisions``, ``_decision_index``,
+ ``_entity_index``, and ``_temporal_index`` before rebuilding so that
+ repeated calls never accumulate stale entries.
+ - Derives ``_decisions[node_id]["metadata"]`` from the full set of
+ node properties, excluding the protected core fields, so that
+ user-supplied metadata survives the round-trip.
+ - Runs under ``self._lock`` when called from load paths; callers that
+ already hold the lock must invoke ``_rebuild_decision_indexes``
+ inside the lock block.
+ """
+ # Always start fresh so repeated loads don't accumulate stale entries.
+ self._decisions: Dict[str, Any] = {}
+ self._decision_index: Dict[str, set] = defaultdict(set)
+ self._entity_index: Dict[str, set] = defaultdict(set)
+ self._temporal_index: List[Tuple[str, float]] = []
+
+ for node in self.nodes.values():
+ if (getattr(node, "node_type", None) or "").lower() != "decision":
+ continue
+
+ # Merge metadata and properties; properties win on collision.
+ meta: Dict[str, Any] = {}
+ meta.update(getattr(node, "metadata", {}) or {})
+ meta.update(getattr(node, "properties", {}) or {})
+
+ # Timestamp: keep whatever was stored (float epoch or ISO string).
+ # The temporal index uses it for sorting; downstream code handles
+ # both types via _normalize_timestamp.
+ raw_ts = meta.get("timestamp", 0.0)
+ try:
+ sort_ts = float(raw_ts)
+ except (TypeError, ValueError):
+ sort_ts = 0.0
+
+ # Entities may be stored as a list in meta or inferred from
+ # outgoing "involves" edges if the list field is absent/empty.
+ # _add_decision_to_graph creates entity nodes connected via
+ # "involves" edges; it does NOT store the list as a node property.
+ entities = meta.get("entities") or []
+ if not isinstance(entities, list):
+ entities = []
+ if not entities:
+ # Recover entity list from "involves" edges on this decision node
+ for edge in self._adjacency.get(node.node_id, []):
+ if edge.edge_type == "involves":
+ entities.append(edge.target_id)
+
+ # Everything that isn't a core field is user-supplied metadata.
+ extra_meta = {
+ k: v
+ for k, v in meta.items()
+ if k not in self._DECISION_CORE_FIELDS
+ }
+
+ decision: Dict[str, Any] = {
+ "id": node.node_id,
+ "category": meta.get("category", ""),
+ "scenario": meta.get("scenario", getattr(node, "content", "") or ""),
+ "reasoning": meta.get("reasoning", ""),
+ "outcome": meta.get("outcome", ""),
+ "confidence": float(meta.get("confidence", 0.0) or 0.0),
+ "entities": entities,
+ "decision_maker": meta.get("decision_maker"),
+ "timestamp": raw_ts,
+ "recorded_at": meta.get("recorded_at", ""),
+ "valid_from": getattr(node, "valid_from", None),
+ "valid_until": getattr(node, "valid_until", None),
+ # Preserve all non-core node properties as decision metadata so
+ # that user-supplied fields survive a save → load round-trip.
+ "metadata": extra_meta,
+ }
+
+ self._decisions[node.node_id] = decision
+
+ category = decision["category"]
+ if category:
+ self._decision_index[category].add(node.node_id)
+
+ for entity in entities:
+ self._entity_index[entity].add(node.node_id)
+
+ self._temporal_index.append((node.node_id, sort_ts))
+
+ self._temporal_index.sort(key=lambda x: x[1], reverse=True)
+
+ def _sync_decision_from_node(self, node_id: str) -> None:
+ """Synchronise a single decision index entry from the node store.
+
+ Called after ``add_node_attribute`` mutates a decision node so that
+ ``_decisions`` and the derived indexes stay consistent without
+ requiring a full rebuild of all decisions.
+ """
+ node = self.nodes.get(node_id)
+ if node is None:
+ return
+ if (getattr(node, "node_type", None) or "").lower() != "decision":
+ return
+
+ if not hasattr(self, "_decisions"):
+ # Indexes don't exist yet — a full rebuild is safer.
+ self._rebuild_decision_indexes()
+ return
+
+ # Remove stale index entries for this decision ID.
+ old = self._decisions.get(node_id)
+ if old:
+ old_cat = old.get("category", "")
+ if old_cat and node_id in self._decision_index.get(old_cat, set()):
+ self._decision_index[old_cat].discard(node_id)
+ for ent in old.get("entities", []):
+ self._entity_index[ent].discard(node_id)
+ self._temporal_index = [
+ (nid, ts) for nid, ts in self._temporal_index if nid != node_id
+ ]
+
+ # Rebuild the entry for this node and re-insert index entries.
+ meta: Dict[str, Any] = {}
+ meta.update(getattr(node, "metadata", {}) or {})
+ meta.update(getattr(node, "properties", {}) or {})
+
+ raw_ts = meta.get("timestamp", 0.0)
try:
- # Simple word-based similarity
+ sort_ts = float(raw_ts)
+ except (TypeError, ValueError):
+ sort_ts = 0.0
+
+ entities = meta.get("entities") or []
+ if not isinstance(entities, list):
+ entities = []
+ if not entities:
+ # Recover entity list from "involves" edges
+ for edge in self._adjacency.get(node_id, []):
+ if edge.edge_type == "involves":
+ entities.append(edge.target_id)
+
+ extra_meta = {
+ k: v for k, v in meta.items() if k not in self._DECISION_CORE_FIELDS
+ }
+
+ decision: Dict[str, Any] = {
+ "id": node_id,
+ "category": meta.get("category", ""),
+ "scenario": meta.get("scenario", getattr(node, "content", "") or ""),
+ "reasoning": meta.get("reasoning", ""),
+ "outcome": meta.get("outcome", ""),
+ "confidence": float(meta.get("confidence", 0.0) or 0.0),
+ "entities": entities,
+ "decision_maker": meta.get("decision_maker"),
+ "timestamp": raw_ts,
+ "recorded_at": meta.get("recorded_at", ""),
+ "valid_from": getattr(node, "valid_from", None),
+ "valid_until": getattr(node, "valid_until", None),
+ "metadata": extra_meta,
+ }
+
+ self._decisions[node_id] = decision
+ if decision["category"]:
+ self._decision_index[decision["category"]].add(node_id)
+ for ent in entities:
+ self._entity_index[ent].add(node_id)
+ self._temporal_index.append((node_id, sort_ts))
+ self._temporal_index.sort(key=lambda x: x[1], reverse=True)
+
+ @staticmethod
+ def _char_bigrams(text: str) -> set:
+ """Character bigrams over whitespace-stripped text (CJK fallback).
+
+ Strips whitespace so CJK characters without word-separating spaces are
+ treated as a contiguous character sequence rather than a single token.
+ """
+ chars = "".join(text.lower().split())
+ return {chars[i:i + 2] for i in range(len(chars) - 1)}
+
+ @staticmethod
+ def _looks_cjk(text: str) -> bool:
+ """True if text contains CJK/Japanese/Korean script characters.
+
+ Used to gate the character-bigram similarity fallback so it only
+ activates for scripts where whitespace tokenisation doesn't work.
+ """
+ for ch in text:
+ code = ord(ch)
+ if (
+ 0x4E00 <= code <= 0x9FFF # CJK Unified Ideographs
+ or 0x3400 <= code <= 0x4DBF # CJK Extension A
+ or 0x3040 <= code <= 0x30FF # Hiragana + Katakana
+ or 0xAC00 <= code <= 0xD7A3 # Hangul Syllables
+ or 0x1100 <= code <= 0x11FF # Hangul Jamo
+ ):
+ return True
+ return False
+
+ def _calculate_decision_content_similarity(self, scenario: str, decision: Dict[str, Any]) -> float:
+ """Calculate content similarity between scenario and decision.
+
+ Uses word-level Jaccard for space-separated languages. For text where
+ whitespace tokenisation is unreliable (CJK/Japanese/Korean scripts, or
+ a query with no whitespace at all) a character-bigram Jaccard is
+ computed over the *stripped* character sequences instead.
+
+ The bigram fallback only activates when whitespace tokenisation would
+ not help — i.e. the query is CJK-like or has at most one whitespace
+ token — so it never contributes for ordinary multi-word English
+ queries, where incidental bigram overlap between unrelated sentences
+ would otherwise inflate scores.
+
+ The bigram side uses *Jaccard* (|A∩B|/|A∪B|), not the overlap
+ coefficient, so a 2-character query whose single bigram happens to
+ appear anywhere in a long document does not silently receive a score of
+ 1.0. A minimum bigram set size of 3 is required before the bigram
+ signal contributes; this prevents 1- and 2-character English queries
+ from polluting results while still allowing 3-character CJK phrases (2
+ bigrams) to match.
+ """
+ try:
+ decision_text = (
+ f"{decision['scenario']} {decision['reasoning']} "
+ f"{' '.join(decision['entities'])}"
+ )
+
+ # --- word-level Jaccard (primary metric for Latin/space-delimited) ---
scenario_words = set(scenario.lower().split())
- decision_text = f"{decision['scenario']} {decision['reasoning']} {' '.join(decision['entities'])}"
decision_words = set(decision_text.lower().split())
-
- intersection = scenario_words.intersection(decision_words)
- union = scenario_words.union(decision_words)
-
- return len(intersection) / len(union) if union else 0.0
-
- except Exception as e:
+ word_union = scenario_words | decision_words
+ word_sim = (
+ len(scenario_words & decision_words) / len(word_union)
+ if word_union
+ else 0.0
+ )
+
+ # --- character-bigram Jaccard (CJK / very-short-query fallback) ---
+ # Only used when whitespace tokenisation can't do the job: CJK-like
+ # scripts, or a query that is a single whitespace token (no spaces
+ # to split on). Ordinary multi-word English queries rely on
+ # word_sim alone, so incidental bigram overlap between unrelated
+ # sentences can never inflate their score.
+ bigram_sim = 0.0
+ needs_bigram_fallback = (
+ self._looks_cjk(scenario) or len(scenario.split()) <= 1
+ )
+ if needs_bigram_fallback:
+ scenario_bigrams = self._char_bigrams(scenario)
+ decision_bigrams = self._char_bigrams(decision_text)
+
+ # Require at least 3 bigrams in the query before the bigram
+ # signal is used. A 2-char query produces only 1 bigram; that
+ # single bigram is far too likely to appear as a substring of
+ # any English word and would produce a spuriously high overlap
+ # coefficient. 3 bigrams correspond to a 4-char stripped query
+ # (e.g. two CJK characters produce 1 bigram each → need ≥3
+ # chars stripped).
+ if len(scenario_bigrams) >= 3 and decision_bigrams:
+ bigram_union = scenario_bigrams | decision_bigrams
+ bigram_sim = (
+ len(scenario_bigrams & decision_bigrams) / len(bigram_union)
+ if bigram_union
+ else 0.0
+ )
+
+ return max(word_sim, bigram_sim)
+
+ except Exception:
self.logger.exception("Content similarity calculation failed")
return 0.0
diff --git a/semantica/context/graph_schema.py b/semantica/context/graph_schema.py
index 6b46e642..9a34fb89 100644
--- a/semantica/context/graph_schema.py
+++ b/semantica/context/graph_schema.py
@@ -6,6 +6,7 @@ including node labels, relationship types, and indexes for graph databases.
"""
import json
+import warnings
from typing import Dict, Any, List
from ..graph_store import GraphStore
@@ -460,11 +461,25 @@ def drop_decision_schema(graph_store: GraphStore) -> None:
"""
Drop decision tracking schema (for cleanup/testing).
+ .. deprecated::
+ ``drop_decision_schema()`` is deprecated and will be removed in a future
+ major version. It has no callers inside Semantica; issue the DROP
+ CONSTRAINT / DROP INDEX / DETACH DELETE statements directly against your
+ :class:`~semantica.graph_store.GraphStore` instead.
+
Args:
graph_store: Graph database instance
"""
logger = get_logger(__name__)
+ warnings.warn(
+ "drop_decision_schema() is deprecated and will be removed in a future "
+ "major version. Issue the DROP CONSTRAINT / DROP INDEX / DETACH DELETE "
+ "statements directly against your GraphStore instead.",
+ DeprecationWarning,
+ stacklevel=2,
+ )
+
try:
# Drop constraints
constraints = [
diff --git a/semantica/explorer/dependencies.py b/semantica/explorer/dependencies.py
index 7a97faeb..862ce03f 100644
--- a/semantica/explorer/dependencies.py
+++ b/semantica/explorer/dependencies.py
@@ -2,8 +2,9 @@
Semantica Explorer : FastAPI Dependencies
Provides ``Depends()``-compatible callables for injecting the
-current ``GraphSession`` and ``ConnectionManager`` into route handlers,
-and for enforcing API-key authentication on protected routes.
+current ``GraphSession`` into route handlers, and for enforcing API-key
+authentication on protected routes. WebSocket manager access is handled
+directly via ``app.state.ws_manager``.
"""
import hmac
@@ -14,7 +15,6 @@ from fastapi import Request, HTTPException, Security, status
from fastapi.security.api_key import APIKeyHeader
from .session import GraphSession
-from .ws import ConnectionManager
_api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
@@ -80,13 +80,3 @@ def get_session(request: Request) -> GraphSession:
detail="GraphSession not initialized."
)
return request.app.state.session
-
-
-def get_ws_manager(request: Request) -> ConnectionManager:
- """Retrieve the ConnectionManager stored on ``app.state``."""
- if not hasattr(request.app.state, "ws_manager") or request.app.state.ws_manager is None:
- raise HTTPException(
- status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
- detail="WebSocket manager not initialized.",
- )
- return request.app.state.ws_manager
diff --git a/semantica/explorer/routes/graph.py b/semantica/explorer/routes/graph.py
index bdd29741..d620bc15 100644
--- a/semantica/explorer/routes/graph.py
+++ b/semantica/explorer/routes/graph.py
@@ -78,66 +78,6 @@ def _parse_bbox(raw_bbox: Optional[str]) -> Optional[tuple[float, float, float,
return min_x, min_y, max_x, max_y
-def _coerce_embedding_vector(value: object) -> Optional[List[float]]:
- if isinstance(value, dict):
- # Probe keys in priority order: generic first, then framework-specific.
- # Must stay aligned with the top-level keys in _extract_node_embeddings.
- for key in ("embedding", "embeddings", "vector", "values", "node2vec", "semantic"):
- nested = _coerce_embedding_vector(value.get(key))
- if nested is not None:
- return nested
- return None
-
- if not isinstance(value, (list, tuple)):
- return None
-
- vector: List[float] = []
- for item in value:
- try:
- vector.append(float(item))
- except (TypeError, ValueError):
- return None
-
- return vector if vector else None
-
-
-def _extract_node_embeddings(graph_dict: dict) -> dict[str, List[float]]:
- """Extract embeddings from graph dictionary."""
- # Top-level keys to probe on each entity (and its metadata/properties dicts).
- # Priority: generic names first, then KG-extras-specific names.
- # Must stay aligned with the inner probe list in _coerce_embedding_vector.
- embedding_keys = (
- "embedding",
- "embeddings",
- "vector",
- "node_embedding",
- "node2vec_embedding",
- "semantic_embedding",
- "reasoning_embedding",
- )
-
- embeddings: dict[str, List[float]] = {}
- for entity in graph_dict.get("entities") or graph_dict.get("nodes") or []:
- if not isinstance(entity, dict):
- continue
- node_id = entity.get("id") or entity.get("node_id")
- if not node_id:
- continue
-
- metadata = entity.get("metadata") if isinstance(entity.get("metadata"), dict) else {}
- properties = entity.get("properties") if isinstance(entity.get("properties"), dict) else {}
-
- for key in embedding_keys:
- vector = _coerce_embedding_vector(
- entity.get(key, metadata.get(key, properties.get(key)))
- )
- if vector is not None:
- embeddings[str(node_id)] = vector
- break
-
- return embeddings
-
-
def _get_cached_embeddings(session: GraphSession) -> dict[str, List[float]]:
"""Get embeddings from session cache for optimal performance."""
return session.get_cached_embeddings()
diff --git a/semantica/explorer/routes/ontology.py b/semantica/explorer/routes/ontology.py
index b50eb206..81877e50 100644
--- a/semantica/explorer/routes/ontology.py
+++ b/semantica/explorer/routes/ontology.py
@@ -456,10 +456,6 @@ class DraftResponse(BaseModel):
updated_at: str
-class ProposalState(BaseModel):
- state: Literal["draft", "proposed", "approved", "published", "rejected"]
-
-
class ProposalRequest(BaseModel):
draft_id: str
ontology_uri: str
diff --git a/semantica/explorer/schemas.py b/semantica/explorer/schemas.py
index 13f2ac3f..808bceed 100644
--- a/semantica/explorer/schemas.py
+++ b/semantica/explorer/schemas.py
@@ -8,11 +8,6 @@ from typing import Any, Dict, List, Literal, Optional, Tuple
from pydantic import BaseModel, Field, field_validator
-class ErrorResponse(BaseModel):
- detail: str
- status_code: int = 500
-
-
class NodeResponse(BaseModel):
id: str
type: str
@@ -187,12 +182,6 @@ class ComplianceResponse(BaseModel):
violations: List[Dict[str, Any]] = Field(default_factory=list)
-class TemporalSnapshotResponse(BaseModel):
- timestamp: str
- active_nodes: List[NodeResponse]
- active_node_count: int
-
-
class TemporalDiffResponse(BaseModel):
from_time: str
to_time: str
@@ -256,13 +245,6 @@ class ExportRequest(BaseModel):
node_ids: Optional[List[str]] = None
-class ExportResponse(BaseModel):
- format: str
- content_type: str
- filename: str
- size_bytes: int = 0
-
-
class ImportResponse(BaseModel):
status: str = "success"
message: str = "Import successful"
@@ -272,11 +254,6 @@ class ImportResponse(BaseModel):
edges_imported: Optional[int] = None
-class StandardMessageResponse(BaseModel):
- status: str
- message: str
-
-
class AnnotationCreate(BaseModel):
node_id: str
content: str
diff --git a/semantica/export/rdf_exporter.py b/semantica/export/rdf_exporter.py
index dffd4951..68e28bd3 100644
--- a/semantica/export/rdf_exporter.py
+++ b/semantica/export/rdf_exporter.py
@@ -738,6 +738,22 @@ class RDFSerializer:
# node to signal that valid_until is OPEN/unbounded. This keeps the
# interval well-formed while remaining human- and machine-readable.
+ @staticmethod
+ def _escape_turtle_literal(value: str) -> str:
+ """Escape a string value for safe embedding in a Turtle string literal.
+
+ Backslash must be escaped first, then the double quote and the
+ recognized control characters (newline, carriage return, tab), per the
+ RDF 1.1 Turtle grammar for STRING_LITERAL_QUOTE.
+ """
+ return (
+ value.replace("\\", "\\\\")
+ .replace('"', '\\"')
+ .replace("\n", "\\n")
+ .replace("\r", "\\r")
+ .replace("\t", "\\t")
+ )
+
def serialize_to_turtle(self, rdf_data: Dict[str, Any], **options) -> str:
"""
Serialize RDF to Turtle format.
@@ -807,7 +823,7 @@ class RDFSerializer:
clauses = [
f"a <{self._as_turtle_iri(entity_type, merged_namespaces)}>",
- f'semantica:text "{text}"',
+ f'semantica:text "{self._escape_turtle_literal(text)}"',
]
if confidence is None:
self.logger.warning(
@@ -999,7 +1015,7 @@ class RDFSerializer:
lines.append(f" time:hasEnd <{end_id}> .")
lines.append(f"<{end_id}> a time:Instant ;")
lines.append(
- f' time:inXSDDateTimeStamp "{until_val}"^^xsd:dateTimeStamp .'
+ f' time:inXSDDateTimeStamp "{self._escape_turtle_literal(until_val)}"^^xsd:dateTimeStamp .'
)
else:
lines[-1] = (
@@ -1008,7 +1024,7 @@ class RDFSerializer:
lines.append(f"<{begin_id}> a time:Instant ;")
lines.append(
- f' time:inXSDDateTimeStamp "{from_val}"^^xsd:dateTimeStamp .'
+ f' time:inXSDDateTimeStamp "{self._escape_turtle_literal(from_val)}"^^xsd:dateTimeStamp .'
)
lines.append("")
@@ -1305,7 +1321,7 @@ class RDFSerializer:
# Text property
text = entity.get("text") or entity.get("label", "")
if text:
- safe_text = text.replace('"', '\\"').replace("\n", "\\n")
+ safe_text = self._escape_turtle_literal(text)
lines.append(
f'{subject} {expand_uri("semantica:text")} "{safe_text}" .'
)
diff --git a/semantica/ingest/ontology_ingestor.py b/semantica/ingest/ontology_ingestor.py
index 3b8638f9..0d63a770 100644
--- a/semantica/ingest/ontology_ingestor.py
+++ b/semantica/ingest/ontology_ingestor.py
@@ -40,7 +40,7 @@ from pathlib import Path
from typing import Any, Dict, List, Optional, Union
import rdflib
-from rdflib import RDF, RDFS, OWL, Graph
+from rdflib import RDF, RDFS, OWL, Dataset, Graph
from ..utils.exceptions import ProcessingError, ValidationError
from ..utils.logging import get_logger
@@ -106,15 +106,24 @@ class OntologyIngestor:
raise ValidationError(f"File not found: {file_path}")
self.progress.update_tracking(tracking_id, message="Parsing RDF graph...")
- g = Graph()
-
+ # `Dataset`, not `Graph`: a JSON-LD document with a top-level `@id` *and*
+ # `@graph` places its terms in a NAMED graph. `Graph.parse()` loads only the
+ # default graph and discards the rest without an error, so every class and
+ # property in such a document was dropped while the load reported success.
+ # Same migration #757 made for JenaStore; the ingest path was not covered by it.
+ # `default_union=True` makes the Dataset itself present triples from every
+ # graph as one merged view (it is an rdflib.Graph subclass, so it satisfies
+ # _convert_to_dict()'s Graph-typed contract directly) instead of copying every
+ # quad into a second in-memory Graph.
+ ds = Dataset(default_union=True)
+
# Use provided format or let rdflib guess based on extension
parse_kwargs = kwargs.copy()
if format:
parse_kwargs['format'] = format
-
+
try:
- g.parse(file_path, **parse_kwargs)
+ ds.parse(file_path, **parse_kwargs)
except Exception as e:
# Fallback: try to guess format from extension if not provided and initial parse failed
if not format:
@@ -130,12 +139,14 @@ class OntologyIngestor:
guessed_fmt = fmt_map.get(ext)
if guessed_fmt:
self.logger.info(f"Retrying with guessed format: {guessed_fmt}")
- g.parse(file_path, format=guessed_fmt, **kwargs)
+ ds.parse(file_path, format=guessed_fmt, **kwargs)
else:
raise e
else:
raise e
+ g = ds
+
self.progress.update_tracking(tracking_id, message="Converting to internal format...")
# Determine format for metadata
diff --git a/semantica/mcp_server/__init__.py b/semantica/mcp_server/__init__.py
index 19fe0bf4..05b9d322 100644
--- a/semantica/mcp_server/__init__.py
+++ b/semantica/mcp_server/__init__.py
@@ -62,6 +62,13 @@ logging.basicConfig(stream=sys.stderr, level=_log_level,
format="%(asctime)s [semantica-mcp] %(levelname)s %(message)s")
log = logging.getLogger("semantica.mcp_server")
+# MCP stdio framing IS stdout: a progress bar or other console renderer writing
+# to stdout would interleave with the JSON-RPC stream and hang every client
+# (observed 2026-08-20: export_graph over MCP timed out at 300s while the same
+# call returned in <1s directly). Force the progress trackers off for this
+# process — stdout is not a console here.
+os.environ["SEMANTICA_DISABLE_PROGRESS"] = "1"
+
# ── lazy graph session ──────────────────────────────────────────────────────
_graph: Any = None
@@ -74,7 +81,7 @@ def _get_graph():
kg_path = os.environ.get("SEMANTICA_KG_PATH")
if kg_path and os.path.exists(kg_path):
try:
- _graph.load(kg_path)
+ _graph.load_from_file(kg_path)
log.info("Loaded graph from %s", kg_path)
except Exception as exc:
log.warning("Could not load graph from %s: %s", kg_path, exc)
@@ -86,30 +93,57 @@ def _get_graph():
# ══════════════════════════════════════════════════════════════════════════════
def _tool_extract_entities(args: dict) -> dict:
- """Extract named entities from text."""
+ """Extract named entities from text.
+
+ Optional ``model`` (spaCy pipeline, e.g. ``zh_core_web_sm`` for Chinese)
+ and ``language`` allow non-English NER; defaults to the Semantica English
+ pipeline when omitted. ``method`` defaults to ``ml`` (spaCy); other
+ options are ``huggingface``, ``llm``, ``pattern``.
+ """
text = args.get("text", "")
if not text:
return {"error": "text is required"}
from semantica.semantic_extract import NamedEntityRecognizer
- entities = NamedEntityRecognizer().extract_entities(text)
+ init_kwargs = {}
+ for k in ("model", "language", "confidence_threshold"):
+ if args.get(k) is not None:
+ init_kwargs[k] = args[k]
+ method = args.get("method", "ml")
+ ner = NamedEntityRecognizer(methods=[method], **init_kwargs)
+ entities = ner.extract_entities(text)
return {
"entities": [
- {"label": getattr(e, "label", str(e)),
- "type": getattr(e, "type", None),
- "start": getattr(e, "start", None),
- "end": getattr(e, "end", None)}
+ {"text": getattr(e, "text", ""),
+ "label": getattr(e, "label", ""),
+ "type": getattr(e, "label", None),
+ "start": getattr(e, "start_char", getattr(e, "start", None)),
+ "end": getattr(e, "end_char", getattr(e, "end", None)),
+ "confidence": getattr(e, "confidence", 1.0)}
for e in (entities or [])
]
}
def _tool_extract_relations(args: dict) -> dict:
- """Extract relations and triplets from text."""
+ """Extract relations and triplets from text.
+
+ Optional ``model``/``language`` enable non-English extraction.
+ ``method`` defaults to ``pattern``; ``dependency`` uses spaCy syntactic
+ parsing (requires a spaCy model, e.g. ``zh_core_web_sm``).
+ """
text = args.get("text", "")
if not text:
return {"error": "text is required"}
- from semantica.semantic_extract import RelationExtractor, TripletExtractor
- relations = RelationExtractor().extract_relations(text)
+ from semantica.semantic_extract import NamedEntityRecognizer, RelationExtractor, TripletExtractor
+ rel_kwargs = {}
+ ner_kwargs = {}
+ for k in ("model", "language"):
+ if args.get(k) is not None:
+ rel_kwargs[k] = args[k]
+ ner_kwargs[k] = args[k]
+ method = args.get("method", "pattern")
+ entities = NamedEntityRecognizer(methods=["ml"], **ner_kwargs).extract_entities(text) or []
+ relations = RelationExtractor(method=method, **rel_kwargs).extract_relations(text, entities)
triplets = TripletExtractor().extract_triplets(text)
return {
"relations": [
@@ -156,10 +190,12 @@ def _tool_query_decisions(args: dict) -> dict:
graph = _get_graph()
try:
if query:
- results = graph.find_similar_decisions(query, max_results=limit)
+ results = graph.find_similar_decisions(query, max_results=limit, min_similarity=0.05)
elif category:
nodes = graph.find_nodes(node_type="decision")
- results = [n for n in nodes if n.get("category") == category][:limit]
+ results = [n for n in nodes
+ if n.get("category") == category
+ or n.get("metadata", {}).get("category") == category][:limit]
else:
results = graph.find_nodes(node_type="decision")[:limit]
return {"decisions": results if isinstance(results, list) else list(results)}
@@ -175,7 +211,9 @@ def _tool_find_precedents(args: dict) -> dict:
max_results = int(args.get("max_results", 5))
graph = _get_graph()
try:
- precedents = graph.find_similar_decisions(scenario, max_results=max_results)
+ min_similarity = float(args.get("min_similarity", 0.05))
+ precedents = graph.find_similar_decisions(
+ scenario, max_results=max_results, min_similarity=min_similarity)
return {"precedents": precedents if isinstance(precedents, list) else list(precedents)}
except Exception as exc:
return {"error": str(exc), "precedents": []}
@@ -260,16 +298,28 @@ def _tool_get_graph_analytics(args: dict) -> dict:
return {"error": str(exc)}
+_EXPORT_GRAPH_FORMATS = ("turtle", "ttl", "nt", "xml", "json-ld", "json")
+
+
def _tool_export_graph(args: dict) -> dict:
"""Export the current knowledge graph to a serialised format."""
fmt = args.get("format", "json-ld")
+ if fmt not in _EXPORT_GRAPH_FORMATS:
+ return {
+ "error": f"Unsupported format '{fmt}'. Supported: {', '.join(_EXPORT_GRAPH_FORMATS)}"
+ }
graph = _get_graph()
try:
- from semantica.export import RDFExporter, JSONExporter
+ from semantica.export import RDFExporter
+ # The exporters consume the canonical kg dict, not the ContextGraph
+ # object (regression: the old code passed the object straight through,
+ # so every branch failed — JSONExporter.export() with no file_path on
+ # the json branch, AttributeError on the RDF branches).
+ kg = graph.to_kg_dict()
if fmt in ("turtle", "ttl", "nt", "xml", "json-ld"):
- result = RDFExporter().export_to_rdf(graph, format=fmt)
+ result = RDFExporter().export_to_rdf(kg, format=fmt)
else:
- result = JSONExporter().export(graph)
+ result = json.dumps(kg, indent=2, ensure_ascii=False)
return {"format": fmt, "data": result}
except Exception as exc:
return {"error": str(exc)}
@@ -290,6 +340,160 @@ def _tool_get_graph_summary(args: dict) -> dict:
return {"error": str(exc), "graph_ready": False}
+def _tool_update_node(args: dict) -> dict:
+ """Update properties of an existing node and persist to SEMANTICA_KG_PATH.
+
+ Common use: mark an action node's status (todo/doing/done) with an
+ optional note. The graph is mutated in-memory then saved back to the
+ file it was loaded from, so changes survive server restarts.
+ """
+ node_id = args.get("node_id", "")
+ if not node_id:
+ return {"error": "node_id is required"}
+ properties = args.get("properties", {})
+ if not isinstance(properties, dict) or not properties:
+ return {"error": "properties (non-empty object) is required"}
+ graph = _get_graph()
+ try:
+ if not graph.find_node(node_id):
+ return {"error": f"node '{node_id}' not found"}
+ graph.add_node_attribute(node_id, properties)
+ # Persist back to disk so the change survives restarts
+ kg_path = os.environ.get("SEMANTICA_KG_PATH")
+ if kg_path:
+ graph.save_to_file(kg_path)
+ persisted = True
+ else:
+ persisted = False
+ updated = graph.find_node(node_id)
+ return {
+ "status": "updated",
+ "node_id": node_id,
+ "properties": {k: (updated.get("metadata") or {}).get(k) for k in properties},
+ "persisted": persisted,
+ }
+ except Exception as exc:
+ return {"error": str(exc)}
+
+
+def _tool_delete_node(args: dict) -> dict:
+ """Archive a node (soft delete) and persist to SEMANTICA_KG_PATH.
+
+ The node is kept in the graph for history but marked status='archived'.
+ Use to retire an action you no longer actively track.
+ """
+ node_id = args.get("node_id", "")
+ if not node_id:
+ return {"error": "node_id is required"}
+ graph = _get_graph()
+ try:
+ if not graph.find_node(node_id):
+ return {"error": f"node '{node_id}' not found"}
+ graph.add_node_attribute(node_id, {"status": "archived"})
+ kg_path = os.environ.get("SEMANTICA_KG_PATH")
+ if kg_path:
+ graph.save_to_file(kg_path)
+ return {"status": "archived", "node_id": node_id, "persisted": bool(kg_path)}
+ except Exception as exc:
+ return {"error": str(exc)}
+
+
+def _tool_query_graph(args: dict) -> dict:
+ """Query the live knowledge graph: node detail, neighbours, or keyword search.
+
+ mode:
+ - "node" : get one node by id (needs node_id)
+ - "neighbors": traverse up to `depth` hops from node_id (default depth=1)
+ - "search" : keyword search over node id+content (needs query)
+ """
+ graph = _get_graph()
+ mode = args.get("mode", "neighbors")
+ try:
+ if mode == "node":
+ node_id = args.get("node_id", "")
+ if not node_id:
+ return {"error": "node_id is required"}
+ node = graph.find_node(node_id)
+ return {"node": node}
+
+ if mode == "neighbors":
+ node_id = args.get("node_id", "")
+ if not node_id:
+ return {"error": "node_id is required"}
+ depth = int(args.get("depth", 1))
+ rel_types = args.get("relationship_types")
+ if isinstance(rel_types, str):
+ rel_types = [rel_types]
+ rel_set = set(rel_types) if rel_types else None
+ limit = args.get("limit")
+ limit = int(limit) if limit is not None else None
+ depth = min(max(depth, 1), 5)
+ # Out-edges (multi-hop) via get_neighbors
+ nb = graph.get_neighbors(
+ node_id, hops=depth, relationship_types=rel_types, limit=limit,
+ )
+ out = [
+ {"id": n.get("id"), "type": n.get("type"),
+ "content": n.get("content"),
+ "relationship": n.get("relationship"),
+ "direction": "out", "hop": n.get("hop", 1)}
+ for n in (nb or [])
+ ]
+ # In-edges (1-hop): scan edges whose target == node_id.
+ # Deduplicate by source node so that multiple edges between the
+ # same pair of nodes (different edge types) produce one entry.
+ # Stop early once we have already collected `limit` inbound results
+ # (if a limit is set) to avoid scanning the full edge list.
+ inb = []
+ seen_inbound = set()
+ for e in graph.find_edges():
+ if e.get("target") != node_id:
+ continue
+ if rel_set is not None and e.get("type") not in rel_set:
+ continue
+ src_id = e.get("source")
+ if src_id in seen_inbound:
+ continue
+ seen_inbound.add(src_id)
+ src = graph.find_node(src_id) or {}
+ inb.append({"id": src_id, "type": src.get("type"),
+ "content": src.get("content"),
+ "relationship": e.get("type"),
+ "direction": "in", "hop": 1})
+ # Early-exit: we already have `limit` inbound results; the
+ # combined list will be truncated to `limit` anyway.
+ if limit is not None and len(inb) >= limit:
+ break
+ neighbors = out + inb
+ # Apply final limit. Use ``is not None`` so limit=0 (zero results)
+ # is honoured correctly; ``if limit:`` would treat 0 as falsy.
+ if limit is not None:
+ neighbors = neighbors[:limit]
+ return {"node_id": node_id, "depth": depth, "neighbors": neighbors}
+
+ if mode == "search":
+ q = (args.get("query") or "").lower()
+ if not q:
+ return {"error": "query is required"}
+ node_type = args.get("node_type")
+ limit = int(args.get("limit", 50))
+ nodes = graph.find_nodes(node_type=node_type) if node_type else graph.find_nodes()
+ hits = []
+ for n in nodes:
+ # Check limit BEFORE appending so limit=0 returns empty.
+ if len(hits) >= limit:
+ break
+ blob = f"{n.get('id','')} {n.get('content','')}".lower()
+ if q in blob:
+ hits.append({"id": n.get("id"), "type": n.get("type"),
+ "content": n.get("content")})
+ return {"query": q, "results": hits, "total": len(hits)}
+
+ return {"error": f"unknown mode '{mode}': use node|neighbors|search"}
+ except Exception as exc:
+ return {"error": str(exc)}
+
+
# ══════════════════════════════════════════════════════════════════════════════
# MCP protocol tables
# ══════════════════════════════════════════════════════════════════════════════
@@ -301,7 +505,11 @@ TOOLS = [
"inputSchema": {
"type": "object",
"properties": {
- "text": {"type": "string", "description": "Input text to extract entities from"}
+ "text": {"type": "string", "description": "Input text to extract entities from"},
+ "model": {"type": "string", "description": "spaCy model name, e.g. 'zh_core_web_sm' for Chinese, 'en_core_web_sm' for English. Defaults to English pipeline."},
+ "language": {"type": "string", "description": "Language code, e.g. 'zh', 'en'."},
+ "method": {"type": "string", "description": "Extraction method: 'ml' (spaCy, default), 'huggingface', 'llm', 'pattern'."},
+ "confidence_threshold": {"type": "number", "description": "Minimum confidence 0-1 (default 0.5)."}
},
"required": ["text"],
},
@@ -313,7 +521,10 @@ TOOLS = [
"inputSchema": {
"type": "object",
"properties": {
- "text": {"type": "string", "description": "Input text to extract relations from"}
+ "text": {"type": "string", "description": "Input text to extract relations from"},
+ "model": {"type": "string", "description": "spaCy model name for dependency parsing, e.g. 'zh_core_web_sm'."},
+ "language": {"type": "string", "description": "Language code, e.g. 'zh'."},
+ "method": {"type": "string", "description": "Extraction method: 'pattern' (default), 'dependency', 'cooccurrence', 'huggingface', 'llm'."}
},
"required": ["text"],
},
@@ -441,7 +652,7 @@ TOOLS = [
"properties": {
"format": {
"type": "string",
- "enum": ["turtle", "ttl", "nt", "xml", "json-ld", "json"],
+ "enum": list(_EXPORT_GRAPH_FORMATS),
"description": "Export format (default: json-ld)",
}
},
@@ -454,6 +665,48 @@ TOOLS = [
"inputSchema": {"type": "object", "properties": {}},
"_handler": _tool_get_graph_summary,
},
+ {
+ "name": "query_graph",
+ "description": "Query the live knowledge graph: get a node, traverse its neighbours (up to 5 hops), or keyword-search nodes by id+content.",
+ "inputSchema": {
+ "type": "object",
+ "properties": {
+ "mode": {"type": "string", "description": "node | neighbors | search (default: neighbors)"},
+ "node_id": {"type": "string", "description": "Node id (required for node/neighbors mode)"},
+ "depth": {"type": "integer", "description": "Hop depth for neighbors (1-5, default 1)"},
+ "relationship_types": {"type": "array", "items": {"type": "string"}, "description": "Optional filter by edge type(s)"},
+ "query": {"type": "string", "description": "Keyword for search mode (matched against node id+content)"},
+ "node_type": {"type": "string", "description": "Optional node_type filter for search mode"},
+ "limit": {"type": "integer", "description": "Max results for neighbors/search"}
+ },
+ },
+ "_handler": _tool_query_graph,
+ },
+ {
+ "name": "update_node",
+ "description": "Update properties of an existing node (e.g. mark an action todo/doing/done with a note) and persist to SEMANTICA_KG_PATH.",
+ "inputSchema": {
+ "type": "object",
+ "properties": {
+ "node_id": {"type": "string", "description": "Node id to update"},
+ "properties": {"type": "object", "description": "Property key-values to merge onto the node, e.g. {\"status\":\"done\",\"updated_at\":\"2026-08-13\",\"note\":\"...\"}"}
+ },
+ "required": ["node_id", "properties"],
+ },
+ "_handler": _tool_update_node,
+ },
+ {
+ "name": "delete_node",
+ "description": "Archive a node (soft delete: marks status='archived', keeps it for history) and persist to SEMANTICA_KG_PATH. Use to retire an action you no longer track.",
+ "inputSchema": {
+ "type": "object",
+ "properties": {
+ "node_id": {"type": "string", "description": "Node id to delete"}
+ },
+ "required": ["node_id"],
+ },
+ "_handler": _tool_delete_node,
+ },
]
RESOURCES = [
diff --git a/semantica/pipeline/pipeline_builder.py b/semantica/pipeline/pipeline_builder.py
index d2044cbe..c64d5e64 100644
--- a/semantica/pipeline/pipeline_builder.py
+++ b/semantica/pipeline/pipeline_builder.py
@@ -272,7 +272,19 @@ class PipelineBuilder:
step_name = step_config.get("name")
step_type = step_config.get("type")
if step_name and step_type:
- self.add_step(step_name, step_type, **step_config.get("config", {}))
+ step = self.add_step(
+ step_name, step_type, **step_config.get("config", {})
+ )
+ step.dependencies = list(
+ step_config.get("dependencies", step.dependencies)
+ )
+ step.delta_mode = step_config.get("delta_mode", step.delta_mode)
+ step.base_version_id = step_config.get(
+ "base_version_id", step.base_version_id
+ )
+ step.target_version_id = step_config.get(
+ "target_version_id", step.target_version_id
+ )
# Set parallelism if specified
if "parallelism" in pipeline_config:
@@ -398,14 +410,29 @@ class PipelineSerializer:
Returns:
Serialized pipeline
+
+ Notes:
+ Step handlers are runtime callables and are intentionally omitted from
+ the serialized representation. They must be rebound after deserialization.
"""
+ reserved_config_keys = {
+ "handler",
+ "dependencies",
+ "delta_mode",
+ "base_version_id",
+ "target_version_id",
+ }
pipeline_data = {
"name": pipeline.name,
"steps": [
{
"name": step.name,
"type": step.step_type,
- "config": step.config,
+ "config": {
+ key: value
+ for key, value in step.config.items()
+ if key not in reserved_config_keys
+ },
"dependencies": step.dependencies,
"delta_mode": getattr(step, "delta_mode", False),
"base_version_id": getattr(step, "base_version_id", None),
@@ -445,6 +472,18 @@ class PipelineSerializer:
else:
pipeline_data = serialized_pipeline
+ # Runtime handlers are process-local and cannot be reconstructed safely
+ # from serialized data. Copy before sanitizing so dict inputs are not mutated.
+ pipeline_data = dict(pipeline_data)
+ sanitized_steps = []
+ for step_data in pipeline_data.get("steps", []):
+ sanitized_step = dict(step_data)
+ step_config = dict(sanitized_step.get("config", {}))
+ step_config.pop("handler", None)
+ sanitized_step["config"] = step_config
+ sanitized_steps.append(sanitized_step)
+ pipeline_data["steps"] = sanitized_steps
+
# Reconstruct pipeline
builder = PipelineBuilder(**self.config)
pipeline = builder.build_pipeline(pipeline_data, **options)
diff --git a/semantica/seed/seed_manager.py b/semantica/seed/seed_manager.py
index 3e6ebbf8..1a31175e 100644
--- a/semantica/seed/seed_manager.py
+++ b/semantica/seed/seed_manager.py
@@ -482,8 +482,8 @@ class SeedDataManager:
List of loaded data records as dictionaries
Raises:
- ProcessingError: If API request fails, response parsing fails, or
- requests library is not available
+ ProcessingError: If the API request fails (connection error,
+ timeout, non-2xx status) or the response cannot be parsed
Example:
>>> records = manager.load_from_api(
@@ -559,10 +559,6 @@ class SeedDataManager:
self.logger.info(f"Loaded {len(records)} records from API: {full_url}")
return records
- except (ImportError, OSError):
- raise ProcessingError(
- "requests library not available. Install with: pip install requests"
- )
except Exception as e:
raise ProcessingError(f"Failed to load from API: {e}") from e
diff --git a/semantica/semantic_extract/methods.py b/semantica/semantic_extract/methods.py
index 31ca3590..00482df1 100644
--- a/semantica/semantic_extract/methods.py
+++ b/semantica/semantic_extract/methods.py
@@ -140,6 +140,47 @@ _result_cache = ExtractionCache(
if not config.get("cache_enabled", True):
_result_cache.enabled = False
+# Generation kwargs that affect provider output and must therefore be part of
+# the cache key. This is the union of every generation-affecting parameter
+# read across providers.py, including params picked up outside _add_if_set
+# (e.g. AnthropicProvider's manual pass-through loop). Sensitive values
+# (api_key, token, etc.) are already filtered out by
+# ExtractionCache._generate_key, so they need not be excluded here.
+_GENERATION_CACHE_KEYS = frozenset({
+ "max_tokens",
+ "max_completion_tokens",
+ "temperature",
+ "top_p",
+ "top_k",
+ "seed",
+ "frequency_penalty",
+ "presence_penalty",
+ "stop",
+ "stop_sequences", # Anthropic/Gemini spelling of "stop"
+ "logit_bias",
+ "user",
+ "system", # Anthropic system prompt
+ "metadata", # Anthropic request metadata
+ "candidate_count", # Gemini
+ "repeat_penalty", # Ollama
+ "num_ctx", # Ollama
+ "context_window", # Ollama alias for num_ctx
+})
+
+
+def _generation_cache_params(kwargs: dict) -> dict:
+ """Return the subset of *kwargs* that affects generation output.
+
+ Only keys listed in ``_GENERATION_CACHE_KEYS`` are included so that
+ irrelevant or sensitive caller kwargs do not pollute the cache key.
+ Values that are ``None`` are omitted; a caller passing
+ ``temperature=None`` is equivalent to not passing it at all.
+ """
+ return {
+ k: v for k, v in kwargs.items()
+ if k in _GENERATION_CACHE_KEYS and v is not None
+ }
+
# Try to import spaCy
from ..utils.helpers import safe_import
@@ -957,6 +998,7 @@ def extract_entities_llm(
"max_text_length": max_text_length,
"structured_output_mode": structured_output_mode,
"entity_types": kwargs.get("entity_types"),
+ **_generation_cache_params(kwargs),
}
cached_result = _result_cache.get("entities", text, **cache_params)
if cached_result is not None:
@@ -1124,47 +1166,6 @@ Text to extract from:
return []
-def _parse_entity_result(result: Any, provider: str, model: Optional[str]) -> List[Entity]:
- """Helper to parse raw LLM result into Entity objects."""
- entities = []
- items = []
-
- if isinstance(result, list):
- items = result
- elif isinstance(result, dict):
- # Handle cases where LLM wraps the list in a key
- for key in ["entities", "data", "results"]:
- if key in result and isinstance(result[key], list):
- items = result[key]
- break
- if not items and "text" in result: # Single object instead of list
- items = [result]
-
- for item in items:
- if not isinstance(item, dict):
- continue
-
- text = item.get("text", "")
- if not text:
- continue
-
- entities.append(
- Entity(
- text=text,
- label=item.get("label", "UNKNOWN"),
- start_char=item.get("start", 0),
- end_char=item.get("end", 0),
- confidence=item.get("confidence", 0.9),
- metadata={
- "provider": provider,
- "model": model,
- "extraction_method": "llm",
- },
- )
- )
- return entities
-
-
def _extract_entities_chunked(
text: str,
provider: str,
@@ -1747,7 +1748,8 @@ def extract_relations_llm(
"relation_types": kwargs.get("relation_types"),
"extract_temporal_bounds": extract_temporal_bounds,
# Include entities hash/str in cache key implicitly via **cache_params
- "entities_hash": hash(tuple(sorted([e.text for e in entities]))) if entities else 0
+ "entities_hash": hash(tuple(sorted([e.text for e in entities]))) if entities else 0,
+ **_generation_cache_params(kwargs),
}
cached_result = _result_cache.get("relations", text, **cache_params)
if cached_result is not None:
@@ -1947,13 +1949,10 @@ Entities found in text: {entities_str}"""
"[methods.extract_relations_llm] Calling llm.generate_typed (%s/%s)...",
provider, model,
)
- # Only forward minimal, safe parameters to provider calls
- call_kwargs = {}
- if "temperature" in kwargs:
- call_kwargs["temperature"] = kwargs["temperature"]
- if "verbose" in kwargs:
- call_kwargs["verbose"] = kwargs["verbose"]
-
+ # Forward all caller-supplied generation kwargs so they reach
+ # generate_typed and the underlying provider API. max_retries is
+ # always set from the explicit parameter.
+ call_kwargs = kwargs.copy()
call_kwargs["max_retries"] = max_retries
# Select schema based on whether temporal extraction is requested
@@ -2405,7 +2404,8 @@ def extract_triplets_llm(
"triplet_types": kwargs.get("triplet_types"),
# Include entities/relations hash in cache key implicitly via **cache_params
"entities_hash": hash(tuple(sorted([e.text for e in entities]))) if entities else 0,
- "relations_hash": hash(tuple(sorted([str(r) for r in relations]))) if relations else 0
+ "relations_hash": hash(tuple(sorted([str(r) for r in relations]))) if relations else 0,
+ **_generation_cache_params(kwargs),
}
cached_result = _result_cache.get("triplets", text, **cache_params)
if cached_result is not None:
@@ -2559,48 +2559,6 @@ Text to extract from:
return []
-def _parse_triplet_result(result: Any, provider: str, model: Optional[str]) -> List[Triplet]:
- """Helper to parse raw LLM result into Triplet objects."""
- triplets = []
- items = []
-
- if isinstance(result, list):
- items = result
- elif isinstance(result, dict):
- for key in ["triplets", "data", "results"]:
- if key in result and isinstance(result[key], list):
- items = result[key]
- break
- if not items and "subject" in result:
- items = [result]
-
- for item in items:
- if not isinstance(item, dict):
- continue
-
- subject = item.get("subject", "")
- predicate = item.get("predicate", "")
- obj = item.get("object", "")
-
- if not subject or not predicate or not obj:
- continue
-
- triplets.append(
- Triplet(
- subject=str(subject),
- predicate=str(predicate),
- object=str(obj),
- confidence=item.get("confidence", 0.9),
- metadata={
- "provider": provider,
- "model": model,
- "extraction_method": "llm",
- },
- )
- )
- return triplets
-
-
def _extract_triplets_chunked(
text: str,
provider: str,
diff --git a/semantica/triplet_store/methods.py b/semantica/triplet_store/methods.py
index 12380aa8..dc999c20 100644
--- a/semantica/triplet_store/methods.py
+++ b/semantica/triplet_store/methods.py
@@ -101,7 +101,6 @@ from .triplet_store import TripletStore
# Global store registry
_global_stores: Dict[str, TripletStore] = {}
_default_store_id: Optional[str] = None
-_global_query_engine: Optional[QueryEngine] = None
_global_bulk_loader: Optional[BulkLoader] = None
@@ -131,25 +130,6 @@ def _get_store(store_id: Optional[str] = None) -> TripletStore:
return _global_stores[target_id]
-def _get_query_engine() -> QueryEngine:
- """Get or create global QueryEngine instance."""
- global _global_query_engine
- if _global_query_engine is None:
- # We need a store backend for the engine, but QueryEngine in this module
- # seems to be initialized with config in the old code.
- # In the new code, TripletStore has its own query_engine.
- # If we use this standalone function, we might need to rely on the store's engine.
- # But let's keep a standalone one if needed, or better, delegate to store.
- config = triplet_store_config.get_all()
- # QueryEngine now expects a backend, but we can initialize it without one
- # if we pass the backend at execution time?
- # Checking QueryEngine implementation... it takes `store_backend` in __init__.
- # So we can't easily have a global one without a store.
- # We'll rely on the store's engine.
- pass
- return None # Deprecated use of global engine
-
-
def _get_bulk_loader() -> BulkLoader:
"""Get or create global BulkLoader instance."""
global _global_bulk_loader
diff --git a/semantica/triplet_store/oxigraph_store.py b/semantica/triplet_store/oxigraph_store.py
index 5262c00a..4ba94f03 100644
--- a/semantica/triplet_store/oxigraph_store.py
+++ b/semantica/triplet_store/oxigraph_store.py
@@ -42,6 +42,10 @@ class OxigraphStore:
ProcessingError: If the store cannot be opened.
"""
self.logger = get_logger("oxigraph_store")
+ # Accept storage_path as an alias for path (matches the convention used
+ # by other Semantica stores). Pop it so it isn't left in self.config.
+ if path is None and "storage_path" in config:
+ path = config.pop("storage_path")
self.config = config
self.path = path if path is not None else config.get("path")
@@ -74,24 +78,65 @@ class OxigraphStore:
) from exc
def add_triplet(self, triplet: Triplet, **options) -> Dict[str, Any]:
- """Add one triplet to the default graph or ``options['graph']``."""
- return self.add_triplets([triplet], **options)
+ """Add one triplet to the default graph or ``options['graph']``.
- def add_triplets(self, triplets: List[Triplet], **options) -> Dict[str, Any]:
- """Add triplets in one native Oxigraph batch."""
+ The write is committed to the store's in-memory state immediately.
+ pyoxigraph's background threads will persist it to disk shortly
+ afterward; call :meth:`flush` explicitly if you need a synchronous
+ durability guarantee before reopening or crashing.
+ """
try:
graph_name = self._graph_name(options.get("graph"))
- quads = [self._to_quad(triplet, graph_name) for triplet in triplets]
- self.store.extend(quads)
+ self.store.extend([self._to_quad(triplet, graph_name)])
return {
"success": True,
- "triplets_loaded": len(triplets),
+ "triplets_loaded": 1,
"graph": options.get("graph"),
}
except Exception as exc:
self.logger.error(f"Oxigraph load failed: {exc}")
raise ProcessingError(f"Oxigraph load failed: {exc}") from exc
+ def add_triplets(self, triplets: List[Triplet], **options) -> Dict[str, Any]:
+ """Add triplets in one native Oxigraph batch.
+
+ The batch is written transactionally and then explicitly flushed to
+ disk before returning. This makes the full batch durable without
+ requiring a separate :meth:`flush` call. In-memory stores skip the
+ flush (there is nothing to sync).
+
+ For high-volume imports the :class:`~.bulk_loader.BulkLoader` splits
+ work into chunks and calls this method once per chunk, so each chunk
+ lands as one atomic, durable unit.
+ """
+ try:
+ graph_name = self._graph_name(options.get("graph"))
+ quads = [self._to_quad(triplet, graph_name) for triplet in triplets]
+ self.store.extend(quads)
+ except Exception as exc:
+ self.logger.error(f"Oxigraph load failed: {exc}")
+ raise ProcessingError(f"Oxigraph load failed: {exc}") from exc
+
+ # Flush is kept outside the write try/except so that a flush I/O error
+ # does not produce a misleading "load failed" message when extend()
+ # already committed the batch successfully.
+ if self.path is not None:
+ try:
+ self.flush()
+ except OSError as exc:
+ self.logger.warning(
+ f"Oxigraph flush failed after successful write: {exc}"
+ )
+ raise ProcessingError(
+ f"Oxigraph flush failed after successful write: {exc}"
+ ) from exc
+
+ return {
+ "success": True,
+ "triplets_loaded": len(triplets),
+ "graph": options.get("graph"),
+ }
+
def bulk_load(self, triplets: List[Triplet], **options) -> Dict[str, Any]:
"""Load a batch of triplets using Oxigraph's native bulk operation."""
return self.add_triplets(triplets, **options)
diff --git a/tests/context/test_decision_persistence_pr967.py b/tests/context/test_decision_persistence_pr967.py
new file mode 100644
index 00000000..dd96fed4
--- /dev/null
+++ b/tests/context/test_decision_persistence_pr967.py
@@ -0,0 +1,1041 @@
+"""
+Regression tests for PR #967 — Decision persistence / index consistency,
+CJK similarity, query_graph limit, and MCP tool correctness.
+
+These tests cover every bug confirmed by the pre-PR investigation and every
+new correctness issue introduced or left behind by the PR:
+
+ 1. save → load → find_similar_decisions (core persistence invariant)
+ 2. save → load → metadata preservation
+ 3. save → load → all decision analytics callable
+ 4. Repeated load clears stale indexes (no ghost decisions)
+ 5. In-memory decisions cleared when loading into a graph that already has
+ decisions recorded in-memory
+ 6. Category filtering via find_nodes / query_decisions
+ 7. Decision index stays consistent after add_node_attribute mutation
+ 8. CJK similarity — short CJK query matches relevant text
+ 9. Bigram spike regression — 2-char English query must NOT produce 1.0
+ 10. English similarity still works normally
+ 11. Empty / 1-char input safety
+ 12. query_graph limit=0 returns empty (not unlimited)
+ 13. query_graph limit=None returns all
+ 14. query_graph limit=1 caps combined results
+ 15. query_graph inbound-only topology
+ 16. query_graph outbound-only topology
+ 17. query_graph mixed inbound + outbound
+ 18. from_dict also rebuilds decision indexes
+ 19. MCP _get_graph loads from SEMANTICA_KG_PATH
+ 20. update_node smoke test + decision index sync
+ 21. delete_node soft-archive smoke test
+ 22. update_node / delete_node persistence after reload
+ 23. entity extraction returns surface text
+"""
+
+import json
+import os
+import sys
+import tempfile
+import unittest
+from collections import defaultdict
+from unittest.mock import patch
+
+# Make sure the repo root is importable even when running from the tests dir.
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")))
+
+from semantica.context.context_graph import ContextGraph
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _decision_graph() -> ContextGraph:
+ """Return a ContextGraph with three decisions pre-recorded."""
+ g = ContextGraph(advanced_analytics=False)
+ g.record_decision(
+ category="loan_approval",
+ scenario="High-income applicant with perfect credit history",
+ reasoning="Credit score 800+, stable employment for 10 years",
+ outcome="approved",
+ confidence=0.95,
+ entities=["applicant_123", "bank_abc"],
+ decision_maker="underwriter",
+ metadata={"risk_tier": "low", "custom_flag": True},
+ )
+ g.record_decision(
+ category="loan_approval",
+ scenario="Self-employed applicant with variable income",
+ reasoning="Good credit but income variability poses moderate risk",
+ outcome="conditional_approval",
+ confidence=0.72,
+ entities=["applicant_456"],
+ decision_maker="underwriter",
+ )
+ g.record_decision(
+ category="fraud_detection",
+ scenario="Unusual transaction pattern detected in account",
+ reasoning="Multiple small transactions in rapid succession across geographies",
+ outcome="flagged",
+ confidence=0.88,
+ entities=["account_789", "transaction_seq"],
+ decision_maker="fraud_engine",
+ )
+ return g
+
+
+# ---------------------------------------------------------------------------
+# Part 1: Core persistence invariant — save → load → find_similar_decisions
+# ---------------------------------------------------------------------------
+
+class TestDecisionPersistenceRoundTrip(unittest.TestCase):
+ """save → load must produce decision-query-equivalent behaviour."""
+
+ def test_find_similar_decisions_after_reload(self):
+ """Core invariant: similarity search works after save/load."""
+ g = _decision_graph()
+ with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
+ path = f.name
+ try:
+ g.save_to_file(path)
+ g2 = ContextGraph(advanced_analytics=False)
+ g2.load_from_file(path)
+
+ # Query that matches the loan_approval decisions
+ results = g2.find_similar_decisions(
+ "credit history approval", max_results=5, min_similarity=0.01
+ )
+ self.assertGreater(len(results), 0, "Expected at least one match after reload")
+ # Each result must be a dict with a decision key
+ self.assertIn("decision", results[0])
+ finally:
+ os.unlink(path)
+
+ def test_find_precedents_by_scenario_after_reload(self):
+ """find_precedents_by_scenario must not return [] after reload."""
+ g = _decision_graph()
+ with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
+ path = f.name
+ try:
+ g.save_to_file(path)
+ g2 = ContextGraph(advanced_analytics=False)
+ g2.load_from_file(path)
+
+ results = g2.find_precedents_by_scenario(
+ "applicant credit history loan",
+ similarity_threshold=0.01,
+ )
+ self.assertGreater(len(results), 0)
+ finally:
+ os.unlink(path)
+
+ def test_decision_count_after_reload(self):
+ """_decisions must be populated for statistics calls after reload."""
+ g = _decision_graph()
+ with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
+ path = f.name
+ try:
+ g.save_to_file(path)
+ g2 = ContextGraph(advanced_analytics=False)
+ g2.load_from_file(path)
+
+ stats = g2.get_decision_insights()
+ # Should not be the "No decisions" sentinel
+ self.assertNotEqual(stats, {"message": "No decisions recorded yet"})
+ self.assertEqual(stats.get("total_decisions", 0), 3)
+ finally:
+ os.unlink(path)
+
+ def test_decisions_dict_populated_after_reload(self):
+ """_decisions must exist and contain 3 entries after reload."""
+ g = _decision_graph()
+ with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
+ path = f.name
+ try:
+ g.save_to_file(path)
+ g2 = ContextGraph(advanced_analytics=False)
+ g2.load_from_file(path)
+
+ self.assertTrue(hasattr(g2, "_decisions"))
+ self.assertEqual(len(g2._decisions), 3)
+ finally:
+ os.unlink(path)
+
+
+# ---------------------------------------------------------------------------
+# Part 2: Metadata preservation across save/load
+# ---------------------------------------------------------------------------
+
+class TestDecisionMetadataPreservation(unittest.TestCase):
+
+ def test_custom_metadata_survives_reload(self):
+ """User-supplied metadata must survive a save → load round-trip."""
+ g = ContextGraph(advanced_analytics=False)
+ did = g.record_decision(
+ category="test",
+ scenario="Testing metadata preservation",
+ reasoning="Verifying that custom fields survive reload",
+ outcome="pass",
+ confidence=0.9,
+ metadata={"foo": "bar", "priority": 42},
+ )
+ with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
+ path = f.name
+ try:
+ g.save_to_file(path)
+ g2 = ContextGraph(advanced_analytics=False)
+ g2.load_from_file(path)
+
+ reloaded_decision = g2._decisions.get(did)
+ self.assertIsNotNone(reloaded_decision, "_decisions must contain the decision after reload")
+ # Metadata should contain the custom fields
+ meta = reloaded_decision.get("metadata", {})
+ self.assertEqual(meta.get("foo"), "bar")
+ self.assertEqual(meta.get("priority"), 42)
+ finally:
+ os.unlink(path)
+
+ def test_core_fields_preserved_after_reload(self):
+ """All core decision fields must survive a round-trip unchanged."""
+ g = ContextGraph(advanced_analytics=False)
+ did = g.record_decision(
+ category="compliance",
+ scenario="Regulatory check for derivative trade",
+ reasoning="Trade complies with Dodd-Frank Section 732",
+ outcome="compliant",
+ confidence=0.85,
+ entities=["trader_X", "instrument_Y"],
+ decision_maker="compliance_engine",
+ )
+ recorded_at_before = g._decisions[did]["recorded_at"]
+ self.assertTrue(recorded_at_before, "recorded_at must be set at record time")
+
+ with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
+ path = f.name
+ try:
+ g.save_to_file(path)
+ g2 = ContextGraph(advanced_analytics=False)
+ g2.load_from_file(path)
+
+ dec = g2._decisions.get(did)
+ self.assertIsNotNone(dec)
+ self.assertEqual(dec["category"], "compliance")
+ self.assertIn("Regulatory check", dec["scenario"])
+ self.assertEqual(dec["outcome"], "compliant")
+ self.assertAlmostEqual(dec["confidence"], 0.85, places=3)
+ self.assertIn("trader_X", dec["entities"])
+ self.assertEqual(dec["decision_maker"], "compliance_engine")
+ self.assertEqual(dec["recorded_at"], recorded_at_before,
+ "recorded_at must survive a save -> load round trip")
+ finally:
+ os.unlink(path)
+
+
+# ---------------------------------------------------------------------------
+# Part 3: Repeated load clears stale indexes
+# ---------------------------------------------------------------------------
+
+class TestRepeatedLoadClearsStaleIndexes(unittest.TestCase):
+
+ def test_second_load_replaces_first(self):
+ """Loading file B into a graph that already loaded file A must leave
+ only B's decisions visible — no ghost decisions from A."""
+ g_a = ContextGraph(advanced_analytics=False)
+ g_a.record_decision(
+ category="cat_A",
+ scenario="Decision from file A",
+ reasoning="Reason A",
+ outcome="outcome_A",
+ confidence=0.9,
+ )
+
+ g_b = ContextGraph(advanced_analytics=False)
+ g_b.record_decision(
+ category="cat_B",
+ scenario="Decision from file B",
+ reasoning="Reason B",
+ outcome="outcome_B",
+ confidence=0.8,
+ )
+
+ with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as fa, \
+ tempfile.NamedTemporaryFile(suffix=".json", delete=False) as fb:
+ path_a, path_b = fa.name, fb.name
+ try:
+ g_a.save_to_file(path_a)
+ g_b.save_to_file(path_b)
+
+ target = ContextGraph(advanced_analytics=False)
+
+ # Load A
+ target.load_from_file(path_a)
+ self.assertEqual(len(target._decisions), 1)
+ cats_after_a = {d["category"] for d in target._decisions.values()}
+ self.assertIn("cat_A", cats_after_a)
+
+ # Load B into same instance — must replace A entirely
+ target.load_from_file(path_b)
+ self.assertEqual(len(target._decisions), 1,
+ "Stale cat_A decision must not persist after loading B")
+ cats_after_b = {d["category"] for d in target._decisions.values()}
+ self.assertIn("cat_B", cats_after_b)
+ self.assertNotIn("cat_A", cats_after_b)
+ finally:
+ os.unlink(path_a)
+ os.unlink(path_b)
+
+ def test_load_into_graph_with_in_memory_decisions(self):
+ """Loading a file into a graph that already has in-memory decisions
+ must produce indexes that reflect ONLY the file's decisions."""
+ g = ContextGraph(advanced_analytics=False)
+ # Record an in-memory decision first
+ g.record_decision(
+ category="in_memory",
+ scenario="Decision recorded before load",
+ reasoning="Testing stale index reset",
+ outcome="ok",
+ confidence=0.5,
+ )
+
+ # Now create a graph file with different content
+ g_file = ContextGraph(advanced_analytics=False)
+ g_file.record_decision(
+ category="from_file",
+ scenario="Decision loaded from file",
+ reasoning="This is what should survive",
+ outcome="loaded",
+ confidence=0.7,
+ )
+ with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
+ path = f.name
+ try:
+ g_file.save_to_file(path)
+ g.load_from_file(path)
+
+ cats = {d["category"] for d in g._decisions.values()}
+ self.assertIn("from_file", cats)
+ self.assertNotIn("in_memory", cats,
+ "In-memory decision must be evicted after load_from_file")
+ finally:
+ os.unlink(path)
+
+
+# ---------------------------------------------------------------------------
+# Part 4: Category filtering
+# ---------------------------------------------------------------------------
+
+class TestCategoryFiltering(unittest.TestCase):
+
+ def test_decision_index_correct_after_reload(self):
+ """_decision_index must map categories correctly after reload."""
+ g = _decision_graph()
+ with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
+ path = f.name
+ try:
+ g.save_to_file(path)
+ g2 = ContextGraph(advanced_analytics=False)
+ g2.load_from_file(path)
+
+ loan_ids = g2._decision_index.get("loan_approval", set())
+ fraud_ids = g2._decision_index.get("fraud_detection", set())
+
+ self.assertEqual(len(loan_ids), 2, "Expected 2 loan_approval decisions")
+ self.assertEqual(len(fraud_ids), 1, "Expected 1 fraud_detection decision")
+ # No overlap
+ self.assertTrue(loan_ids.isdisjoint(fraud_ids))
+ finally:
+ os.unlink(path)
+
+ def test_find_nodes_category_in_metadata(self):
+ """find_nodes returns category inside 'metadata', not at top level."""
+ g = ContextGraph(advanced_analytics=False)
+ g.record_decision(
+ category="risk_check",
+ scenario="Scenario",
+ reasoning="Reasoning",
+ outcome="pass",
+ confidence=0.9,
+ )
+ nodes = g.find_nodes(node_type="decision")
+ self.assertGreater(len(nodes), 0)
+ # Category must be accessible via metadata key
+ found = any(
+ n.get("metadata", {}).get("category") == "risk_check"
+ for n in nodes
+ )
+ self.assertTrue(found, "category must be in n['metadata']['category']")
+ # Must NOT be at top level (that is the bug that was fixed)
+ top_level = any(n.get("category") == "risk_check" for n in nodes)
+ self.assertFalse(top_level, "category must NOT appear at the top level of find_nodes result")
+
+
+# ---------------------------------------------------------------------------
+# Part 5: Decision index consistency after mutation
+# ---------------------------------------------------------------------------
+
+class TestDecisionIndexMutationSync(unittest.TestCase):
+
+ def test_add_node_attribute_syncs_decision_index(self):
+ """After add_node_attribute on a decision node, _decisions must reflect
+ the new values without requiring a reload."""
+ g = ContextGraph(advanced_analytics=False)
+ did = g.record_decision(
+ category="original_cat",
+ scenario="Original scenario",
+ reasoning="Original reasoning",
+ outcome="original",
+ confidence=0.6,
+ )
+ # Verify original state
+ self.assertIn(did, g._decision_index.get("original_cat", set()))
+
+ # Mutate via add_node_attribute
+ g.add_node_attribute(did, {"confidence": 0.95, "custom_note": "reviewed"})
+
+ # _decisions must reflect updated confidence
+ updated = g._decisions.get(did)
+ self.assertIsNotNone(updated)
+ self.assertAlmostEqual(updated["confidence"], 0.95, places=3)
+ # custom_note should appear in metadata
+ self.assertEqual(updated["metadata"].get("custom_note"), "reviewed")
+
+ def test_update_node_does_not_leave_stale_index(self):
+ """update_node (via add_node_attribute) must not break category lookup."""
+ g = ContextGraph(advanced_analytics=False)
+ did = g.record_decision(
+ category="cat_original",
+ scenario="Some scenario",
+ reasoning="Some reasoning",
+ outcome="ok",
+ confidence=0.7,
+ )
+ # The decision should be findable by category
+ results_before = g.find_precedents_by_scenario(
+ "Some scenario", similarity_threshold=0.01
+ )
+ self.assertGreater(len(results_before), 0)
+
+ # Mutate some non-index fields
+ g.add_node_attribute(did, {"status": "reviewed", "reviewer": "alice"})
+
+ # Decision should still be findable after mutation
+ results_after = g.find_precedents_by_scenario(
+ "Some scenario", similarity_threshold=0.01
+ )
+ self.assertGreater(len(results_after), 0)
+
+
+# ---------------------------------------------------------------------------
+# Part 6: from_dict also rebuilds decision indexes
+# ---------------------------------------------------------------------------
+
+class TestFromDictDecisionIndexes(unittest.TestCase):
+
+ def test_from_dict_populates_decision_indexes(self):
+ """from_dict must rebuild _decisions, _decision_index, etc."""
+ g = _decision_graph()
+ d = g.to_dict()
+
+ g2 = ContextGraph(advanced_analytics=False)
+ g2.from_dict(d)
+
+ self.assertTrue(hasattr(g2, "_decisions"))
+ self.assertEqual(len(g2._decisions), 3)
+ self.assertGreater(len(g2._decision_index), 0)
+
+ def test_from_dict_repeated_call_clears_stale(self):
+ """Calling from_dict twice must not accumulate ghost entries."""
+ g1 = ContextGraph(advanced_analytics=False)
+ g1.record_decision(
+ category="x", scenario="s", reasoning="r", outcome="o", confidence=0.5
+ )
+ g2 = ContextGraph(advanced_analytics=False)
+ g2.record_decision(
+ category="y", scenario="s2", reasoning="r2", outcome="o2", confidence=0.6
+ )
+
+ target = ContextGraph(advanced_analytics=False)
+ target.from_dict(g1.to_dict())
+ self.assertEqual(len(target._decisions), 1)
+ cats = {d["category"] for d in target._decisions.values()}
+ self.assertIn("x", cats)
+
+ target.from_dict(g2.to_dict())
+ self.assertEqual(len(target._decisions), 1)
+ cats2 = {d["category"] for d in target._decisions.values()}
+ self.assertIn("y", cats2)
+ self.assertNotIn("x", cats2)
+
+
+# ---------------------------------------------------------------------------
+# Part 7: CJK similarity
+# ---------------------------------------------------------------------------
+
+class TestCJKSimilarity(unittest.TestCase):
+
+ def _sim(self, scenario, decision_scenario, decision_reasoning="", entities=None):
+ """Helper: compute _calculate_decision_content_similarity directly."""
+ g = ContextGraph(advanced_analytics=False)
+ decision = {
+ "scenario": decision_scenario,
+ "reasoning": decision_reasoning,
+ "entities": entities or [],
+ }
+ return g._calculate_decision_content_similarity(scenario, decision)
+
+ def test_cjk_two_char_query_matches_relevant_text(self):
+ """A 2-character CJK query should match text containing those chars."""
+ # 中文 (Chinese text) — 2 chars, produces 1 bigram: not enough for
+ # bigram signal. But a 3-char query should work.
+ # Use a 4-char CJK phrase (→ 3 bigrams) to activate the bigram path.
+ query = "中文审批" # 4 CJK chars → 3 bigrams
+ doc_scenario = "中文审批流程 贷款决策"
+ sim = self._sim(query, doc_scenario)
+ self.assertGreater(sim, 0.0, "CJK query must produce a non-zero similarity")
+
+ def test_cjk_irrelevant_text_low_similarity(self):
+ """A CJK query must NOT produce high similarity with unrelated text."""
+ query = "中文审批"
+ unrelated = "Python programming language feature request"
+ sim = self._sim(query, unrelated)
+ # Some accidental bigram overlap is possible with stripped chars, but
+ # should be significantly less than 1.0
+ self.assertLess(sim, 0.5)
+
+ def test_cjk_identical_text_high_similarity(self):
+ """Identical CJK text must produce similarity close to 1.0."""
+ text = "中文审批流程决策" # 8 chars → 7 bigrams
+ sim = self._sim(text, text)
+ self.assertGreater(sim, 0.9)
+
+
+# ---------------------------------------------------------------------------
+# Part 8: Bigram spike regression (2-char English query must NOT give 1.0)
+# ---------------------------------------------------------------------------
+
+class TestBigramSpikeRegression(unittest.TestCase):
+
+ def _sim(self, scenario, doc_scenario):
+ g = ContextGraph(advanced_analytics=False)
+ return g._calculate_decision_content_similarity(
+ scenario, {"scenario": doc_scenario, "reasoning": "", "entities": []}
+ )
+
+ def test_two_char_english_query_no_spike(self):
+ """A 2-char English query must NOT receive similarity 1.0 merely
+ because those chars appear as a substring in the document text."""
+ # 'in' is a 2-char query → 1 bigram → below the 3-bigram threshold.
+ # Word-based Jaccard also gives 0.0 ('in' not a word in the doc).
+ sim = self._sim("in", "interest rate decision analysis")
+ self.assertLess(sim, 0.5,
+ "2-char English query 'in' must not spike to 1.0")
+
+ def test_single_char_query_safe(self):
+ """A single-character query must return 0.0 without crashing."""
+ sim = self._sim("a", "apple analysis algorithm")
+ self.assertEqual(sim, 0.0)
+
+ def test_empty_query_safe(self):
+ """An empty query must return 0.0 without crashing."""
+ sim = self._sim("", "some decision text here")
+ self.assertEqual(sim, 0.0)
+
+ def test_normal_english_similarity_preserved(self):
+ """Normal English word overlap must still produce reasonable scores."""
+ sim = self._sim(
+ "credit approval loan applicant",
+ "loan applicant credit history approval decision",
+ )
+ self.assertGreater(sim, 0.3, "Normal English similarity must remain reasonable")
+
+ def test_common_bigram_substring_below_threshold(self):
+ """2-char queries 'al', 'ba', 'at' must not produce similarity 1.0."""
+ for q in ("al", "ba", "at", "re"):
+ sim = self._sim(q, "algorithm alignment base rate attention")
+ self.assertLess(sim, 0.5,
+ f"2-char query {q!r} must not produce high similarity")
+
+ def test_unrelated_multiword_english_queries_score_zero(self):
+ """The bigram fallback must not activate for ordinary multi-word
+ English queries -- it exists only for CJK/single-token queries where
+ whitespace tokenisation can't help. Unrelated multi-word English
+ sentences must score 0.0, not a nonzero incidental bigram overlap."""
+ sim = self._sim(
+ "employee vacation request approval process",
+ "Server infrastructure migration to cloud provider",
+ )
+ self.assertEqual(sim, 0.0,
+ "Unrelated multi-word English queries must not "
+ "receive a nonzero score from bigram overlap")
+
+
+# ---------------------------------------------------------------------------
+# Part 9: query_graph limit semantics
+# ---------------------------------------------------------------------------
+
+class TestQueryGraphLimitSemantics(unittest.TestCase):
+ """Tests for _tool_query_graph limit correctness."""
+
+ def _make_graph_and_patch(self):
+ """Build a simple graph and patch _get_graph to return it."""
+ g = ContextGraph(advanced_analytics=False)
+ g.add_node("center", "hub", label="Center")
+ for i in range(5):
+ g.add_node(f"out_{i}", "spoke", label=f"Spoke {i}")
+ g.add_edge("center", f"out_{i}", "connects")
+ for i in range(3):
+ g.add_node(f"in_{i}", "feeder", label=f"Feeder {i}")
+ g.add_edge(f"in_{i}", "center", "feeds")
+ return g
+
+ def test_limit_none_returns_all(self):
+ """limit=None must return all neighbours (outbound + inbound)."""
+ from semantica.mcp_server import _tool_query_graph
+ g = self._make_graph_and_patch()
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ result = _tool_query_graph({
+ "mode": "neighbors",
+ "node_id": "center",
+ "depth": 1,
+ "limit": None,
+ })
+ neighbors = result.get("neighbors", [])
+ self.assertGreaterEqual(len(neighbors), 5, "Should include all outbound")
+
+ def test_limit_zero_returns_empty(self):
+ """limit=0 must return an empty neighbors list, not bypass the cap."""
+ from semantica.mcp_server import _tool_query_graph
+ g = self._make_graph_and_patch()
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ result = _tool_query_graph({
+ "mode": "neighbors",
+ "node_id": "center",
+ "depth": 1,
+ "limit": 0,
+ })
+ neighbors = result.get("neighbors", [])
+ self.assertEqual(neighbors, [],
+ "limit=0 must produce an empty result, not bypass the cap")
+
+ def test_limit_one_caps_result(self):
+ """limit=1 must return exactly 1 neighbour regardless of total."""
+ from semantica.mcp_server import _tool_query_graph
+ g = self._make_graph_and_patch()
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ result = _tool_query_graph({
+ "mode": "neighbors",
+ "node_id": "center",
+ "depth": 1,
+ "limit": 1,
+ })
+ neighbors = result.get("neighbors", [])
+ self.assertEqual(len(neighbors), 1)
+
+ def test_limit_larger_than_available(self):
+ """limit > total results must return all available without error."""
+ from semantica.mcp_server import _tool_query_graph
+ g = self._make_graph_and_patch()
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ result = _tool_query_graph({
+ "mode": "neighbors",
+ "node_id": "center",
+ "depth": 1,
+ "limit": 1000,
+ })
+ self.assertNotIn("error", result)
+ neighbors = result.get("neighbors", [])
+ # center has 5 outbound + 3 inbound = 8 total
+ self.assertGreaterEqual(len(neighbors), 5)
+
+ def test_outbound_only_topology(self):
+ """Nodes with only outbound edges must return outbound neighbours."""
+ from semantica.mcp_server import _tool_query_graph
+ g = ContextGraph(advanced_analytics=False)
+ g.add_node("source", "hub")
+ g.add_node("dest1", "leaf")
+ g.add_node("dest2", "leaf")
+ g.add_edge("source", "dest1", "points_to")
+ g.add_edge("source", "dest2", "points_to")
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ result = _tool_query_graph({
+ "mode": "neighbors",
+ "node_id": "source",
+ "depth": 1,
+ })
+ neighbors = result.get("neighbors", [])
+ directions = {n["direction"] for n in neighbors}
+ self.assertIn("out", directions)
+
+ def test_inbound_only_topology(self):
+ """Nodes with only inbound edges must return inbound neighbours."""
+ from semantica.mcp_server import _tool_query_graph
+ g = ContextGraph(advanced_analytics=False)
+ g.add_node("sink", "hub")
+ g.add_node("src1", "feeder")
+ g.add_node("src2", "feeder")
+ g.add_edge("src1", "sink", "feeds")
+ g.add_edge("src2", "sink", "feeds")
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ result = _tool_query_graph({
+ "mode": "neighbors",
+ "node_id": "sink",
+ "depth": 1,
+ })
+ neighbors = result.get("neighbors", [])
+ directions = {n["direction"] for n in neighbors}
+ self.assertIn("in", directions)
+ self.assertNotIn("out", directions)
+
+ def test_mixed_inbound_outbound(self):
+ """A node with both inbound and outbound edges returns both directions."""
+ from semantica.mcp_server import _tool_query_graph
+ g = ContextGraph(advanced_analytics=False)
+ g.add_node("mid", "hub")
+ g.add_node("up", "parent")
+ g.add_node("down", "child")
+ g.add_edge("up", "mid", "parent_of")
+ g.add_edge("mid", "down", "child_of")
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ result = _tool_query_graph({
+ "mode": "neighbors",
+ "node_id": "mid",
+ "depth": 1,
+ })
+ neighbors = result.get("neighbors", [])
+ directions = {n["direction"] for n in neighbors}
+ self.assertIn("in", directions)
+ self.assertIn("out", directions)
+
+
+# ---------------------------------------------------------------------------
+# Part 10: MCP _get_graph loads from SEMANTICA_KG_PATH
+# ---------------------------------------------------------------------------
+
+class TestMCPGetGraphLoadsFromPath(unittest.TestCase):
+
+ def test_get_graph_loads_kg_path(self):
+ """When SEMANTICA_KG_PATH is set, _get_graph must load it."""
+ import semantica.mcp_server as mcp_mod
+
+ g = ContextGraph(advanced_analytics=False)
+ g.add_node("kg_node_1", "entity", label="Loaded from file")
+ g.record_decision(
+ category="test_load",
+ scenario="Testing KG path load",
+ reasoning="Verifying MCP server auto-load",
+ outcome="verified",
+ confidence=0.99,
+ )
+
+ with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
+ path = f.name
+ try:
+ g.save_to_file(path)
+
+ # Reset the module-level _graph so _get_graph re-initialises
+ original_graph = mcp_mod._graph
+ mcp_mod._graph = None
+ try:
+ with patch.dict(os.environ, {"SEMANTICA_KG_PATH": path}):
+ loaded_graph = mcp_mod._get_graph()
+ self.assertTrue(loaded_graph.has_node("kg_node_1"),
+ "Graph must contain node from persisted file")
+ # Decision indexes must also be rebuilt
+ self.assertTrue(
+ hasattr(loaded_graph, "_decisions") and loaded_graph._decisions,
+ "Decision indexes must be rebuilt when loading from SEMANTICA_KG_PATH"
+ )
+ finally:
+ mcp_mod._graph = original_graph
+ finally:
+ os.unlink(path)
+
+
+# ---------------------------------------------------------------------------
+# Part 11: update_node / delete_node smoke tests + decision sync
+# ---------------------------------------------------------------------------
+
+class TestUpdateDeleteNodeMCP(unittest.TestCase):
+
+ def _fresh_graph_with_decision(self):
+ g = ContextGraph(advanced_analytics=False)
+ g.add_node("task_1", "task", label="A task node")
+ did = g.record_decision(
+ category="project",
+ scenario="Scope definition for Q3",
+ reasoning="Requirements complete",
+ outcome="approved",
+ confidence=0.9,
+ )
+ return g, did
+
+ def test_update_node_returns_updated_properties(self):
+ """update_node must reflect new property values in its response."""
+ from semantica.mcp_server import _tool_update_node
+ g, _ = self._fresh_graph_with_decision()
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ result = _tool_update_node({
+ "node_id": "task_1",
+ "properties": {"status": "done", "note": "completed by alice"},
+ })
+ self.assertEqual(result.get("status"), "updated")
+ self.assertEqual(result.get("node_id"), "task_1")
+ # Verify node actually updated in graph
+ node = g.find_node("task_1")
+ self.assertEqual((node.get("metadata") or {}).get("status"), "done")
+
+ def test_update_node_nonexistent_returns_error(self):
+ """update_node on a nonexistent node must return an error dict."""
+ from semantica.mcp_server import _tool_update_node
+ g, _ = self._fresh_graph_with_decision()
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ result = _tool_update_node({
+ "node_id": "does_not_exist",
+ "properties": {"status": "done"},
+ })
+ self.assertIn("error", result)
+
+ def test_delete_node_soft_archives(self):
+ """delete_node must mark the node status='archived', not remove it."""
+ from semantica.mcp_server import _tool_delete_node
+ g, _ = self._fresh_graph_with_decision()
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ result = _tool_delete_node({"node_id": "task_1"})
+ self.assertEqual(result.get("status"), "archived")
+ # Node must still exist
+ node = g.find_node("task_1")
+ self.assertIsNotNone(node, "Node must still exist after soft-delete")
+ self.assertEqual((node.get("metadata") or {}).get("status"), "archived")
+
+ def test_delete_node_nonexistent_returns_error(self):
+ """delete_node on a nonexistent node must return an error dict."""
+ from semantica.mcp_server import _tool_delete_node
+ g, _ = self._fresh_graph_with_decision()
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ result = _tool_delete_node({"node_id": "ghost_id"})
+ self.assertIn("error", result)
+
+ def test_update_decision_node_syncs_index(self):
+ """update_node on a decision node must keep _decisions consistent."""
+ from semantica.mcp_server import _tool_update_node
+ g, did = self._fresh_graph_with_decision()
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ _tool_update_node({
+ "node_id": did,
+ "properties": {"status": "reviewed", "reviewer": "bob"},
+ })
+ # _decisions must reflect the new metadata
+ dec = g._decisions.get(did)
+ self.assertIsNotNone(dec)
+ self.assertEqual(dec["metadata"].get("reviewer"), "bob")
+
+ def test_update_delete_persist_after_reload(self):
+ """Changes made by update_node / delete_node must survive save → load."""
+ from semantica.mcp_server import _tool_update_node, _tool_delete_node
+ g, _ = self._fresh_graph_with_decision()
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ _tool_update_node({"node_id": "task_1",
+ "properties": {"status": "done"}})
+ _tool_delete_node.__wrapped__ = None # noop; we call the real fn below
+
+ # Manually save and reload
+ with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
+ path = f.name
+ try:
+ g.save_to_file(path)
+ g2 = ContextGraph(advanced_analytics=False)
+ g2.load_from_file(path)
+ node = g2.find_node("task_1")
+ self.assertIsNotNone(node)
+ self.assertEqual((node.get("metadata") or {}).get("status"), "done")
+ finally:
+ os.unlink(path)
+
+
+# ---------------------------------------------------------------------------
+# Part 12: MCP entity extraction surface text
+# ---------------------------------------------------------------------------
+
+class TestEntityExtractionSurfaceText(unittest.TestCase):
+
+ def test_extract_entities_returns_text_field(self):
+ """extract_entities must include a 'text' key with the surface form."""
+ from semantica.mcp_server import _tool_extract_entities
+
+ # Minimal smoke test: verify the response shape regardless of whether
+ # spaCy models are available. If no entities are found we skip the
+ # assertion on content but still verify no crash and no missing key
+ # structure.
+ try:
+ result = _tool_extract_entities({"text": "Apple announced new iPhone"})
+ except Exception as exc:
+ self.skipTest(f"NER dependency unavailable: {exc}")
+
+ if "error" in result:
+ # spaCy model not installed in this environment — acceptable skip
+ self.skipTest(f"NER not available: {result['error']}")
+
+ entities = result.get("entities", [])
+ for ent in entities:
+ self.assertIn("text", ent,
+ "Each entity must have a 'text' key with the surface form")
+ self.assertIn("label", ent,
+ "Each entity must have a 'label' key (NER category)")
+ self.assertIn("start", ent)
+ self.assertIn("end", ent)
+
+ def test_extract_entities_missing_text_returns_error(self):
+ """extract_entities with no text must return an error dict."""
+ from semantica.mcp_server import _tool_extract_entities
+ result = _tool_extract_entities({})
+ self.assertIn("error", result)
+
+ def test_extract_relations_missing_text_returns_error(self):
+ """extract_relations with no text must return an error dict."""
+ from semantica.mcp_server import _tool_extract_relations
+ result = _tool_extract_relations({})
+ self.assertIn("error", result)
+
+ def test_extract_relations_with_text_does_not_raise(self):
+ """extract_relations must not raise TypeError for missing `entities`
+ (RelationExtractor.extract_relations requires an `entities` arg;
+ the tool must supply one, e.g. by running NER first)."""
+ from semantica.mcp_server import _tool_extract_relations
+
+ try:
+ result = _tool_extract_relations({"text": "Apple announced new iPhone"})
+ except Exception as exc:
+ self.fail(f"extract_relations raised unexpectedly: {exc!r}")
+
+ self.assertNotIn("error", result,
+ "extract_relations should not error on valid text input")
+ self.assertIn("relations", result)
+ self.assertIn("triplets", result)
+
+
+# ---------------------------------------------------------------------------
+# Part 13: query_graph node / search modes
+# ---------------------------------------------------------------------------
+
+class TestQueryGraphNodeAndSearch(unittest.TestCase):
+
+ def _make_graph(self):
+ g = ContextGraph(advanced_analytics=False)
+ g.add_node("alpha", "concept", label="Alpha Concept")
+ g.add_node("beta", "concept", label="Beta Concept")
+ g.add_edge("alpha", "beta", "relates_to")
+ return g
+
+ def test_node_mode_existing(self):
+ """node mode must return the node dict for an existing id."""
+ from semantica.mcp_server import _tool_query_graph
+ g = self._make_graph()
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ result = _tool_query_graph({"mode": "node", "node_id": "alpha"})
+ self.assertIn("node", result)
+ self.assertIsNotNone(result["node"])
+
+ def test_node_mode_missing_id(self):
+ """node mode with no node_id must return an error."""
+ from semantica.mcp_server import _tool_query_graph
+ g = self._make_graph()
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ result = _tool_query_graph({"mode": "node"})
+ self.assertIn("error", result)
+
+ def test_search_mode_finds_matching(self):
+ """search mode must return nodes whose id or content contains the query."""
+ from semantica.mcp_server import _tool_query_graph
+ g = self._make_graph()
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ result = _tool_query_graph({"mode": "search", "query": "alpha"})
+ hits = result.get("results", [])
+ self.assertGreater(len(hits), 0)
+ ids = [h["id"] for h in hits]
+ self.assertIn("alpha", ids)
+
+ def test_search_mode_limit_respected(self):
+ """search mode must respect the limit parameter."""
+ from semantica.mcp_server import _tool_query_graph
+ g = ContextGraph(advanced_analytics=False)
+ for i in range(20):
+ g.add_node(f"item_{i}", "thing", label=f"item {i}")
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ result = _tool_query_graph({"mode": "search", "query": "item", "limit": 3})
+ self.assertLessEqual(len(result.get("results", [])), 3)
+
+ def test_search_mode_limit_zero_returns_empty(self):
+ """search mode with limit=0 must return no results."""
+ from semantica.mcp_server import _tool_query_graph
+ g = ContextGraph(advanced_analytics=False)
+ for i in range(5):
+ g.add_node(f"alpha_{i}", "thing", label=f"alpha item {i}")
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ result = _tool_query_graph({"mode": "search", "query": "alpha", "limit": 0})
+ hits = result.get("results", [])
+ self.assertEqual(hits, [], f"limit=0 must return empty, got {len(hits)} results")
+
+ def test_unknown_mode_returns_error(self):
+ """An unknown mode string must return an error."""
+ from semantica.mcp_server import _tool_query_graph
+ g = self._make_graph()
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ result = _tool_query_graph({"mode": "invalid_mode"})
+ self.assertIn("error", result)
+
+ def test_inbound_no_duplicates_when_multiple_edges(self):
+ """Multiple edges between same source→target must produce only one
+ inbound entry for the source node."""
+ from semantica.mcp_server import _tool_query_graph
+ g = ContextGraph(advanced_analytics=False)
+ g.add_node("hub", "center")
+ g.add_node("src", "node")
+ g.add_edge("src", "hub", "type_A")
+ g.add_edge("src", "hub", "type_B")
+ with patch("semantica.mcp_server._get_graph", return_value=g):
+ result = _tool_query_graph({"mode": "neighbors", "node_id": "hub", "depth": 1})
+ in_ids = [n["id"] for n in result.get("neighbors", []) if n.get("direction") == "in"]
+ self.assertEqual(in_ids.count("src"), 1,
+ "src must appear exactly once even with two edges")
+
+
+# ---------------------------------------------------------------------------
+# Part 14: clear() resets decision indexes
+# ---------------------------------------------------------------------------
+
+class TestClearResetsDecisionIndexes(unittest.TestCase):
+
+ def test_clear_removes_decision_indexes(self):
+ """clear() must reset _decisions so that decision queries return empty."""
+ g = ContextGraph(advanced_analytics=False)
+ g.record_decision(
+ category="test", scenario="s", reasoning="r", outcome="o", confidence=0.9
+ )
+ self.assertTrue(hasattr(g, "_decisions"))
+ self.assertEqual(len(g._decisions), 1)
+
+ g.clear()
+
+ # After clear, _decisions must be empty
+ self.assertEqual(len(getattr(g, "_decisions", {})), 0,
+ "_decisions must be empty after clear()")
+ # find_similar_decisions must return empty
+ results = g.find_similar_decisions("s", min_similarity=0.01)
+ self.assertEqual(results, [],
+ "find_similar_decisions must return [] after clear()")
+
+ def test_clear_then_record_works(self):
+ """clear() followed by record_decision must work correctly."""
+ g = ContextGraph(advanced_analytics=False)
+ g.record_decision(category="old", scenario="s", reasoning="r", outcome="o", confidence=0.9)
+ g.clear()
+ did = g.record_decision(
+ category="new", scenario="fresh decision", reasoning="fresh",
+ outcome="ok", confidence=0.8
+ )
+ self.assertEqual(len(g._decisions), 1)
+ self.assertIn(did, g._decisions)
+ self.assertEqual(g._decisions[did]["category"], "new")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/explorer/test_decision_route_timestamp.py b/tests/explorer/test_decision_route_timestamp.py
new file mode 100644
index 00000000..4b568619
--- /dev/null
+++ b/tests/explorer/test_decision_route_timestamp.py
@@ -0,0 +1,237 @@
+"""Regression tests for the /api/decisions 422 bug.
+
+``ContextGraph.record_decision()`` stores ``timestamp`` as a POSIX float
+(``datetime.now().timestamp()``). ``DecisionResponse.timestamp`` is typed
+``Optional[str]``. Without the ``_normalize_timestamp`` field-validator on
+``DecisionResponse`` the raw float fails Pydantic validation and every decision
+endpoint returns 422.
+
+The validator lives on ``DecisionResponse`` in ``semantica/explorer/schemas.py``
+and converts float/int epochs to ISO-8601 strings via
+``datetime.fromtimestamp(value, tz=timezone.utc).isoformat()``.
+
+``_node_to_decision()`` must pass the raw stored value through unchanged so the
+validator can do its job. A route-level ``str()`` cast would pre-empt the
+validator and produce raw numeric strings instead of ISO-8601, breaking the API
+contract and all callers that call ``datetime.fromisoformat()`` on the result.
+
+Each test below is written so that it *fails* when the route-level ``str()``
+cast is present (i.e. it would have caught the regression introduced by the
+incorrect fix).
+"""
+
+import math
+from datetime import datetime
+
+import pytest
+from fastapi.testclient import TestClient
+from pydantic import ValidationError
+
+from semantica.context.context_graph import ContextGraph
+from semantica.explorer.app import create_app
+from semantica.explorer.routes.decisions import _node_to_decision
+from semantica.explorer.schemas import DecisionResponse
+from semantica.explorer.session import GraphSession
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _decision_node(timestamp):
+ """Minimal node dict as returned by the graph session layer."""
+ return {
+ "id": "d-test",
+ "type": "decision",
+ "properties": {
+ "category": "loan_underwriting",
+ "scenario": "A-7291 review",
+ "reasoning": "DTI within policy",
+ "outcome": "approved",
+ "confidence": 0.94,
+ "timestamp": timestamp,
+ },
+ }
+
+
+def _recorded_client():
+ """TestClient backed by a graph built with real record_decision() calls.
+
+ This is the path that was broken in production: record_decision() stores
+ timestamp as a float epoch, which must come out the other side as an
+ ISO-8601 string, not a raw numeric string.
+ """
+ graph = ContextGraph(advanced_analytics=False)
+ graph.record_decision(
+ category="credit_application",
+ scenario="Personal loan, $85k income, 31% DTI",
+ reasoning="Income meets threshold; employment stable",
+ outcome="proceed_to_underwriting",
+ confidence=0.88,
+ entities=["applicant_A7291"],
+ )
+ return TestClient(create_app(session=GraphSession(graph)))
+
+
+# ---------------------------------------------------------------------------
+# Unit tests: _node_to_decision() → DecisionResponse
+#
+# Each assertion must fail when the route contains the incorrect str() cast:
+# timestamp=None if ... is None else str(properties.get("timestamp"))
+# because that cast turns floats into numeric strings such as "1786513069.69"
+# rather than ISO-8601 strings such as "2026-08-12T05:37:49+00:00".
+# ---------------------------------------------------------------------------
+
+def test_float_timestamp_becomes_iso8601():
+ """A POSIX-float epoch must be normalised to an ISO-8601 string.
+
+ Fails with the str() cast because str(1786513069.694965) ==
+ '1786513069.694965', which is not a valid isoformat string.
+ """
+ decision = _node_to_decision(_decision_node(timestamp=1786513069.694965))
+
+ assert isinstance(decision.timestamp, str)
+ # Must parse as a valid ISO-8601 datetime — this is the key assertion that
+ # the incorrect str() cast breaks.
+ parsed = datetime.fromisoformat(decision.timestamp)
+ # Round-trip: parsed timestamp must be within 1 s of the original epoch.
+ assert abs(parsed.timestamp() - 1786513069.694965) < 1.0
+
+
+def test_int_timestamp_becomes_iso8601():
+ """An integer epoch (no sub-second component) must also become ISO-8601.
+
+ Fails with the str() cast because str(1786513069) == '1786513069'.
+ """
+ decision = _node_to_decision(_decision_node(timestamp=1786513069))
+
+ assert isinstance(decision.timestamp, str)
+ parsed = datetime.fromisoformat(decision.timestamp)
+ assert abs(parsed.timestamp() - 1786513069) < 1.0
+
+
+def test_none_timestamp_stays_none():
+ """A stored None must remain None, not become the string 'None'."""
+ decision = _node_to_decision(_decision_node(timestamp=None))
+
+ assert decision.timestamp is None
+
+
+def test_missing_timestamp_key_stays_none():
+ """A node without a timestamp key at all must not raise and must be None."""
+ node = {
+ "id": "d-no-ts",
+ "type": "decision",
+ "properties": {"category": "x", "outcome": "y"},
+ }
+
+ decision = _node_to_decision(node)
+
+ assert decision.timestamp is None
+
+
+def test_iso_string_passes_through_unchanged():
+ """An already-ISO-8601 string must be returned verbatim."""
+ iso = "2026-08-12T10:04:20+00:00"
+ decision = _node_to_decision(_decision_node(timestamp=iso))
+
+ assert decision.timestamp == iso
+
+
+def test_nan_timestamp_raises_validation_error():
+ """NaN must be rejected by the validator, not silently accepted.
+
+ With the str() cast, str(nan) == 'nan' bypasses the validator's finiteness
+ check and is silently accepted — this test would pass the incorrect version
+ of the code if it expected 'nan', but it correctly expects a ValidationError.
+ """
+ with pytest.raises(ValidationError):
+ _node_to_decision(_decision_node(timestamp=float("nan")))
+
+
+def test_inf_timestamp_raises_validation_error():
+ """Positive infinity must be rejected, not silently accepted as 'inf'."""
+ with pytest.raises(ValidationError):
+ _node_to_decision(_decision_node(timestamp=float("inf")))
+
+
+def test_negative_inf_timestamp_raises_validation_error():
+ """Negative infinity must be rejected, not silently accepted as '-inf'."""
+ with pytest.raises(ValidationError):
+ _node_to_decision(_decision_node(timestamp=float("-inf")))
+
+
+def test_out_of_range_epoch_raises_validation_error():
+ """A millisecond epoch accidentally passed as seconds must be rejected.
+
+ With the str() cast, str(1723600000000) is silently accepted as a string.
+ The validator correctly raises ValidationError for out-of-range epochs.
+ """
+ with pytest.raises(ValidationError):
+ _node_to_decision(_decision_node(timestamp=1723600000000))
+
+
+# ---------------------------------------------------------------------------
+# Integration tests: full HTTP path through TestClient
+#
+# These exercise the complete production path:
+# record_decision() → float stored in graph → HTTP GET → JSON response
+#
+# They are the definitive check: if the route emits numeric strings instead of
+# ISO-8601 the fromisoformat() assertion below fails immediately.
+# ---------------------------------------------------------------------------
+
+def test_list_decisions_float_timestamp_serialised_as_iso8601():
+ """GET /api/decisions must return ISO-8601 timestamps for all decisions.
+
+ This is the exact production failure path. record_decision() stores
+ timestamp as a float; the endpoint must return an ISO-8601 string, not a
+ raw numeric string like '1786513069.69'.
+ """
+ with _recorded_client() as client:
+ response = client.get("/api/decisions")
+
+ assert response.status_code == 200
+ payload = response.json()
+ assert len(payload) >= 1
+
+ for item in payload:
+ ts = item["timestamp"]
+ assert isinstance(ts, str), f"timestamp must be str, got {type(ts)}"
+ # This is the line that fails when the str() cast is present:
+ datetime.fromisoformat(ts)
+
+
+def test_get_decision_float_timestamp_serialised_as_iso8601():
+ """GET /api/decisions/{id} must return an ISO-8601 timestamp."""
+ with _recorded_client() as client:
+ decision_id = client.get("/api/decisions").json()[0]["decision_id"]
+ response = client.get(f"/api/decisions/{decision_id}")
+
+ assert response.status_code == 200
+ ts = response.json()["timestamp"]
+ assert isinstance(ts, str)
+ datetime.fromisoformat(ts)
+
+
+def test_get_precedents_float_timestamp_serialised_as_iso8601():
+ """GET /api/decisions/{id}/precedents must return ISO-8601 timestamps."""
+ graph = ContextGraph(advanced_analytics=False)
+ for i in range(3):
+ graph.record_decision(
+ category="risk",
+ scenario=f"loan assessment scenario {i}",
+ reasoning="standard criteria",
+ outcome="approved",
+ confidence=0.9,
+ )
+
+ with TestClient(create_app(session=GraphSession(graph))) as client:
+ decision_id = client.get("/api/decisions").json()[0]["decision_id"]
+ response = client.get(f"/api/decisions/{decision_id}/precedents")
+
+ assert response.status_code == 200
+ for item in response.json():
+ ts = item["timestamp"]
+ assert isinstance(ts, str)
+ datetime.fromisoformat(ts)
diff --git a/tests/export/test_rdf_literal_escaping.py b/tests/export/test_rdf_literal_escaping.py
new file mode 100644
index 00000000..8a3c2354
--- /dev/null
+++ b/tests/export/test_rdf_literal_escaping.py
@@ -0,0 +1,107 @@
+"""Regression tests for #1098: Turtle/N-Triples literal escaping."""
+import pytest
+
+from semantica.export.rdf_exporter import RDFExporter, RDFSerializer
+
+
+@pytest.fixture
+def serializer():
+ return RDFSerializer()
+
+
+class TestTurtleLiteralEscaping:
+ def test_quote_in_text_is_escaped(self, serializer):
+ kg = {
+ "entities": [{"id": "e1", "text": 'He said "hello"', "type": "ORG"}],
+ "relationships": [],
+ }
+ turtle = serializer.serialize_to_turtle(kg)
+ assert '"He said \\"hello\\""' in turtle
+
+ def test_backslash_in_text_is_escaped(self, serializer):
+ kg = {
+ "entities": [{"id": "e1", "text": r"path\to\file", "type": "ORG"}],
+ "relationships": [],
+ }
+ turtle = serializer.serialize_to_turtle(kg)
+ assert r"path\\to\\file" in turtle
+
+ def test_newline_in_text_is_escaped(self, serializer):
+ kg = {
+ "entities": [{"id": "e1", "text": "line1\nline2", "type": "ORG"}],
+ "relationships": [],
+ }
+ turtle = serializer.serialize_to_turtle(kg)
+ assert "line1\\nline2" in turtle
+
+ def test_tab_in_text_is_escaped(self, serializer):
+ kg = {
+ "entities": [{"id": "e1", "text": "a\tb", "type": "ORG"}],
+ "relationships": [],
+ }
+ turtle = serializer.serialize_to_turtle(kg)
+ assert "a\\tb" in turtle
+
+ def test_plain_text_unchanged(self, serializer):
+ kg = {
+ "entities": [{"id": "e1", "text": "Apple Inc.", "type": "ORG"}],
+ "relationships": [],
+ }
+ turtle = serializer.serialize_to_turtle(kg)
+ assert 'semantica:text "Apple Inc."' in turtle
+
+
+class TestNTriplesLiteralEscaping:
+ def test_quote_in_text_is_escaped(self, serializer):
+ kg = {
+ "entities": [{"id": "e1", "text": 'He said "hello"', "type": "ORG"}],
+ "relationships": [],
+ }
+ ntriples = serializer.serialize_to_ntriples(kg)
+ assert '\\"hello\\"' in ntriples
+
+ def test_backslash_in_text_is_escaped(self, serializer):
+ kg = {
+ "entities": [{"id": "e1", "text": r"path\to\file", "type": "ORG"}],
+ "relationships": [],
+ }
+ ntriples = serializer.serialize_to_ntriples(kg)
+ assert r"path\\to\\file" in ntriples
+
+ def test_newline_in_text_is_escaped(self, serializer):
+ kg = {
+ "entities": [{"id": "e1", "text": "line1\nline2", "type": "ORG"}],
+ "relationships": [],
+ }
+ ntriples = serializer.serialize_to_ntriples(kg)
+ assert "line1\\nline2" in ntriples
+
+ def test_tab_in_text_is_escaped(self, serializer):
+ kg = {
+ "entities": [{"id": "e1", "text": "a\tb", "type": "ORG"}],
+ "relationships": [],
+ }
+ ntriples = serializer.serialize_to_ntriples(kg)
+ assert "a\\tb" in ntriples
+
+
+class TestOWLTimeLiteralEscaping:
+ """Timestamp literals in OWL-Time turtle output must also be escaped."""
+
+ def test_owl_time_timestamps_are_escaped(self):
+ exporter = RDFExporter()
+ kg = {
+ "entities": [],
+ "relationships": [
+ {
+ "id": "r1",
+ "source_id": "a",
+ "target_id": "b",
+ "type": "works_for",
+ "valid_from": "2020-01-01T00:00:00Z",
+ "valid_until": None,
+ }
+ ],
+ }
+ turtle = exporter.export_to_rdf(kg, format="turtle", include_temporal=True)
+ assert 'time:inXSDDateTimeStamp "2020-01-01T00:00:00Z"' in turtle
\ No newline at end of file
diff --git a/tests/ingest/test_ontology_named_graph.py b/tests/ingest/test_ontology_named_graph.py
new file mode 100644
index 00000000..baa8dea9
--- /dev/null
+++ b/tests/ingest/test_ontology_named_graph.py
@@ -0,0 +1,92 @@
+"""A JSON-LD ontology whose terms live in a named graph must not be silently dropped.
+
+A JSON-LD document with a top-level ``@id`` *and* ``@graph`` places its terms in a NAMED
+graph. ``rdflib.Graph.parse()`` loads only the default graph and discards the rest without
+raising, so every class and property in such a document disappeared while the load reported
+success — see issue #1129 for the reproduction through the public API.
+
+This is the same ``Graph`` -> ``Dataset`` migration #757 made for ``JenaStore`` (#756); the
+ingest path was not covered by it.
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from semantica.ingest.ontology_ingestor import OntologyIngestor
+
+NAMED_GRAPH_ONTOLOGY = {
+ "@context": {
+ "ex": "https://example.org/ns#",
+ "owl": "http://www.w3.org/2002/07/owl#",
+ "rdfs": "http://www.w3.org/2000/01/rdf-schema#",
+ },
+ "@id": "https://example.org/ns",
+ "@type": "owl:Ontology",
+ "@graph": [
+ {"@id": "ex:Thing", "@type": "owl:Class", "rdfs:label": "Thing"},
+ {"@id": "ex:Other", "@type": "owl:Class", "rdfs:label": "Other"},
+ {
+ "@id": "ex:relatesTo",
+ "@type": "owl:ObjectProperty",
+ "rdfs:domain": {"@id": "ex:Thing"},
+ "rdfs:range": {"@id": "ex:Other"},
+ },
+ ],
+}
+
+DEFAULT_GRAPH_ONTOLOGY = {
+ "@context": NAMED_GRAPH_ONTOLOGY["@context"],
+ "@graph": [
+ {"@id": "ex:Thing", "@type": "owl:Class", "rdfs:label": "Thing"},
+ {"@id": "ex:Other", "@type": "owl:Class", "rdfs:label": "Other"},
+ ],
+}
+
+
+def _write(tmp_path, name, document):
+ path = tmp_path / name
+ path.write_text(json.dumps(document), encoding="utf-8")
+ return path
+
+
+def test_terms_in_a_named_graph_are_ingested(tmp_path):
+ """The regression: two classes and one object property, all inside the named graph."""
+ path = _write(tmp_path, "named.jsonld", NAMED_GRAPH_ONTOLOGY)
+
+ data = OntologyIngestor().ingest_ontology(path).data
+
+ assert len(data["classes"]) == 2, (
+ "classes inside a JSON-LD named graph were dropped; the ingestor is reading only "
+ "the default graph"
+ )
+ assert len(data["properties"]) == 1
+ assert {c["uri"] for c in data["classes"]} == {
+ "https://example.org/ns#Thing",
+ "https://example.org/ns#Other",
+ }
+
+
+def test_terms_in_the_default_graph_still_work(tmp_path):
+ """Canary for the test above: a document *without* a top-level ``@id`` keeps its terms
+ in the default graph and always parsed correctly. If this stopped passing, the fix would
+ have traded one blind spot for another."""
+ path = _write(tmp_path, "default.jsonld", DEFAULT_GRAPH_ONTOLOGY)
+
+ data = OntologyIngestor().ingest_ontology(path).data
+
+ assert len(data["classes"]) == 2
+
+
+@pytest.mark.parametrize("document", [NAMED_GRAPH_ONTOLOGY, DEFAULT_GRAPH_ONTOLOGY])
+def test_metadata_reports_what_was_actually_read(tmp_path, document):
+ """Whatever the shape of the document, the counts reported have to match the terms
+ returned — a load that says it succeeded while returning nothing is what made #1129
+ cost an afternoon to find."""
+ path = _write(tmp_path, "any.jsonld", document)
+
+ result = OntologyIngestor().ingest_ontology(path)
+
+ assert result.data["classes"], "reported success with zero classes"
diff --git a/tests/integrations/agno/test_load_urls_ssrf.py b/tests/integrations/agno/test_load_urls_ssrf.py
new file mode 100644
index 00000000..28ad1cbe
--- /dev/null
+++ b/tests/integrations/agno/test_load_urls_ssrf.py
@@ -0,0 +1,278 @@
+"""SSRF regression tests for AgnoKnowledgeGraph.load_urls().
+
+Prior to the fix, load_urls() used urllib.request.urlopen with only a
+scheme check — private/loopback/link-local/metadata IPs were not blocked
+and redirects were followed without re-validation.
+
+These tests exercise the real SSRF guard (no mock of request_with_ssrf_guard
+itself) by patching at the socket.getaddrinfo level, confirming that
+blocked addresses never reach the network layer.
+"""
+
+from __future__ import annotations
+
+import socket
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# conftest.py installs the full agno stub before this file is collected.
+from integrations.agno.knowledge_graph import AgnoKnowledgeGraph
+
+from semantica.utils.exceptions import ValidationError
+
+
+# ---------------------------------------------------------------------------
+# Minimal fakes so AgnoKnowledgeGraph.__init__ succeeds without real imports.
+# ---------------------------------------------------------------------------
+class _FakeNER:
+ def extract_entities(self, text):
+ return []
+
+
+class _FakeRelExtractor:
+ def extract_relations(self, text, entities=None):
+ return []
+
+
+class _FakeGraphBuilder:
+ def build(self, sources):
+ pass
+
+
+class _FakeContextGraph:
+ def find_nodes(self, label=None):
+ return []
+
+ def get_neighbors(self, node_id=None, hops=1):
+ return []
+
+
+def _make_kg() -> AgnoKnowledgeGraph:
+ return AgnoKnowledgeGraph(
+ graph_builder=_FakeGraphBuilder(),
+ ner_extractor=_FakeNER(),
+ relation_extractor=_FakeRelExtractor(),
+ context_graph=_FakeContextGraph(),
+ )
+
+
+def _public_getaddrinfo(host, *args, **kwargs):
+ """Stub that makes every hostname resolve to a public IP."""
+ return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))]
+
+
+# ---------------------------------------------------------------------------
+# Tests: blocked addresses must never be fetched
+# ---------------------------------------------------------------------------
+
+class TestLoadUrlsBlockedAddresses:
+ """load_urls() must silently skip (warn) any URL that fails the SSRF guard."""
+
+ @pytest.mark.parametrize("url", [
+ "http://127.0.0.1/secret",
+ "http://127.0.0.1:9200/", # common internal service port
+ "http://0.0.0.0/",
+ "http://169.254.169.254/latest/meta-data/",
+ "http://169.254.169.254/computeMetadata/v1/",
+ "http://10.0.0.1/internal",
+ "http://10.255.255.255/",
+ "http://172.16.0.1/",
+ "http://172.31.255.255/",
+ "http://192.168.0.1/admin",
+ "http://192.168.100.200/",
+ "http://[::1]/ipv6-loopback",
+ "http://[fc00::1]/ipv6-ula",
+ "http://[fe80::1]/ipv6-link-local",
+ ])
+ def test_blocked_ip_never_reaches_network(self, url):
+ """Blocked addresses must raise ValidationError inside the guard,
+ which load_urls() catches and logs — _ingest_text must NOT be called."""
+ kg = _make_kg()
+ with patch.object(kg, "_ingest_text") as mock_ingest:
+ kg.load_urls([url])
+ mock_ingest.assert_not_called()
+
+ def test_localhost_hostname_blocked(self):
+ kg = _make_kg()
+ with patch.object(kg, "_ingest_text") as mock_ingest:
+ kg.load_urls(["http://localhost/admin"])
+ mock_ingest.assert_not_called()
+
+ def test_localhost_subdomain_blocked(self):
+ kg = _make_kg()
+ with patch.object(kg, "_ingest_text") as mock_ingest:
+ kg.load_urls(["http://foo.localhost/"])
+ mock_ingest.assert_not_called()
+
+ def test_hostname_resolving_to_private_ip_blocked(self):
+ """A hostname that resolves to a private IP must be blocked even though
+ the URL string itself looks like a normal hostname."""
+ def _internal_getaddrinfo(host, *a, **kw):
+ return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.0.5", 0))]
+
+ kg = _make_kg()
+ with patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_internal_getaddrinfo):
+ with patch.object(kg, "_ingest_text") as mock_ingest:
+ kg.load_urls(["http://internal.corp/secret"])
+ mock_ingest.assert_not_called()
+
+ def test_hostname_resolving_to_metadata_ip_blocked(self):
+ def _meta_getaddrinfo(host, *a, **kw):
+ return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("169.254.169.254", 0))]
+
+ kg = _make_kg()
+ with patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_meta_getaddrinfo):
+ with patch.object(kg, "_ingest_text") as mock_ingest:
+ kg.load_urls(["http://metadata.internal/v1/token"])
+ mock_ingest.assert_not_called()
+
+
+class TestLoadUrlsNonHttpSchemes:
+ """Non-HTTP(S) schemes must be rejected."""
+
+ @pytest.mark.parametrize("url", [
+ "file:///etc/passwd",
+ "file://localhost/etc/shadow",
+ "ftp://example.com/file.txt",
+ "gopher://example.com/1",
+ "dict://example.com/",
+ "sftp://example.com/data",
+ ])
+ def test_non_http_scheme_blocked(self, url):
+ kg = _make_kg()
+ with patch.object(kg, "_ingest_text") as mock_ingest:
+ kg.load_urls([url])
+ mock_ingest.assert_not_called()
+
+
+class TestLoadUrlsRedirects:
+ """Redirects to private/blocked addresses must be rejected."""
+
+ def test_redirect_to_loopback_blocked(self):
+ """A public first hop that redirects to loopback must be blocked."""
+ redirect = MagicMock()
+ redirect.status_code = 302
+ redirect.headers = {"Location": "http://127.0.0.1/secret"}
+ redirect.close = MagicMock()
+
+ kg = _make_kg()
+ with patch(
+ "semantica.ingest.ssrf.socket.getaddrinfo",
+ return_value=[(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))],
+ ):
+ # Patch requests.Session so the first hop returns our redirect mock.
+ # The guard sees the 302, then validates the Location — 127.0.0.1 is
+ # blocked without a second network call.
+ with patch("semantica.ingest.ssrf.requests.Session") as MockSession:
+ mock_session = MockSession.return_value
+ mock_session.adapters = {}
+ mock_session.headers = {}
+ mock_session.auth = None
+ mock_session.trust_env = True
+ mock_session.request.return_value = redirect
+
+ with patch.object(kg, "_ingest_text") as mock_ingest:
+ kg.load_urls(["https://example.com/start"])
+ mock_ingest.assert_not_called()
+
+ def test_redirect_to_metadata_ip_blocked(self):
+ """Redirect to cloud metadata endpoint must be blocked."""
+ redirect = MagicMock()
+ redirect.status_code = 301
+ redirect.headers = {"Location": "http://169.254.169.254/latest/meta-data/"}
+ redirect.close = MagicMock()
+
+ kg = _make_kg()
+ with patch(
+ "semantica.ingest.ssrf.socket.getaddrinfo",
+ return_value=[(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))],
+ ):
+ with patch("semantica.ingest.ssrf.requests.Session") as MockSession:
+ mock_session = MockSession.return_value
+ mock_session.adapters = {}
+ mock_session.headers = {}
+ mock_session.auth = None
+ mock_session.trust_env = True
+ mock_session.request.return_value = redirect
+
+ with patch.object(kg, "_ingest_text") as mock_ingest:
+ kg.load_urls(["https://example.com/redirect-me"])
+ mock_ingest.assert_not_called()
+
+
+class TestLoadUrlsValidUrls:
+ """Valid public URLs must succeed and call _ingest_text."""
+
+ def test_valid_public_url_ingested(self):
+ """A URL resolving to a public IP must be fetched and ingested."""
+ ok_response = MagicMock()
+ ok_response.status_code = 200
+ ok_response.headers = {}
+ ok_response.text = "This is the document content."
+
+ kg = _make_kg()
+ with patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo):
+ with patch("semantica.ingest.ssrf.requests.Session") as MockSession:
+ mock_session = MockSession.return_value
+ mock_session.adapters = {}
+ mock_session.headers = {}
+ mock_session.auth = None
+ mock_session.trust_env = True
+ mock_session.request.return_value = ok_response
+
+ with patch.object(kg, "_ingest_text") as mock_ingest:
+ kg.load_urls(["https://example.com/doc.txt"])
+
+ mock_ingest.assert_called_once_with(
+ "This is the document content.", source="https://example.com/doc.txt"
+ )
+
+ def test_multiple_urls_each_independently_validated(self):
+ """Each URL in the list is independently validated; one blocked URL
+ must not prevent valid subsequent URLs from being ingested."""
+ ok_response = MagicMock()
+ ok_response.status_code = 200
+ ok_response.headers = {}
+ ok_response.text = "Valid content."
+
+ def _selective_getaddrinfo(host, *a, **kw):
+ if host == "internal.corp":
+ return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.0.5", 0))]
+ return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))]
+
+ kg = _make_kg()
+ with patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_selective_getaddrinfo):
+ with patch("semantica.ingest.ssrf.requests.Session") as MockSession:
+ mock_session = MockSession.return_value
+ mock_session.adapters = {}
+ mock_session.headers = {}
+ mock_session.auth = None
+ mock_session.trust_env = True
+ mock_session.request.return_value = ok_response
+
+ with patch.object(kg, "_ingest_text") as mock_ingest:
+ kg.load_urls([
+ "http://internal.corp/secret", # blocked
+ "https://example.com/public.txt", # allowed
+ ])
+
+ # Only the valid URL triggers ingestion
+ mock_ingest.assert_called_once_with("Valid content.", source="https://example.com/public.txt")
+
+ def test_failed_fetch_does_not_raise(self):
+ """A network failure on a valid URL must log a warning, not raise."""
+ import requests as _requests
+
+ kg = _make_kg()
+ with patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo):
+ with patch("semantica.ingest.ssrf.requests.Session") as MockSession:
+ mock_session = MockSession.return_value
+ mock_session.adapters = {}
+ mock_session.headers = {}
+ mock_session.auth = None
+ mock_session.trust_env = True
+ mock_session.request.side_effect = _requests.exceptions.ConnectionError("refused")
+
+ # Must not raise; failure is logged and skipped
+ kg.load_urls(["https://example.com/unreachable"])
diff --git a/tests/integrations/openclaw/__init__.py b/tests/integrations/openclaw/__init__.py
new file mode 100644
index 00000000..6def8d9a
--- /dev/null
+++ b/tests/integrations/openclaw/__init__.py
@@ -0,0 +1 @@
+# tests/integrations/openclaw package
diff --git a/tests/integrations/openclaw/test_mcp_tool_ssrf.py b/tests/integrations/openclaw/test_mcp_tool_ssrf.py
new file mode 100644
index 00000000..4c75c772
--- /dev/null
+++ b/tests/integrations/openclaw/test_mcp_tool_ssrf.py
@@ -0,0 +1,290 @@
+"""SSRF hardening tests for OpenClawKGTool.
+
+OpenClawKGTool is designed to speak to a locally-running Semantica REST server
+(default: http://localhost:8000). The fix validates base_url at construction
+time so that obviously wrong schemes (file://, ftp://, gopher://, etc.) and
+malformed URLs are rejected immediately, while localhost and other private
+addresses remain valid because allow_private_ips=True is the correct posture
+for this tool's intended use case.
+
+These are construction-time tests; per-request SSRF guarding is not the
+contract of this tool (its threat model is operator-configured base_url, not
+untrusted per-call URLs).
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from integrations.openclaw.mcp_tool import OpenClawKGTool
+from semantica.utils.exceptions import ValidationError
+
+
+class TestOpenClawKGToolBaseUrlValidation:
+ """base_url is validated at __init__ time."""
+
+ # ------------------------------------------------------------------
+ # Valid base_urls — all must construct without raising
+ # ------------------------------------------------------------------
+
+ @pytest.mark.parametrize("url", [
+ "http://localhost:8000",
+ "http://localhost",
+ "http://127.0.0.1:8000",
+ "http://127.0.0.1",
+ "https://localhost:8443",
+ "http://0.0.0.0:8000",
+ "http://192.168.1.10:8000", # LAN Semantica server
+ "http://10.0.0.5:8000", # corporate intranet deployment
+ "https://semantica.internal/api",
+ "https://semantica.example.com",
+ ])
+ def test_valid_base_url_accepted(self, url):
+ """All reasonable operator-configured base_urls must be accepted."""
+ tool = OpenClawKGTool(base_url=url)
+ assert tool.base_url == url.rstrip("/")
+
+ # ------------------------------------------------------------------
+ # Invalid schemes — must raise at construction
+ # ------------------------------------------------------------------
+
+ @pytest.mark.parametrize("url", [
+ "file:///etc/passwd",
+ "file://localhost/etc/shadow",
+ "ftp://example.com/",
+ "gopher://example.com/1",
+ "dict://example.com/",
+ "sftp://example.com/",
+ "ldap://example.com/",
+ "javascript:alert(1)",
+ ])
+ def test_invalid_scheme_rejected(self, url):
+ """Non-HTTP(S) schemes must be rejected at construction time."""
+ with pytest.raises((ValidationError, ValueError)):
+ OpenClawKGTool(base_url=url)
+
+ # ------------------------------------------------------------------
+ # Malformed URLs
+ # ------------------------------------------------------------------
+
+ def test_empty_string_rejected(self):
+ with pytest.raises((ValidationError, ValueError)):
+ OpenClawKGTool(base_url="")
+
+ def test_no_scheme_rejected(self):
+ """A bare hostname without a scheme must be rejected."""
+ with pytest.raises((ValidationError, ValueError)):
+ OpenClawKGTool(base_url="localhost:8000")
+
+ def test_whitespace_only_rejected(self):
+ with pytest.raises((ValidationError, ValueError)):
+ OpenClawKGTool(base_url=" ")
+
+ # ------------------------------------------------------------------
+ # Default is the documented localhost value
+ # ------------------------------------------------------------------
+
+ def test_default_base_url_is_localhost(self):
+ """The default must remain http://localhost:8000 for backward compat."""
+ tool = OpenClawKGTool()
+ assert tool.base_url == "http://localhost:8000"
+
+ def test_trailing_slash_stripped(self):
+ """base_url trailing slash must be stripped so paths concatenate cleanly."""
+ tool = OpenClawKGTool(base_url="http://localhost:8000/")
+ assert tool.base_url == "http://localhost:8000"
+
+ def test_multiple_trailing_slashes_stripped(self):
+ tool = OpenClawKGTool(base_url="http://localhost:8000///")
+ assert tool.base_url == "http://localhost:8000"
+
+ def test_leading_and_trailing_whitespace_stripped(self):
+ """Whitespace around a valid URL must be stripped before storage so
+ _post/_get don't build requests with space-padded URLs like
+ ' http://localhost:8000 /extract'."""
+ tool = OpenClawKGTool(base_url=" http://localhost:8000 ")
+ assert tool.base_url == "http://localhost:8000"
+
+ def test_whitespace_plus_trailing_slash_both_stripped(self):
+ tool = OpenClawKGTool(base_url=" http://localhost:8000/ ")
+ assert tool.base_url == "http://localhost:8000"
+
+
+class TestOpenClawKGToolFallbackValidation:
+ """When semantica.ingest.ssrf is unavailable (ImportError path), the fallback
+ must perform the same structural checks as validate_url_for_request:
+ non-empty string, http/https scheme, netloc present, hostname present.
+
+ The fallback is exercised by temporarily hiding semantica.ingest.ssrf
+ from sys.modules so the import inside __init__ raises ImportError.
+ """
+
+ @staticmethod
+ def _hide_ssrf(monkeypatch):
+ """Return a context in which semantica.ingest.ssrf appears unimportable."""
+ import sys
+ monkeypatch.setitem(sys.modules, "semantica.ingest.ssrf", None)
+
+ # ------------------------------------------------------------------
+ # Valid URLs must still be accepted in the fallback path
+ # ------------------------------------------------------------------
+
+ @pytest.mark.parametrize("url", [
+ "http://localhost:8000",
+ "http://127.0.0.1:8000",
+ "https://semantica.example.com",
+ ])
+ def test_fallback_valid_url_accepted(self, url, monkeypatch):
+ self._hide_ssrf(monkeypatch)
+ tool = OpenClawKGTool(base_url=url)
+ assert tool.base_url == url.rstrip("/")
+
+ # ------------------------------------------------------------------
+ # Malformed URLs that the fallback previously let through
+ # ------------------------------------------------------------------
+
+ @pytest.mark.parametrize("url", [
+ "http://", # scheme only, no netloc or hostname
+ "https://", # same
+ "http:///path", # empty hostname (netloc is present but hostname is None)
+ ])
+ def test_fallback_no_netloc_rejected(self, url, monkeypatch):
+ """URLs with a valid scheme but missing netloc/hostname must be rejected
+ in the fallback path, matching validate_url_for_request's behaviour."""
+ self._hide_ssrf(monkeypatch)
+ with pytest.raises((ValidationError, ValueError)):
+ OpenClawKGTool(base_url=url)
+
+ def test_fallback_empty_string_rejected(self, monkeypatch):
+ self._hide_ssrf(monkeypatch)
+ with pytest.raises((ValidationError, ValueError)):
+ OpenClawKGTool(base_url="")
+
+ def test_fallback_whitespace_only_rejected(self, monkeypatch):
+ self._hide_ssrf(monkeypatch)
+ with pytest.raises((ValidationError, ValueError)):
+ OpenClawKGTool(base_url=" ")
+
+ def test_fallback_invalid_scheme_rejected(self, monkeypatch):
+ self._hide_ssrf(monkeypatch)
+ with pytest.raises((ValidationError, ValueError)):
+ OpenClawKGTool(base_url="file:///etc/passwd")
+
+ def test_fallback_no_scheme_rejected(self, monkeypatch):
+ self._hide_ssrf(monkeypatch)
+ with pytest.raises((ValidationError, ValueError)):
+ OpenClawKGTool(base_url="localhost:8000")
+
+ def test_fallback_whitespace_padded_valid_url_stored_clean(self, monkeypatch):
+ """Whitespace around a valid URL must be stripped before storage in the
+ fallback path too — same guarantee as the normal path."""
+ self._hide_ssrf(monkeypatch)
+ tool = OpenClawKGTool(base_url=" http://localhost:8000 ")
+ assert tool.base_url == "http://localhost:8000"
+
+
+class TestOpenClawKGToolEndpointConstruction:
+ """Verify that per-method URLs are assembled from base_url + hardcoded paths.
+
+ The endpoint strings are always literals defined in the class body —
+ they are not caller-supplied — so these tests confirm the URL assembly
+ logic is correct rather than testing SSRF guards on the endpoints.
+
+ All HTTP calls are mocked so no real network connection is made.
+ """
+
+ def _mock_session(self, status: int = 200, body: bytes = b"{}") -> "MagicMock":
+ """Return a mock session whose post/get return a minimal JSON response."""
+ from unittest.mock import MagicMock
+ mock_resp = MagicMock()
+ mock_resp.status_code = status
+ mock_resp.raise_for_status = MagicMock()
+ mock_resp.json.return_value = {}
+ session = MagicMock()
+ session.post.return_value = mock_resp
+ session.get.return_value = mock_resp
+ return session
+
+ def test_post_url_constructed_from_base_url(self):
+ """_post must call session.post with the exact URL base_url+endpoint,
+ the supplied payload as json=, and the tool timeout. No real connection."""
+ from unittest.mock import patch
+
+ tool = OpenClawKGTool(base_url="http://localhost:8000")
+ mock_session = self._mock_session()
+
+ with patch.object(tool, "_get_session", return_value=mock_session):
+ tool._post("/extract", {"text": "hello"})
+
+ mock_session.post.assert_called_once_with(
+ "http://localhost:8000/extract",
+ json={"text": "hello"},
+ timeout=30,
+ )
+
+ def test_post_url_with_custom_base_url(self):
+ """base_url is reflected correctly in the outbound URL for _post."""
+ from unittest.mock import patch
+
+ tool = OpenClawKGTool(base_url="http://192.168.1.10:9000")
+ mock_session = self._mock_session()
+
+ with patch.object(tool, "_get_session", return_value=mock_session):
+ tool._post("/decisions", {"decision": "deploy"})
+
+ mock_session.post.assert_called_once_with(
+ "http://192.168.1.10:9000/decisions",
+ json={"decision": "deploy"},
+ timeout=30,
+ )
+
+ def test_get_url_constructed_from_base_url(self):
+ """_get must call session.get with the exact URL base_url+endpoint,
+ params={} when none are supplied, and the tool timeout."""
+ from unittest.mock import patch
+
+ tool = OpenClawKGTool(base_url="http://localhost:8000")
+ mock_session = self._mock_session()
+
+ with patch.object(tool, "_get_session", return_value=mock_session):
+ tool._get("/analytics")
+
+ mock_session.get.assert_called_once_with(
+ "http://localhost:8000/analytics",
+ params={},
+ timeout=30,
+ )
+
+ def test_get_url_with_params(self):
+ """_get must forward supplied params to session.get."""
+ from unittest.mock import patch
+
+ tool = OpenClawKGTool(base_url="http://localhost:8000")
+ mock_session = self._mock_session()
+
+ with patch.object(tool, "_get_session", return_value=mock_session):
+ tool._get("/decisions/search", {"q": "deploy", "limit": 5})
+
+ mock_session.get.assert_called_once_with(
+ "http://localhost:8000/decisions/search",
+ params={"q": "deploy", "limit": 5},
+ timeout=30,
+ )
+
+ def test_custom_timeout_forwarded(self):
+ """A non-default timeout must reach session.post and session.get."""
+ from unittest.mock import patch
+
+ tool = OpenClawKGTool(base_url="http://localhost:8000", timeout=60)
+ mock_session = self._mock_session()
+
+ with patch.object(tool, "_get_session", return_value=mock_session):
+ tool._post("/extract", {"text": "x"})
+ tool._get("/analytics")
+
+ assert mock_session.post.call_args.kwargs["timeout"] == 60
+ assert mock_session.get.call_args.kwargs["timeout"] == 60
+
+ def test_repr_includes_base_url(self):
+ tool = OpenClawKGTool(base_url="http://localhost:9000")
+ assert "http://localhost:9000" in repr(tool)
diff --git a/tests/pipeline/test_pipeline_serializer.py b/tests/pipeline/test_pipeline_serializer.py
new file mode 100644
index 00000000..fe616ddd
--- /dev/null
+++ b/tests/pipeline/test_pipeline_serializer.py
@@ -0,0 +1,103 @@
+import copy
+import json
+
+import pytest
+
+from semantica.pipeline.pipeline_builder import PipelineBuilder, PipelineSerializer
+
+
+@pytest.mark.parametrize("serialization_format", ["dict", "json"])
+def test_roundtrip_preserves_dependencies_and_delta_metadata(serialization_format):
+ builder = PipelineBuilder()
+ builder.add_step("extract", "source")
+ builder.add_step(
+ "index",
+ "sink",
+ delta_mode=True,
+ base_version_id="v1",
+ target_version_id="v2",
+ )
+ builder.connect_steps("extract", "index")
+ pipeline = builder.build("incremental-index")
+
+ serializer = PipelineSerializer()
+ serialized = serializer.serialize_pipeline(pipeline, format=serialization_format)
+ restored = serializer.deserialize_pipeline(serialized)
+
+ index_step = next(step for step in restored.steps if step.name == "index")
+ assert index_step.dependencies == ["extract"]
+ assert index_step.delta_mode is True
+ assert index_step.base_version_id == "v1"
+ assert index_step.target_version_id == "v2"
+
+
+@pytest.mark.parametrize("serialization_format", ["dict", "json"])
+def test_serialization_omits_runtime_handlers(serialization_format):
+ def handler(data, **config):
+ return data
+
+ builder = PipelineBuilder()
+ builder.add_step("extract", "source", handler=handler, batch_size=10)
+ pipeline = builder.build("handler-pipeline")
+
+ serializer = PipelineSerializer()
+ serialized = serializer.serialize_pipeline(pipeline, format=serialization_format)
+ serialized_data = (
+ json.loads(serialized) if isinstance(serialized, str) else serialized
+ )
+
+ assert serialized_data["steps"][0]["config"] == {"batch_size": 10}
+
+ restored = serializer.deserialize_pipeline(serialized)
+ assert restored.steps[0].handler is None
+ assert restored.steps[0].config == {"batch_size": 10}
+
+
+def test_deserialization_ignores_legacy_stringified_handler():
+ serialized = json.dumps(
+ {
+ "name": "legacy-handler-pipeline",
+ "steps": [
+ {
+ "name": "extract",
+ "type": "source",
+ "config": {
+ "handler": "