Merge pull request #410 from Hawksight-AI/semantic-extract

feat(semantic-extract): temporal metadata extraction from text (#400)

CHANGELOG.md

@@ -7,6 +7,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+- **Temporal Metadata Extraction from Text** (PR #400 by @KaifAhmad1):
+  - Added `extract_temporal_bounds: bool = False` parameter to `extract_relations_llm()`. When `True`, the LLM prompt is extended with a calibrated confidence scale and four few-shot examples; each returned `Relation` gains `valid_from`, `valid_until`, `temporal_confidence` (0.0–1.0), and `temporal_source_text` in its `metadata` dict. Default `False` preserves 100% backward compatibility.
+  - Confidence scale anchors baked into the prompt: `1.00` = full ISO date, `0.90` = year+month, `0.85` = year only, `0.75` = quarter, `0.65` = named season/approximate range, `0.50` = vague relative with computable anchor, `0.35` = highly vague, `0.00` = no temporal signal. LLMs self-report certainty rather than clustering near 1.0.
+  - Low temporal confidence (< 0.5) with a non-null date logs a `WARNING`; signal is never suppressed — callers decide how to filter.
+  - Cache key now includes the `extract_temporal_bounds` flag to prevent cross-mode cache pollution.
+  - Flag propagated through `_extract_relations_chunked()` so long-text chunked extraction also carries temporal metadata.
+  - Added `RelationWithTemporalOut` and `RelationsWithTemporalResponse` Pydantic schemas in `semantica/semantic_extract/schemas.py`. A separate schema is required because `RelationOut` uses `extra="ignore"`, which silently drops any undeclared field including the four temporal fields.
+  - New `semantica/kg/temporal_normalizer.py` — `TemporalNormalizer` class (zero LLM calls, pure regex + `dateutil` arithmetic):
+    - `normalize(value)` → `(valid_from, valid_until)` UTC `datetime` tuple or `None`. Resolution order: ISO 8601 full parse → partial-date regex (year-only, month+year, YYYY-MM, Q[1-4] YYYY) → ambiguous-slash-date detection → domain phrase map → relative phrase resolution via `relativedelta`.
+    - `normalize_phrase(phrase)` → metadata dict `{"maps_to": ..., "type": ..., "domain": [...]}` or `None` — exact match then regex-pattern keys.
+    - Ambiguous `DD/MM/YYYY`-style inputs issue `TemporalAmbiguityWarning` and return `None` — never silently guesses locale.
+    - Unparseable inputs return `None` with a debug log — never raise.
+    - Relative phrases (`"last year"`, `"three months ago"`, etc.) raise `ValueError` if `reference_date` is `None` rather than guessing.
+    - Default phrase map covers 13 domains: General/Policy (`effective date`, `effective from/as of/beginning`, `in force until`, `retroactive to`, `sunset clause`), Healthcare (`approval date`, `expiry date`, `market authorization`), Cybersecurity (`incident window`, `campaign period`), Supply Chain (`certification valid through`), Finance (`trading halt`), Energy (`commissioned date`, `decommissioned date`).
+    - User-supplied `phrase_map` is merged over defaults at construction (`{**defaults, **user_map}`) — custom entries win without forking the library.
+  - Added `TemporalAmbiguityWarning(UserWarning)` to `semantica/utils/exceptions.py`.
+  - Exported `TemporalNormalizer` from `semantica/kg/__init__.py`.
+  - Added 53 new tests in `tests/semantic_extract/test_temporal_extraction.py`; zero real LLM calls, suite runs in ~3.5 s. All 873 existing tests continue to pass.
+
 - **Fix: OllamaProvider ignores `base_url`** (PR #408 by @AlexeyMyslin, fixed by @KaifAhmad1):
   - `OllamaProvider._init_client()` was assigning the raw `ollama` module to `self.client` instead of instantiating `ollama.Client(host=self.base_url)`, causing all requests to silently hit `localhost:11434` regardless of the `base_url` passed by the user
   - Fixed by replacing `self.client = ollama` with `self.client = ollama.Client(host=self.base_url)` — remote Ollama servers (e.g. `http://192.168.1.3:11434`) are now reachable

semantica/kg/__init__.py

@@ -127,6 +127,7 @@ from .temporal_query import (
     TemporalVersionManager,
 )
 from .temporal_model import BiTemporalFact, TemporalBound
+from .temporal_normalizer import TemporalNormalizer
 
 __all__ = [
     # Core Classes
@@ -140,6 +141,7 @@ __all__ = [
     "TemporalVersionManager",
     "TemporalBound",
     "BiTemporalFact",
+    "TemporalNormalizer",
     "AlgorithmTrackerWithProvenance",
     "ProvenanceTracker",
     # Enhanced Graph Algorithms

semantica/kg/temporal_normalizer.py

@@ -0,0 +1,391 @@
+"""
+Temporal Normalizer
+
+Deterministic resolution of temporal phrases extracted from text into
+UTC datetime intervals. Zero LLM calls — pure regex and date arithmetic.
+
+Usage::
+
+    from semantica.kg import TemporalNormalizer
+    from datetime import datetime, timezone
+
+    tn = TemporalNormalizer(reference_date=datetime(2025, 3, 25, tzinfo=timezone.utc))
+    start, end = tn.normalize("Q2 2021")   # → (2021-04-01, 2021-06-30)
+    start, end = tn.normalize("last year") # → (2024-01-01, 2024-12-31)
+    info = tn.normalize_phrase("expiry date")  # → {"maps_to": "valid_until", ...}
+"""
+
+from __future__ import annotations
+
+import calendar
+import logging
+import re
+import warnings
+from datetime import datetime, timezone
+from typing import Any, Callable, Dict, Optional, Tuple
+
+from dateutil.relativedelta import relativedelta
+
+from ..utils.exceptions import TemporalAmbiguityWarning
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Compiled regex patterns for structured date formats
+# ---------------------------------------------------------------------------
+
+_RE_YEAR_ONLY = re.compile(r"^\s*(\d{4})\s*$")
+_RE_MONTH_YEAR_WORD = re.compile(
+    r"^\s*(january|february|march|april|may|june|july|august|september|october|november|december|"
+    r"jan|feb|mar|apr|jun|jul|aug|sep|oct|nov|dec)\s+(\d{4})\s*$",
+    re.IGNORECASE,
+)
+_RE_YEAR_MONTH_ISO = re.compile(r"^\s*(\d{4})-(\d{1,2})\s*$")
+_RE_QUARTER = re.compile(r"^\s*Q([1-4])\s+(\d{4})\s*$", re.IGNORECASE)
+_RE_AMBIGUOUS_SLASH = re.compile(r"^\s*\d{1,2}/\d{1,2}/\d{4}\s*$")
+
+_MONTH_NAMES: Dict[str, int] = {
+    "january": 1, "jan": 1,
+    "february": 2, "feb": 2,
+    "march": 3, "mar": 3,
+    "april": 4, "apr": 4,
+    "may": 5,
+    "june": 6, "jun": 6,
+    "july": 7, "jul": 7,
+    "august": 8, "aug": 8,
+    "september": 9, "sep": 9,
+    "october": 10, "oct": 10,
+    "november": 11, "nov": 11,
+    "december": 12, "dec": 12,
+}
+
+_QUARTER_BOUNDS: Dict[int, Tuple[int, int, int, int]] = {
+    # quarter → (from_month, from_day, until_month, until_day)
+    1: (1, 1, 3, 31),
+    2: (4, 1, 6, 30),
+    3: (7, 1, 9, 30),
+    4: (10, 1, 12, 31),
+}
+
+
+# ---------------------------------------------------------------------------
+# Small date-arithmetic helpers
+# ---------------------------------------------------------------------------
+
+def _utc(year: int, month: int, day: int) -> datetime:
+    return datetime(year, month, day, tzinfo=timezone.utc)
+
+
+def _last_day_of_month(year: int, month: int) -> int:
+    return calendar.monthrange(year, month)[1]
+
+
+def _this_quarter(ref: datetime) -> Tuple[datetime, datetime]:
+    q = (ref.month - 1) // 3 + 1
+    fm, fd, um, ud = _QUARTER_BOUNDS[q]
+    return _utc(ref.year, fm, fd), _utc(ref.year, um, ud)
+
+
+def _last_quarter(ref: datetime) -> Tuple[datetime, datetime]:
+    q = (ref.month - 1) // 3 + 1
+    prev_q = q - 1 if q > 1 else 4
+    year = ref.year if q > 1 else ref.year - 1
+    fm, fd, um, ud = _QUARTER_BOUNDS[prev_q]
+    return _utc(year, fm, fd), _utc(year, um, ud)
+
+
+def _last_month(ref: datetime) -> Tuple[datetime, datetime]:
+    first = ref.replace(day=1) - relativedelta(months=1)
+    last_day = _last_day_of_month(first.year, first.month)
+    return _utc(first.year, first.month, 1), _utc(first.year, first.month, last_day)
+
+
+# ---------------------------------------------------------------------------
+# Default phrase map
+# ---------------------------------------------------------------------------
+# Keys: lowercase canonical phrases (or regex patterns prefixed with "r:").
+# Values: callables (ref: datetime) → (valid_from, valid_until).
+#
+# Domain-specific terms that carry no self-contained date (e.g. "approval date")
+# return (ref, ref) as a placeholder so callers can distinguish
+# "known temporal term, date needs context" from "unrecognised phrase".
+# ---------------------------------------------------------------------------
+
+def _phrase_entry(maps_to: str, type_: str, **extra: Any) -> Dict[str, Any]:
+    return {"maps_to": maps_to, "type": type_, **extra}
+
+
+# Phrase map entries also carry metadata for normalize_phrase()
+_DEFAULT_PHRASE_META: Dict[str, Dict[str, Any]] = {
+    # ── Relative references ─────────────────────────────────────────────
+    "last year":        _phrase_entry("valid_from", "relative"),
+    "this year":        _phrase_entry("valid_from", "relative"),
+    "last quarter":     _phrase_entry("valid_from", "relative"),
+    "this quarter":     _phrase_entry("valid_from", "relative"),
+    "last month":       _phrase_entry("valid_from", "relative"),
+    "this month":       _phrase_entry("valid_from", "relative"),
+    "three months ago": _phrase_entry("valid_from", "relative"),
+    "six months ago":   _phrase_entry("valid_from", "relative"),
+    "two years ago":    _phrase_entry("valid_from", "relative"),
+    # ── General / Policy ────────────────────────────────────────────────
+    "r:effective\\s+(as\\s+of|from|beginning|date)":
+        _phrase_entry("valid_from", "start", domain=["General", "Policy"]),
+    "in force until":
+        _phrase_entry("valid_until", "end", domain=["Policy", "Regulatory"]),
+    "retroactive to":
+        _phrase_entry("valid_from", "start", retroactive=True, domain=["Regulatory", "Finance"]),
+    "sunset clause":
+        _phrase_entry("valid_until", "sunset", domain=["Policy"]),
+    # ── Healthcare / Drug Discovery ──────────────────────────────────────
+    "approval date":
+        _phrase_entry("valid_from", "start", domain=["Healthcare", "Drug Discovery"]),
+    "expiry date":
+        _phrase_entry("valid_until", "end", domain=["Healthcare", "Supply Chain"]),
+    "market authorization":
+        _phrase_entry("valid_from", "start", domain=["Drug Discovery", "Healthcare"]),
+    # ── Cybersecurity ────────────────────────────────────────────────────
+    "incident window":
+        _phrase_entry("window", "window", domain=["Cybersecurity"]),
+    "campaign period":
+        _phrase_entry("window", "window", domain=["Cybersecurity"]),
+    # ── Supply Chain ─────────────────────────────────────────────────────
+    "certification valid through":
+        _phrase_entry("valid_until", "end", domain=["Supply Chain"]),
+    # ── Finance ──────────────────────────────────────────────────────────
+    "trading halt":
+        _phrase_entry("window", "window", domain=["Finance"]),
+    # ── Energy ───────────────────────────────────────────────────────────
+    "commissioned date":
+        _phrase_entry("valid_from", "start", domain=["Energy"]),
+    "decommissioned date":
+        _phrase_entry("valid_until", "end", domain=["Energy"]),
+}
+
+# Separate callable map for date resolution (subset of the above)
+def _build_default_callable_map() -> Dict[str, Callable[[datetime], Tuple[datetime, datetime]]]:
+    return {
+        "last year": lambda ref: (
+            _utc(ref.year - 1, 1, 1),
+            _utc(ref.year - 1, 12, 31),
+        ),
+        "this year": lambda ref: (
+            _utc(ref.year, 1, 1),
+            _utc(ref.year, 12, 31),
+        ),
+        "last quarter": _last_quarter,
+        "this quarter": _this_quarter,
+        "last month": _last_month,
+        "this month": lambda ref: (
+            _utc(ref.year, ref.month, 1),
+            _utc(ref.year, ref.month, _last_day_of_month(ref.year, ref.month)),
+        ),
+        "three months ago": lambda ref: (
+            (ref - relativedelta(months=3)).replace(day=1, hour=0, minute=0, second=0, microsecond=0),
+            ((ref - relativedelta(months=3)).replace(day=1) + relativedelta(months=1) - relativedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0),
+        ),
+        "six months ago": lambda ref: (
+            (ref - relativedelta(months=6)).replace(day=1, hour=0, minute=0, second=0, microsecond=0),
+            ((ref - relativedelta(months=6)).replace(day=1) + relativedelta(months=1) - relativedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0),
+        ),
+        "two years ago": lambda ref: (
+            _utc(ref.year - 2, 1, 1),
+            _utc(ref.year - 2, 12, 31),
+        ),
+    }
+
+
+# ---------------------------------------------------------------------------
+# TemporalNormalizer
+# ---------------------------------------------------------------------------
+
+class TemporalNormalizer:
+    """
+    Deterministic resolution of temporal phrases into UTC datetime intervals.
+
+    Zero LLM calls. All resolution is done via regex patterns and Python
+    date arithmetic (``dateutil.relativedelta``).
+
+    Args:
+        reference_date: Anchor for relative phrases like "last year". When
+            ``None`` and a relative phrase is encountered, :meth:`normalize`
+            raises :class:`ValueError`.
+        phrase_map: Optional dict that extends or overrides the default
+            domain phrase map. Keys are lowercase phrases (or regex patterns
+            prefixed with ``"r:"``). Values are callables
+            ``(reference_date: datetime) -> (start: datetime, end: datetime)``.
+    """
+
+    def __init__(
+        self,
+        reference_date: Optional[datetime] = None,
+        phrase_map: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        self.reference_date = reference_date
+        # Build the callable resolution map (relative dates + user overrides)
+        self._callable_map: Dict[str, Callable[[datetime], Tuple[datetime, datetime]]] = (
+            _build_default_callable_map()
+        )
+        if phrase_map:
+            self._callable_map.update(phrase_map)
+        # Phrase metadata map (for normalize_phrase)
+        self._phrase_meta: Dict[str, Dict[str, Any]] = {**_DEFAULT_PHRASE_META}
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    def normalize(self, value: Optional[str]) -> Optional[Tuple[datetime, datetime]]:
+        """
+        Resolve a temporal string to a ``(valid_from, valid_until)`` interval.
+
+        Resolution order:
+        1. ``None`` / empty → ``None``
+        2. ISO 8601 full datetime / date → point interval ``(dt, dt)``
+        3. Partial date patterns: year-only, month+year, quarter+year
+        4. Ambiguous slash-date (``DD/MM/YYYY`` vs ``MM/DD/YYYY``) →
+           issues :class:`~semantica.utils.exceptions.TemporalAmbiguityWarning`
+           and returns ``None``
+        5. Phrase map / domain phrase lookup
+        6. Relative phrase via callable map (requires ``reference_date``)
+        7. Unparseable → ``None`` (debug log, never raises)
+
+        Returns:
+            Tuple of UTC datetimes ``(start, end)`` or ``None``.
+        """
+        if value is None:
+            return None
+        value_stripped = value.strip()
+        if not value_stripped:
+            return None
+
+        # 1. ISO 8601 parse
+        iso_result = self._try_iso(value_stripped)
+        if iso_result is not None:
+            return iso_result
+
+        # 2. Partial date patterns
+        partial_result = self._try_partial_date(value_stripped)
+        if partial_result is not None:
+            return partial_result
+
+        # 3. Ambiguous slash date — warn, return None
+        if _RE_AMBIGUOUS_SLASH.match(value_stripped):
+            warnings.warn(
+                f"Temporal expression {value_stripped!r} is ambiguous (day/month ordering unknown). "
+                "Provide locale or use ISO 8601 format (YYYY-MM-DD).",
+                TemporalAmbiguityWarning,
+                stacklevel=2,
+            )
+            return None
+
+        # 4. Relative phrase / callable map
+        callable_result = self._try_callable(value_stripped)
+        if callable_result is not None:
+            return callable_result
+
+        logger.debug("Could not parse temporal value: %r", value_stripped)
+        return None
+
+    def normalize_phrase(self, phrase: str) -> Optional[Dict[str, Any]]:
+        """
+        Look up a temporal phrase in the domain phrase map.
+
+        Checks exact match first, then regex patterns (keys prefixed with
+        ``"r:"``). Returns the metadata dict if matched, ``None`` otherwise.
+
+        Args:
+            phrase: Lowercase phrase to look up (case-insensitive internally).
+
+        Returns:
+            Dict with at minimum ``{"maps_to": ..., "type": ...}`` or ``None``.
+        """
+        normalized = phrase.strip().lower()
+
+        # Exact match
+        if normalized in self._phrase_meta:
+            return self._phrase_meta[normalized]
+
+        # Regex pattern match (keys prefixed with "r:")
+        for key, meta in self._phrase_meta.items():
+            if key.startswith("r:"):
+                pattern = key[2:]
+                if re.search(pattern, normalized, re.IGNORECASE):
+                    return meta
+
+        logger.debug("Unrecognized temporal phrase: %r", phrase)
+        return None
+
+    # ------------------------------------------------------------------
+    # Private helpers
+    # ------------------------------------------------------------------
+
+    def _try_iso(self, value: str) -> Optional[Tuple[datetime, datetime]]:
+        """Attempt ISO 8601 parse. Returns point interval on success."""
+        normalized = value
+        if normalized.endswith("Z"):
+            normalized = normalized[:-1] + "+00:00"
+        try:
+            dt = datetime.fromisoformat(normalized)
+            if dt.tzinfo is None:
+                dt = dt.replace(tzinfo=timezone.utc)
+            return (dt, dt)
+        except ValueError:
+            return None
+
+    def _try_partial_date(self, value: str) -> Optional[Tuple[datetime, datetime]]:
+        """Try partial date patterns: YYYY, Month YYYY, YYYY-MM, Q[1-4] YYYY."""
+        # Year only
+        m = _RE_YEAR_ONLY.match(value)
+        if m:
+            year = int(m.group(1))
+            return _utc(year, 1, 1), _utc(year, 12, 31)
+
+        # Month YYYY (word)
+        m = _RE_MONTH_YEAR_WORD.match(value)
+        if m:
+            month = _MONTH_NAMES[m.group(1).lower()]
+            year = int(m.group(2))
+            last = _last_day_of_month(year, month)
+            return _utc(year, month, 1), _utc(year, month, last)
+
+        # YYYY-MM (ISO partial)
+        m = _RE_YEAR_MONTH_ISO.match(value)
+        if m:
+            year, month = int(m.group(1)), int(m.group(2))
+            if 1 <= month <= 12:
+                last = _last_day_of_month(year, month)
+                return _utc(year, month, 1), _utc(year, month, last)
+
+        # Q[1-4] YYYY
+        m = _RE_QUARTER.match(value)
+        if m:
+            q, year = int(m.group(1)), int(m.group(2))
+            fm, fd, um, ud = _QUARTER_BOUNDS[q]
+            return _utc(year, fm, fd), _utc(year, um, ud)
+
+        return None
+
+    def _try_callable(self, value: str) -> Optional[Tuple[datetime, datetime]]:
+        """Try the relative phrase callable map."""
+        key = value.lower()
+
+        # Exact match
+        if key in self._callable_map:
+            if self.reference_date is None:
+                raise ValueError(
+                    f"reference_date is required to resolve relative temporal expression: {value!r}"
+                )
+            return self._callable_map[key](self.reference_date)
+
+        # Regex pattern match (keys prefixed with "r:")
+        for map_key, fn in self._callable_map.items():
+            if map_key.startswith("r:"):
+                pattern = map_key[2:]
+                if re.search(pattern, key, re.IGNORECASE):
+                    if self.reference_date is None:
+                        raise ValueError(
+                            f"reference_date is required to resolve relative temporal expression: {value!r}"
+                        )
+                    return fn(self.reference_date)
+
+        return None

semantica/semantic_extract/methods.py

@@ -122,7 +122,12 @@ from .cache import ExtractionCache
 from .config import config
 
 try:
-    from .schemas import EntitiesResponse, RelationsResponse, TripletsResponse
+    from .schemas import (
+        EntitiesResponse,
+        RelationsResponse,
+        RelationsWithTemporalResponse,
+        TripletsResponse,
+    )
     SCHEMAS_AVAILABLE = True
 except ImportError:
     SCHEMAS_AVAILABLE = False
@@ -1664,11 +1669,12 @@ def extract_relations_llm(
     max_text_length: Optional[int] = None,
     structured_output_mode: str = "typed",
     max_retries: int = 3,
+    extract_temporal_bounds: bool = False,
     **kwargs,
 ) -> List[Relation]:
     """
     LLM-based relation extraction.
-    
+
     Args:
         text: Input text
         entities: Pre-extracted entities
@@ -1677,6 +1683,10 @@ def extract_relations_llm(
         silent_fail: If True, return empty list on error. If False (default), raise exception.
         max_text_length: Maximum text length before auto-chunking. None = provider default.
         max_retries: Maximum number of retries for LLM calls (default: 3)
+        extract_temporal_bounds: If True, extend the prompt to extract temporal validity
+            per relation. Each relation's metadata gains: valid_from, valid_until,
+            temporal_confidence (0.0–1.0), and temporal_source_text. Low confidence (<0.5)
+            produces a warning log but is not suppressed. Default False.
         **kwargs: Additional options
     """
     # Support llm_model parameter to disambiguate from ML model
@@ -1691,6 +1701,7 @@ def extract_relations_llm(
         "structured_output_mode": structured_output_mode,
         "max_retries": max_retries,
         "relation_types": kwargs.get("relation_types"),
+        "extract_temporal_bounds": extract_temporal_bounds,
         # Include entities hash/str in cache key implicitly via **cache_params
         "entities_hash": hash(tuple(sorted([e.text for e in entities]))) if entities else 0
     }
@@ -1759,9 +1770,10 @@ def extract_relations_llm(
     if len(text) > max_text_length:
         logger.info(f"Text length ({len(text)}) exceeds limit for relations. Chunking...")
         return _extract_relations_chunked(
-            text, entities, provider=provider, model=model, 
-            silent_fail=silent_fail, max_text_length=max_text_length, 
+            text, entities, provider=provider, model=model,
+            silent_fail=silent_fail, max_text_length=max_text_length,
             max_retries=max_retries,
+            extract_temporal_bounds=extract_temporal_bounds,
             **kwargs
         )
 
@@ -1777,7 +1789,7 @@ def extract_relations_llm(
         )
 
     entities_str = ", ".join([f"{e.text} ({e.label})" for e in prompt_entities])
-    
+
     # Use custom relation types if provided
     relation_types = kwargs.get("relation_types")
     if relation_types:
@@ -1790,16 +1802,18 @@ If a relation doesn't fit any of the preferred types, use the most appropriate t
         relation_types_instruction = """
 Extract meaningful relationships between entities. Use appropriate relation types that accurately describe how entities are connected.
 Common relation types include: related_to, part_of, located_in, created_by, uses, depends_on, interacts_with, and similar variations."""
-    
+
     verbose_mode = kwargs.get("verbose", False)
     if verbose_mode:
         import sys
         print(f"    [methods.extract_relations_llm] Constructing prompt for {len(prompt_entities)} entities...", flush=True, file=sys.stdout)
-    
+
     if not SCHEMAS_AVAILABLE:
         raise ImportError("Pydantic schemas not available. Install pydantic/instructor to use LLM extraction.")
 
-    prompt = f"""Extract relations between entities from the provided text.
+    # ── Base prompt (always included) ───────────────────────────────────────
+    if not extract_temporal_bounds:
+        prompt = f"""Extract relations between entities from the provided text.
 Return the result as a JSON object with a "relations" key containing the list of relations.
 Each relation must have 'subject', 'predicate', and 'object' fields.
 
@@ -1820,115 +1834,59 @@ Instructions:
 Text to extract from:
 {text}
 Entities found in text: {entities_str}"""
-
-
-    if not entities:
-        error_msg = "No entities provided for relation extraction. Relations require existing entities."
-        logger.error(error_msg)
-        if not silent_fail:
-            raise ProcessingError(error_msg)
-        return []
-
-    # Pass api_key if provided in kwargs
-    provider_kwargs = kwargs.copy()
-    
-    # Check if api_key is provided but empty, or not provided at all
-    if "api_key" not in provider_kwargs or not provider_kwargs["api_key"]:
-        import os
-        env_key = f"{provider.upper()}_API_KEY"
-        api_key = os.getenv(env_key)
-        if api_key:
-            provider_kwargs["api_key"] = api_key
-            
-    # Remove None/empty API key if still present to avoid provider errors
-    if "api_key" in provider_kwargs and not provider_kwargs["api_key"]:
-        del provider_kwargs["api_key"]
-
-    # 2. PROVIDER VALIDATION
-    try:
-        llm = create_provider(provider, model=model, **provider_kwargs)
-        if not llm.is_available():
-            error_msg = f"{provider} provider not available for relation extraction (key missing?)."
-            logger.error(error_msg)
-            if not silent_fail:
-                raise ProcessingError(error_msg)
-            return []
-    except Exception as e:
-        error_msg = f"Failed to create {provider} provider for relations: {e}"
-        logger.error(error_msg)
-        if not silent_fail:
-            raise ProcessingError(error_msg) from e
-        return []
-
-    # 3. TEXT LENGTH CHECK AND CHUNKING
-    if max_text_length is None:
-        # Default limits for chunking only - NOT for LLM generation
-        max_text_length = {
-            "groq": 64000,
-            "openai": 64000,
-            "gemini": 64000,
-            "anthropic": 64000,
-            "deepseek": 64000,
-        }.get(provider.lower(), 32000)
-    
-    if len(text) > max_text_length:
-        logger.info(f"Text length ({len(text)}) exceeds limit for relations. Chunking...")
-        return _extract_relations_chunked(
-            text, entities, provider=provider, model=model, 
-            silent_fail=silent_fail, max_text_length=max_text_length, 
-            max_retries=max_retries,
-            **kwargs
-        )
-
-    original_entities = entities
-    # Use a fixed internal default for prompt entity cap (do not accept overrides from kwargs)
-    max_entities_prompt = 80
-    prompt_entities = original_entities
-    if max_entities_prompt > 0 and len(original_entities) > max_entities_prompt:
-        prompt_entities = filter_entities_for_text(
-            text,
-            original_entities,
-            max_keep=max_entities_prompt,
-        )
-
-    entities_str = ", ".join([f"{e.text} ({e.label})" for e in prompt_entities])
-    
-    # Use custom relation types if provided
-    relation_types = kwargs.get("relation_types")
-    if relation_types:
-        relation_types_str = ", ".join(relation_types)
-        relation_types_instruction = f"""
-Preferred relation types: {relation_types_str}.
-You may also use related or similar relation types if they better capture the relationship (e.g., variations, synonyms, or domain-specific relations).
-If a relation doesn't fit any of the preferred types, use the most appropriate type from the preferred list or a closely related type that accurately describes the relationship."""
     else:
-        relation_types_instruction = """
-Extract meaningful relationships between entities. Use appropriate relation types that accurately describe how entities are connected.
-Common relation types include: related_to, part_of, located_in, created_by, uses, depends_on, interacts_with, and similar variations."""
-    
-    verbose_mode = kwargs.get("verbose", False)
-    if verbose_mode:
-        import sys
-        print(f"    [methods.extract_relations_llm] Constructing prompt for {len(prompt_entities)} entities...", flush=True, file=sys.stdout)
-    
-    if not SCHEMAS_AVAILABLE:
-        raise ImportError("Pydantic schemas not available. Install pydantic/instructor to use LLM extraction.")
+        # ── Temporal-extended prompt ─────────────────────────────────────────
+        prompt = f"""Extract relations between entities from the provided text, along with temporal validity information for each relation.
+Return the result as a JSON object with a "relations" key. Each relation must have:
+'subject', 'predicate', 'object', 'confidence', 'valid_from', 'valid_until', 'temporal_confidence', 'temporal_source_text'.
 
-    prompt = f"""Extract relations between entities from the provided text.
-Return the result as a JSON object with a "relations" key containing the list of relations.
-Each relation must have 'subject', 'predicate', and 'object' fields.
+TEMPORAL EXTRACTION RULES:
+- valid_from: ISO 8601 date or exact phrase from the text for when this relation became valid. Set to null if no temporal signal is present.
+- valid_until: ISO 8601 date or exact phrase for when this relation ceased. Set to null if open-ended or absent.
+- temporal_confidence (float 0.0–1.0) — calibrated as follows:
+    1.00 = full ISO date ("2022-03-15", "March 15, 2022")
+    0.90 = explicit year + month ("March 2022", "2022-03")
+    0.85 = explicit year only ("in 2022", "since 2021", "from 2019")
+    0.75 = quarter ("Q3 2023", "Q2 2021")
+    0.65 = named season or approximate range ("summer 2022", "early 2020s", "mid-2022")
+    0.50 = vague relative with computable anchor ("last year", "three months ago")
+    0.35 = highly vague relative ("recently", "years ago", "in the past")
+    0.00 = no temporal signal present for this relation
+- temporal_source_text: the EXACT verbatim substring from the source text that contains the temporal signal. Set to null when temporal_confidence is 0.0.
 
-Example output (JSON format only):
+IMPORTANT: Do NOT invent or guess dates. If the text contains no temporal signal for a relation, set valid_from and valid_until to null and temporal_confidence to 0.0.
+
+Few-shot examples (do NOT include these in your output):
+  Text: "Apple acquired Beats in May 2014."
+  → valid_from: "2014-05-01", valid_until: null, temporal_confidence: 0.90, temporal_source_text: "May 2014"
+
+  Text: "The CEO has led the company since Q3 2020."
+  → valid_from: "Q3 2020", valid_until: null, temporal_confidence: 0.75, temporal_source_text: "since Q3 2020"
+
+  Text: "Last year, Google partnered with Samsung."
+  → valid_from: "last year", valid_until: null, temporal_confidence: 0.50, temporal_source_text: "Last year"
+
+  Text: "The firm was under enhanced supervision between Q2 and Q4 2021."
+  → valid_from: "Q2 2021", valid_until: "Q4 2021", temporal_confidence: 0.75, temporal_source_text: "between Q2 and Q4 2021"
+
+  Text: "Microsoft develops Windows."
+  → valid_from: null, valid_until: null, temporal_confidence: 0.00, temporal_source_text: null
+
+Example JSON output format:
 {{
   "relations": [
-    {{"subject": "Entity A", "predicate": "related_to", "object": "Entity B", "confidence": 0.95}},
-    {{"subject": "Subject Entity", "predicate": "action_verb", "object": "Object Entity", "confidence": 0.90}}
+    {{
+      "subject": "Apple", "predicate": "acquired", "object": "Beats",
+      "confidence": 0.97,
+      "valid_from": "2014-05-01", "valid_until": null,
+      "temporal_confidence": 0.90, "temporal_source_text": "May 2014"
+    }}
   ]
 }}
 
 Instructions:
 1. Extract relations ONLY from the text provided below.
-2. Do not include any relations from the example above.
+2. Do not include any relations from the examples above.
 3. Use the provided entities list as a reference for subjects and objects.
 4. {relation_types_instruction}
 
@@ -1938,24 +1896,25 @@ Entities found in text: {entities_str}"""
 
     try:
         # Use typed generation with Pydantic schema
-        # Pass kwargs to allow max_tokens and other parameters to be used
         if verbose_mode:
-             import sys
-             print(f"    [methods.extract_relations_llm] Calling llm.generate_typed ({provider}/{model})...", flush=True, file=sys.stdout)
+            import sys
+            print(f"    [methods.extract_relations_llm] Calling llm.generate_typed ({provider}/{model})...", flush=True, file=sys.stdout)
         # Only forward minimal, safe parameters to provider calls
         call_kwargs = {}
         if "temperature" in kwargs:
             call_kwargs["temperature"] = kwargs["temperature"]
         if "verbose" in kwargs:
             call_kwargs["verbose"] = kwargs["verbose"]
-        
+
         call_kwargs["max_retries"] = max_retries
 
-        result_obj = llm.generate_typed(prompt, schema=RelationsResponse, **call_kwargs)
+        # Select schema based on whether temporal extraction is requested
+        active_schema = RelationsWithTemporalResponse if extract_temporal_bounds else RelationsResponse
+        result_obj = llm.generate_typed(prompt, schema=active_schema, **call_kwargs)
         if verbose_mode:
-             import sys
-             print(f"    [methods.extract_relations_llm] Received response from {provider}.", flush=True, file=sys.stdout)
-        
+            import sys
+            print(f"    [methods.extract_relations_llm] Received response from {provider}.", flush=True, file=sys.stdout)
+
         # Convert back to internal Relation format (robust across providers)
         # Normalize typed result to a plain dict compatible with _parse_relation_result
         try:
@@ -1972,12 +1931,16 @@ Entities found in text: {entities_str}"""
                     elif isinstance(r, dict):
                         rel_items.append(r)
                     else:
-                        # Best-effort attribute access
+                        # Best-effort attribute access — include temporal fields when present
                         rel_items.append({
                             "subject": getattr(r, "subject", ""),
                             "object": getattr(r, "object", ""),
                             "predicate": getattr(r, "predicate", "related_to"),
                             "confidence": getattr(r, "confidence", 0.9),
+                            "valid_from": getattr(r, "valid_from", None),
+                            "valid_until": getattr(r, "valid_until", None),
+                            "temporal_confidence": getattr(r, "temporal_confidence", 0.0),
+                            "temporal_source_text": getattr(r, "temporal_source_text", None),
                         })
                 parsed = {"relations": rel_items}
             else:
@@ -1986,7 +1949,11 @@ Entities found in text: {entities_str}"""
             parsed = result_obj
 
         # Use common parser to build internal Relation objects
-        relations = _parse_relation_result(parsed, original_entities, text, provider, model, extraction_method="llm_typed")
+        relations = _parse_relation_result(
+            parsed, original_entities, text, provider, model,
+            extraction_method="llm_typed",
+            extract_temporal_bounds=extract_temporal_bounds,
+        )
 
         # If typed path returned no relations, attempt a structured JSON fallback
         if not relations:
@@ -1995,7 +1962,11 @@ Entities found in text: {entities_str}"""
                     import sys
                     print("    [methods.extract_relations_llm] Typed result empty, attempting structured JSON fallback...", flush=True, file=sys.stdout)
                 raw_json = llm.generate_structured(prompt, **call_kwargs)
-                relations = _parse_relation_result(raw_json, original_entities, text, provider, model, extraction_method="llm_typed")
+                relations = _parse_relation_result(
+                    raw_json, original_entities, text, provider, model,
+                    extraction_method="llm_typed",
+                    extract_temporal_bounds=extract_temporal_bounds,
+                )
             except Exception as _e:
                 # Keep relations as empty if fallback fails
                 pass
@@ -2003,22 +1974,23 @@ Entities found in text: {entities_str}"""
         logger.info(f"Successfully extracted {len(relations)} relations using {provider}/{model} (typed)")
         _result_cache.set("relations", text, relations, **cache_params)
         return relations
-        
+
     except Exception as e:
         # Check for length/token limit errors
         error_msg_str = str(e).lower()
         if "length" in error_msg_str or "max_tokens" in error_msg_str:
             logger.warning(f"LLM output truncated due to length limit. Reducing chunk size and retrying... ({e})")
-            
+
             # Determine new chunk size (halve it)
             current_max = max_text_length or len(text)
             new_max = current_max // 2
-            
-            if new_max > 100: # Minimum viable chunk size check
+
+            if new_max > 100:  # Minimum viable chunk size check
                 return _extract_relations_chunked(
                     text, entities, provider=provider, model=model,
                     silent_fail=silent_fail, max_text_length=new_max,
                     structured_output_mode=structured_output_mode,
+                    extract_temporal_bounds=extract_temporal_bounds,
                     **kwargs
                 )
 
@@ -2038,11 +2010,12 @@ def _parse_relation_result(
     provider: str,
     model: Optional[str],
     extraction_method: str = "llm",
+    extract_temporal_bounds: bool = False,
 ) -> List[Relation]:
     """Helper to parse raw LLM result into Relation objects."""
     relations = []
     items = []
-    
+
     if isinstance(result, list):
         items = result
     elif isinstance(result, dict):
@@ -2056,13 +2029,13 @@ def _parse_relation_result(
     for item in items:
         if not isinstance(item, dict):
             continue
-            
+
         subject_text = item.get("subject", "")
         object_text = item.get("object", "")
-        
+
         if not subject_text or not object_text:
             continue
-            
+
         # Ensure they are strings
         subject_text = str(subject_text)
         object_text = str(object_text)
@@ -2085,6 +2058,33 @@ def _parse_relation_result(
                 confidence=0.8, metadata={"synthetic": True},
             )
 
+        metadata: dict = {
+            "provider": provider,
+            "model": model,
+            "extraction_method": extraction_method,
+        }
+
+        if extract_temporal_bounds:
+            temporal_confidence = float(item.get("temporal_confidence") or 0.0)
+            valid_from = item.get("valid_from")
+            valid_until = item.get("valid_until")
+            temporal_source_text = item.get("temporal_source_text")
+
+            metadata["valid_from"] = valid_from
+            metadata["valid_until"] = valid_until
+            metadata["temporal_confidence"] = temporal_confidence
+            metadata["temporal_source_text"] = temporal_source_text
+
+            if temporal_confidence < 0.5 and (valid_from is not None or valid_until is not None):
+                logger.warning(
+                    "Low temporal confidence (%.2f) for '%s' (%s → %s). Source: %r",
+                    temporal_confidence,
+                    item.get("predicate", ""),
+                    item.get("subject", ""),
+                    item.get("object", ""),
+                    temporal_source_text,
+                )
+
         relations.append(
             Relation(
                 subject=subject_entity,
@@ -2092,11 +2092,7 @@ def _parse_relation_result(
                 object=object_entity,
                 confidence=item.get("confidence", 0.9),
                 context=text,
-                metadata={
-                    "provider": provider,
-                    "model": model,
-                    "extraction_method": extraction_method,
-                },
+                metadata=metadata,
             )
         )
     return relations
@@ -2111,6 +2107,7 @@ def _extract_relations_chunked(
     max_text_length: int,
     structured_output_mode: str = "typed",
     max_retries: int = 3,
+    extract_temporal_bounds: bool = False,
     **kwargs
 ) -> List[Relation]:
     """Internal helper to extract relations from long text by chunking."""
@@ -2154,6 +2151,7 @@ def _extract_relations_chunked(
                 max_text_length=len(chunk.text) + 1,
                 structured_output_mode=structured_output_mode,
                 max_retries=max_retries,
+                extract_temporal_bounds=extract_temporal_bounds,
                 **limited_kwargs
             )
             future_to_chunk[future] = i

semantica/semantic_extract/schemas.py

@@ -125,3 +125,60 @@ class RelationsResponse(BaseModel):
 class TripletsResponse(BaseModel):
     """Wrapper for list of triplets."""
     triplets: List[TripletOut] = Field(default_factory=list)
+
+
+class RelationWithTemporalOut(BaseModel):
+    """Schema for relation extraction output with temporal validity bounds."""
+    model_config = ConfigDict(populate_by_name=True, extra="ignore")
+
+    subject: str = Field(..., description="Source entity text")
+    object: str = Field(..., description="Target entity text")
+    predicate: str = Field(..., description="Relation type or predicate")
+    confidence: float = Field(0.9, description="Confidence score between 0 and 1")
+    metadata: dict = Field(default_factory=dict, description="Additional metadata including provenance")
+
+    valid_from: Optional[str] = Field(
+        None,
+        description="ISO 8601 date or natural-language phrase for when this relation became valid. Null if no temporal signal in text.",
+    )
+    valid_until: Optional[str] = Field(
+        None,
+        description="ISO 8601 date or phrase for when this relation ceased to be valid. Null if open-ended or not stated.",
+    )
+    temporal_confidence: float = Field(
+        0.0,
+        description="Confidence that temporal information was present and correctly extracted. 0.0 if no temporal signal.",
+    )
+    temporal_source_text: Optional[str] = Field(
+        None,
+        description="Exact verbatim substring from the source text containing the temporal signal. Null when temporal_confidence is 0.0.",
+    )
+
+    @model_validator(mode="before")
+    @classmethod
+    def handle_aliases(cls, data):
+        if isinstance(data, dict):
+            if "subject" not in data and "source" in data:
+                data["subject"] = data["source"]
+            if "object" not in data and "target" in data:
+                data["object"] = data["target"]
+            if "predicate" not in data and "label" in data:
+                data["predicate"] = data["label"]
+        return data
+
+    @field_validator("confidence", "temporal_confidence", mode="before")
+    @classmethod
+    def normalize_confidence(cls, v):
+        if isinstance(v, str):
+            try:
+                v = float(v)
+            except ValueError:
+                return 0.0
+        if isinstance(v, (int, float)):
+            return max(0.0, min(1.0, float(v)))
+        return 0.0
+
+
+class RelationsWithTemporalResponse(BaseModel):
+    """Wrapper for list of relations with temporal validity bounds."""
+    relations: List[RelationWithTemporalOut] = Field(default_factory=list)

semantica/utils/exceptions.py

@@ -167,6 +167,18 @@ class TemporalValidationError(ValidationError):
         self.error_code = "SEM001T"
 
 
+class TemporalAmbiguityWarning(UserWarning):
+    """
+    Warning raised when a temporal expression is ambiguous and cannot be
+    resolved without additional locale or context information.
+
+    Example: "03/04/2022" is ambiguous without knowing whether day-first or
+    month-first ordering applies. Use ``warnings.catch_warnings()`` to handle.
+    """
+
+    pass
+
+
 class ProcessingError(SemanticaError):
     """
     Exception raised for data processing errors.

tests/semantic_extract/test_temporal_extraction.py

@@ -0,0 +1,675 @@
+"""
+Tests for temporal metadata extraction (Issue #400).
+
+Covers:
+  - extract_temporal_bounds=True/False flag on extract_relations_llm()
+  - TemporalNormalizer: relative dates, partial dates, ambiguity, domain phrases,
+    custom phrase map
+  - Full pipeline: extract → normalize → BiTemporalFact
+
+All LLM calls are mocked. No real API keys required. Suite runs in < 5 s.
+"""
+
+import os
+import sys
+import unittest
+import warnings
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
+
+# ── Mock optional heavyweight dependencies before any semantica import ──────
+sys.modules.setdefault("spacy", MagicMock())
+sys.modules.setdefault("instructor", MagicMock())
+_openai_mock = MagicMock()
+sys.modules.setdefault("openai", _openai_mock)
+sys.modules.setdefault("groq", MagicMock())
+sys.modules.setdefault("sentence_transformers", MagicMock())
+sys.modules.setdefault("transformers", MagicMock())
+sys.modules.setdefault("torch", MagicMock())
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")))
+
+from semantica.semantic_extract.methods import extract_relations_llm
+from semantica.semantic_extract.ner_extractor import Entity
+from semantica.semantic_extract.schemas import (
+    RelationsResponse,
+    RelationsWithTemporalResponse,
+)
+from semantica.kg.temporal_normalizer import TemporalNormalizer
+from semantica.utils.exceptions import TemporalAmbiguityWarning
+
+
+# ── Helpers ─────────────────────────────────────────────────────────────────
+
+def _make_entities():
+    return [
+        Entity(text="Apple", label="ORG", start_char=0, end_char=5),
+        Entity(text="Beats", label="ORG", start_char=15, end_char=20),
+    ]
+
+
+def _ref_date():
+    return datetime(2025, 6, 15, tzinfo=timezone.utc)
+
+
+# ============================================================================
+# Part 1 – extract_relations_llm() temporal flag
+# ============================================================================
+
+class TestTemporalExtractionFlag(unittest.TestCase):
+
+    def setUp(self):
+        from semantica.semantic_extract.methods import _result_cache
+        _result_cache.clear()
+
+    @patch("semantica.semantic_extract.methods.create_provider")
+    def test_extract_temporal_bounds_true_adds_four_fields(self, mock_create):
+        """With extract_temporal_bounds=True all four temporal keys appear in metadata."""
+        mock_prov = MagicMock()
+        mock_prov.is_available.return_value = True
+        mock_prov.generate_typed.return_value = RelationsWithTemporalResponse(
+            relations=[
+                {
+                    "subject": "Apple",
+                    "predicate": "acquired",
+                    "object": "Beats",
+                    "confidence": 0.97,
+                    "valid_from": "2014-05-01",
+                    "valid_until": None,
+                    "temporal_confidence": 0.90,
+                    "temporal_source_text": "May 2014",
+                }
+            ]
+        )
+        mock_create.return_value = mock_prov
+
+        rels = extract_relations_llm(
+            "Apple acquired Beats in May 2014.",
+            _make_entities(),
+            provider="openai",
+            extract_temporal_bounds=True,
+        )
+
+        self.assertEqual(len(rels), 1)
+        meta = rels[0].metadata
+        self.assertIn("valid_from", meta)
+        self.assertIn("valid_until", meta)
+        self.assertIn("temporal_confidence", meta)
+        self.assertIn("temporal_source_text", meta)
+        self.assertEqual(meta["valid_from"], "2014-05-01")
+        self.assertIsNone(meta["valid_until"])
+        self.assertAlmostEqual(meta["temporal_confidence"], 0.90, places=2)
+        self.assertEqual(meta["temporal_source_text"], "May 2014")
+
+    @patch("semantica.semantic_extract.methods.create_provider")
+    def test_extract_temporal_bounds_false_output_identical(self, mock_create):
+        """With extract_temporal_bounds=False (default) no temporal keys appear."""
+        mock_prov = MagicMock()
+        mock_prov.is_available.return_value = True
+        mock_prov.generate_typed.return_value = RelationsResponse(
+            relations=[
+                {
+                    "subject": "Apple",
+                    "predicate": "acquired",
+                    "object": "Beats",
+                    "confidence": 0.97,
+                }
+            ]
+        )
+        mock_create.return_value = mock_prov
+
+        rels = extract_relations_llm(
+            "Apple acquired Beats.",
+            _make_entities(),
+            provider="openai",
+        )
+
+        self.assertEqual(len(rels), 1)
+        meta = rels[0].metadata
+        self.assertNotIn("valid_from", meta)
+        self.assertNotIn("valid_until", meta)
+        self.assertNotIn("temporal_confidence", meta)
+        self.assertNotIn("temporal_source_text", meta)
+
+    @patch("semantica.semantic_extract.methods.create_provider")
+    def test_no_temporal_signal_returns_zero_confidence_and_null_dates(self, mock_create):
+        """When LLM returns no temporal signal, confidence=0.0 and dates are null."""
+        mock_prov = MagicMock()
+        mock_prov.is_available.return_value = True
+        mock_prov.generate_typed.return_value = RelationsWithTemporalResponse(
+            relations=[
+                {
+                    "subject": "Apple",
+                    "predicate": "owns",
+                    "object": "Beats",
+                    "confidence": 0.95,
+                    "valid_from": None,
+                    "valid_until": None,
+                    "temporal_confidence": 0.0,
+                    "temporal_source_text": None,
+                }
+            ]
+        )
+        mock_create.return_value = mock_prov
+
+        rels = extract_relations_llm(
+            "Apple owns Beats.",
+            _make_entities(),
+            provider="openai",
+            extract_temporal_bounds=True,
+        )
+
+        meta = rels[0].metadata
+        self.assertIsNone(meta["valid_from"])
+        self.assertIsNone(meta["valid_until"])
+        self.assertEqual(meta["temporal_confidence"], 0.0)
+        self.assertIsNone(meta["temporal_source_text"])
+
+    @patch("semantica.semantic_extract.methods.create_provider")
+    def test_low_temporal_confidence_logs_warning(self, mock_create):
+        """temporal_confidence < 0.5 with non-null date logs a WARNING."""
+        mock_prov = MagicMock()
+        mock_prov.is_available.return_value = True
+        mock_prov.generate_typed.return_value = RelationsWithTemporalResponse(
+            relations=[
+                {
+                    "subject": "Apple",
+                    "predicate": "partnered_with",
+                    "object": "Beats",
+                    "confidence": 0.80,
+                    "valid_from": "recently",
+                    "valid_until": None,
+                    "temporal_confidence": 0.35,
+                    "temporal_source_text": "recently",
+                }
+            ]
+        )
+        mock_create.return_value = mock_prov
+
+        with self.assertLogs("semantica", level="WARNING") as cm:
+            extract_relations_llm(
+                "Apple recently partnered with Beats.",
+                _make_entities(),
+                provider="openai",
+                extract_temporal_bounds=True,
+            )
+        self.assertTrue(any("Low temporal confidence" in line for line in cm.output))
+
+    @patch("semantica.semantic_extract.methods.create_provider")
+    def test_correct_schema_used_when_temporal_true(self, mock_create):
+        """generate_typed is called with RelationsWithTemporalResponse when flag=True."""
+        mock_prov = MagicMock()
+        mock_prov.is_available.return_value = True
+        mock_prov.generate_typed.return_value = RelationsWithTemporalResponse(relations=[])
+        mock_create.return_value = mock_prov
+
+        extract_relations_llm(
+            "Some text.",
+            _make_entities(),
+            provider="openai",
+            extract_temporal_bounds=True,
+        )
+
+        call_kwargs = mock_prov.generate_typed.call_args[1]
+        self.assertIs(call_kwargs["schema"], RelationsWithTemporalResponse)
+
+    @patch("semantica.semantic_extract.methods.create_provider")
+    def test_correct_schema_used_when_temporal_false(self, mock_create):
+        """generate_typed is called with RelationsResponse when flag=False."""
+        mock_prov = MagicMock()
+        mock_prov.is_available.return_value = True
+        mock_prov.generate_typed.return_value = RelationsResponse(relations=[])
+        mock_create.return_value = mock_prov
+
+        extract_relations_llm(
+            "Some text.",
+            _make_entities(),
+            provider="openai",
+        )
+
+        call_kwargs = mock_prov.generate_typed.call_args[1]
+        self.assertIs(call_kwargs["schema"], RelationsResponse)
+
+
+# ============================================================================
+# Part 2 – TemporalNormalizer: relative dates
+# ============================================================================
+
+class TestTemporalNormalizerRelativeDates(unittest.TestCase):
+
+    def setUp(self):
+        self.ref = _ref_date()  # 2025-06-15
+        self.tn = TemporalNormalizer(reference_date=self.ref)
+
+    def test_last_year(self):
+        result = self.tn.normalize("last year")
+        self.assertIsNotNone(result)
+        start, end = result
+        self.assertEqual(start.year, 2024)
+        self.assertEqual(start.month, 1)
+        self.assertEqual(start.day, 1)
+        self.assertEqual(end.year, 2024)
+        self.assertEqual(end.month, 12)
+        self.assertEqual(end.day, 31)
+
+    def test_this_year(self):
+        result = self.tn.normalize("this year")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0].year, 2025)
+        self.assertEqual(result[0].month, 1)
+        self.assertEqual(result[1].month, 12)
+
+    def test_three_months_ago(self):
+        result = self.tn.normalize("three months ago")
+        self.assertIsNotNone(result)
+        # ref is 2025-06-15; three months ago → March 2025
+        self.assertEqual(result[0].year, 2025)
+        self.assertEqual(result[0].month, 3)
+
+    def test_last_quarter(self):
+        # ref is 2025-06-15 (Q2) → last quarter = Q1 2025
+        result = self.tn.normalize("last quarter")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0].month, 1)
+        self.assertEqual(result[1].month, 3)
+        self.assertEqual(result[0].year, 2025)
+
+    def test_this_quarter(self):
+        # ref is 2025-06-15 (Q2) → Q2 2025
+        result = self.tn.normalize("this quarter")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0].month, 4)
+        self.assertEqual(result[1].month, 6)
+
+    def test_last_month(self):
+        # ref June 2025 → May 2025
+        result = self.tn.normalize("last month")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0].month, 5)
+        self.assertEqual(result[0].year, 2025)
+
+    def test_two_years_ago(self):
+        result = self.tn.normalize("two years ago")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0].year, 2023)
+        self.assertEqual(result[1].year, 2023)
+
+    def test_no_reference_date_raises_value_error(self):
+        tn = TemporalNormalizer()
+        with self.assertRaises(ValueError):
+            tn.normalize("last year")
+
+    def test_none_input_returns_none(self):
+        self.assertIsNone(self.tn.normalize(None))
+
+    def test_empty_string_returns_none(self):
+        self.assertIsNone(self.tn.normalize(""))
+
+    def test_whitespace_string_returns_none(self):
+        self.assertIsNone(self.tn.normalize("   "))
+
+
+# ============================================================================
+# Part 3 – TemporalNormalizer: partial / structured dates
+# ============================================================================
+
+class TestTemporalNormalizerPartialDates(unittest.TestCase):
+
+    def setUp(self):
+        self.tn = TemporalNormalizer(reference_date=datetime(2025, 3, 25, tzinfo=timezone.utc))
+
+    def test_year_only(self):
+        result = self.tn.normalize("2021")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0], datetime(2021, 1, 1, tzinfo=timezone.utc))
+        self.assertEqual(result[1], datetime(2021, 12, 31, tzinfo=timezone.utc))
+
+    def test_month_year_word(self):
+        result = self.tn.normalize("March 2022")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0].year, 2022)
+        self.assertEqual(result[0].month, 3)
+        self.assertEqual(result[0].day, 1)
+        self.assertEqual(result[1].day, 31)
+
+    def test_month_year_word_abbreviated(self):
+        result = self.tn.normalize("Dec 2023")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0].month, 12)
+        self.assertEqual(result[1].day, 31)
+
+    def test_year_month_iso_partial(self):
+        result = self.tn.normalize("2022-03")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0].year, 2022)
+        self.assertEqual(result[0].month, 3)
+        self.assertEqual(result[0].day, 1)
+        self.assertEqual(result[1].day, 31)
+
+    def test_q1_2024(self):
+        result = self.tn.normalize("Q1 2024")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0], datetime(2024, 1, 1, tzinfo=timezone.utc))
+        self.assertEqual(result[1], datetime(2024, 3, 31, tzinfo=timezone.utc))
+
+    def test_q2_2021(self):
+        result = self.tn.normalize("Q2 2021")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0], datetime(2021, 4, 1, tzinfo=timezone.utc))
+        self.assertEqual(result[1], datetime(2021, 6, 30, tzinfo=timezone.utc))
+
+    def test_q3_2023(self):
+        result = self.tn.normalize("Q3 2023")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0], datetime(2023, 7, 1, tzinfo=timezone.utc))
+        self.assertEqual(result[1], datetime(2023, 9, 30, tzinfo=timezone.utc))
+
+    def test_q4_2022(self):
+        result = self.tn.normalize("Q4 2022")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0], datetime(2022, 10, 1, tzinfo=timezone.utc))
+        self.assertEqual(result[1], datetime(2022, 12, 31, tzinfo=timezone.utc))
+
+    def test_iso_full_date_returns_point(self):
+        result = self.tn.normalize("2022-03-15")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0].year, 2022)
+        self.assertEqual(result[0].month, 3)
+        self.assertEqual(result[0].day, 15)
+        # Point interval: start == end
+        self.assertEqual(result[0], result[1])
+
+    def test_iso_datetime_with_z(self):
+        result = self.tn.normalize("2022-03-15T00:00:00Z")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0].year, 2022)
+
+    def test_unparseable_returns_none(self):
+        result = self.tn.normalize("sometime in the medieval period")
+        self.assertIsNone(result)
+
+    def test_none_returns_none(self):
+        self.assertIsNone(self.tn.normalize(None))
+
+
+# ============================================================================
+# Part 4 – TemporalNormalizer: ambiguous formats
+# ============================================================================
+
+class TestTemporalNormalizerAmbiguity(unittest.TestCase):
+
+    def setUp(self):
+        self.tn = TemporalNormalizer(reference_date=datetime(2025, 3, 25, tzinfo=timezone.utc))
+
+    def test_ambiguous_slash_date_raises_warning_and_returns_none(self):
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+            result = self.tn.normalize("03/04/2022")
+        self.assertIsNone(result)
+        ambig = [x for x in w if issubclass(x.category, TemporalAmbiguityWarning)]
+        self.assertEqual(len(ambig), 1)
+        self.assertIn("ambiguous", str(ambig[0].message).lower())
+
+    def test_iso_hyphenated_date_not_flagged_as_ambiguous(self):
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+            result = self.tn.normalize("2022-03-04")
+        ambig = [x for x in w if issubclass(x.category, TemporalAmbiguityWarning)]
+        self.assertEqual(len(ambig), 0)
+        self.assertIsNotNone(result)
+
+    def test_ambiguous_date_does_not_raise_exception(self):
+        # Must not raise, only warn
+        try:
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore")
+                self.tn.normalize("01/12/2023")
+        except Exception as e:
+            self.fail(f"normalize() raised unexpectedly: {e}")
+
+
+# ============================================================================
+# Part 5 – TemporalNormalizer: domain phrase map
+# ============================================================================
+
+class TestTemporalNormalizerDomainPhrases(unittest.TestCase):
+
+    def setUp(self):
+        self.tn = TemporalNormalizer(reference_date=datetime(2025, 3, 25, tzinfo=timezone.utc))
+
+    def _assert_recognized(self, phrase):
+        result = self.tn.normalize_phrase(phrase)
+        self.assertIsNotNone(result, f"Expected phrase {phrase!r} to be recognized but got None")
+        return result
+
+    # General / Policy
+    def test_effective_date_recognized(self):
+        r = self._assert_recognized("effective date")
+        self.assertEqual(r["maps_to"], "valid_from")
+
+    def test_effective_from_regex_recognized(self):
+        r = self._assert_recognized("effective from")
+        self.assertEqual(r["maps_to"], "valid_from")
+
+    def test_effective_as_of_regex_recognized(self):
+        r = self._assert_recognized("effective as of")
+        self.assertEqual(r["maps_to"], "valid_from")
+
+    def test_in_force_until_recognized(self):
+        r = self._assert_recognized("in force until")
+        self.assertEqual(r["maps_to"], "valid_until")
+
+    def test_retroactive_to_recognized(self):
+        r = self._assert_recognized("retroactive to")
+        self.assertTrue(r.get("retroactive"))
+
+    def test_sunset_clause_recognized(self):
+        r = self._assert_recognized("sunset clause")
+        self.assertEqual(r["maps_to"], "valid_until")
+
+    # Healthcare / Drug Discovery
+    def test_approval_date_recognized(self):
+        r = self._assert_recognized("approval date")
+        self.assertEqual(r["maps_to"], "valid_from")
+        self.assertIn("Healthcare", r.get("domain", []))
+
+    def test_expiry_date_recognized(self):
+        r = self._assert_recognized("expiry date")
+        self.assertEqual(r["maps_to"], "valid_until")
+
+    def test_market_authorization_recognized(self):
+        r = self._assert_recognized("market authorization")
+        self.assertEqual(r["maps_to"], "valid_from")
+        self.assertIn("Drug Discovery", r.get("domain", []))
+
+    # Cybersecurity
+    def test_incident_window_recognized(self):
+        r = self._assert_recognized("incident window")
+        self.assertIn("Cybersecurity", r.get("domain", []))
+
+    def test_campaign_period_recognized(self):
+        r = self._assert_recognized("campaign period")
+        self.assertIn("Cybersecurity", r.get("domain", []))
+
+    # Supply Chain
+    def test_certification_valid_through_recognized(self):
+        r = self._assert_recognized("certification valid through")
+        self.assertEqual(r["maps_to"], "valid_until")
+        self.assertIn("Supply Chain", r.get("domain", []))
+
+    # Finance
+    def test_trading_halt_recognized(self):
+        r = self._assert_recognized("trading halt")
+        self.assertIn("Finance", r.get("domain", []))
+
+    # Energy
+    def test_commissioned_date_recognized(self):
+        r = self._assert_recognized("commissioned date")
+        self.assertEqual(r["maps_to"], "valid_from")
+        self.assertIn("Energy", r.get("domain", []))
+
+    def test_decommissioned_date_recognized(self):
+        r = self._assert_recognized("decommissioned date")
+        self.assertEqual(r["maps_to"], "valid_until")
+        self.assertIn("Energy", r.get("domain", []))
+
+    def test_unrecognized_phrase_returns_none(self):
+        result = self.tn.normalize_phrase("totally unknown phrase xyz")
+        self.assertIsNone(result)
+
+
+# ============================================================================
+# Part 6 – TemporalNormalizer: custom phrase map
+# ============================================================================
+
+class TestTemporalNormalizerCustomPhraseMap(unittest.TestCase):
+
+    def setUp(self):
+        ref = datetime(2025, 3, 25, tzinfo=timezone.utc)
+        self.tn = TemporalNormalizer(
+            reference_date=ref,
+            phrase_map={
+                "fiscal year 2024": lambda r: (
+                    datetime(2024, 4, 1, tzinfo=timezone.utc),
+                    datetime(2025, 3, 31, tzinfo=timezone.utc),
+                )
+            },
+        )
+
+    def test_custom_phrase_resolved(self):
+        result = self.tn.normalize("fiscal year 2024")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0].year, 2024)
+        self.assertEqual(result[0].month, 4)
+        self.assertEqual(result[1].year, 2025)
+        self.assertEqual(result[1].month, 3)
+
+    def test_default_phrase_still_works_alongside_custom(self):
+        result = self.tn.normalize("last year")
+        self.assertIsNotNone(result)
+        self.assertEqual(result[0].year, 2024)
+
+    def test_custom_phrase_overrides_default_when_same_key(self):
+        # Override "last year" to a custom sentinel
+        sentinel_start = datetime(2000, 1, 1, tzinfo=timezone.utc)
+        sentinel_end = datetime(2000, 12, 31, tzinfo=timezone.utc)
+        tn = TemporalNormalizer(
+            reference_date=datetime(2025, 3, 25, tzinfo=timezone.utc),
+            phrase_map={"last year": lambda r: (sentinel_start, sentinel_end)},
+        )
+        result = tn.normalize("last year")
+        self.assertEqual(result[0], sentinel_start)
+
+
+# ============================================================================
+# Part 7 – Full pipeline: extract → normalize → BiTemporalFact
+# ============================================================================
+
+class TestFullPipelineTemporalToBiTemporal(unittest.TestCase):
+
+    def setUp(self):
+        from semantica.semantic_extract.methods import _result_cache
+        _result_cache.clear()
+
+    @patch("semantica.semantic_extract.methods.create_provider")
+    def test_full_pipeline_explicit_date(self, mock_create):
+        """extract_relations(temporal=True) → normalize → BiTemporalFact."""
+        from semantica.kg.temporal_model import BiTemporalFact
+
+        mock_prov = MagicMock()
+        mock_prov.is_available.return_value = True
+        mock_prov.generate_typed.return_value = RelationsWithTemporalResponse(
+            relations=[
+                {
+                    "subject": "Apple",
+                    "predicate": "acquired",
+                    "object": "Beats",
+                    "confidence": 0.97,
+                    "valid_from": "2014-05-01",
+                    "valid_until": None,
+                    "temporal_confidence": 0.90,
+                    "temporal_source_text": "May 2014",
+                }
+            ]
+        )
+        mock_create.return_value = mock_prov
+
+        entities = [
+            Entity(text="Apple", label="ORG", start_char=0, end_char=5),
+            Entity(text="Beats", label="ORG", start_char=15, end_char=20),
+        ]
+        rels = extract_relations_llm(
+            "Apple acquired Beats in May 2014.",
+            entities,
+            provider="openai",
+            extract_temporal_bounds=True,
+        )
+        self.assertEqual(len(rels), 1)
+
+        meta = rels[0].metadata
+        ref = datetime(2025, 3, 25, tzinfo=timezone.utc)
+        tn = TemporalNormalizer(reference_date=ref)
+
+        vf = tn.normalize(meta["valid_from"])
+        vu = tn.normalize(meta["valid_until"])
+
+        self.assertIsNotNone(vf)
+        self.assertIsNone(vu)
+        self.assertEqual(vf[0].year, 2014)
+        self.assertEqual(vf[0].month, 5)
+        self.assertEqual(vf[0].day, 1)
+
+        # Feed into BiTemporalFact
+        fact = BiTemporalFact.from_relationship({
+            "valid_from": "2014-05-01T00:00:00Z",
+            "valid_until": None,
+        })
+        self.assertIsNotNone(fact.valid_from)
+        self.assertEqual(fact.valid_from.year, 2014)
+        self.assertEqual(fact.valid_from.month, 5)
+
+    @patch("semantica.semantic_extract.methods.create_provider")
+    def test_full_pipeline_quarter_expression(self, mock_create):
+        """Pipeline with Q-expression normalizes to correct quarter bounds."""
+        mock_prov = MagicMock()
+        mock_prov.is_available.return_value = True
+        mock_prov.generate_typed.return_value = RelationsWithTemporalResponse(
+            relations=[
+                {
+                    "subject": "Apple",
+                    "predicate": "supervised",
+                    "object": "Beats",
+                    "confidence": 0.85,
+                    "valid_from": "Q2 2021",
+                    "valid_until": "Q4 2021",
+                    "temporal_confidence": 0.75,
+                    "temporal_source_text": "between Q2 and Q4 2021",
+                }
+            ]
+        )
+        mock_create.return_value = mock_prov
+
+        entities = [
+            Entity(text="Apple", label="ORG", start_char=0, end_char=5),
+            Entity(text="Beats", label="ORG", start_char=15, end_char=20),
+        ]
+        rels = extract_relations_llm(
+            "Apple supervised Beats between Q2 and Q4 2021.",
+            entities,
+            provider="openai",
+            extract_temporal_bounds=True,
+        )
+        meta = rels[0].metadata
+        tn = TemporalNormalizer(reference_date=datetime(2025, 3, 25, tzinfo=timezone.utc))
+
+        vf = tn.normalize(meta["valid_from"])
+        vu = tn.normalize(meta["valid_until"])
+
+        self.assertEqual(vf[0], datetime(2021, 4, 1, tzinfo=timezone.utc))
+        self.assertEqual(vf[1], datetime(2021, 6, 30, tzinfo=timezone.utc))
+        self.assertEqual(vu[0], datetime(2021, 10, 1, tzinfo=timezone.utc))
+        self.assertEqual(vu[1], datetime(2021, 12, 31, tzinfo=timezone.utc))
+
+
+if __name__ == "__main__":
+    unittest.main()