Merge pull request #752 from Sameer6305/feat/322-construct-templates

Add SPARQL CONSTRUCT query templates (Blazegraph-only)

CHANGELOG.md

@@ -11,6 +11,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **SPARQL CONSTRUCT query templates** (#752, #322) by @Sameer6305
+  - Added parameterized, injection-safe `CONSTRUCT` templates (`ConstructTemplate`, `ParameterDescriptor`, `ConstructTemplateRegistry`)
+  - Implemented Blazegraph-only execution support for now (Jena/RDF4J tracked in #754)
+  - Added pipeline integration via the `construct_template` step type
+
 - **Databricks Connector (Unity Catalog + Delta Lake ingestion)** (#747) by @KaifAhmad1
   - Added `DatabricksIngestor` (`semantica/ingest/databricks_ingestor.py`), mirroring `SnowflakeIngestor`'s structure and public API shape: a `DatabricksConnector` connection handler, a `DatabricksData` dataclass, and an optional-import guard for `databricks-sdk`/`databricks-sql-connector`
   - Supports personal access token and OAuth M2M (service principal `client_id`/`client_secret`) authentication, configurable via constructor args or `DATABRICKS_*` environment variables

docs/reference/pipeline.md

@@ -495,6 +495,40 @@ result   = engine.execute_pipeline(
   Delta detection uses SHA-256 checksums on source content. Only sources whose checksum differs from `base_version_id` are passed to downstream steps. For pipelines that run hourly or daily against a growing corpus, delta mode eliminates redundant re-embedding and re-extraction.
 </Note>
 
+## SPARQL CONSTRUCT Template Steps
+
+Use the `"construct_template"` step type to render and execute a [SPARQL CONSTRUCT template](triplet_store#sparql-construct-templates) as part of a pipeline. `store_backend` and `construct_template_registry` are execution-time resources, not step config — pass them to `execute_pipeline()`, the same way `delta_mode` steps receive `version_manager` and `triplet_store`:
+
+```python
+from semantica.pipeline import PipelineBuilder, ExecutionEngine
+from semantica.triplet_store.construct_templates import construct_template_step_handler
+
+builder = PipelineBuilder()
+builder.add_step(
+    "apply_person_template",
+    "construct_template",
+    handler=construct_template_step_handler,
+    template_name="person_to_foaf",
+    params={"subject": "http://ex.org/p1", "name": "Alice", "age": 30},
+    target_graph="http://ex.org/graphs/people",
+)
+pipeline = builder.build("person_pipeline")
+
+engine = ExecutionEngine()
+result = engine.execute_pipeline(
+    pipeline,
+    data=None,
+    store_backend=store,                   # required: a BlazegraphStore instance
+    construct_template_registry=registry,  # required: holds the registered template
+)
+
+triplets = result.output   # List[Triplet], already persisted via store.add_triplets
+```
+
+<Note>
+  `construct_template` steps raise `ProcessingError` if `store_backend` or `construct_template_registry` is missing from `execute_pipeline()`'s options, and `ValidationError` if `template_name` isn't registered.
+</Note>
+
 ## Schemas
 
 <AccordionGroup>

docs/reference/triplet_store.md

@@ -277,6 +277,65 @@ store.execute_query("""
   **`execute_query()` returns `QueryResult`, not a list.** Iterate `result.bindings`, not `result` directly. Each binding is a dict mapping variable name → `{"value": ..., "type": ...}`.
 </Warning>
 
+## SPARQL CONSTRUCT Templates
+
+`semantica.triplet_store.construct_templates` provides parameterized SPARQL `CONSTRUCT` query templates: define a reusable query once, substitute typed parameters safely, and persist the resulting triples in one call. This is available for the **Blazegraph backend only** (see [Backends](#backends) above) — `BlazegraphStore.execute_sparql()` is the only backend with CONSTRUCT-aware RDF parsing.
+
+```python
+from semantica.triplet_store.construct_templates import (
+    ConstructTemplate,
+    ParameterDescriptor,
+    ConstructTemplateRegistry,
+    render_construct_template,
+    execute_construct_template,
+)
+from semantica.triplet_store import BlazegraphStore
+
+# Define and register a template
+template = ConstructTemplate(
+    name="person_to_foaf",
+    description="Maps a person record subject to a foaf:name triple",
+    construct_query="""
+        PREFIX foaf: <http://xmlns.com/foaf/0.1/>
+        CONSTRUCT { {{subject}} foaf:name {{name}} ; foaf:age {{age}} }
+        WHERE { {{subject}} a <http://ex.org/Person> }
+    """,
+    parameters=[
+        ParameterDescriptor(name="subject", type="uri", required=True),
+        ParameterDescriptor(name="name", type="literal", required=True),
+        ParameterDescriptor(
+            name="age", type="typed-literal", required=False, default=0,
+            datatype="xsd:integer",
+        ),
+    ],
+)
+
+registry = ConstructTemplateRegistry()
+registry.register(template)
+
+# Render only: inspect the substituted SPARQL string, no network call
+sparql = render_construct_template(
+    registry.get("person_to_foaf"),
+    params={"subject": "http://ex.org/p1", "name": "Alice", "age": 30},
+)
+
+# Render + execute + persist in one call
+store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph", namespace="kb")
+triplets = execute_construct_template(
+    template=registry.get("person_to_foaf"),
+    params={"subject": "http://ex.org/p1", "name": "Alice", "age": 30},
+    store_backend=store,
+    target_graph="http://ex.org/graphs/people",
+)
+# triplets: List[Triplet], already persisted via store.add_triplets
+```
+
+Each `ParameterDescriptor.type` controls how its value is rendered: `"uri"` values are validated against an allowlist and wrapped in `<...>`, `"literal"` values are escaped and quoted, and `"typed-literal"` values require a `datatype` (e.g. `"xsd:integer"`) and render unquoted for numeric/boolean XSD types. Placeholders use `{{param}}` rather than SPARQL's own `?param` syntax so template placeholders are never confused with real SPARQL variables in the query body.
+
+<Note>
+  CONSTRUCT templates are Blazegraph-only. `execute_construct_template()` raises `ProcessingError` if `store_backend` does not implement both `execute_sparql()` and `add_triplets()`.
+</Note>
+
 ## SPARQL Result Pagination
 
 For large result sets, paginate with LIMIT and OFFSET:

semantica/pipeline/pipeline_validator.py

@@ -85,13 +85,26 @@ class PipelineValidator:
         return self.validate_pipeline(pipeline, **options)
 
     def validate_pipeline(
-        self, pipeline: Union["Pipeline", "PipelineBuilder"], **options
+        self,
+        pipeline: Union["Pipeline", "PipelineBuilder"],
+        construct_template_registry: Optional[Any] = None,
+        **options,
     ) -> ValidationResult:
         """
         Validate entire pipeline.
 
         Args:
             pipeline: Pipeline object or builder
+            construct_template_registry: Optional ConstructTemplateRegistry
+                instance.  When provided, steps whose step_type is
+                "construct_template" are additionally validated: the
+                template_name is checked for existence in the registry and
+                step.config["params"] is checked for all required template
+                parameters.  When None (the default), a WARNING-level issue
+                is added for each construct_template step noting that
+                template existence could not be checked — existing callers
+                that do not pass this argument see zero behavior change for
+                any other step type.
             **options: Additional options
 
         Returns:
@@ -134,7 +147,10 @@ class PipelineValidator:
                     message=f"Validating {len(pipeline.steps)} pipeline steps...",
                 )
                 for step in pipeline.steps:
-                    step_result = self.validate_step(step)
+                    step_result = self.validate_step(
+                        step,
+                        _construct_template_registry=construct_template_registry,
+                    )
                     if not step_result.valid:
                         errors.extend(step_result.errors)
                     warnings.extend(step_result.warnings)
@@ -205,7 +221,19 @@ class PipelineValidator:
 
         Args:
             step: Pipeline step
-            **constraints: Validation constraints
+            **constraints: Validation constraints.  The following keys are
+                understood by this method and consumed internally; all others
+                are available for future extension:
+
+                allow_no_handler (bool, default False): suppress the
+                    "has no handler" warning.
+
+                _construct_template_registry (ConstructTemplateRegistry | None,
+                    default None): registry forwarded by validate_pipeline
+                    for construct_template step-type validation.  Prefixed
+                    with '_' to signal internal plumbing — callers invoking
+                    validate_step directly should use validate_pipeline's
+                    construct_template_registry keyword argument instead.
 
         Returns:
             Validation result
@@ -227,6 +255,47 @@ class PipelineValidator:
         if not step.config:
             warnings.append(f"Step '{step.name}' has no configuration")
 
+        # --- construct_template step-type-specific validation ---
+        # This is the first step-type-specific check in this validator.
+        # Future step-type-specific checks should follow the same pattern:
+        # extract a registry/context object from constraints via a
+        # '_<type>_registry' key forwarded by validate_pipeline.
+        if step.step_type == "construct_template":
+            registry = constraints.get("_construct_template_registry")
+            if registry is None:
+                warnings.append(
+                    f"Step '{step.name}' (construct_template): no "
+                    f"construct_template_registry provided — template "
+                    f"existence and required parameters could not be checked."
+                )
+            else:
+                template_name = step.config.get("template_name")
+                template = registry.get(template_name) if template_name else None
+                if not template_name or template is None:
+                    errors.append(
+                        f"Step '{step.name}' (construct_template): "
+                        f"template_name {template_name!r} is not registered "
+                        f"in the provided construct_template_registry."
+                    )
+                else:
+                    # Static required-param check — mirrors render_construct_template's
+                    # exact runtime logic: required=True means the param is mandatory
+                    # regardless of whether a default is declared (render only uses
+                    # default when required=False, so a required param with a default
+                    # still raises ValidationError at execution time).
+                    provided_params = step.config.get("params") or {}
+                    missing = [
+                        d.name
+                        for d in template.parameters
+                        if d.required and d.name not in provided_params
+                    ]
+                    if missing:
+                        errors.append(
+                            f"Step '{step.name}' (construct_template): missing "
+                            f"required parameter(s) for template "
+                            f"{template_name!r}: {missing}."
+                        )
+
         return ValidationResult(
             valid=len(errors) == 0, errors=errors, warnings=warnings
         )

semantica/triplet_store/blazegraph_store.py

@@ -32,11 +32,35 @@ from typing import Any, Dict, List, Optional
 from urllib.parse import urljoin, urlparse
 
 import requests
+from rdflib import Graph, Literal
 
 from ..semantic_extract.triplet_extractor import Triplet
 from ..utils.exceptions import ProcessingError, ValidationError
 from ..utils.logging import get_logger
 from ..utils.progress_tracker import get_progress_tracker
+from . import sparql_escaping
+
+# Matches CONSTRUCT only as the actual SPARQL query-form keyword: anchored
+# from the start of the string, optionally preceded by PREFIX/BASE
+# declarations, then requires CONSTRUCT as the first non-whitespace keyword.
+# This prevents false-positives from SELECT/ASK queries that merely contain
+# the word "CONSTRUCT" inside a string literal or comment (e.g. a literal
+# value of '"please CONSTRUCT this"' or a comment line).
+_CONSTRUCT_QUERY_RE = re.compile(
+    r"""
+    \A                       # anchor to start of string
+    (?:                      # skip zero or more of:
+        \s+                  # whitespace
+      | \#[^\n]*             # comments (until newline)
+      | PREFIX\s+[\w\-]*:\s*<[^>]*>  # PREFIX declaration
+      | BASE\s+<[^>]*>       # BASE declaration
+    )*
+    \s*                      # any remaining whitespace before the query form
+    CONSTRUCT                # the actual query-form keyword
+    \b                       # must be followed by a non-word character
+    """,
+    re.IGNORECASE | re.VERBOSE,
+)
 
 
 class BlazegraphStore:
@@ -111,16 +135,49 @@ class BlazegraphStore:
         """Get SPARQL Update endpoint URL."""
         return urljoin(self.endpoint, f"/blazegraph/namespace/{self.namespace}/sparql")
 
+    def _is_construct_query(self, query: str) -> bool:
+        """
+        Detect whether `query` is a SPARQL CONSTRUCT query.
+
+        This is a dispatch helper local to BlazegraphStore, distinct from
+        QueryEngine._validate_query (which already treats CONSTRUCT as one of
+        its valid_keywords and therefore requires no change — CONSTRUCT
+        queries already pass validation today). This helper only decides
+        which HTTP Accept header and response parser execute_sparql uses; it
+        does not gate query validity.
+        """
+        return _CONSTRUCT_QUERY_RE.search(query) is not None
+
     def execute_sparql(self, query: str, **options) -> Dict[str, Any]:
         """
         Execute SPARQL query.
 
         Args:
             query: SPARQL query string
-            **options: Additional options
+            **options: Additional options:
+                - result_format: Optional[Literal["bindings", "construct"]].
+                  If omitted, auto-detected via _is_construct_query(query).
 
         Returns:
-            Query results
+            Query results. For non-CONSTRUCT queries (or when result_format
+            resolves to "bindings"), the existing shape is unchanged:
+                {"success": bool, "bindings": [...], "variables": [...], "metadata": {...}}
+            For CONSTRUCT queries (or result_format="construct"), the shape is:
+                {"success": bool, "bindings": [], "variables": [], "triples": [...],
+                 "metadata": {...}}
+            where "triples" is a list of (subject, predicate, object, metadata)
+            4-tuples parsed from the Turtle response via rdflib. subject and
+            predicate are always plain strings. object is the literal's
+            lexical value or the IRI string. metadata is a dict that is empty
+            ({}) for URIs and plain untyped/unlang-tagged literals, and
+            otherwise contains "datatype" (the datatype IRI as a string) and/
+            or "language" (the RFC 5646 language tag) for literals that carry
+            that information — preserving what would otherwise be lost by
+            collapsing every rdflib term down to str(term).
+
+        Raises:
+            ProcessingError: if not connected, the HTTP request fails, or (for
+                CONSTRUCT queries) the response body fails to parse as Turtle.
         """
         tracking_id = self.progress_tracker.start_tracking(
             module="triplet_store",
@@ -137,6 +194,70 @@ class BlazegraphStore:
 
             sparql_endpoint = self._get_sparql_endpoint()
 
+            result_format = options.get("result_format")
+            if result_format is None:
+                result_format = "construct" if self._is_construct_query(query) else "bindings"
+
+            if result_format == "construct":
+                self.progress_tracker.update_tracking(
+                    tracking_id, message="Sending CONSTRUCT query to Blazegraph endpoint..."
+                )
+                response = requests.post(
+                    sparql_endpoint,
+                    data={"query": query},
+                    headers={
+                        "Content-Type": "application/x-www-form-urlencoded",
+                        "Accept": "text/turtle",
+                    },
+                    timeout=self.timeout,
+                    auth=(self.username, self.password)
+                    if self.username and self.password
+                    else None,
+                )
+
+                response.raise_for_status()
+
+                self.progress_tracker.update_tracking(
+                    tracking_id, message="Parsing CONSTRUCT response as Turtle..."
+                )
+                graph = Graph()
+                try:
+                    graph.parse(data=response.content, format="turtle")
+                except Exception as parse_error:
+                    raise ProcessingError(
+                        f"Failed to parse CONSTRUCT response as Turtle: {parse_error}"
+                    ) from parse_error
+
+                triples = []
+                for s, p, o in graph:
+                    obj_metadata: Dict[str, Any] = {}
+                    if isinstance(o, Literal):
+                        if o.datatype is not None:
+                            obj_metadata["datatype"] = str(o.datatype)
+                        if o.language is not None:
+                            obj_metadata["language"] = str(o.language)
+                    triples.append((str(s), str(p), str(o), obj_metadata))
+
+                result = {
+                    "success": True,
+                    "bindings": [],
+                    "variables": [],
+                    "triples": triples,
+                    "metadata": {
+                        "query": query,
+                        "endpoint": sparql_endpoint,
+                        "result_format": "construct",
+                    },
+                }
+
+                self.progress_tracker.stop_tracking(
+                    tracking_id,
+                    status="completed",
+                    message=f"CONSTRUCT query executed: {len(triples)} triples",
+                )
+                return result
+
+            # Non-CONSTRUCT path — unchanged from prior behavior.
             self.progress_tracker.update_tracking(
                 tracking_id, message="Sending query to Blazegraph endpoint..."
             )
@@ -307,31 +428,12 @@ class BlazegraphStore:
         - Known prefixed names: ``xsd:integer``, ``rdf:langString``, etc.
 
         Raises ValueError for anything else.
+
+        Delegates to the shared sparql_escaping.resolve_datatype_iri so this
+        logic has one canonical implementation shared with the CONSTRUCT
+        template renderer (semantica/triplet_store/construct_templates.py).
         """
-        datatype = str(datatype)
-
-        # Already angle-bracketed — validate the inner IRI contains no whitespace
-        if datatype.startswith("<") and datatype.endswith(">"):
-            inner = datatype[1:-1]
-            if not inner or re.search(r"[\s<>\"{}|\\^`]", inner):
-                raise ValueError(f"Invalid datatype IRI: {datatype!r}")
-            return datatype
-
-        # Full absolute IRI without brackets
-        parsed = urlparse(datatype)
-        if parsed.scheme in {"http", "https", "urn"} and not re.search(r"[\s<>\"{}|\\^`]", datatype):
-            return f"<{datatype}>"
-
-        # Prefixed form — expand known prefixes only
-        if ":" in datatype:
-            prefix, local = datatype.split(":", 1)
-            if prefix in self._KNOWN_PREFIXES and re.match(r"^[A-Za-z0-9_\-\.]+$", local):
-                return f"<{self._KNOWN_PREFIXES[prefix]}{local}>"
-
-        raise ValueError(
-            f"Unsupported datatype {datatype!r}: use a full IRI (http/https/urn), "
-            f"an angle-bracketed IRI, or a known prefix (xsd/rdf/rdfs/owl/skos)."
-        )
+        return sparql_escaping.resolve_datatype_iri(datatype)
 
     def _is_uri_value(self, value: str) -> bool:
         """Detect if a value should be serialized as an IRI."""
@@ -346,15 +448,13 @@ class BlazegraphStore:
         return not re.search(r"\s", value)
 
     def _escape_literal(self, value: str) -> str:
-        """Escape string literal for SPARQL."""
-        return (
-            str(value)
-            .replace("\\", "\\\\")
-            .replace("\"", "\\\"")
-            .replace("\n", "\\n")
-            .replace("\r", "\\r")
-            .replace("\t", "\\t")
-        )
+        """Escape string literal for SPARQL.
+
+        Delegates to the shared sparql_escaping.escape_literal so this logic
+        has one canonical implementation shared with the CONSTRUCT template
+        renderer (semantica/triplet_store/construct_templates.py).
+        """
+        return sparql_escaping.escape_literal(value)
 
     def add_triplet(self, triplet: Triplet, **options) -> Dict[str, Any]:
         """Add single triplet."""

semantica/triplet_store/construct_templates.py

@@ -0,0 +1,824 @@
+"""
+SPARQL CONSTRUCT Query Templates (Blazegraph-only)
+
+This module provides parameterized SPARQL CONSTRUCT query templates: a
+`ConstructTemplate` defines a reusable CONSTRUCT query body with `{{param}}`
+placeholders, `ConstructTemplateRegistry` stores/retrieves templates by name,
+and `render_construct_template` safely substitutes parameter values into a
+validated, injection-safe SPARQL string.
+
+This module is intentionally Blazegraph-only. `execute_construct_template`
+renders a template, executes it via `store_backend.execute_sparql(...,
+result_format="construct")` (the Blazegraph CONSTRUCT-aware extension),
+converts the parsed RDF triples into `Triplet` objects, and persists them via
+`store_backend.add_triplets`.
+
+All literal escaping, URI allowlist validation, and datatype-IRI resolution
+is delegated to the shared `semantica.triplet_store.sparql_escaping` module —
+no escaping/validation logic is reimplemented here.
+
+Main Classes:
+    - ParameterDescriptor: Declares one template parameter's name/type/validation.
+    - ConstructTemplate: Named, reusable CONSTRUCT query definition.
+    - ConstructTemplateRegistry: Stores and retrieves ConstructTemplate instances.
+
+Main Functions:
+    - render_construct_template: Safely render a ConstructTemplate into SPARQL.
+    - execute_construct_template: Render, execute, parse, and persist in one call.
+
+Author: Semantica Contributors
+License: MIT
+"""
+
+import re
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Literal, Optional, Tuple
+
+from ..semantic_extract.triplet_extractor import Triplet
+from ..utils.exceptions import ProcessingError, ValidationError
+from ..utils.logging import get_logger
+from .sparql_escaping import escape_literal, resolve_datatype_iri, validate_uri
+
+ParameterKind = Literal["uri", "literal", "typed-literal"]
+
+# XSD local names (case-insensitive) that render as unquoted numeric/boolean
+# literals rather than quoted "<value>"^^<iri> literals. Matched against the
+# local name of the *resolved* datatype IRI so this works whether the
+# descriptor's datatype was given as a prefixed name (xsd:integer), a full
+# IRI, or a bracketed IRI.
+_INTEGER_LOCAL_NAMES = frozenset({"integer", "int", "long", "short"})
+_DECIMAL_LOCAL_NAMES = frozenset({"decimal", "double", "float"})
+_BOOLEAN_LOCAL_NAMES = frozenset({"boolean"})
+_NUMERIC_UNQUOTED_LOCAL_NAMES = _INTEGER_LOCAL_NAMES | _DECIMAL_LOCAL_NAMES | _BOOLEAN_LOCAL_NAMES
+
+_CONSTRUCT_KEYWORD_RE = re.compile(r"\bCONSTRUCT\b", re.IGNORECASE)
+_WHERE_KEYWORD_RE = re.compile(r"\bWHERE\b", re.IGNORECASE)
+_PLACEHOLDER_RE = re.compile(r"\{\{|\}\}")
+
+
+@dataclass
+class ParameterDescriptor:
+    """Describes one substitution parameter accepted by a ConstructTemplate."""
+
+    name: str
+    """Placeholder name as it appears in the query body, e.g. "subject" for {{subject}}."""
+
+    type: ParameterKind = "literal"
+    """One of "uri" | "literal" | "typed-literal"."""
+
+    required: bool = True
+    """If True and no value/default is supplied at render time, render raises ValidationError."""
+
+    default: Optional[Any] = None
+    """Used when the caller omits this parameter and required=False."""
+
+    datatype: Optional[str] = None
+    """Only meaningful when type == "typed-literal". An XSD datatype token accepted by
+    the shared resolve_datatype_iri, e.g. "xsd:integer", "xsd:dateTime", or a full IRI.
+    Required when type == "typed-literal"; render_construct_template (and
+    ConstructTemplateRegistry.register) raise ValidationError if type == "typed-literal"
+    and datatype is None."""
+
+    language: Optional[str] = None
+    """Only meaningful when type == "literal". RFC 5646 language tag, e.g. "en"."""
+
+
+@dataclass
+class ConstructTemplate:
+    """Parameterized SPARQL CONSTRUCT query template."""
+
+    name: str
+    """Unique registry key, e.g. "person_to_foaf"."""
+
+    description: str
+    """Human-readable summary, shown by list()/get_template_info()."""
+
+    construct_query: str
+    """Full CONSTRUCT query body containing {{param}} placeholders, e.g.:
+    "CONSTRUCT { <{{subject}}> foaf:name {{name}} } WHERE { ... }"
+    Must contain the CONSTRUCT keyword — enforced at register() time, not at
+    dataclass construction time."""
+
+    parameters: List[ParameterDescriptor] = field(default_factory=list)
+    """Ordered list of accepted parameters. Order has no runtime meaning, only used for
+    documentation / get_template_info() output."""
+
+    target_graph: Optional[str] = None
+    """Optional default named-graph IRI used when render_construct_template's/
+    execute_construct_template's target_graph argument is not supplied. This value,
+    like any caller-supplied target_graph, is ALWAYS passed through the same
+    validate_uri/escape path as a "uri"-typed parameter before being interpolated —
+    never through a raw f-string."""
+
+    metadata: dict = field(default_factory=dict)
+    """Free-form, mirrors PipelineTemplate.metadata (e.g. {"category": "rdf_mapping"})."""
+
+
+class ConstructTemplateRegistry:
+    """
+    CONSTRUCT template management system (Blazegraph-only).
+
+    Method shape mirrors PipelineTemplateManager:
+        PipelineTemplateManager.register_template(template) -> None
+        PipelineTemplateManager.get_template(name) -> Optional[PipelineTemplate]
+        PipelineTemplateManager.list_templates(category=None) -> List[str]
+
+    Unlike PipelineTemplateManager.register_template (which silently overwrites
+    by name), this registry rejects duplicate names with ValidationError —
+    CONSTRUCT queries execute against real triple stores, so silent overwrite
+    is a correctness hazard.
+    """
+
+    def __init__(self, config: Optional[Dict[str, Any]] = None, **kwargs):
+        """
+        Initialize template registry.
+
+        Args:
+            config: Configuration dictionary.
+            **kwargs: Additional configuration options.
+        """
+        self.logger = get_logger("construct_template_registry")
+        self.config = config or {}
+        self.config.update(kwargs)
+        self.templates: Dict[str, ConstructTemplate] = {}
+
+    def register(self, template: ConstructTemplate) -> None:
+        """
+        Register a CONSTRUCT template.
+
+        Args:
+            template: ConstructTemplate to register.
+
+        Raises:
+            ValidationError: if template.name is already registered,
+                construct_query does not contain the CONSTRUCT keyword, or any
+                ParameterDescriptor with type == "typed-literal" is missing
+                datatype. Validation happens before any mutation — a failed
+                register() call never partially overwrites the registry.
+        """
+        if template.name in self.templates:
+            raise ValidationError(
+                f"Template already registered: {template.name!r}. "
+                f"Remove it first via remove() if you intend to replace it."
+            )
+
+        if not _CONSTRUCT_KEYWORD_RE.search(template.construct_query):
+            raise ValidationError(
+                f"Template {template.name!r}: construct_query does not contain "
+                f"the CONSTRUCT keyword."
+            )
+
+        for descriptor in template.parameters:
+            if descriptor.type == "typed-literal" and descriptor.datatype is None:
+                raise ValidationError(
+                    f"Template {template.name!r}: parameter {descriptor.name!r} has "
+                    f"type='typed-literal' but no datatype declared."
+                )
+
+        self.templates[template.name] = template
+        self.logger.info(f"Registered CONSTRUCT template: {template.name}")
+
+    def get(self, name: str) -> Optional[ConstructTemplate]:
+        """
+        Get template by name.
+
+        Mirrors PipelineTemplateManager.get_template(template_name) -> Optional[PipelineTemplate].
+        """
+        return self.templates.get(name)
+
+    def list(self, category: Optional[str] = None) -> List[str]:
+        """
+        List registered template names, optionally filtered by metadata["category"].
+
+        Mirrors PipelineTemplateManager.list_templates(category=None) -> List[str].
+        """
+        if category:
+            return [
+                name
+                for name, template in self.templates.items()
+                if template.metadata.get("category") == category
+            ]
+        return list(self.templates.keys())
+
+    def remove(self, name: str) -> bool:
+        """
+        Remove a template by name.
+
+        Returns:
+            True if a template was removed, False if name was not registered.
+        """
+        if name in self.templates:
+            del self.templates[name]
+            self.logger.info(f"Removed CONSTRUCT template: {name}")
+            return True
+        return False
+
+    # --- PipelineTemplateManager-name aliases, for call-site consistency ---
+
+    def register_template(self, template: ConstructTemplate) -> None:
+        """Alias for register(), matching PipelineTemplateManager.register_template."""
+        self.register(template)
+
+    def get_template(self, template_name: str) -> Optional[ConstructTemplate]:
+        """Alias for get(), matching PipelineTemplateManager.get_template."""
+        return self.get(template_name)
+
+    def list_templates(self, category: Optional[str] = None) -> List[str]:
+        """Alias for list(), matching PipelineTemplateManager.list_templates."""
+        return self.list(category)
+
+    def get_template_info(self, template_name: str) -> Optional[Dict[str, Any]]:
+        """
+        Get template information.
+
+        Mirrors PipelineTemplateManager.get_template_info.
+
+        Returns:
+            {"name", "description", "parameter_count", "target_graph", "metadata"}
+            or None if template_name is not registered.
+        """
+        template = self.get(template_name)
+        if not template:
+            return None
+
+        return {
+            "name": template.name,
+            "description": template.description,
+            "parameter_count": len(template.parameters),
+            "target_graph": template.target_graph,
+            "metadata": template.metadata,
+        }
+
+
+# ---------------------------------------------------------------------------
+# render_construct_template and its internal helpers
+# ---------------------------------------------------------------------------
+
+
+def _local_name_of_datatype_iri(datatype_iri: str) -> str:
+    """Extract the lower-cased local name from a resolved (bracketed) datatype IRI."""
+    inner = datatype_iri[1:-1] if datatype_iri.startswith("<") and datatype_iri.endswith(">") else datatype_iri
+    if "#" in inner:
+        return inner.rsplit("#", 1)[1].lower()
+    if "/" in inner:
+        return inner.rsplit("/", 1)[1].lower()
+    return inner.lower()
+
+
+def _render_numeric_literal(value: Any, local_name: str, descriptor_name: str) -> str:
+    """
+    Render an unquoted numeric/boolean literal for a "typed-literal" parameter
+    whose resolved datatype local name is in _NUMERIC_UNQUOTED_LOCAL_NAMES.
+
+    Raises:
+        ValidationError: if value cannot be coerced to the declared datatype.
+    """
+    if local_name in _INTEGER_LOCAL_NAMES:
+        if isinstance(value, bool):
+            raise ValidationError(
+                f"Parameter {descriptor_name!r}: expected an integer value for "
+                f"datatype local name {local_name!r}, got boolean {value!r}."
+            )
+        try:
+            return str(int(str(value)))
+        except (TypeError, ValueError):
+            raise ValidationError(
+                f"Parameter {descriptor_name!r}: value {value!r} is not a valid "
+                f"integer for its declared XSD datatype."
+            )
+
+    if local_name in _DECIMAL_LOCAL_NAMES:
+        try:
+            return str(float(str(value)))
+        except (TypeError, ValueError):
+            raise ValidationError(
+                f"Parameter {descriptor_name!r}: value {value!r} is not a valid "
+                f"decimal/double/float for its declared XSD datatype."
+            )
+
+    if local_name in _BOOLEAN_LOCAL_NAMES:
+        if isinstance(value, bool):
+            return "true" if value else "false"
+        normalized = str(value).strip().lower()
+        if normalized in ("true", "1"):
+            return "true"
+        if normalized in ("false", "0"):
+            return "false"
+        raise ValidationError(
+            f"Parameter {descriptor_name!r}: value {value!r} is not a valid "
+            f"boolean for its declared XSD datatype."
+        )
+
+    raise ValidationError(
+        f"Parameter {descriptor_name!r}: datatype local name {local_name!r} is not "
+        f"a recognized unquoted-numeric XSD datatype."
+    )
+
+
+def _find_matching_brace(text: str, open_index: int) -> int:
+    """
+    Given the index of an opening '{' in text, return the index of its matching '}'.
+
+    Skips over double-quoted string literal content while scanning, so a '{'
+    or '}' character that appears *inside* a rendered literal value (e.g. a
+    "typed-literal"/"literal" parameter whose value contains a brace
+    character — braces are not among the characters escape_literal escapes,
+    since SPARQL/Turtle string literals don't require it) does not corrupt
+    the brace-depth count. Quote-escaping is assumed to follow
+    escape_literal's convention (\\\\ and \\" are the only two-character
+    escapes that can produce a literal backslash or double-quote), so an
+    unescaped '"' reliably toggles in/out of string-literal content.
+    """
+    depth = 0
+    in_string = False
+    i = open_index
+    while i < len(text):
+        ch = text[i]
+        if in_string:
+            if ch == "\\":
+                # Skip the escaped character (e.g. \" or \\) without
+                # inspecting it, matching escape_literal's escaping scheme.
+                i += 2
+                continue
+            if ch == '"':
+                in_string = False
+        else:
+            if ch == '"':
+                in_string = True
+            elif ch == "{":
+                depth += 1
+            elif ch == "}":
+                depth -= 1
+                if depth == 0:
+                    return i
+        i += 1
+    raise ValidationError("Unbalanced braces in construct_query: no matching '}' found.")
+
+
+def _split_construct_query(query_body: str) -> Tuple[str, str, str]:
+    """
+    Split a fully-substituted CONSTRUCT query into (preamble, construct_clause,
+    where_body), so the WHERE body can be re-wrapped in a GRAPH clause for
+    target_graph support.
+
+    Args:
+        query_body: Fully-substituted query string (all {{name}} tokens
+            already replaced).
+
+    Returns:
+        preamble: Everything before the CONSTRUCT keyword (e.g. PREFIX
+            declarations), stripped of leading/trailing whitespace. Preserved
+            verbatim so PREFIX declarations are not silently dropped when
+            target_graph wrapping rewrites the CONSTRUCT/WHERE structure.
+        construct_clause: The raw "{ ... }" template graph pattern immediately
+            following CONSTRUCT, braces included.
+        where_body: The raw text *inside* the "{ ... }" following WHERE,
+            braces excluded.
+
+    Raises:
+        ValidationError: if the CONSTRUCT/WHERE structure cannot be located.
+    """
+    construct_match = _CONSTRUCT_KEYWORD_RE.search(query_body)
+    if not construct_match:
+        raise ValidationError("construct_query does not contain the CONSTRUCT keyword.")
+    preamble = query_body[: construct_match.start()].strip()
+
+    construct_open = query_body.find("{", construct_match.end())
+    if construct_open == -1:
+        raise ValidationError("construct_query is missing '{' after the CONSTRUCT keyword.")
+    construct_close = _find_matching_brace(query_body, construct_open)
+    construct_clause = query_body[construct_open : construct_close + 1]
+
+    where_match = _WHERE_KEYWORD_RE.search(query_body, construct_close + 1)
+    if not where_match:
+        raise ValidationError("construct_query is missing a WHERE clause.")
+    where_open = query_body.find("{", where_match.end())
+    if where_open == -1:
+        raise ValidationError("construct_query is missing '{' after the WHERE keyword.")
+    where_close = _find_matching_brace(query_body, where_open)
+    where_body = query_body[where_open + 1 : where_close]
+
+    return preamble, construct_clause, where_body
+
+
+def render_construct_template(
+    template: ConstructTemplate,
+    params: Dict[str, Any],
+    target_graph: Optional[str] = None,
+) -> str:
+    """
+    Render a ConstructTemplate's construct_query into a safe, executable SPARQL
+    CONSTRUCT query string.
+
+    Args:
+        template: The ConstructTemplate to render.
+        params: Values for each {{name}} placeholder, keyed by ParameterDescriptor.name.
+            Values for missing optional parameters fall back to ParameterDescriptor.default.
+        target_graph: Named graph IRI to wrap the CONSTRUCT query in (via a GRAPH
+            clause around the WHERE body). If None, falls back to
+            template.target_graph. If both are None, no graph wrapping is applied.
+
+    Returns:
+        Fully-substituted SPARQL CONSTRUCT query string, safe to pass directly to
+        BlazegraphStore.execute_sparql.
+
+    Raises:
+        ValidationError: on any of:
+            - a required ParameterDescriptor has no value in params and no default
+            - a "uri"-typed parameter value fails validate_uri
+            - a "typed-literal"-typed parameter is missing template-declared datatype
+            - a "typed-literal"-typed parameter value cannot be coerced to that datatype
+            - target_graph (explicit arg or template.target_graph) fails validate_uri
+            - construct_query references a {{placeholder}} with no matching
+              ParameterDescriptor (detected as an unresolved placeholder after
+              substitution)
+    """
+    # Step 1: Resolve effective parameter values (params override, then default).
+    resolved: Dict[str, Any] = {}
+    for descriptor in template.parameters:
+        if descriptor.name in params:
+            resolved[descriptor.name] = params[descriptor.name]
+        elif not descriptor.required:
+            resolved[descriptor.name] = descriptor.default
+        else:
+            raise ValidationError(
+                f"Missing required parameter: {descriptor.name!r} "
+                f"(template {template.name!r})"
+            )
+
+    # Reject any keys in params that don't correspond to a declared
+    # ParameterDescriptor — silently ignoring unknown parameters would let
+    # a caller's typo (e.g. "subjcet" instead of "subject") go unnoticed
+    # while the mistyped value is simply dropped.
+    declared_names = {descriptor.name for descriptor in template.parameters}
+    unexpected_keys = sorted(set(params) - declared_names)
+    if unexpected_keys:
+        raise ValidationError(
+            f"Unexpected parameter(s) for template {template.name!r}: "
+            f"{unexpected_keys}. Declared parameters: {sorted(declared_names)}."
+        )
+
+    # Step 2: Render each parameter value according to its declared type.
+    rendered_values: Dict[str, str] = {}
+    for descriptor in template.parameters:
+        value = resolved[descriptor.name]
+
+        if descriptor.type == "uri":
+            safe_uri = validate_uri(value)
+            rendered_values[descriptor.name] = f"<{safe_uri}>"
+
+        elif descriptor.type == "literal":
+            escaped = escape_literal(value)
+            if descriptor.language is not None:
+                rendered_values[descriptor.name] = f'"{escaped}"@{descriptor.language}'
+            else:
+                rendered_values[descriptor.name] = f'"{escaped}"'
+
+        elif descriptor.type == "typed-literal":
+            if descriptor.datatype is None:
+                raise ValidationError(
+                    f"typed-literal parameter requires datatype: {descriptor.name!r} "
+                    f"(template {template.name!r})"
+                )
+            try:
+                datatype_iri = resolve_datatype_iri(descriptor.datatype)
+            except ValueError as exc:
+                raise ValidationError(
+                    f"Parameter {descriptor.name!r}: {exc}"
+                ) from exc
+
+            local_name = _local_name_of_datatype_iri(datatype_iri)
+            if local_name in _NUMERIC_UNQUOTED_LOCAL_NAMES:
+                rendered_values[descriptor.name] = _render_numeric_literal(
+                    value, local_name, descriptor.name
+                )
+            else:
+                escaped = escape_literal(value)
+                rendered_values[descriptor.name] = f'"{escaped}"^^{datatype_iri}'
+
+        else:
+            raise ValidationError(
+                f"Parameter {descriptor.name!r}: unknown parameter type {descriptor.type!r} "
+                f"(template {template.name!r})"
+            )
+
+    # Step 3: Substitute {{name}} tokens in construct_query.
+    query_body = template.construct_query
+    for name, rendered in rendered_values.items():
+        query_body = query_body.replace("{{" + name + "}}", rendered)
+
+    if _PLACEHOLDER_RE.search(query_body):
+        raise ValidationError(
+            f"Unresolved placeholder(s) in construct_query for template "
+            f"{template.name!r}: query still contains '{{{{' / '}}}}' tokens "
+            f"with no matching ParameterDescriptor."
+        )
+
+    # Step 4: Resolve effective target_graph — SAME validate_uri path as any
+    # "uri" parameter. No raw f-string interpolation is used here.
+    effective_graph = target_graph if target_graph is not None else template.target_graph
+    if effective_graph is not None:
+        safe_graph_uri = validate_uri(effective_graph)
+        wrapped_graph_token = f"<{safe_graph_uri}>"
+
+        preamble, construct_clause, where_body = _split_construct_query(query_body)
+        graph_wrapped_query = (
+            f"CONSTRUCT {construct_clause} "
+            f"WHERE {{ GRAPH {wrapped_graph_token} {{ {where_body} }} }}"
+        )
+        rendered_query = f"{preamble}\n{graph_wrapped_query}" if preamble else graph_wrapped_query
+    else:
+        rendered_query = query_body
+
+    return rendered_query
+
+
+def execute_construct_template(
+    template: ConstructTemplate,
+    params: Dict[str, Any],
+    store_backend: Any,
+    target_graph: Optional[str] = None,
+    **options: Any,
+) -> List[Triplet]:
+    """
+    Render, execute, parse, and persist a CONSTRUCT template in one call.
+
+    Args:
+        template: ConstructTemplate to execute.
+        params: Parameter values, forwarded to render_construct_template.
+        store_backend: A BlazegraphStore instance (Blazegraph-only; duck-typed
+            via hasattr(store_backend, "execute_sparql"), but this function
+            additionally requires store_backend to expose add_triplets — a
+            plain SPARQL-only backend without write support is rejected with
+            ProcessingError).
+        target_graph: Forwarded to render_construct_template; also used as
+            the `graph` option when persisting results via add_triplets so
+            constructed triples land in the same named graph they were
+            scoped to at query time.
+        **options: Forwarded to both store_backend.execute_sparql and
+            store_backend.add_triplets (e.g. timeout overrides). If options
+            contains "result_format" and/or "graph", those keys are
+            overridden by this function's own required values
+            ("construct" and effective_graph respectively) rather than
+            raising a duplicate-keyword-argument error — both keys are
+            load-bearing internal details of what this function does, so a
+            caller-supplied value for either is silently superseded, not an
+            error condition.
+
+    Returns:
+        The List[Triplet] that were constructed AND successfully persisted
+        via add_triplets. Order matches the order the store backend's
+        execute_sparql yielded triples in its "triples" key.
+
+    Raises:
+        ValidationError: propagated from render_construct_template.
+        ProcessingError: if store_backend lacks execute_sparql/add_triplets,
+            or if persistence via add_triplets does not report success.
+
+    Exception-propagation convention:
+        This function does not wrap or catch exceptions raised by
+        render_construct_template or store_backend.execute_sparql — each
+        sub-layer is responsible for raising its own correctly-typed
+        exception (ValidationError for rendering failures; ProcessingError
+        for execution failures, as BlazegraphStore.execute_sparql already
+        does internally for connection/request/Turtle-parse errors). Adding
+        a second wrapping layer here would only obscure the original error
+        with no new information. The one exception this function DOES raise
+        itself is the add_triplets write-failure case immediately below,
+        because add_triplets signals failure via a returned dict rather than
+        an exception, so there is no pre-existing typed exception to let
+        propagate.
+
+    Why store_backend.execute_sparql is called directly instead of
+    QueryEngine.execute_query (investigated for issue #322 item on reusing
+    "Generic SPARQL execution ... QueryEngine.execute_query" — this is a
+    deliberate choice, not an oversight):
+        Routing through QueryEngine.execute_query was evaluated and found to
+        introduce three concrete regressions against this function's already
+        -tested behavior:
+        1. QueryEngine.optimize_query's whitespace-collapse
+           (" ".join(query.split())) corrupts literal content. Verified: a
+           literal parameter value of "value:   three   spaces   " comes
+           back from optimize_query as "value: three spaces " — the
+           collapsing operates on the whole query string with no awareness
+           of quoted-string boundaries, silently altering the literal's
+           actual content. This breaks the escaping guarantees
+           render_construct_template exists to provide (Property 1).
+        2. QueryEngine.execute_query caches results keyed only on
+           normalized query text (enable_caching=True by default). A
+           CONSTRUCT query's correct results depend on the live state of the
+           graph at query time; repeated execute_construct_template calls
+           with identical params (a normal usage pattern — same template
+           re-run periodically) would silently return a stale cached
+           QueryResult instead of re-querying, causing incorrect
+           persistence via add_triplets.
+        3. QueryEngine.execute_query wraps its entire body in a blanket
+           `except Exception: raise ProcessingError(...)`, re-typing every
+           exception regardless of origin. This directly conflicts with the
+           exception-propagation convention documented and tested above
+           (e.g. a raw ConnectionError from store_backend.execute_sparql
+           must propagate as ConnectionError, not get silently re-wrapped
+           into a differently-worded ProcessingError).
+        Fixing this properly would require QueryEngine itself to support a
+        "do not touch this already-rendered, already-safe query" mode
+        (disabling optimize_query and caching for CONSTRUCT) and to stop
+        re-wrapping already-correctly-typed exceptions — changes to a
+        shared, backend-agnostic module used by other query paths, which is
+        out of scope for this Blazegraph-only feature per the issue's own
+        no-scope-creep guidance. Calling store_backend.execute_sparql
+        directly is therefore the correct choice today, not a gap to close
+        casually.
+    """
+    if not (hasattr(store_backend, "execute_sparql") and hasattr(store_backend, "add_triplets")):
+        raise ProcessingError(
+            "store_backend must support both execute_sparql and add_triplets "
+            "to use execute_construct_template."
+        )
+
+    # Step 1: Render (raises ValidationError unchanged on any failure).
+    rendered_query = render_construct_template(template, params, target_graph)
+
+    # Step 2: Execute via Blazegraph's CONSTRUCT-aware path. result_format is
+    # a load-bearing internal detail of this function (CONSTRUCT parsing
+    # requires it); if a caller's own **options happens to contain
+    # "result_format", the explicit value here must win rather than raising
+    # "got multiple values for keyword argument" — so it is popped out of a
+    # local copy of options and re-applied explicitly.
+    execute_options = dict(options)
+    execute_options.pop("result_format", None)
+    # Deliberately calling store_backend.execute_sparql directly, NOT
+    # QueryEngine.execute_query — see "Why store_backend.execute_sparql is
+    # called directly instead of QueryEngine.execute_query" in this
+    # function's docstring before routing through QueryEngine here.
+    query_result = store_backend.execute_sparql(
+        rendered_query, result_format="construct", **execute_options
+    )
+
+    if not query_result.get("success", False):
+        raise ProcessingError(
+            f"CONSTRUCT query execution failed for template {template.name!r}: "
+            f"{query_result}"
+        )
+
+    # Step 3: Convert parsed RDF triples to Triplet objects. store_backend is
+    # a BlazegraphStore instance, whose execute_sparql returns Dict[str, Any]
+    # with a "triples" key for CONSTRUCT queries: a list of
+    # (subject, predicate, object, metadata) 4-tuples (see
+    # BlazegraphStore.execute_sparql). object_metadata carries "datatype"
+    # and/or "language" for literals that have that information — those are
+    # folded into the resulting Triplet's own metadata under the
+    # "datatype"/"lang" keys, which is exactly what
+    # BlazegraphStore._format_object_for_sparql reads
+    # (metadata.get("datatype") / metadata.get("lang")) when re-serializing
+    # a Triplet back to SPARQL, so a typed/lang-tagged literal round-trips
+    # correctly through add_triplets instead of being silently flattened to
+    # an untyped plain string.
+    raw_triples = query_result.get("triples", [])
+    triplets: List[Triplet] = []
+    for s, p, o, object_metadata in raw_triples:
+        triplet_metadata = {"source": "construct_template", "template": template.name}
+        if object_metadata.get("datatype"):
+            triplet_metadata["datatype"] = object_metadata["datatype"]
+        if object_metadata.get("language"):
+            triplet_metadata["lang"] = object_metadata["language"]
+
+        triplets.append(
+            Triplet(
+                subject=str(s),
+                predicate=str(p),
+                object=str(o),
+                # confidence=1.0 (explicit, matching Triplet's own default):
+                # a CONSTRUCT query is a deterministic graph transformation
+                # over already-persisted RDF/SPARQL-computed data, not a
+                # probabilistic extraction (unlike NER/LLM-based Triplet
+                # extraction, where confidence reflects genuine estimation
+                # uncertainty). There is no meaningful uncertainty to encode
+                # here, so full confidence is the correct value, not an
+                # accidental default.
+                confidence=1.0,
+                metadata=triplet_metadata,
+            )
+        )
+
+    # Step 4: Persist via add_triplets (same write path as any other bulk
+    # load). Same reasoning as the result_format pop above: "graph" is the
+    # explicit target_graph/template.target_graph resolution this function
+    # exists to enforce, so a caller-supplied "graph" in **options must not
+    # crash the call or silently bypass that resolution — pop it before
+    # forwarding and let the computed effective_graph win.
+    effective_graph = target_graph if target_graph is not None else template.target_graph
+    add_triplets_options = dict(options)
+    add_triplets_options.pop("graph", None)
+    write_result = store_backend.add_triplets(
+        triplets, graph=effective_graph, **add_triplets_options
+    )
+
+    if not write_result.get("success", False):
+        raise ProcessingError(
+            f"Failed to persist constructed triples for template "
+            f"{template.name!r}: {write_result}"
+        )
+
+    return triplets
+
+
+def construct_template_step_handler(data: Any, **options: Any) -> List[Triplet]:
+    """
+    Pipeline step handler for the "construct_template" step type.
+
+    Resolves `store_backend` and `construct_template_registry` from execution
+    options, looks up the named template, and delegates to
+    execute_construct_template. Mirrors the exact resolution pattern used for
+    `triplet_store`/`version_manager` in
+    ExecutionEngine._execute_step's delta_mode handling: call-time options
+    first, engine config fallback second, ProcessingError if either is
+    missing.
+
+    Called by ExecutionEngine._execute_step as
+    `step.handler(data, **step.config, **options)` — no change to
+    PipelineStep or _execute_step is required, since step.config and
+    **options already flow through generically for any step type.
+
+    Args:
+        data: Pipeline data flowing in (unused by this step type — a
+            construct_template step's output is independent of upstream
+            step data, exactly like other non-delta step types).
+        **options: The merged step.config + execution-time options dict, as
+            passed by ExecutionEngine._execute_step's
+            `step.handler(data, **step.config, **options)` call. Expected
+            keys:
+                - template_name: str — required, key into
+                  construct_template_registry.
+                - params: Dict[str, Any] — required, forwarded to
+                  render_construct_template.
+                - target_graph: Optional[str] — optional.
+                - store_backend: Any — resolved from options directly, or
+                  from an "engine_config" dict passed alongside it (see
+                  below); required.
+                - construct_template_registry: ConstructTemplateRegistry —
+                  resolved the same way; required.
+                - engine_config: Optional[Dict[str, Any]] — fallback source
+                  for store_backend/construct_template_registry when not
+                  present directly in options, mirroring the *shape* of
+                  self.config.get(...) in ExecutionEngine._execute_step's
+                  delta_mode handling. One real difference from delta_mode:
+                  delta_mode's fallback runs inside _execute_step itself (a
+                  bound ExecutionEngine method with direct access to
+                  self.config), whereas construct_template_step_handler is a
+                  plain function with no such access — by design, this step
+                  type requires no pre-handler interception in
+                  _execute_step (see Requirement 7.5), so self.config is
+                  simply never threaded to any step.handler call today. In
+                  practice this means store_backend/construct_template_registry
+                  should normally be passed as call-time options via
+                  ExecutionEngine.execute_pipeline(pipeline, data,
+                  store_backend=..., construct_template_registry=...),
+                  which flow through untouched to this handler. The
+                  engine_config parameter exists for callers who explicitly
+                  forward their own ExecutionEngine(config=...) dict into
+                  execute_pipeline's options (e.g. engine_config=engine.config)
+                  and want the same two-tier resolution shape as delta_mode.
+                - step_name: Optional[str] — used only to name the failing
+                  step in error messages; defaults to "construct_template"
+                  if not supplied (design.md's own pseudocode references
+                  step.name inside this handler, but the actual
+                  step.handler(data, **step.config, **options) invocation
+                  never passes step.name to a standalone handler function —
+                  this default covers that gap without requiring any
+                  ExecutionEngine/_execute_step change).
+
+    Returns:
+        The List[Triplet] returned by execute_construct_template.
+
+    Raises:
+        ProcessingError: if store_backend or construct_template_registry
+            cannot be resolved from options (or its "engine_config" fallback).
+        ValidationError: if template_name is not registered in the resolved
+            construct_template_registry.
+    """
+    step_name = options.get("step_name", "construct_template")
+    engine_config = options.get("engine_config") or {}
+
+    store_backend = options.get("store_backend") or engine_config.get("store_backend")
+    construct_template_registry = options.get(
+        "construct_template_registry"
+    ) or engine_config.get("construct_template_registry")
+
+    if not store_backend or not construct_template_registry:
+        raise ProcessingError(
+            f"Step '{step_name}' requires 'store_backend' and "
+            f"'construct_template_registry' in execution options for "
+            f"construct_template processing."
+        )
+
+    template_name = options.get("template_name")
+    template = construct_template_registry.get(template_name)
+    if template is None:
+        raise ValidationError(f"Unknown construct template: {template_name!r}")
+
+    return execute_construct_template(
+        template=template,
+        params=options.get("params", {}),
+        store_backend=store_backend,
+        target_graph=options.get("target_graph"),
+    )

semantica/triplet_store/query_engine.py

@@ -49,6 +49,15 @@ class QueryResult:
     variables: List[str]
     execution_time: float = 0.0
     metadata: Dict[str, Any] = field(default_factory=dict)
+    triples: List[tuple] = field(default_factory=list)
+    """Populated only for CONSTRUCT queries. Each element is a (subject,
+    predicate, object, metadata) 4-tuple, taken directly from the store
+    backend's execute_sparql "triples" key (see BlazegraphStore.execute_sparql
+    CONSTRUCT path). subject/predicate/object are strings; metadata is a dict
+    that is empty ({}) for URIs and plain untyped/unlang-tagged literals, and
+    otherwise carries "datatype" and/or "language" keys for literals that
+    have that information, so it is not silently lost. Empty list for all
+    SELECT/ASK/DESCRIBE queries and for backends without CONSTRUCT support."""
 
 
 @dataclass
@@ -183,6 +192,7 @@ class QueryEngine:
                 bindings=result_data.get("bindings", []),
                 variables=result_data.get("variables", []),
                 execution_time=execution_time,
+                triples=result_data.get("triples", []),
                 metadata={
                     **result_data.get("metadata", {}),
                     "optimized": optimized_query != prepared_query,

semantica/triplet_store/sparql_escaping.py

@@ -0,0 +1,159 @@
+"""
+Shared SPARQL literal-escaping and URI-validation primitives.
+
+This module centralizes string-escaping, datatype-IRI resolution, and URI
+allowlist validation logic that was previously duplicated (or would have
+been duplicated) between ``BlazegraphStore`` and the CONSTRUCT template
+renderer. ``BlazegraphStore._escape_literal`` and
+``BlazegraphStore._resolve_datatype_iri`` delegate to ``escape_literal`` and
+``resolve_datatype_iri`` here without any change in behavior; ``validate_uri``
+is new and is used by ``render_construct_template`` for both ``"uri"``-typed
+parameters and ``target_graph`` values.
+
+Author: Semantica Contributors
+License: MIT
+"""
+
+import re
+from typing import FrozenSet
+from urllib.parse import urlparse
+
+from ..utils.exceptions import ValidationError
+
+# Allowed URI schemes for validate_uri's allowlist (Requirement 3.2/4.4).
+_ALLOWED_URI_SCHEMES: FrozenSet[str] = frozenset({"http", "https", "urn"})
+
+# Known prefix expansions for XSD and common RDF vocabularies.
+# Identical table to BlazegraphStore._KNOWN_PREFIXES.
+KNOWN_PREFIXES: dict = {
+    "xsd": "http://www.w3.org/2001/XMLSchema#",
+    "rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#",
+    "rdfs": "http://www.w3.org/2000/01/rdf-schema#",
+    "owl": "http://www.w3.org/2002/07/owl#",
+    "skos": "http://www.w3.org/2004/02/skos/core#",
+}
+
+# RFC 5646 language tag: primary subtag optionally followed by '-' + subtags.
+# Identical pattern to BlazegraphStore._LANG_TAG_RE.
+LANG_TAG_RE = re.compile(r"^[a-zA-Z]{1,8}(-[a-zA-Z0-9]{1,8})*$")
+
+# Disallowed-character check shared by validate_uri and resolve_datatype_iri,
+# identical to the character class used throughout BlazegraphStore.
+_DISALLOWED_URI_CHARS_RE = re.compile(r"[\s<>\"{}|\\^`]")
+
+
+def escape_literal(value: str) -> str:
+    """
+    Escape a string literal for safe inclusion inside SPARQL double quotes.
+
+    This is a byte-for-byte copy of BlazegraphStore._escape_literal's
+    transformation:
+        \\ -> \\\\ , " -> \\" , \\n -> \\n , \\r -> \\r , \\t -> \\t
+
+    Args:
+        value: Raw literal value (converted via str() first, matching the
+            original method's behavior of accepting non-str inputs).
+
+    Returns:
+        Escaped string safe to place inside a SPARQL/Turtle double-quoted
+        literal.
+    """
+    return (
+        str(value)
+        .replace("\\", "\\\\")
+        .replace("\"", "\\\"")
+        .replace("\n", "\\n")
+        .replace("\r", "\\r")
+        .replace("\t", "\\t")
+    )
+
+
+def resolve_datatype_iri(datatype: str) -> str:
+    """
+    Expand a datatype string to a validated SPARQL IRI token.
+
+    This is a byte-for-byte copy of BlazegraphStore._resolve_datatype_iri's
+    logic and exception behavior (raises ValueError, not ValidationError, to
+    match the original method's existing contract).
+
+    Accepts:
+    - Already-wrapped IRIs: ``<http://...>``
+    - Full IRIs:            ``http://...`` / ``https://...`` / ``urn:...``
+    - Known prefixed names: ``xsd:integer``, ``rdf:langString``, etc.
+
+    Raises:
+        ValueError: for anything else, or for malformed/unsafe IRIs.
+
+    Returns:
+        An angle-bracketed IRI token, e.g. ``<http://www.w3.org/2001/XMLSchema#integer>``.
+    """
+    datatype = str(datatype)
+
+    # Already angle-bracketed — validate the inner IRI contains no whitespace
+    if datatype.startswith("<") and datatype.endswith(">"):
+        inner = datatype[1:-1]
+        if not inner or _DISALLOWED_URI_CHARS_RE.search(inner):
+            raise ValueError(f"Invalid datatype IRI: {datatype!r}")
+        return datatype
+
+    # Full absolute IRI without brackets
+    parsed = urlparse(datatype)
+    if parsed.scheme in {"http", "https", "urn"} and not _DISALLOWED_URI_CHARS_RE.search(datatype):
+        return f"<{datatype}>"
+
+    # Prefixed form — expand known prefixes only
+    if ":" in datatype:
+        prefix, local = datatype.split(":", 1)
+        if prefix in KNOWN_PREFIXES and re.match(r"^[A-Za-z0-9_\-\.]+$", local):
+            return f"<{KNOWN_PREFIXES[prefix]}{local}>"
+
+    raise ValueError(
+        f"Unsupported datatype {datatype!r}: use a full IRI (http/https/urn), "
+        f"an angle-bracketed IRI, or a known prefix (xsd/rdf/rdfs/owl/skos)."
+    )
+
+
+def validate_uri(
+    value: str,
+    *,
+    allowed_schemes: FrozenSet[str] = _ALLOWED_URI_SCHEMES,
+) -> str:
+    """
+    Validate that `value` is a safe absolute IRI using urllib.parse, and
+    return it unchanged (no bracket-wrapping — callers wrap with <...> at
+    render time).
+
+    Uses urllib.parse.urlparse(value):
+      - scheme must be in allowed_schemes (default {"http", "https", "urn"})
+      - the value must not contain whitespace or any of
+        < > " { } | \\ ^ ` characters (same disallowed-character set used by
+        resolve_datatype_iri / BlazegraphStore's existing IRI checks)
+
+    Args:
+        value: Candidate URI/IRI string.
+        allowed_schemes: Set of permitted URI schemes.
+
+    Returns:
+        The validated `value`, unchanged.
+
+    Raises:
+        ValidationError: if value is empty, not a string, has a scheme not in
+            allowed_schemes, or contains a disallowed character.
+    """
+    if not isinstance(value, str) or not value:
+        raise ValidationError(f"Invalid URI: value must be a non-empty string, got {value!r}")
+
+    if _DISALLOWED_URI_CHARS_RE.search(value):
+        raise ValidationError(
+            f"Invalid URI {value!r}: contains disallowed character(s) "
+            f"(whitespace or one of < > \" {{ }} | \\ ^ `)"
+        )
+
+    parsed = urlparse(value)
+    if parsed.scheme not in allowed_schemes:
+        raise ValidationError(
+            f"Invalid URI {value!r}: scheme {parsed.scheme!r} is not allowed "
+            f"(allowed: {sorted(allowed_schemes)})"
+        )
+
+    return value

tests/pipeline/test_pipeline_validator_construct.py

@@ -0,0 +1,374 @@
+"""
+Tests for issue #3: construct_template_registry support in PipelineValidator.
+
+Covers four new behaviours and one explicit regression guard:
+  1. No registry provided → WARNING-level issue, not an error (result still valid).
+  2. Registry provided, template_name absent → ERROR.
+  3. Registry provided, required param absent from step.config["params"] → ERROR.
+  4. Registry provided, everything valid → no errors/warnings for that step.
+  5. Regression: pipelines containing only non-construct_template steps see
+     identical validation output whether or not construct_template_registry is passed.
+"""
+
+import unittest
+from unittest.mock import MagicMock, patch
+
+from semantica.pipeline.pipeline_builder import Pipeline, PipelineStep
+from semantica.pipeline.pipeline_validator import PipelineValidator
+from semantica.triplet_store.construct_templates import (
+    ConstructTemplate,
+    ConstructTemplateRegistry,
+    ParameterDescriptor,
+)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_MINIMAL_QUERY = "CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }"
+
+
+def _make_registry(*templates):
+    """Return a ConstructTemplateRegistry pre-loaded with the given templates."""
+    reg = ConstructTemplateRegistry()
+    for t in templates:
+        reg.register(t)
+    return reg
+
+
+def _make_template(name, *param_descriptors):
+    return ConstructTemplate(
+        name=name,
+        description="test template",
+        construct_query=_MINIMAL_QUERY,
+        parameters=list(param_descriptors),
+    )
+
+
+def _make_pipeline(*steps):
+    """Wrap PipelineStep objects in a minimal Pipeline."""
+    return Pipeline(name="test_pipeline", steps=list(steps))
+
+
+def _make_step(name, step_type, config=None):
+    return PipelineStep(name=name, step_type=step_type, config=config or {})
+
+
+# ---------------------------------------------------------------------------
+# Test class
+# ---------------------------------------------------------------------------
+
+
+class TestPipelineValidatorConstructRegistry(unittest.TestCase):
+
+    def setUp(self):
+        self.tracker_patcher = patch(
+            "semantica.utils.progress_tracker.get_progress_tracker"
+        )
+        mock_get_tracker = self.tracker_patcher.start()
+        mock_tracker = MagicMock()
+        mock_get_tracker.return_value = mock_tracker
+
+    def tearDown(self):
+        self.tracker_patcher.stop()
+
+    # ------------------------------------------------------------------
+    # Test 1: registry not provided -> warning, not error
+    # ------------------------------------------------------------------
+
+    def test_no_registry_yields_warning_not_error(self):
+        """
+        When construct_template_registry is None (the default), a construct_template
+        step should produce exactly one WARNING mentioning the inability to check
+        template existence, and the result should still be valid (no errors).
+        """
+        step = _make_step(
+            "build_foaf",
+            "construct_template",
+            config={"template_name": "person_to_foaf", "params": {}},
+        )
+        pipeline = _make_pipeline(step)
+        validator = PipelineValidator()
+
+        result = validator.validate_pipeline(pipeline)  # no registry kwarg
+
+        self.assertTrue(result.valid, msg=f"Expected valid; errors={result.errors}")
+        self.assertEqual(result.errors, [], msg="Expected no errors")
+        # At least one warning about the missing registry
+        registry_warnings = [
+            w for w in result.warnings if "construct_template_registry" in w
+        ]
+        self.assertGreater(
+            len(registry_warnings),
+            0,
+            msg=f"Expected a warning about missing registry; warnings={result.warnings}",
+        )
+
+    # ------------------------------------------------------------------
+    # Test 2: registry provided, template_name absent -> error
+    # ------------------------------------------------------------------
+
+    def test_missing_template_name_in_registry_yields_error(self):
+        """
+        When a registry is provided but step.config["template_name"] is not
+        registered in it, validate_pipeline must produce an ERROR and result.valid
+        must be False.
+        """
+        # Registry has "other_template", not "person_to_foaf"
+        registry = _make_registry(
+            _make_template("other_template")
+        )
+        step = _make_step(
+            "build_foaf",
+            "construct_template",
+            config={"template_name": "person_to_foaf", "params": {}},
+        )
+        pipeline = _make_pipeline(step)
+        validator = PipelineValidator()
+
+        result = validator.validate_pipeline(
+            pipeline, construct_template_registry=registry
+        )
+
+        self.assertFalse(result.valid, msg="Expected invalid result")
+        template_errors = [e for e in result.errors if "person_to_foaf" in e]
+        self.assertGreater(
+            len(template_errors),
+            0,
+            msg=f"Expected error mentioning 'person_to_foaf'; errors={result.errors}",
+        )
+
+    def test_none_template_name_in_config_yields_error(self):
+        """
+        Edge case: step.config has no 'template_name' key at all.
+        Registry is provided; should still produce an error.
+        """
+        registry = _make_registry(_make_template("some_template"))
+        step = _make_step(
+            "bad_step",
+            "construct_template",
+            config={"params": {}},  # no template_name key
+        )
+        pipeline = _make_pipeline(step)
+        validator = PipelineValidator()
+
+        result = validator.validate_pipeline(
+            pipeline, construct_template_registry=registry
+        )
+
+        self.assertFalse(result.valid)
+        self.assertTrue(any("template_name" in e or "None" in e for e in result.errors),
+                        msg=f"Expected error about missing/None template_name; errors={result.errors}")
+
+    # ------------------------------------------------------------------
+    # Test 3: registry provided, required param absent -> error
+    # ------------------------------------------------------------------
+
+    def test_missing_required_param_yields_error(self):
+        """
+        When the registry is provided, the template is found, but a required
+        parameter is absent from step.config["params"], validate_pipeline must
+        produce an ERROR listing the missing parameter.
+        """
+        template = _make_template(
+            "person_to_foaf",
+            ParameterDescriptor(name="subject", type="uri", required=True),
+            ParameterDescriptor(name="name", type="literal", required=True),
+            ParameterDescriptor(name="lang", type="literal", required=False, default="en"),
+        )
+        registry = _make_registry(template)
+        # Provides "subject" but omits the required "name"
+        step = _make_step(
+            "build_foaf",
+            "construct_template",
+            config={
+                "template_name": "person_to_foaf",
+                "params": {"subject": "http://example.org/alice"},
+            },
+        )
+        pipeline = _make_pipeline(step)
+        validator = PipelineValidator()
+
+        result = validator.validate_pipeline(
+            pipeline, construct_template_registry=registry
+        )
+
+        self.assertFalse(result.valid, msg="Expected invalid result")
+        missing_errors = [e for e in result.errors if "name" in e]
+        self.assertGreater(
+            len(missing_errors),
+            0,
+            msg=f"Expected error about missing 'name' param; errors={result.errors}",
+        )
+
+    def test_required_param_with_default_still_flagged_when_absent(self):
+        """
+        Correctness of the corrected check: required=True AND d.name not in
+        provided_params -> error, even when d.default is not None.
+        This verifies the 'and d.default is None' condition was NOT included,
+        matching render_construct_template's actual runtime behavior.
+        """
+        template = _make_template(
+            "tricky_template",
+            ParameterDescriptor(
+                name="subject",
+                type="literal",
+                required=True,
+                default="fallback_value",  # default exists but required=True
+            ),
+        )
+        registry = _make_registry(template)
+        step = _make_step(
+            "tricky_step",
+            "construct_template",
+            config={
+                "template_name": "tricky_template",
+                "params": {},  # subject not supplied
+            },
+        )
+        pipeline = _make_pipeline(step)
+        validator = PipelineValidator()
+
+        result = validator.validate_pipeline(
+            pipeline, construct_template_registry=registry
+        )
+
+        # Must be an error even though default="fallback_value" exists,
+        # because render_construct_template raises on required=True + no caller value.
+        self.assertFalse(result.valid, msg=(
+            "Expected invalid: required=True param with a default should still be "
+            "flagged if absent from step.config['params']"
+        ))
+        self.assertTrue(
+            any("subject" in e for e in result.errors),
+            msg=f"Expected error mentioning 'subject'; errors={result.errors}",
+        )
+
+    # ------------------------------------------------------------------
+    # Test 4: all valid -> no errors, no construct-related warnings
+    # ------------------------------------------------------------------
+
+    def test_all_valid_yields_no_issues(self):
+        """
+        Registry provided, template found, all required params supplied ->
+        no errors, no construct_template-related warnings.
+        """
+        template = _make_template(
+            "person_to_foaf",
+            ParameterDescriptor(name="subject", type="uri", required=True),
+            ParameterDescriptor(name="name", type="literal", required=True),
+            ParameterDescriptor(name="lang", type="literal", required=False, default="en"),
+        )
+        registry = _make_registry(template)
+        step = _make_step(
+            "build_foaf",
+            "construct_template",
+            config={
+                "template_name": "person_to_foaf",
+                "params": {
+                    "subject": "http://example.org/alice",
+                    "name": "Alice",
+                    # "lang" intentionally omitted -- optional, should not trigger error
+                },
+            },
+        )
+        pipeline = _make_pipeline(step)
+        validator = PipelineValidator()
+
+        result = validator.validate_pipeline(
+            pipeline, construct_template_registry=registry
+        )
+
+        self.assertEqual(result.errors, [], msg=f"Expected no errors; got {result.errors}")
+        # No construct-specific warnings (handler/config warnings from generic check
+        # are fine -- the step has no handler and that's expected in this test fixture)
+        construct_warnings = [
+            w for w in result.warnings if "construct_template_registry" in w
+        ]
+        self.assertEqual(
+            construct_warnings, [],
+            msg=f"Expected no registry-related warnings; got {result.warnings}",
+        )
+
+    # ------------------------------------------------------------------
+    # Test 5: regression -- non-construct_template steps unaffected
+    # ------------------------------------------------------------------
+
+    def test_non_construct_steps_unchanged(self):
+        """
+        Regression guard: validate_pipeline's output must be identical for a
+        pipeline containing only non-construct_template steps, whether or not
+        construct_template_registry is passed.
+
+        Both calls (with and without the registry kwarg) must produce the same
+        valid/errors/warnings, confirming zero behavior change for existing
+        callers of any other step type.
+        """
+        steps = [
+            _make_step("ingest", "file_ingest", config={"path": "/data"}),
+            _make_step("parse", "document_parse", config={"format": "pdf"}),
+            _make_step("embed", "embedding", config={"model": "openai"}),
+        ]
+        pipeline = _make_pipeline(*steps)
+        validator = PipelineValidator()
+
+        # Call without registry (existing behavior)
+        result_without = validator.validate_pipeline(pipeline)
+        # Call with a registry (should not affect these steps at all)
+        dummy_registry = ConstructTemplateRegistry()
+        result_with = validator.validate_pipeline(
+            pipeline, construct_template_registry=dummy_registry
+        )
+
+        self.assertEqual(
+            result_without.valid,
+            result_with.valid,
+            msg="valid flag changed for non-construct_template pipeline",
+        )
+        self.assertEqual(
+            result_without.errors,
+            result_with.errors,
+            msg="errors changed for non-construct_template pipeline",
+        )
+        self.assertEqual(
+            result_without.warnings,
+            result_with.warnings,
+            msg="warnings changed for non-construct_template pipeline",
+        )
+
+    def test_mixed_pipeline_only_construct_step_gets_warning(self):
+        """
+        A pipeline with mixed step types: only the construct_template step should
+        receive the 'no registry' warning; other steps should be unaffected.
+        """
+        steps = [
+            _make_step("ingest", "file_ingest", config={"path": "/data"}),
+            _make_step(
+                "build_graph",
+                "construct_template",
+                config={"template_name": "some_template", "params": {}},
+            ),
+            _make_step("embed", "embedding", config={"model": "openai"}),
+        ]
+        pipeline = _make_pipeline(*steps)
+        validator = PipelineValidator()
+
+        result = validator.validate_pipeline(pipeline)  # no registry
+
+        self.assertTrue(result.valid, msg=f"Expected valid; errors={result.errors}")
+        # Exactly one construct-registry warning (for "build_graph")
+        registry_warnings = [
+            w for w in result.warnings if "construct_template_registry" in w
+        ]
+        self.assertEqual(
+            len(registry_warnings),
+            1,
+            msg=f"Expected exactly 1 registry warning; warnings={result.warnings}",
+        )
+        # The warning should mention the step name
+        self.assertIn("build_graph", registry_warnings[0])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/triplet_store/test_blazegraph_store.py

@@ -1,7 +1,7 @@
 import unittest
 import os
 import sys
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
 if PROJECT_ROOT not in sys.path:
@@ -9,6 +9,7 @@ if PROJECT_ROOT not in sys.path:
 
 from semantica.semantic_extract.triplet_extractor import Triplet
 from semantica.triplet_store.blazegraph_store import BlazegraphStore
+from semantica.utils.exceptions import ProcessingError
 
 
 class TestBlazegraphStoreSerialization(unittest.TestCase):
@@ -221,5 +222,334 @@ class TestBlazegraphStoreSerialization(unittest.TestCase):
         self.assertEqual(obj, "\"Colour\"@en-GB")
 
 
+from semantica.triplet_store import sparql_escaping
+
+
+class TestSparqlEscapingExtractionParity(unittest.TestCase):
+    """
+    Regression tests proving BlazegraphStore._escape_literal /
+    BlazegraphStore._resolve_datatype_iri are byte-for-byte identical in
+    behavior to the extracted sparql_escaping.escape_literal /
+    sparql_escaping.resolve_datatype_iri functions, across every branch of
+    both original methods.
+    """
+
+    # --- escape_literal: every special character + combinations + plain text ---
+
+    ESCAPE_LITERAL_CASES = [
+        "",  # empty string
+        "plain text, no special chars",
+        "back\\slash",  # backslash
+        'embedded "double" quote',  # double quote
+        "line1\nline2",  # newline
+        "line1\rline2",  # carriage return
+        "tab\there",  # tab
+        "\\\"\n\r\t",  # all five special characters combined
+        'mix \\ and " and \n and \r and \t together',
+        42,  # non-str input (both methods call str(value) first)
+        None,
+    ]
+
+    def test_escape_literal_matches_blazegraph_store_for_every_branch(self):
+        with patch.object(BlazegraphStore, "_connect", autospec=True):
+            store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph")
+        for value in self.ESCAPE_LITERAL_CASES:
+            with self.subTest(value=value):
+                self.assertEqual(
+                    store._escape_literal(value),
+                    sparql_escaping.escape_literal(value),
+                )
+
+    # --- resolve_datatype_iri: every branch (bracketed, full IRI, prefixed,
+    # invalid bracketed, invalid full IRI-with-whitespace, unknown prefix) ---
+
+    RESOLVE_DATATYPE_IRI_VALID_CASES = [
+        "<http://www.w3.org/2001/XMLSchema#integer>",  # already bracketed
+        "http://www.w3.org/2001/XMLSchema#integer",  # full IRI, no brackets
+        "https://example.org/type",  # https scheme
+        "urn:isbn:0451450523",  # urn scheme
+        "xsd:integer",  # known prefix
+        "rdf:langString",  # known prefix
+        "rdfs:label",  # known prefix
+        "owl:Thing",  # known prefix
+        "skos:Concept",  # known prefix
+    ]
+
+    RESOLVE_DATATYPE_IRI_ERROR_CASES = [
+        "<>",  # empty bracketed IRI
+        "<http://example.org/type with space>",  # bracketed IRI with whitespace
+        "<http://example.org/type<injected>",  # bracketed IRI with disallowed char
+        "http://example.org/type CLEAR ALL",  # full IRI with whitespace
+        "myns:customType",  # unknown prefix
+        "not_a_uri_no_colon",  # no scheme, no colon-prefixed form matches
+        "javascript:alert(1)",  # disallowed scheme, not http/https/urn, no known prefix match
+    ]
+
+    def test_resolve_datatype_iri_matches_blazegraph_store_for_valid_cases(self):
+        with patch.object(BlazegraphStore, "_connect", autospec=True):
+            store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph")
+        for datatype in self.RESOLVE_DATATYPE_IRI_VALID_CASES:
+            with self.subTest(datatype=datatype):
+                self.assertEqual(
+                    store._resolve_datatype_iri(datatype),
+                    sparql_escaping.resolve_datatype_iri(datatype),
+                )
+
+    def test_resolve_datatype_iri_matches_blazegraph_store_for_error_cases(self):
+        with patch.object(BlazegraphStore, "_connect", autospec=True):
+            store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph")
+        for datatype in self.RESOLVE_DATATYPE_IRI_ERROR_CASES:
+            with self.subTest(datatype=datatype):
+                store_exc = None
+                shared_exc = None
+                try:
+                    store._resolve_datatype_iri(datatype)
+                except ValueError as exc:
+                    store_exc = exc
+                try:
+                    sparql_escaping.resolve_datatype_iri(datatype)
+                except ValueError as exc:
+                    shared_exc = exc
+
+                self.assertIsNotNone(store_exc, f"Expected ValueError from store for {datatype!r}")
+                self.assertIsNotNone(shared_exc, f"Expected ValueError from shared module for {datatype!r}")
+                self.assertEqual(str(store_exc), str(shared_exc))
+
+
+def _make_connected_store() -> BlazegraphStore:
+    """Create a BlazegraphStore instance bypassing the real _connect() call,
+    with .connected forced True (mirrors the state execute_sparql requires)."""
+    with patch.object(BlazegraphStore, "_connect", autospec=True):
+        store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph")
+    store.connected = True
+    return store
+
+
+class TestBlazegraphStoreConstructExtension(unittest.TestCase):
+    """
+    Tests for the Blazegraph CONSTRUCT extension: _is_construct_query
+    detection, execute_sparql's CONSTRUCT branch (Accept header, Turtle
+    parsing via rdflib, triples shape, ProcessingError on malformed Turtle),
+    and Property 9 (non-CONSTRUCT queries are byte-for-byte unaffected).
+    """
+
+    # --- _is_construct_query detection ---
+
+    def test_is_construct_query_detects_uppercase_keyword(self):
+        store = _make_connected_store()
+        self.assertTrue(store._is_construct_query("CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }"))
+
+    def test_is_construct_query_detects_lowercase_keyword(self):
+        store = _make_connected_store()
+        self.assertTrue(store._is_construct_query("construct { ?s ?p ?o } where { ?s ?p ?o }"))
+
+    def test_is_construct_query_detects_mixed_case_keyword(self):
+        store = _make_connected_store()
+        self.assertTrue(store._is_construct_query("Construct { ?s ?p ?o } Where { ?s ?p ?o }"))
+
+    def test_is_construct_query_detects_complex_preambles(self):
+        # Permanent regression tests covering edge cases discovered during
+        # regex stress-testing (issue #7): multiline declarations, empty
+        # prefix namespaces, and inline comments embedded in the preamble.
+        store = _make_connected_store()
+        cases = {
+            "multiline_prefix": "PREFIX foaf:\n  <http://xmlns.com/foaf/0.1/>\nCONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }",
+            "empty_prefix_namespace": "PREFIX : <http://ex.org/> CONSTRUCT { ?s ?p ?o }",
+            "inline_comment": "PREFIX ex: <http://ex.org/>\n# inline comment\nCONSTRUCT { ?s ?p ?o }",
+        }
+        for name, query in cases.items():
+            with self.subTest(case=name):
+                self.assertTrue(store._is_construct_query(query))
+
+    def test_is_construct_query_false_for_select(self):
+        store = _make_connected_store()
+        self.assertFalse(store._is_construct_query("SELECT ?s WHERE { ?s ?p ?o }"))
+
+    def test_is_construct_query_false_for_ask(self):
+        store = _make_connected_store()
+        self.assertFalse(store._is_construct_query("ASK { ?s ?p ?o }"))
+
+    def test_is_construct_query_does_not_match_substring_inside_identifier(self):
+        # "CONSTRUCTOR" contains "CONSTRUCT" as a substring but must not
+        # match due to \b word-boundary anchoring.
+        store = _make_connected_store()
+        self.assertFalse(
+            store._is_construct_query("SELECT ?s WHERE { ?s <urn:p> \"CONSTRUCTOR\" }")
+        )
+
+    # --- execute_sparql CONSTRUCT path ---
+
+    def test_execute_sparql_construct_sends_turtle_accept_header(self):
+        store = _make_connected_store()
+        mock_response = MagicMock()
+        mock_response.content = (
+            b"@prefix ex: <http://ex.org/> .\n"
+            b'ex:s1 ex:p1 "value1" .\n'
+        )
+        mock_response.raise_for_status = MagicMock()
+
+        with patch("semantica.triplet_store.blazegraph_store.requests.post", return_value=mock_response) as mock_post:
+            store.execute_sparql("CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }")
+
+        _, kwargs = mock_post.call_args
+        self.assertEqual(kwargs["headers"]["Accept"], "text/turtle")
+        self.assertEqual(
+            kwargs["headers"]["Content-Type"], "application/x-www-form-urlencoded"
+        )
+
+    def test_execute_sparql_construct_parses_triples_from_fixed_turtle_fixture(self):
+        store = _make_connected_store()
+        turtle_fixture = (
+            b"@prefix ex: <http://ex.org/> .\n"
+            b'ex:s1 ex:p1 "value1" .\n'
+            b"ex:s1 ex:p2 ex:o2 .\n"
+        )
+        mock_response = MagicMock()
+        mock_response.content = turtle_fixture
+        mock_response.raise_for_status = MagicMock()
+
+        with patch("semantica.triplet_store.blazegraph_store.requests.post", return_value=mock_response):
+            result = store.execute_sparql("CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }")
+
+        self.assertTrue(result["success"])
+        self.assertEqual(result["bindings"], [])
+        self.assertEqual(result["variables"], [])
+        self.assertEqual(result["metadata"]["result_format"], "construct")
+
+        # "triples" is now a list of (s, p, o, metadata) 4-tuples. Both
+        # triples here are plain untyped literals/URIs, so metadata is {}.
+        triples = {(s, p, o) for s, p, o, _metadata in result["triples"]}
+        self.assertIn(("http://ex.org/s1", "http://ex.org/p1", "value1"), triples)
+        self.assertIn(("http://ex.org/s1", "http://ex.org/p2", "http://ex.org/o2"), triples)
+        self.assertEqual(len(result["triples"]), 2)
+        for _s, _p, _o, metadata in result["triples"]:
+            self.assertEqual(metadata, {})
+
+    def test_execute_sparql_construct_result_format_option_forces_construct_path(self):
+        # Even for a query that doesn't literally contain "CONSTRUCT",
+        # result_format="construct" should force the Turtle-parsing path.
+        store = _make_connected_store()
+        mock_response = MagicMock()
+        mock_response.content = b'@prefix ex: <http://ex.org/> .\nex:s1 ex:p1 "v" .\n'
+        mock_response.raise_for_status = MagicMock()
+
+        with patch("semantica.triplet_store.blazegraph_store.requests.post", return_value=mock_response) as mock_post:
+            result = store.execute_sparql("SELECT ?s WHERE { ?s ?p ?o }", result_format="construct")
+
+        _, kwargs = mock_post.call_args
+        self.assertEqual(kwargs["headers"]["Accept"], "text/turtle")
+        self.assertIn("triples", result)
+
+    def test_execute_sparql_construct_malformed_turtle_raises_processing_error(self):
+        store = _make_connected_store()
+        mock_response = MagicMock()
+        # Deliberately invalid Turtle syntax.
+        mock_response.content = b"this is { not [ valid turtle syntax at all !!!"
+        mock_response.raise_for_status = MagicMock()
+
+        with patch("semantica.triplet_store.blazegraph_store.requests.post", return_value=mock_response):
+            with self.assertRaises(ProcessingError) as ctx:
+                store.execute_sparql("CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }")
+
+        # Must be a ProcessingError, not a raw rdflib exception leaking out.
+        self.assertIsInstance(ctx.exception, ProcessingError)
+        self.assertNotIsInstance(ctx.exception, (SyntaxError, ValueError))
+
+    def test_execute_sparql_construct_handles_literal_with_braces_in_valid_turtle(self):
+        # Adversarial case implied by the brace-matching bug found in the
+        # template-string layer (construct_templates._find_matching_brace):
+        # confirm rdflib itself parses a *valid* Turtle literal containing
+        # brace characters correctly, since this is a different parsing
+        # layer (real Turtle syntax, not our {{param}} template string).
+        store = _make_connected_store()
+        turtle_fixture = (
+            b"@prefix ex: <http://ex.org/> .\n"
+            b'ex:s1 ex:p1 "text with { and } braces inside" .\n'
+        )
+        mock_response = MagicMock()
+        mock_response.content = turtle_fixture
+        mock_response.raise_for_status = MagicMock()
+
+        with patch("semantica.triplet_store.blazegraph_store.requests.post", return_value=mock_response):
+            result = store.execute_sparql("CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }")
+
+        self.assertEqual(len(result["triples"]), 1)
+        subject, predicate, obj, metadata = result["triples"][0]
+        self.assertEqual(subject, "http://ex.org/s1")
+        self.assertEqual(predicate, "http://ex.org/p1")
+        self.assertEqual(obj, "text with { and } braces inside")
+        self.assertEqual(metadata, {})
+
+    # --- Property 9: non-CONSTRUCT queries are byte-for-byte unaffected ---
+
+    def test_execute_sparql_select_query_response_shape_unchanged(self):
+        store = _make_connected_store()
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "head": {"vars": ["s", "p", "o"]},
+            "results": {
+                "bindings": [
+                    {
+                        "s": {"type": "uri", "value": "http://ex.org/s1"},
+                        "p": {"type": "uri", "value": "http://ex.org/p1"},
+                        "o": {"type": "literal", "value": "v1"},
+                    }
+                ]
+            },
+        }
+        mock_response.raise_for_status = MagicMock()
+
+        with patch("semantica.triplet_store.blazegraph_store.requests.post", return_value=mock_response) as mock_post:
+            result = store.execute_sparql("SELECT ?s ?p ?o WHERE { ?s ?p ?o }")
+
+        # Accept header must NOT be sent for a non-CONSTRUCT query — request
+        # shape is byte-for-byte identical to pre-CONSTRUCT-extension behavior.
+        _, kwargs = mock_post.call_args
+        self.assertEqual(kwargs["headers"], {"Content-Type": "application/x-www-form-urlencoded"})
+        self.assertNotIn("Accept", kwargs["headers"])
+
+        # Response shape must be exactly the pre-existing shape: no "triples"
+        # key at all (not even an empty list) for a plain SELECT response.
+        self.assertEqual(
+            result,
+            {
+                "success": True,
+                "bindings": mock_response.json.return_value["results"]["bindings"],
+                "variables": ["s", "p", "o"],
+                "metadata": {
+                    "query": "SELECT ?s ?p ?o WHERE { ?s ?p ?o }",
+                    "endpoint": store._get_sparql_endpoint(),
+                },
+            },
+        )
+        self.assertNotIn("triples", result)
+
+    def test_execute_sparql_ask_query_uses_bindings_path_not_construct(self):
+        store = _make_connected_store()
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"head": {}, "boolean": True}
+        mock_response.raise_for_status = MagicMock()
+
+        with patch("semantica.triplet_store.blazegraph_store.requests.post", return_value=mock_response) as mock_post:
+            store.execute_sparql("ASK { ?s ?p ?o }")
+
+        _, kwargs = mock_post.call_args
+        self.assertNotIn("Accept", kwargs["headers"])
+
+
+class TestQueryEngineConstructValidation(unittest.TestCase):
+    """
+    Confirms QueryEngine._validate_query requires zero changes for CONSTRUCT
+    support — CONSTRUCT was already a valid keyword before this feature.
+    """
+
+    def test_construct_query_passes_validation_unchanged(self):
+        from semantica.triplet_store.query_engine import QueryEngine
+
+        engine = QueryEngine()
+        query = "CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }"
+        self.assertTrue(engine._validate_query(query))
+
+
 if __name__ == "__main__":
     unittest.main()

tests/triplet_store/test_query_engine.py

@@ -0,0 +1,97 @@
+"""
+Tests for semantica.triplet_store.query_engine, focused on the
+QueryResult.triples field added for CONSTRUCT query support.
+"""
+
+import os
+import sys
+import unittest
+from unittest.mock import MagicMock
+
+PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+if PROJECT_ROOT not in sys.path:
+    sys.path.insert(0, PROJECT_ROOT)
+
+from semantica.triplet_store.query_engine import QueryEngine, QueryResult
+
+
+class TestQueryResultTriplesField(unittest.TestCase):
+    def test_query_result_triples_defaults_to_empty_list(self):
+        result = QueryResult(bindings=[], variables=[])
+        self.assertEqual(result.triples, [])
+
+    def test_query_result_triples_field_is_independent_per_instance(self):
+        # default_factory=list must produce a fresh list per instance, not a
+        # shared mutable default.
+        r1 = QueryResult(bindings=[], variables=[])
+        r2 = QueryResult(bindings=[], variables=[])
+        r1.triples.append(("s", "p", "o"))
+        self.assertEqual(r1.triples, [("s", "p", "o")])
+        self.assertEqual(r2.triples, [])
+
+    def test_query_result_accepts_explicit_triples(self):
+        triples = [("http://ex.org/s1", "http://ex.org/p1", "v1")]
+        result = QueryResult(bindings=[], variables=[], triples=triples)
+        self.assertEqual(result.triples, triples)
+
+
+class FakeConstructBackend:
+    """Fake store_backend whose execute_sparql returns a CONSTRUCT-shaped result."""
+
+    supports_named_graphs = True
+
+    def execute_sparql(self, query, **options):
+        return {
+            "success": True,
+            "bindings": [],
+            "variables": [],
+            "triples": [
+                ("http://ex.org/s1", "http://ex.org/p1", "v1"),
+                ("http://ex.org/s2", "http://ex.org/p2", "v2"),
+            ],
+            "metadata": {"query": query, "result_format": "construct"},
+        }
+
+
+class FakeBindingsBackend:
+    """Fake store_backend whose execute_sparql returns a plain bindings result
+    with no "triples" key at all, matching pre-CONSTRUCT-extension backends."""
+
+    supports_named_graphs = True
+
+    def execute_sparql(self, query, **options):
+        return {
+            "success": True,
+            "bindings": [{"s": {"value": "http://ex.org/s1"}}],
+            "variables": ["s"],
+            "metadata": {"query": query},
+        }
+
+
+class TestExecuteQueryPopulatesTriples(unittest.TestCase):
+    def test_execute_query_populates_triples_from_construct_backend(self):
+        engine = QueryEngine(enable_caching=False, enable_optimization=False)
+        result = engine.execute_query(
+            "CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }", FakeConstructBackend()
+        )
+        self.assertEqual(
+            result.triples,
+            [
+                ("http://ex.org/s1", "http://ex.org/p1", "v1"),
+                ("http://ex.org/s2", "http://ex.org/p2", "v2"),
+            ],
+        )
+
+    def test_execute_query_defaults_triples_to_empty_list_when_backend_omits_key(self):
+        engine = QueryEngine(enable_caching=False, enable_optimization=False)
+        result = engine.execute_query(
+            "SELECT ?s WHERE { ?s ?p ?o }", FakeBindingsBackend()
+        )
+        self.assertEqual(result.triples, [])
+        # Non-CONSTRUCT behavior otherwise unaffected.
+        self.assertEqual(result.bindings, [{"s": {"value": "http://ex.org/s1"}}])
+        self.assertEqual(result.variables, ["s"])
+
+
+if __name__ == "__main__":
+    unittest.main()