diff --git a/mcp/server.py b/mcp/server.py
index 3cd935be..bf84dd24 100644
--- a/mcp/server.py
+++ b/mcp/server.py
@@ -93,7 +93,14 @@ def _handle_tools_call(req_id: Any, params: dict) -> dict:
         result = tool["_handler"](args)
     except Exception as exc:
         log.exception("Tool %s raised an exception", name)
-        return _err(req_id, _INTERNAL_ERROR, str(exc))
+        # The exception's class name (e.g. "ValidationError", "TimeoutError")
+        # is safe to surface — unlike str(exc), it never carries paths,
+        # connection strings, or other internal detail — and lets the
+        # client distinguish failure kinds without a full message.
+        return _err(
+            req_id, _INTERNAL_ERROR,
+            f"Tool '{name}' failed ({type(exc).__name__}). See server logs for details.",
+        )
 
     # MCP spec: content must be a list of content items
     return _ok(req_id, {
@@ -171,7 +178,10 @@ class SemanticaMCPServer:
             log.exception("Unhandled error in method %s", method)
             if req_id is None:
                 return None
-            return _err(req_id, _INTERNAL_ERROR, str(exc))
+            return _err(
+                req_id, _INTERNAL_ERROR,
+                f"Method '{method}' failed ({type(exc).__name__}). See server logs for details.",
+            )
 
     # ------------------------------------------------------------------
     def run(self) -> None:
diff --git a/semantica/cli.py b/semantica/cli.py
index 7b944dc6..403971e0 100644
--- a/semantica/cli.py
+++ b/semantica/cli.py
@@ -20,7 +20,7 @@ if sys.platform == "win32":
         sys.stderr.reconfigure(encoding="utf-8", errors="replace")
 
 from dataclasses import asdict, dataclass, field, is_dataclass
-from pathlib import Path
+from pathlib import Path, PurePosixPath, PureWindowsPath
 from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Sequence, Tuple
 
 import yaml
@@ -3933,15 +3933,68 @@ def backup_restore(cli_ctx: CLIContext, source: str, local_dry: bool) -> None:
 
         try:
             if _tf.is_tarfile(str(work_path)):
-                restore_root = Path.cwd()
+                restore_root = Path.cwd().resolve()
                 with _tf.open(str(work_path), "r:*") as tar:
                     # Dry-run listing was already handled above; extract now
                     for member in tar.getmembers():
                         # Strip the leading "semantica-backup/" prefix
                         member.name = member.name.replace("semantica-backup/", "", 1)
-                        if member.name:
-                            tar.extract(member, path=str(restore_root))
-                            console.print(f"  restored: {member.name}")
+                        if not member.name:
+                            continue
+
+                        # Reject members whose resolved path escapes the
+                        # restore root (path traversal / absolute paths),
+                        # regardless of the "semantica-backup/" prefix.
+                        member_path = (restore_root / member.name).resolve()
+                        try:
+                            member_path.relative_to(restore_root)
+                        except ValueError:
+                            raise click.ClickException(
+                                f"Refusing to restore '{member.name}': "
+                                "path escapes the restore directory."
+                            )
+
+                        # Reject symlink/hardlink members whose target
+                        # escapes the restore root. Checked two ways:
+                        # lexically (linkname itself, so an absolute path or
+                        # a literal ".." segment is rejected outright, with
+                        # no dependence on what else does or doesn't already
+                        # exist on disk) and by resolution (catches any
+                        # remaining traversal the lexical check misses).
+                        if member.issym() or member.islnk():
+                            linkname = member.linkname or ""
+                            linkname_parts = PurePosixPath(
+                                linkname.replace("\\", "/")
+                            ).parts
+                            if (
+                                not linkname
+                                or os.path.isabs(linkname)
+                                or PureWindowsPath(linkname).is_absolute()
+                                or ".." in linkname_parts
+                            ):
+                                raise click.ClickException(
+                                    f"Refusing to restore '{member.name}': "
+                                    "link target is absolute or traverses "
+                                    "out of the archive."
+                                )
+                            link_target = (
+                                member_path.parent / linkname
+                            ).resolve()
+                            try:
+                                link_target.relative_to(restore_root)
+                            except ValueError:
+                                raise click.ClickException(
+                                    f"Refusing to restore '{member.name}': "
+                                    "link target escapes the restore directory."
+                                )
+
+                        extract_kwargs: Dict[str, Any] = {"path": str(restore_root)}
+                        if hasattr(_tf, "data_filter"):
+                            # Python >=3.12: also reject device files, and
+                            # further harden the traversal/ownership checks.
+                            extract_kwargs["filter"] = "data"
+                        tar.extract(member, **extract_kwargs)
+                        console.print(f"  restored: {member.name}")
             elif src.is_dir():
                 restore_root = Path.cwd()
                 for f in src.rglob("*"):
diff --git a/semantica/export/report_generator.py b/semantica/export/report_generator.py
index d4b67920..fcee11a1 100644
--- a/semantica/export/report_generator.py
+++ b/semantica/export/report_generator.py
@@ -23,6 +23,7 @@ Author: Semantica Contributors
 License: MIT
 """
 
+import html
 import json
 from datetime import datetime
 from pathlib import Path
@@ -362,7 +363,7 @@ class ReportGenerator:
             '  <meta name="viewport" content="width=device-width, initial-scale=1.0">'
         )
         title = data.get("title", "Report")
-        lines.append(f"  <title>{title}</title>")
+        lines.append(f"  <title>{html.escape(str(title))}</title>")
         lines.append("  <style>")
         lines.append("    body { font-family: Arial, sans-serif; margin: 20px; }")
         lines.append("    h1 { color: #333; }")
@@ -380,11 +381,14 @@ class ReportGenerator:
 
         # Title
         title = data.get("title", "Report")
-        lines.append(f"  <h1>{title}</h1>")
+        lines.append(f"  <h1>{html.escape(str(title))}</h1>")
 
         # Generated at
         if "generated_at" in data:
-            lines.append(f'  <p><strong>Generated:</strong> {data["generated_at"]}</p>')
+            lines.append(
+                f'  <p><strong>Generated:</strong> '
+                f'{html.escape(str(data["generated_at"]))}</p>'
+            )
 
         # Summary
         if "summary" in data:
@@ -393,10 +397,13 @@ class ReportGenerator:
             if isinstance(summary, dict):
                 lines.append("  <ul>")
                 for key, value in summary.items():
-                    lines.append(f"    <li><strong>{key}:</strong> {value}</li>")
+                    lines.append(
+                        f"    <li><strong>{html.escape(str(key))}:</strong> "
+                        f"{html.escape(str(value))}</li>"
+                    )
                 lines.append("  </ul>")
             else:
-                lines.append(f"  <p>{summary}</p>")
+                lines.append(f"  <p>{html.escape(str(summary))}</p>")
 
         # Metrics
         if "metrics" in data:
@@ -483,7 +490,10 @@ class ReportGenerator:
             else:
                 value_str = str(value)
 
-            lines.append(f"    <tr><td>{key}</td><td>{value_str}</td></tr>")
+            lines.append(
+                f"    <tr><td>{html.escape(str(key))}</td>"
+                f"<td>{html.escape(value_str)}</td></tr>"
+            )
 
         lines.append("  </table>")
 
diff --git a/semantica/graph_store/age_store.py b/semantica/graph_store/age_store.py
index 744c7cf5..c9888c74 100644
--- a/semantica/graph_store/age_store.py
+++ b/semantica/graph_store/age_store.py
@@ -66,6 +66,12 @@ except (ImportError, OSError):
 # Helpers
 # ---------------------------------------------------------------------------
 
+# create_index's index_type reaches a raw SQL keyword position (`USING
+# {index_type}`) that can't be bound as a query parameter; only the
+# documented, PostgreSQL-recognized types are allowed through.
+_ALLOWED_INDEX_TYPES = frozenset({"btree", "gin", "hash", "gist", "brin"})
+
+
 def _sanitize_label(label: str) -> str:
     """
     Sanitize a Cypher label to prevent injection.
@@ -1214,6 +1220,11 @@ class ApacheAgeStore:
             safe_label = _sanitize_label(label)
             if not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", property_name):
                 raise ValidationError(f"Invalid property name: '{property_name}'")
+            if index_type not in _ALLOWED_INDEX_TYPES:
+                raise ValidationError(
+                    f"Invalid index_type: {index_type!r}. "
+                    f"Allowed: {sorted(_ALLOWED_INDEX_TYPES)}"
+                )
 
             index_name = options.get(
                 "index_name", f"idx_{self.graph_name}_{safe_label}_{property_name}"
diff --git a/semantica/graph_store/neo4j_store.py b/semantica/graph_store/neo4j_store.py
index 93270efe..bb7d58d1 100644
--- a/semantica/graph_store/neo4j_store.py
+++ b/semantica/graph_store/neo4j_store.py
@@ -503,6 +503,16 @@ class Neo4jStore:
         Returns:
             List of matching nodes
         """
+        # LIMIT can't be bound as a query parameter in a way Neo4j accepts
+        # here, so it's interpolated directly; validate explicitly rather
+        # than trust the `limit: int` type hint, which Python doesn't
+        # enforce at runtime. Done outside the try/except below so a bad
+        # limit raises ValidationError, not a generic ProcessingError.
+        try:
+            limit = int(limit)
+        except (TypeError, ValueError) as exc:
+            raise ValidationError(f"Invalid limit: {limit!r}") from exc
+
         try:
             # Build query
             if labels:
@@ -698,6 +708,15 @@ class Neo4jStore:
         Returns:
             List of matching relationships
         """
+        # See get_nodes: LIMIT is interpolated directly, so validate
+        # explicitly rather than trust the unenforced `limit: int` hint,
+        # outside the try/except below so a bad limit raises
+        # ValidationError, not a generic ProcessingError.
+        try:
+            limit = int(limit)
+        except (TypeError, ValueError) as exc:
+            raise ValidationError(f"Invalid limit: {limit!r}") from exc
+
         try:
             type_filter = f":{sanitize_identifier(rel_type, 'relationship type')}" if rel_type else ""
 
diff --git a/semantica/ingest/db_ingestor.py b/semantica/ingest/db_ingestor.py
index 460e0fcb..33d17bd0 100644
--- a/semantica/ingest/db_ingestor.py
+++ b/semantica/ingest/db_ingestor.py
@@ -29,6 +29,7 @@ License: MIT
 """
 
 import json
+import re
 from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Any, Dict, List, Optional
@@ -38,6 +39,64 @@ from ..utils.exceptions import ProcessingError, ValidationError
 from ..utils.logging import get_logger
 from ..utils.progress_tracker import get_progress_tracker
 
+_IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+
+# Fragments that turn a filter/order clause into a second statement, a
+# data-exfiltration UNION, a time-based blind-injection oracle, or schema
+# enumeration, rather than a boolean/ordering expression.
+_SQL_FRAGMENT_BLOCKLIST_RE = re.compile(
+    r";|--|/\*|\*/|\bunion\b|\binsert\b|\bupdate\b|\bdelete\b|\bdrop\b|"
+    r"\balter\b|\bcreate\b|\bexec\b|\bexecute\b|\bgrant\b|\brevoke\b|"
+    r"\battach\b|\bpragma\b|\bxp_\w+|\bsp_\w+|\binto\s+outfile\b|\bload_file\b|"
+    r"\bsleep\s*\(|\bbenchmark\s*\(|\bpg_sleep\s*\(|\bwaitfor\b|"
+    r"\bdbms_\w+|\butl_\w+|\binformation_schema\b|\bpg_catalog\b",
+    re.IGNORECASE,
+)
+
+
+def _validate_sql_identifier(name: str, kind: str) -> str:
+    """Validate a table/schema name used as a raw SQL identifier.
+
+    ``export_table_data`` interpolates *name* directly into the query text
+    (SQLAlchemy has no bind-parameter syntax for identifiers), so anything
+    outside a plain alphanumeric/underscore identifier is a potential
+    breakout of the surrounding ``"..."`` quoting.
+    """
+    if not isinstance(name, str) or not _IDENTIFIER_RE.match(name):
+        raise ValidationError(
+            f"Invalid {kind}: {name!r}. Must start with a letter or "
+            "underscore and contain only alphanumeric characters and "
+            "underscores."
+        )
+    return name
+
+
+def _validate_sql_fragment(fragment: str, kind: str) -> str:
+    """Reject WHERE/ORDER BY fragments that smuggle a second statement.
+
+    These clauses can't be bound as query parameters (they're arbitrary
+    boolean/ordering expressions, not values), so this blocks the concrete
+    injection primitives (statement separators, comments, UNION, DML/DDL
+    keywords, time-based blind oracles, schema enumeration) rather than
+    parameterizing.
+
+    This is a blocklist, not a grammar: it cannot exhaustively prove
+    *fragment* is safe, only reject known-dangerous constructs, so a
+    boolean-blind subquery expressed with none of the blocked keywords
+    (e.g. ``id = (SELECT 1 FROM t WHERE ...)``) still passes. ``where``/
+    ``order_by`` are a raw-SQL-fragment API by design (see
+    ``export_table_data``'s docstring); treat them as trusted/operator
+    input, not something to expose directly to untrusted end users.
+    """
+    if not isinstance(fragment, str):
+        raise ValidationError(f"Invalid {kind}: must be a string")
+    if _SQL_FRAGMENT_BLOCKLIST_RE.search(fragment):
+        raise ValidationError(
+            f"Invalid {kind}: {fragment!r} contains disallowed SQL "
+            "keywords or statement-boundary characters"
+        )
+    return fragment
+
 
 @dataclass
 class TableData:
@@ -253,8 +312,13 @@ class DataExporter:
             schema: Schema name (for databases with schema support, optional)
             limit: Maximum number of rows to export (optional)
             offset: Row offset for pagination (optional)
-            where: WHERE clause for filtering (optional, e.g., "age > 18")
-            order_by: ORDER BY clause for sorting (optional, e.g., "name ASC")
+            where: WHERE clause for filtering (optional, e.g., "age > 18").
+                Raw SQL, checked against a keyword/character blocklist (see
+                ``_validate_sql_fragment``) but not fully sanitized — treat
+                as trusted/operator input, never pass untrusted end-user
+                text here directly.
+            order_by: ORDER BY clause for sorting (optional, e.g., "name ASC").
+                Same trust requirement as ``where``.
             **options: Additional export options (unused)
 
         Returns:
@@ -269,7 +333,16 @@ class DataExporter:
             ProcessingError: If table export fails
         """
         try:
-            from sqlalchemy import inspect
+            from sqlalchemy import inspect, text
+
+            _validate_sql_identifier(table_name, "table_name")
+            if schema:
+                _validate_sql_identifier(schema, "schema")
+            if where:
+                _validate_sql_fragment(where, "where")
+            if order_by:
+                _validate_sql_fragment(order_by, "order_by")
+
             inspector = inspect(connection)
 
             # Get column information
@@ -760,8 +833,11 @@ class DBIngestor:
             schema: Schema name (for databases with schema support, optional)
             limit: Maximum number of rows to export (optional)
             offset: Row offset for pagination (optional)
-            where: WHERE clause for filtering (optional, e.g., "status = 'active'")
-            order_by: ORDER BY clause for sorting (optional, e.g., "created_at DESC")
+            where: WHERE clause for filtering (optional, e.g., "status = 'active'").
+                Raw SQL passed through to ``export_table_data`` — same trust
+                requirement documented there: not for untrusted end-user text.
+            order_by: ORDER BY clause for sorting (optional, e.g., "created_at DESC").
+                Same trust requirement as ``where``.
             transform: Whether to apply data transformations (default: False)
             **filters: Additional filtering options (merged with above parameters)
 
diff --git a/semantica/ingest/ssrf.py b/semantica/ingest/ssrf.py
index 488ae3cf..f21ca94a 100644
--- a/semantica/ingest/ssrf.py
+++ b/semantica/ingest/ssrf.py
@@ -11,7 +11,7 @@ import concurrent.futures
 import ipaddress
 import socket
 import threading
-from typing import Any, Iterable, Optional
+from typing import Any, Iterable, List, Optional
 from urllib.parse import urljoin, urlparse
 
 import requests
@@ -138,6 +138,7 @@ def _get_dns_executor() -> concurrent.futures.ThreadPoolExecutor:
 BLOCKED_NETWORKS = (
     ipaddress.ip_network("0.0.0.0/8"),
     ipaddress.ip_network("10.0.0.0/8"),
+    ipaddress.ip_network("100.64.0.0/10"),  # CGNAT (RFC 6598) — routable inside carrier/cloud NAT
     ipaddress.ip_network("127.0.0.0/8"),
     ipaddress.ip_network("169.254.0.0/16"),  # link-local / cloud metadata
     ipaddress.ip_network("172.16.0.0/12"),
@@ -262,6 +263,208 @@ def validate_url_for_request(
         )
 
 
+def _resolve_pinned_ips(
+    url: str, *, allow_private_ips: bool
+) -> Optional[List[str]]:
+    """Validate *url* and return every resolved IP for connection pinning.
+
+    ``validate_url_for_request`` and the subsequent connection used to
+    resolve the same hostname independently, which reopens a DNS-rebinding
+    TOCTOU window: a low-TTL or rebinding DNS answer can differ between the
+    validation lookup and the connect-time lookup, so a hostname that
+    validated as public can still connect to a private/internal address.
+    This performs the one resolution that is actually used for both the
+    accept/reject decision *and* the connection (see
+    ``_make_pinned_adapter``), closing that window the same way
+    ``explorer/routes/ontology.py``'s ``_validate_fetch_url`` /
+    ``_make_pinned_session`` pair already does.
+
+    Returns ``None`` when ``allow_private_ips`` is True (the caller
+    explicitly trusts this host, e.g. an operator-configured internal
+    endpoint that may rely on live DNS/service discovery — pinning is
+    skipped so it keeps resolving normally) or when the URL has no host.
+    Otherwise returns the deduplicated, resolution-ordered list of
+    validated IP addresses.
+    """
+    validate_url_for_request(url, allow_private_ips=allow_private_ips)
+    if allow_private_ips:
+        return None
+
+    host = urlparse(url).hostname
+    if not host:
+        return None
+
+    try:
+        literal_ip = ipaddress.ip_address(host)
+    except ValueError:
+        literal_ip = None
+    if literal_ip is not None:
+        return [str(literal_ip)]
+
+    executor = _get_dns_executor()
+    owned_executor = False
+    try:
+        try:
+            future = executor.submit(socket.getaddrinfo, host, None)
+        except RuntimeError:
+            executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
+            owned_executor = True
+            future = executor.submit(socket.getaddrinfo, host, None)
+        resolved: Iterable = future.result(timeout=_DNS_RESOLVE_TIMEOUT_SECONDS)
+    except (socket.gaierror, concurrent.futures.TimeoutError, OSError) as exc:
+        raise ValidationError(
+            f"URL host '{host}' could not be resolved safely "
+            "(DNS error or timeout); request blocked"
+        ) from exc
+    finally:
+        if owned_executor:
+            _shutdown_executor(executor)
+
+    pinned_ips: List[str] = []
+    for info in resolved:
+        addr = ipaddress.ip_address(info[4][0])
+        if _ip_is_blocked(addr):
+            raise ValidationError(
+                f"URL host '{host}' resolves to a blocked (private/loopback/"
+                "link-local) address"
+            )
+        addr_str = str(addr)
+        if addr_str not in pinned_ips:
+            pinned_ips.append(addr_str)
+    if not pinned_ips:
+        raise ValidationError(
+            f"URL host '{host}' could not be resolved to a usable address"
+        )
+    return pinned_ips
+
+
+def _make_pinned_adapter(pinned_ips: List[str], hostname: str) -> "requests.adapters.HTTPAdapter":
+    """Build an HTTPAdapter that connects only to *pinned_ips*.
+
+    Falls back across every pinned address in order (a hostname can have
+    multiple A/AAAA records) while presenting *hostname* as the TLS SNI /
+    certificate identity and outgoing Host header, so DNS resolution is
+    bypassed entirely for the actual connection — mirroring
+    ``explorer/routes/ontology.py``'s ``_make_pinned_session``.
+    """
+    import urllib3.util.connection as _u3_connection
+    from urllib3.exceptions import NewConnectionError
+
+    class _MultiIPConnectionMixin:
+        def _new_conn(self):
+            last_exc: Optional[BaseException] = None
+            for ip in pinned_ips:
+                try:
+                    return _u3_connection.create_connection(
+                        (ip, self.port),
+                        self.timeout,
+                        source_address=self.source_address,
+                        socket_options=self.socket_options,
+                    )
+                except OSError as exc:
+                    last_exc = exc
+                    continue
+            raise NewConnectionError(
+                self,
+                f"Failed to establish a connection to any of {pinned_ips}: {last_exc}",
+            )
+
+    class _PinnedIPHTTPAdapter(requests.adapters.HTTPAdapter):
+        def get_connection_with_tls_context(self, request, verify, proxies=None, cert=None):
+            # A proxy performs its own DNS resolution outside this
+            # process's control, which would silently reopen the exact
+            # rebinding race pinning exists to close. Fail closed instead.
+            if requests.utils.select_proxy(request.url, proxies):
+                raise ValidationError(
+                    "Proxied requests are not supported through the "
+                    "SSRF-guarded request path (a proxy would resolve the "
+                    "host itself and bypass IP pinning)."
+                )
+            host_params, pool_kwargs = self.build_connection_pool_key_attributes(
+                request, verify, cert
+            )
+            if host_params.get("scheme") == "https":
+                pool_kwargs.setdefault("assert_hostname", hostname)
+                pool_kwargs.setdefault("server_hostname", hostname)
+            host_params["host"] = pinned_ips[0]
+            pool = self.poolmanager.connection_from_host(
+                **host_params, pool_kwargs=pool_kwargs
+            )
+            base_connection_cls = pool.ConnectionCls
+            if not issubclass(base_connection_cls, _MultiIPConnectionMixin):
+                pool.ConnectionCls = type(
+                    "_PinnedConnection",
+                    (_MultiIPConnectionMixin, base_connection_cls),
+                    {},
+                )
+            return pool
+
+    return _PinnedIPHTTPAdapter()
+
+
+def _apply_connection_pin(
+    active_session: "requests.Session",
+    url: str,
+    pinned_ips: Optional[List[str]],
+    orig_http_adapter: "requests.adapters.HTTPAdapter",
+    orig_https_adapter: "requests.adapters.HTTPAdapter",
+) -> None:
+    """Mount (or remove) IP pinning on *active_session* for the next hop."""
+    # Session.mount() silently drops whatever adapter it replaces without
+    # closing it. Across a multi-hop redirect chain, each hop gets its own
+    # fresh pinned adapter (a new pool), so failing to close the one from
+    # the previous hop would leak its pooled connection.
+    _current = active_session.adapters.get("http://")
+    if getattr(_current, "_semantica_pinned", False):
+        _current.close()
+
+    parsed = urlparse(url)
+    if pinned_ips:
+        port = parsed.port
+        default_port = _DEFAULT_PORTS.get(parsed.scheme, 80)
+        host_header = (
+            parsed.hostname
+            if port in (None, default_port)
+            else f"{parsed.hostname}:{port}"
+        )
+        adapter = _make_pinned_adapter(pinned_ips, parsed.hostname or "")
+        adapter._semantica_pinned = True
+        active_session.mount("http://", adapter)
+        active_session.mount("https://", adapter)
+        active_session.headers["Host"] = host_header
+    else:
+        active_session.mount("http://", orig_http_adapter)
+        active_session.mount("https://", orig_https_adapter)
+        active_session.headers.pop("Host", None)
+
+
+_SESSION_LOCK_ATTR = "_semantica_ssrf_lock"
+_session_lock_registry_lock = threading.Lock()
+
+
+def _get_session_lock(session: "requests.Session") -> threading.Lock:
+    """Return a lock private to *session*, creating one on first use.
+
+    request_with_ssrf_guard mutates a caller-supplied session's adapters
+    and Host header for the duration of one guarded call (including every
+    redirect hop). Without serializing on the session itself, two guarded
+    calls sharing the same session from different threads could interleave
+    their mount()/restore cycles — one call's request could go out pinned
+    to (or carrying the Host header for) a completely different call's
+    target host. Double-checked locking so concurrent first-use doesn't
+    attach two different locks to the same session.
+    """
+    lock = getattr(session, _SESSION_LOCK_ATTR, None)
+    if lock is not None:
+        return lock
+    with _session_lock_registry_lock:
+        lock = getattr(session, _SESSION_LOCK_ATTR, None)
+        if lock is None:
+            lock = threading.Lock()
+            setattr(session, _SESSION_LOCK_ATTR, lock)
+        return lock
+
+
 def request_with_ssrf_guard(
     method: str,
     url: str,
@@ -317,10 +520,12 @@ def request_with_ssrf_guard(
 
     Session state that was removed is unconditionally restored in a ``finally``
     block so the session is left in its original state after this call returns,
-    regardless of how it exits (normal return, exception, redirect cap).  The
-    loop is sequential and single-threaded within one call, so the mutation is
-    safe as long as the caller does not share the session across concurrent
-    threads (the standard Semantica pattern: one session per ingestor instance).
+    regardless of how it exits (normal return, exception, redirect cap). A
+    caller-supplied session is also serialized on internally (see
+    ``_get_session_lock``): two guarded calls sharing the same session from
+    different threads block on each other for the call's duration rather than
+    interleaving their mutations, so concurrent use of a shared session is
+    safe, if not concurrent.
 
     Once credentials have been stripped for a cross-origin hop they are NOT
     re-added for subsequent hops in the same chain, even if a later hop
@@ -337,13 +542,38 @@ def request_with_ssrf_guard(
     )
     _original_host = (urlparse(url).hostname or "").lower()
 
-    validate_url_for_request(url, allow_private_ips=allow_private_ips)
+    current_pinned_ips = _resolve_pinned_ips(url, allow_private_ips=allow_private_ips)
 
-    requester = session.request if session is not None else requests.request
+    _owns_session = session is None
+    active_session = session if session is not None else requests.Session()
+    requester = active_session.request
     current_url = url
     current_method = method.upper()
     redirects_followed = 0
 
+    # A caller-supplied session is mutated (adapters + Host header, and
+    # potentially auth/trust_env below) for the duration of this call,
+    # including every redirect hop; serialize on the session itself so a
+    # second guarded call sharing it from another thread can't interleave
+    # its own mount()/restore cycle into the middle of this one. An owned
+    # session is private to this call, so no lock is needed. Released in
+    # the outermost `finally` below, alongside the state it protects.
+    _session_lock = None if _owns_session else _get_session_lock(active_session)
+    if _session_lock is not None:
+        _session_lock.acquire()
+
+    # Snapshot the session's pre-existing adapters/Host header so pinning
+    # (mounted per-hop below) can be fully undone when this call returns —
+    # required for a caller-supplied session, which outlives this call.
+    _orig_http_adapter = (
+        active_session.adapters.get("http://") or requests.adapters.HTTPAdapter()
+    )
+    _orig_https_adapter = (
+        active_session.adapters.get("https://") or requests.adapters.HTTPAdapter()
+    )
+    _had_host_header = "Host" in active_session.headers
+    _orig_host_header = active_session.headers.get("Host")
+
     # -- issue #947: snapshot every session-level credential source so we can
     #    restore them unconditionally when this call exits.
     _SENSITIVE = ("Authorization", "Proxy-Authorization")
@@ -369,6 +599,13 @@ def request_with_ssrf_guard(
 
     try:
         while True:
+            _apply_connection_pin(
+                active_session,
+                current_url,
+                current_pinned_ips,
+                _orig_http_adapter,
+                _orig_https_adapter,
+            )
             response = requester(
                 current_method,
                 current_url,
@@ -407,7 +644,9 @@ def request_with_ssrf_guard(
                 if next_host and next_host == _original_host
                 else redirect_allow_private_ips
             )
-            validate_url_for_request(next_url, allow_private_ips=hop_allow_private_ips)
+            current_pinned_ips = _resolve_pinned_ips(
+                next_url, allow_private_ips=hop_allow_private_ips
+            )
 
             # Do not leak sensitive headers or auth handlers to a different
             # origin on redirects.  All four credential sources are cleared:
@@ -472,13 +711,31 @@ def request_with_ssrf_guard(
             redirects_followed += 1
 
     finally:
-        # Unconditionally restore every session credential source we touched,
-        # so the session is in its original state after this call returns or raises.
-        if session is not None:
+        if _owns_session:
+            # No caller holds a reference to this session; just release it.
+            active_session.close()
+        else:
+            # Unconditionally restore every session credential source and
+            # pinning artifact we touched, so the session is in its
+            # original state after this call returns or raises.
             if _session_auth_backup:
                 for _h, _v in _session_auth_backup.items():
-                    session.headers[_h] = _v
+                    active_session.headers[_h] = _v
             # Restore session.auth to whatever it was before this call.
-            session.auth = _session_auth_handler_backup
+            active_session.auth = _session_auth_handler_backup
             # Restore session.trust_env (.netrc / env-proxy lookup flag).
-            session.trust_env = _session_trust_env_backup
+            active_session.trust_env = _session_trust_env_backup
+            # Undo any IP-pinning adapter/Host header mounted for a hop,
+            # closing the last pinned adapter so its pooled connection
+            # isn't leaked (see _apply_connection_pin).
+            _current = active_session.adapters.get("http://")
+            if getattr(_current, "_semantica_pinned", False):
+                _current.close()
+            active_session.mount("http://", _orig_http_adapter)
+            active_session.mount("https://", _orig_https_adapter)
+            if _had_host_header:
+                active_session.headers["Host"] = _orig_host_header
+            else:
+                active_session.headers.pop("Host", None)
+        if _session_lock is not None:
+            _session_lock.release()
diff --git a/semantica/parse/media_parser.py b/semantica/parse/media_parser.py
index af36b684..f34dcabe 100644
--- a/semantica/parse/media_parser.py
+++ b/semantica/parse/media_parser.py
@@ -243,6 +243,14 @@ class MediaParser:
         try:
             import subprocess
 
+            # ffprobe's own argument parser doesn't reliably honor a bare
+            # "--" end-of-options marker, so a filename starting with "-"
+            # could otherwise be parsed as an option; neutralize that by
+            # forcing a relative-path prefix ffprobe can't mistake for a flag.
+            ffprobe_path = str(file_path)
+            if ffprobe_path.startswith("-"):
+                ffprobe_path = f"./{ffprobe_path}"
+
             result = subprocess.run(
                 [
                     "ffprobe",
@@ -252,7 +260,7 @@ class MediaParser:
                     "json",
                     "-show_format",
                     "-show_streams",
-                    str(file_path),
+                    ffprobe_path,
                 ],
                 capture_output=True,
                 text=True,
diff --git a/semantica/triplet_store/anzo_store.py b/semantica/triplet_store/anzo_store.py
index 4ec4adaa..64423739 100644
--- a/semantica/triplet_store/anzo_store.py
+++ b/semantica/triplet_store/anzo_store.py
@@ -383,9 +383,9 @@ class AnzoStore:
         if self._is_uri_value(obj):
             if obj.startswith("<") and obj.endswith(">"):
                 inner = obj[1:-1]
-                if " " in inner or ">" in inner:
-                    raise ValueError(f"IRI contains invalid characters: {obj!r}")
+                sparql_escaping.validate_uri(inner)
                 return obj
+            sparql_escaping.validate_uri(obj)
             return f"<{obj}>"
 
         escaped = sparql_escaping.escape_literal(obj)
diff --git a/tests/ingest/test_auth_header_redirect_security.py b/tests/ingest/test_auth_header_redirect_security.py
index 882a9d59..fab40fe8 100644
--- a/tests/ingest/test_auth_header_redirect_security.py
+++ b/tests/ingest/test_auth_header_redirect_security.py
@@ -254,7 +254,7 @@ class TestAuthHandlerStripping:
         final = _mock_final()
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             side_effect=[redirect, final],
         ) as mock_req:
             request_with_ssrf_guard(
@@ -285,7 +285,7 @@ class TestAuthHandlerStripping:
         final = _mock_final()
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             side_effect=[redirect, final],
         ) as mock_req:
             request_with_ssrf_guard(
@@ -416,7 +416,7 @@ class TestAuthHandlerStripping:
         final = _mock_final()
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             side_effect=[hop1, hop2, final],
         ) as mock_req:
             request_with_ssrf_guard(
@@ -520,7 +520,7 @@ class TestCredentialResurrection:
         final = _mock_final()
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             side_effect=[hop1, hop2, final],
         ) as mock_req:
             request_with_ssrf_guard(
@@ -578,7 +578,7 @@ class TestRedirectTypesAndOriginChanges:
         final = _mock_final()
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             side_effect=[redirect, final],
         ) as mock_req:
             request_with_ssrf_guard(
@@ -600,7 +600,7 @@ class TestRedirectTypesAndOriginChanges:
         final = _mock_final()
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             side_effect=[redirect, final],
         ) as mock_req:
             request_with_ssrf_guard(
@@ -619,7 +619,7 @@ class TestRedirectTypesAndOriginChanges:
         final = _mock_final()
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             side_effect=[redirect, final],
         ) as mock_req:
             request_with_ssrf_guard(
@@ -638,7 +638,7 @@ class TestRedirectTypesAndOriginChanges:
         final = _mock_final()
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             side_effect=[redirect, final],
         ) as mock_req:
             request_with_ssrf_guard(
@@ -657,7 +657,7 @@ class TestRedirectTypesAndOriginChanges:
         final = _mock_final()
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             side_effect=[redirect, final],
         ) as mock_req:
             request_with_ssrf_guard(
@@ -676,7 +676,7 @@ class TestRedirectTypesAndOriginChanges:
         final = _mock_final()
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             side_effect=[redirect, final],
         ) as mock_req:
             request_with_ssrf_guard(
@@ -701,7 +701,7 @@ class TestAllowPrivateIpsOnRedirect:
         redirect = _mock_redirect("http://169.254.169.254/latest/meta-data/")
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             return_value=redirect,
         ):
             with pytest.raises(ValidationError, match="blocked"):
@@ -718,7 +718,7 @@ class TestAllowPrivateIpsOnRedirect:
         final = _mock_final()
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             side_effect=[redirect, final],
         ) as mock_req:
             request_with_ssrf_guard(
@@ -736,7 +736,7 @@ class TestAllowPrivateIpsOnRedirect:
         redirect = _mock_redirect("http://169.254.169.254/latest/meta-data/")
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             side_effect=[redirect, _mock_final()],
         ) as mock_req:
             # allow_private_ips=True with no override: redirect target validation
@@ -784,7 +784,7 @@ class TestMCPClientAuthRedirect:
         )
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             side_effect=[redirect, final],
         ) as mock_req:
             client._send_request_http({"jsonrpc": "2.0", "method": "ping"})
@@ -806,7 +806,7 @@ class TestMCPClientAuthRedirect:
         )
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             side_effect=[redirect, final],
         ) as mock_req:
             client._send_request_http({"jsonrpc": "2.0", "method": "ping"})
@@ -821,7 +821,7 @@ class TestMCPClientAuthRedirect:
         client = MCPClient(url="http://localhost:8000/mcp")
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             return_value=final,
         ) as mock_req:
             client._send_request_http({"jsonrpc": "2.0", "method": "ping"})
@@ -834,7 +834,7 @@ class TestMCPClientAuthRedirect:
         client = MCPClient(url="http://127.0.0.1:9000/mcp")
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             return_value=final,
         ) as mock_req:
             client._send_request_http({"jsonrpc": "2.0", "method": "ping"})
@@ -849,7 +849,7 @@ class TestMCPClientAuthRedirect:
         client = MCPClient(url="http://localhost:8000/mcp")
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             side_effect=[redirect, final],
         ) as mock_req:
             client._send_request_http({"jsonrpc": "2.0", "method": "ping"})
@@ -869,7 +869,7 @@ class TestMCPClientAuthRedirect:
         client = MCPClient(url="https://mcp.example.com/mcp")
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             return_value=redirect,
         ):
             with pytest.raises(ValidationError, match="blocked"):
@@ -888,7 +888,7 @@ class TestMCPClientAuthRedirect:
         )
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             side_effect=[redirect, final],
         ) as mock_req:
             client._send_request_http({"jsonrpc": "2.0", "method": "ping"})
@@ -904,7 +904,7 @@ class TestMCPClientAuthRedirect:
         client = MCPClient(url="https://mcp.example.com/mcp")
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             return_value=hop,
         ):
             with pytest.raises((ValidationError, Exception), match="[Rr]edirect|[Ee]xceeded"):
@@ -920,7 +920,7 @@ class TestMCPClientAuthRedirect:
         )
 
         with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
             return_value=final,
         ) as mock_req:
             client._send_request_http({"jsonrpc": "2.0", "method": "ping"})
diff --git a/tests/ingest/test_cookbook_integration.py b/tests/ingest/test_cookbook_integration.py
index ad470558..ffee8f03 100644
--- a/tests/ingest/test_cookbook_integration.py
+++ b/tests/ingest/test_cookbook_integration.py
@@ -13,7 +13,7 @@ class TestCookbookIntegration:
         # MCPClient._send_request_http now routes through request_with_ssrf_guard,
         # which calls requests.request (not httpx.post / requests.post directly).
         # Patch at the point where the guard issues the actual HTTP call.
-        with patch("semantica.ingest.ssrf.requests.request") as mock_request:
+        with patch("requests.Session.request") as mock_request:
 
             def side_effect(method, url, json=None, **kwargs):
                 if not json:
diff --git a/tests/ingest/test_feed_ingestor.py b/tests/ingest/test_feed_ingestor.py
index 957dedbf..70e676ed 100644
--- a/tests/ingest/test_feed_ingestor.py
+++ b/tests/ingest/test_feed_ingestor.py
@@ -222,7 +222,7 @@ def test_discover_feeds_empty() -> None:
         "semantica.ingest.ssrf.socket.getaddrinfo",
         return_value=[(2, 1, 6, "", ("93.184.216.34", 0))],
     ):
-        with patch("requests.request", side_effect=fake_request):
+        with patch("requests.Session.request", side_effect=fake_request):
             feeds = ingestor.discover_feeds("http://site.com")
 
     assert len(feeds) == 0
@@ -254,7 +254,7 @@ def test_discover_feeds_found() -> None:
         "semantica.ingest.ssrf.socket.getaddrinfo",
         return_value=[(2, 1, 6, "", ("93.184.216.34", 0))],
     ):
-        with patch("requests.request", side_effect=fake_request):
+        with patch("requests.Session.request", side_effect=fake_request):
             feeds = ingestor.discover_feeds("http://site.com")
 
     assert "http://site.com/rss.xml" in feeds
diff --git a/tests/ingest/test_ssrf_protection.py b/tests/ingest/test_ssrf_protection.py
index d7d3ab9b..8d6b9116 100644
--- a/tests/ingest/test_ssrf_protection.py
+++ b/tests/ingest/test_ssrf_protection.py
@@ -447,7 +447,7 @@ class TestSitemapCrawlerSSRF:
         redirect.close = MagicMock()
 
         with patch(
-            "semantica.ingest.ssrf.requests.request", return_value=redirect
+            "requests.Session.request", return_value=redirect
         ), patch(
             "semantica.ingest.ssrf.socket.getaddrinfo",
             return_value=[(None, None, None, None, ("93.184.216.34", 0))],
diff --git a/tests/ingest/test_submodules.py b/tests/ingest/test_submodules.py
index cac9c092..da5a944d 100644
--- a/tests/ingest/test_submodules.py
+++ b/tests/ingest/test_submodules.py
@@ -198,7 +198,7 @@ class TestMCPClient:
         # MCPClient._send_request_http now routes through request_with_ssrf_guard,
         # which calls requests.request (not requests.post) with allow_redirects=False.
         # Patch the requests.request call inside ssrf.py.
-        with patch("semantica.ingest.ssrf.requests.request") as mock_request:
+        with patch("requests.Session.request") as mock_request:
             mock_response = MagicMock()
             mock_response.status_code = 200
 
@@ -226,7 +226,7 @@ class TestMCPClient:
 
     def test_call_tool_mock_check(self):
         # Redo with the corrected patch target.
-        with patch("semantica.ingest.ssrf.requests.request") as mock_request:
+        with patch("requests.Session.request") as mock_request:
             mock_response = MagicMock()
             mock_response.status_code = 200