"""Graph store: Neo4j wrapper with bi-temporal support.

This module provides a thin, testable abstraction over Neo4j for upserting
entities and edges with validity windows, plus hybrid retrieval (vector +
keyword + traversal + temporal filters).

We use the official neo4j driver directly rather than Graphiti's higher-level
API so that Zaxy controls the exact bi-temporal schema and extraction pipeline.
"""

from __future__ import annotations

import re
from dataclasses import dataclass
from importlib import import_module
from pathlib import Path
from typing import Any, Literal

from zaxy.extract import ExtractedEdge, ExtractedEntity, ExtractionResult
from zaxy.projection import ProjectionStore
from zaxy.schema import apply_schema_migrations
from zaxy.security import (
    validate_limit,
    validate_session_id,
    validate_traversal_depth,
    vector_has_signal,
)

_Neo4jPropertyValue = str | int | float | bool | list[str] | list[int] | list[float] | list[bool]
_RELATION_TYPE_RE = re.compile(r"^[a-z][a-z0-9_]{0,63}$")
_PROPERTY_KEY_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,63}$")


def _load_async_neo4j() -> tuple[Any, type[Any], type[Any]]:
    """Load Neo4j's async driver symbols only for the explicit Neo4j backend."""
    try:
        neo4j = import_module("neo4j")
    except ImportError as exc:
        raise RuntimeError('Neo4j backend requires `pip install "zaxy-memory[neo4j]"`') from exc
    return neo4j.AsyncGraphDatabase, neo4j.TrustAll, neo4j.TrustCustomCAs


@dataclass(frozen=True)
class GraphEntity:
    """Entity as stored in Neo4j."""

    name: str
    entity_type: str
    valid_from: str
    valid_to: str | None
    properties: dict[str, Any]
    session_id: str = "default"


@dataclass(frozen=True)
class GraphEdge:
    """Edge as stored in Neo4j."""

    source: str
    target: str
    relation_type: str
    valid_from: str
    valid_to: str | None
    properties: dict[str, Any]


@dataclass(frozen=True)
class SearchResult:
    """A single result from hybrid search."""

    entity: GraphEntity
    score: float
    source: str  # 'vector', 'keyword', 'traversal', 'exact'
    raw_score: float | None = None
    source_weight: float | None = None
    ranking_score: float | None = None
    matched_query: str | None = None
    query_weight: float | None = None
    temporal_score: float | None = None
    temporal_weight: float | None = None
    scoring_profile: str | None = None
    reranker: str | None = None
    rerank_score: float | None = None
    rerank_strategy: str | None = None
    warnings: tuple[str, ...] = ()
    # True when the result came from an exhaustive scoring path; False for
    # approximate paths (HNSW index, quantized candidate selection).
    exact: bool = True


@dataclass(frozen=True)
class GraphEventProjectionStatus:
    """Integrity status for one session's Eventloom graph projection."""

    session_id: str
    event_count: int
    latest_seq: int | None
    latest_hash: str | None
    eventloom_latest_seq: int | None
    eventloom_latest_hash: str | None
    projection_lag: int | None
    latest_hash_matches: bool
    next_event_edges: int
    previous_event_edges: int
    missing_chain_links: int
    integrity_ok: bool

    def to_dict(self) -> dict[str, Any]:
        """Return a stable JSON-serializable representation."""
        return {
            "session_id": self.session_id,
            "event_count": self.event_count,
            "latest_seq": self.latest_seq,
            "latest_hash": self.latest_hash,
            "eventloom_latest_seq": self.eventloom_latest_seq,
            "eventloom_latest_hash": self.eventloom_latest_hash,
            "projection_lag": self.projection_lag,
            "latest_hash_matches": self.latest_hash_matches,
            "next_event_edges": self.next_event_edges,
            "previous_event_edges": self.previous_event_edges,
            "missing_chain_links": self.missing_chain_links,
            "integrity_ok": self.integrity_ok,
        }


@dataclass(frozen=True)
class GraphInferredEdgeMethodStatus:
    """Audit summary for inferred edges generated by one inference method."""

    method: str
    edge_count: int
    relation_types: tuple[str, ...]
    average_confidence: float | None
    minimum_confidence: float | None
    evidence_count: int
    missing_evidence_count: int
    missing_source_event_count: int

    def to_dict(self) -> dict[str, Any]:
        """Return a stable JSON-serializable representation."""
        return {
            "method": self.method,
            "edge_count": self.edge_count,
            "relation_types": list(self.relation_types),
            "average_confidence": self.average_confidence,
            "minimum_confidence": self.minimum_confidence,
            "evidence_count": self.evidence_count,
            "missing_evidence_count": self.missing_evidence_count,
            "missing_source_event_count": self.missing_source_event_count,
        }


@dataclass(frozen=True)
class GraphInferredEdgeSample:
    """Representative inferred edge for operator inspection."""

    source: str
    target: str
    relation_type: str
    confidence: float | None
    method: str
    source_event_seq: int | None
    source_event_hash: str | None
    evidence_keys: tuple[str, ...]

    def to_dict(self) -> dict[str, Any]:
        """Return a stable JSON-serializable representation."""
        return {
            "source": self.source,
            "target": self.target,
            "relation_type": self.relation_type,
            "confidence": self.confidence,
            "method": self.method,
            "source_event_seq": self.source_event_seq,
            "source_event_hash": self.source_event_hash,
            "evidence_keys": list(self.evidence_keys),
        }


@dataclass(frozen=True)
class GraphInferredEdgeStatus:
    """Read-only audit status for inferred graph relationships."""

    session_id: str
    total_edges: int
    method_count: int
    evidence_count: int
    missing_evidence_count: int
    missing_source_event_count: int
    evidence_coverage: float
    methods: tuple[GraphInferredEdgeMethodStatus, ...]
    samples: tuple[GraphInferredEdgeSample, ...]

    def to_dict(self) -> dict[str, Any]:
        """Return a stable JSON-serializable representation."""
        return {
            "session_id": self.session_id,
            "total_edges": self.total_edges,
            "method_count": self.method_count,
            "evidence_count": self.evidence_count,
            "missing_evidence_count": self.missing_evidence_count,
            "missing_source_event_count": self.missing_source_event_count,
            "evidence_coverage": self.evidence_coverage,
            "methods": [method.to_dict() for method in self.methods],
            "samples": [sample.to_dict() for sample in self.samples],
        }


class GraphStore(ProjectionStore):
    """Async Neo4j wrapper for bi-temporal knowledge graph operations.

    Args:
        uri: Bolt URI, e.g. 'bolt://localhost:7687'.
        user: Neo4j username.
        password: Neo4j password.
    """

    def __init__(
        self,
        uri: str,
        user: str,
        password: str,
        ca_cert: str | None = None,
        trust_all: bool = False,
    ) -> None:
        self._uri = uri
        self._auth = (user, password)
        self._ca_cert = ca_cert
        self._trust_all = trust_all
        self._driver: Any | None = None
        self.last_projection_steps: tuple[str, ...] = ()

    async def connect(self) -> None:
        """Initialize the async driver with optional TLS."""
        async_graph_database, trust_all_cls, trust_custom_cas_cls = _load_async_neo4j()
        kwargs: dict[str, Any] = {"auth": self._auth}
        if self._trust_all:
            kwargs["encrypted"] = True
            kwargs["trusted_certificates"] = trust_all_cls()
        elif self._ca_cert:
            if not Path(self._ca_cert).exists():
                raise ValueError(
                    f"NEO4J_CA_CERT does not exist: {self._ca_cert}. "
                    "Unset NEO4J_CA_CERT for plain local bolt:// Neo4j or provide a valid CA file."
                )
            kwargs["encrypted"] = True
            kwargs["trusted_certificates"] = trust_custom_cas_cls(self._ca_cert)
        self._driver = async_graph_database.driver(self._uri, **kwargs)

    async def close(self) -> None:
        """Close the driver."""
        if self._driver:
            await self._driver.close()
            self._driver = None

    async def has_traversal_edges(self, session_id: str = "default") -> bool:
        """Return whether a session has active RELATES edges for traversal."""
        assert self._driver is not None, "Call connect() first"
        records, _, _ = await self._driver.execute_query(
            """
            MATCH (source:Entity {session_id: $session_id})-[r:RELATES]->(target:Entity {session_id: $session_id})
            WHERE r.valid_to IS NULL
              AND source.valid_to IS NULL
              AND target.valid_to IS NULL
            RETURN true AS has_edges
            LIMIT 1
            """,
            session_id=validate_session_id(session_id),
        )
        return bool(records)

    # ------------------------------------------------------------------
    # Schema
    # ------------------------------------------------------------------

    async def init_schema(self) -> None:
        """Create constraints and indexes idempotently.

        Must be called once before ingestion.
        """
        assert self._driver is not None, "Call connect() first"
        await apply_schema_migrations(self._driver)

    async def inspect_event_projection_status(
        self,
        session_id: str,
        *,
        eventloom_latest_seq: int | None = None,
        eventloom_latest_hash: str | None = None,
    ) -> GraphEventProjectionStatus:
        """Inspect Event nodes and hash-chain edges for one projected session."""
        assert self._driver is not None
        safe_session_id = validate_session_id(session_id)
        records, _, _ = await self._driver.execute_query(
            """
            MATCH (e:Event {session_id: $session_id})
            WITH count(e) AS event_count, max(e.seq) AS latest_seq
            OPTIONAL MATCH (latest:Event {session_id: $session_id, seq: latest_seq})
            WITH event_count, latest_seq, latest.hash AS latest_hash
            CALL () {
                MATCH (:Event {session_id: $session_id})-[n:NEXT_EVENT]->(:Event {session_id: $session_id})
                RETURN count(n) AS next_event_edges
            }
            CALL () {
                MATCH (:Event {session_id: $session_id})-[p:PREVIOUS_EVENT]->(:Event {session_id: $session_id})
                RETURN count(p) AS previous_event_edges
            }
            CALL () {
                MATCH (e:Event {session_id: $session_id})
                WHERE e.seq > 1
                OPTIONAL MATCH (prev:Event {session_id: $session_id, hash: e.prev_hash})
                OPTIONAL MATCH (prev)-[n:NEXT_EVENT]->(e)
                OPTIONAL MATCH (e)-[p:PREVIOUS_EVENT]->(prev)
                WITH e, prev, n, p
                WHERE prev IS NULL OR n IS NULL OR p IS NULL
                RETURN count(e) AS missing_chain_links
            }
            RETURN event_count,
                   latest_seq,
                   latest_hash,
                   next_event_edges,
                   previous_event_edges,
                   missing_chain_links
            """,
            session_id=safe_session_id,
        )
        record = records[0] if records else {}
        event_count = _int_record_value(record.get("event_count"))
        latest_seq = _optional_int_record_value(record.get("latest_seq"))
        latest_hash = _optional_str_record_value(record.get("latest_hash"))
        next_event_edges = _int_record_value(record.get("next_event_edges"))
        previous_event_edges = _int_record_value(record.get("previous_event_edges"))
        missing_chain_links = _int_record_value(record.get("missing_chain_links"))
        projection_lag = (
            max(0, eventloom_latest_seq - (latest_seq or 0))
            if eventloom_latest_seq is not None
            else None
        )
        latest_hash_matches = (
            latest_hash == eventloom_latest_hash
            if eventloom_latest_hash is not None
            else True
        )
        integrity_ok = (
            missing_chain_links == 0
            and latest_hash_matches
            and (projection_lag in (None, 0))
            and next_event_edges == max(0, event_count - 1)
            and previous_event_edges == max(0, event_count - 1)
        )
        return GraphEventProjectionStatus(
            session_id=safe_session_id,
            event_count=event_count,
            latest_seq=latest_seq,
            latest_hash=latest_hash,
            eventloom_latest_seq=eventloom_latest_seq,
            eventloom_latest_hash=eventloom_latest_hash,
            projection_lag=projection_lag,
            latest_hash_matches=latest_hash_matches,
            next_event_edges=next_event_edges,
            previous_event_edges=previous_event_edges,
            missing_chain_links=missing_chain_links,
            integrity_ok=integrity_ok,
        )

    async def inspect_inferred_edge_status(
        self,
        session_id: str,
        *,
        limit: int = 10,
    ) -> GraphInferredEdgeStatus:
        """Inspect inferred relationships by method, confidence, and evidence coverage."""
        assert self._driver is not None
        safe_session_id = validate_session_id(session_id)
        safe_limit = validate_limit(limit)
        method_records, _, _ = await self._driver.execute_query(
            """
            MATCH (:Entity)-[r:RELATES]->(:Entity)
            WHERE r.session_id = $session_id
              AND r.inferred = true
            WITH coalesce(r.inference_method, 'unknown') AS method,
                 r,
                 [key IN keys(r) WHERE key STARTS WITH 'evidence_'] AS evidence_keys
            WITH method,
                 count(r) AS edge_count,
                 collect(DISTINCT r.relation_type) AS relation_types,
                 avg(r.confidence) AS average_confidence,
                 min(r.confidence) AS minimum_confidence,
                 sum(CASE WHEN size(evidence_keys) > 0 THEN 1 ELSE 0 END) AS evidence_count,
                 sum(CASE WHEN size(evidence_keys) = 0 THEN 1 ELSE 0 END) AS missing_evidence_count,
                 sum(CASE
                     WHEN r.source_event_seq IS NULL OR r.source_event_hash IS NULL THEN 1
                     ELSE 0
                 END) AS missing_source_event_count
            RETURN method,
                   edge_count,
                   relation_types,
                   average_confidence,
                   minimum_confidence,
                   evidence_count,
                   missing_evidence_count,
                   missing_source_event_count
            ORDER BY edge_count DESC, method ASC
            """,
            session_id=safe_session_id,
        )
        sample_records, _, _ = await self._driver.execute_query(
            """
            MATCH (source:Entity)-[r:RELATES]->(target:Entity)
            WHERE r.session_id = $session_id
              AND r.inferred = true
            WITH source,
                 target,
                 r,
                 [key IN keys(r) WHERE key STARTS WITH 'evidence_'] AS evidence_keys
            RETURN source.name AS source,
                   target.name AS target,
                   r.relation_type AS relation_type,
                   r.confidence AS confidence,
                   coalesce(r.inference_method, 'unknown') AS method,
                   r.source_event_seq AS source_event_seq,
                   r.source_event_hash AS source_event_hash,
                   evidence_keys
            ORDER BY r.source_event_seq DESC, source.name ASC, target.name ASC
            LIMIT $limit
            """,
            session_id=safe_session_id,
            limit=safe_limit,
        )
        methods = tuple(_record_to_inferred_edge_method(record) for record in method_records)
        samples = tuple(_record_to_inferred_edge_sample(record) for record in sample_records)
        total_edges = sum(method.edge_count for method in methods)
        evidence_count = sum(method.evidence_count for method in methods)
        missing_evidence_count = sum(method.missing_evidence_count for method in methods)
        missing_source_event_count = sum(method.missing_source_event_count for method in methods)
        evidence_coverage = evidence_count / total_edges if total_edges else 1.0
        return GraphInferredEdgeStatus(
            session_id=safe_session_id,
            total_edges=total_edges,
            method_count=len(methods),
            evidence_count=evidence_count,
            missing_evidence_count=missing_evidence_count,
            missing_source_event_count=missing_source_event_count,
            evidence_coverage=evidence_coverage,
            methods=methods,
            samples=samples,
        )

    # ------------------------------------------------------------------
    # Ingestion
    # ------------------------------------------------------------------

    async def upsert_extraction(
        self,
        result: ExtractionResult,
        session_id: str = "default",
    ) -> None:
        """Project an ExtractionResult into the graph.

        Entities are versioned by (name, type, valid_from). Reasserting an
        entity closes the version that was current at that time and preserves
        the historical node. Edges are merged by (source, target, type,
        valid_from) so that re-ingestion is idempotent.
        """
        assert self._driver is not None
        safe_session_id = validate_session_id(session_id)
        observed_at = _extraction_observed_at(result)
        projection_steps: list[str] = []

        await self._upsert_provenance_backbone(
            result,
            session_id=safe_session_id,
            observed_at=observed_at,
            projection_steps=projection_steps,
        )

        if result.source_event_type == "projection.retired":
            source_path = _retired_source_path(result)
            if source_path and observed_at:
                await self.retire_source_projections(
                    source_path=source_path,
                    invalid_at=observed_at,
                    session_id=safe_session_id,
                )
                projection_steps.append("source_projection_retirement")

        for ent in result.entities:
            await self._upsert_entity_version_projection(
                ent,
                result,
                session_id=safe_session_id,
                projection_steps=projection_steps,
            )

        for edge in result.edges:
            await self._upsert_relationship_projection(
                edge,
                result,
                session_id=safe_session_id,
                projection_steps=projection_steps,
            )
        self.last_projection_steps = tuple(projection_steps)

    async def _upsert_provenance_backbone(
        self,
        result: ExtractionResult,
        *,
        session_id: str,
        observed_at: str | None,
        projection_steps: list[str],
    ) -> None:
        """Project session, event, and hash-chain provenance links."""
        await self._execute_projection_statement(
            "provenance_backbone",
            projection_steps,
            """
            MERGE (s:Session {id: $session_id})
            ON CREATE SET s.created_at = datetime()
            SET s.updated_at = datetime()
            MERGE (ev:Event {session_id: $session_id, seq: $source_event_seq})
            ON CREATE SET ev.created_at = datetime()
            SET ev.updated_at = datetime(),
                ev.hash = $source_event_hash,
                ev.prev_hash = $source_event_prev_hash,
                ev.type = $source_event_type,
                ev.thread = $source_thread,
                ev.observed_at = CASE
                    WHEN $observed_at IS NULL THEN ev.observed_at
                    ELSE datetime($observed_at)
                END
            MERGE (s)-[r:HAS_EVENT]->(ev)
            ON CREATE SET r.created_at = datetime()
            SET r.updated_at = datetime()
            WITH ev
            CALL (ev) {
                MATCH (prev:Event {session_id: $session_id, hash: $source_event_prev_hash})
                MERGE (prev)-[next:NEXT_EVENT]->(ev)
                ON CREATE SET next.created_at = datetime()
                SET next.updated_at = datetime(),
                    next.session_id = $session_id,
                    next.from_hash = $source_event_prev_hash,
                    next.to_hash = $source_event_hash
                MERGE (ev)-[previous:PREVIOUS_EVENT]->(prev)
                ON CREATE SET previous.created_at = datetime()
                SET previous.updated_at = datetime(),
                    previous.session_id = $session_id,
                    previous.from_hash = $source_event_hash,
                    previous.to_hash = $source_event_prev_hash
                RETURN count(prev) AS previous_event_links
            }
            WITH ev, previous_event_links
            CALL (ev) {
                MATCH (next_event:Event {session_id: $session_id, prev_hash: $source_event_hash})
                MERGE (ev)-[next_rel:NEXT_EVENT]->(next_event)
                ON CREATE SET next_rel.created_at = datetime()
                SET next_rel.updated_at = datetime(),
                    next_rel.session_id = $session_id,
                    next_rel.from_hash = $source_event_hash,
                    next_rel.to_hash = next_event.hash
                MERGE (next_event)-[previous_rel:PREVIOUS_EVENT]->(ev)
                ON CREATE SET previous_rel.created_at = datetime()
                SET previous_rel.updated_at = datetime(),
                    previous_rel.session_id = $session_id,
                    previous_rel.from_hash = next_event.hash,
                    previous_rel.to_hash = $source_event_hash
                RETURN count(next_event) AS next_event_links
            }
            RETURN previous_event_links, next_event_links
            """,
            session_id=session_id,
            source_event_seq=result.source_event_seq,
            source_event_hash=result.source_event_hash,
            source_event_prev_hash=result.source_event_prev_hash,
            source_event_type=result.source_event_type,
            source_thread=result.source_thread,
            observed_at=observed_at,
        )

    async def _upsert_entity_version_projection(
        self,
        entity: ExtractedEntity,
        result: ExtractionResult,
        *,
        session_id: str,
        projection_steps: list[str],
    ) -> None:
        """Project one temporal entity version plus provenance and citations."""
        await self._execute_projection_statement(
            "entity_version",
            projection_steps,
            """
            MERGE (e:Entity {
                session_id: $session_id,
                name: $name,
                entity_type: $entity_type,
                valid_from: datetime($observed_at)
            })
            ON CREATE SET e.created_at = datetime($observed_at)
            SET e.updated_at = datetime($observed_at),
                e.summary = coalesce($summary, e.summary),
                e.embedding = coalesce($embedding, e.embedding),
                e.source_event_seq = $source_event_seq,
                e.source_event_hash = $source_event_hash,
                e.source_event_type = $source_event_type,
                e.source_thread = $source_thread
            SET e += $properties
            WITH e
            OPTIONAL MATCH (prev:Entity {name: $name, entity_type: $entity_type})
            WHERE prev.session_id = $session_id
              AND prev.valid_from < e.valid_from
              AND (prev.valid_to IS NULL OR prev.valid_to > e.valid_from)
            SET e.summary = coalesce(e.summary, prev.summary),
                e.taskId = coalesce(e.taskId, prev.taskId),
                e.task_id = coalesce(e.task_id, prev.task_id),
                e.userId = coalesce(e.userId, prev.userId),
                e.user_id = coalesce(e.user_id, prev.user_id),
                e.goalTitle = coalesce(e.goalTitle, prev.goalTitle),
                e.goal_title = coalesce(e.goal_title, prev.goal_title)
            SET prev.valid_to = e.valid_from,
                prev.updated_at = datetime($observed_at)
            WITH e
            OPTIONAL MATCH (next:Entity {name: $name, entity_type: $entity_type})
            WHERE next.session_id = $session_id
              AND next.valid_from > e.valid_from
            WITH e, min(next.valid_from) AS next_valid_from
            SET e.valid_to = next_valid_from
            WITH e
            CALL (e) {
                OPTIONAL MATCH (prev:Entity {session_id: $session_id, name: $name, entity_type: $entity_type})
                WHERE prev.valid_to = e.valid_from
                WITH e, prev
                WHERE prev IS NOT NULL
                MERGE (prev)-[sb:SUPERSEDED_BY]->(e)
                ON CREATE SET sb.created_at = datetime($observed_at)
                SET sb.updated_at = datetime($observed_at),
                    sb.session_id = $session_id,
                    sb.valid_from = e.valid_from,
                    sb.source_event_seq = $source_event_seq,
                    sb.source_event_hash = $source_event_hash,
                    sb.source_event_type = $source_event_type,
                    sb.source_thread = $source_thread
                MERGE (e)-[pv:PREVIOUS_VERSION]->(prev)
                ON CREATE SET pv.created_at = datetime($observed_at)
                SET pv.updated_at = datetime($observed_at),
                    pv.session_id = $session_id,
                    pv.valid_from = e.valid_from,
                    pv.source_event_seq = $source_event_seq,
                    pv.source_event_hash = $source_event_hash,
                    pv.source_event_type = $source_event_type,
                    pv.source_thread = $source_thread
                RETURN count(prev) AS linked_previous_versions
            }
            WITH e
            CALL (e) {
                OPTIONAL MATCH (next:Entity {session_id: $session_id, name: $name, entity_type: $entity_type})
                WHERE e.valid_to = next.valid_from
                WITH e, next
                WHERE next IS NOT NULL
                MERGE (e)-[next_sb:SUPERSEDED_BY]->(next)
                ON CREATE SET next_sb.created_at = datetime($observed_at)
                SET next_sb.updated_at = datetime($observed_at),
                    next_sb.session_id = $session_id,
                    next_sb.valid_from = next.valid_from,
                    next_sb.source_event_seq = $source_event_seq,
                    next_sb.source_event_hash = $source_event_hash,
                    next_sb.source_event_type = $source_event_type,
                    next_sb.source_thread = $source_thread
                MERGE (next)-[next_pv:PREVIOUS_VERSION]->(e)
                ON CREATE SET next_pv.created_at = datetime($observed_at)
                SET next_pv.updated_at = datetime($observed_at),
                    next_pv.session_id = $session_id,
                    next_pv.valid_from = next.valid_from,
                    next_pv.source_event_seq = $source_event_seq,
                    next_pv.source_event_hash = $source_event_hash,
                    next_pv.source_event_type = $source_event_type,
                    next_pv.source_thread = $source_thread
                RETURN count(next) AS linked_next_versions
            }
            WITH e
            CALL (e) {
                OPTIONAL MATCH (prev:Entity {session_id: $session_id, name: $name, entity_type: $entity_type})
                WHERE prev.valid_to = e.valid_from
                OPTIONAL MATCH (source:Entity)-[old_in:RELATES]->(prev)
                WHERE old_in.session_id = $session_id
                  AND old_in.valid_to IS NULL
                WITH e, source, old_in
                WHERE old_in IS NOT NULL
                MERGE (source)-[new_in:RELATES {
                    relation_type: old_in.relation_type,
                    valid_from: old_in.valid_from
                }]->(e)
                ON CREATE SET new_in.created_at = datetime($observed_at)
                SET new_in.session_id = $session_id,
                    new_in.valid_to = null,
                    new_in.source_event_seq = old_in.source_event_seq,
                    new_in.source_event_hash = old_in.source_event_hash,
                    new_in.source_event_type = old_in.source_event_type,
                    new_in.source_thread = old_in.source_thread
                SET new_in.inferred = coalesce(old_in.inferred, false),
                    new_in.confidence = coalesce(old_in.confidence, 1.0),
                    new_in.inference_method = old_in.inference_method
                RETURN count(new_in) AS copied_incoming_relationships
            }
            WITH e
            CALL (e) {
                OPTIONAL MATCH (prev:Entity {session_id: $session_id, name: $name, entity_type: $entity_type})
                WHERE prev.valid_to = e.valid_from
                OPTIONAL MATCH (prev)-[old_out:RELATES]->(target:Entity)
                WHERE old_out.session_id = $session_id
                  AND old_out.valid_to IS NULL
                WITH e, target, old_out
                WHERE old_out IS NOT NULL
                MERGE (e)-[new_out:RELATES {
                    relation_type: old_out.relation_type,
                    valid_from: old_out.valid_from
                }]->(target)
                ON CREATE SET new_out.created_at = datetime($observed_at)
                SET new_out.session_id = $session_id,
                    new_out.valid_to = null,
                    new_out.source_event_seq = old_out.source_event_seq,
                    new_out.source_event_hash = old_out.source_event_hash,
                    new_out.source_event_type = old_out.source_event_type,
                    new_out.source_thread = old_out.source_thread
                SET new_out.inferred = coalesce(old_out.inferred, false),
                    new_out.confidence = coalesce(old_out.confidence, 1.0),
                    new_out.inference_method = old_out.inference_method
                RETURN count(new_out) AS copied_outgoing_relationships
            }
            WITH e
            MATCH (ev:Event {session_id: $session_id, seq: $source_event_seq})
            MERGE (ev)-[pe:PROJECTED_ENTITY {
                name: $name,
                entity_type: $entity_type,
                valid_from: datetime($observed_at)
            }]->(e)
            ON CREATE SET pe.created_at = datetime($observed_at)
            SET pe.source_event_hash = $source_event_hash,
                pe.source_event_type = $source_event_type
            WITH e, ev
            CALL (e, ev) {
                WITH e, ev
                WITH e, ev WHERE $source_path IS NOT NULL
                MERGE (src:Source {session_id: $session_id, path: $source_path})
                ON CREATE SET src.created_at = datetime($observed_at)
                SET src.updated_at = datetime($observed_at),
                    src.sha256 = coalesce($source_sha256, src.sha256)
                MERGE (e)-[cs:CITES_SOURCE]->(src)
                ON CREATE SET cs.created_at = datetime($observed_at)
                SET cs.updated_at = datetime($observed_at),
                    cs.source_start_line = $source_start_line,
                    cs.source_end_line = $source_end_line,
                    cs.source_sha256 = $source_sha256,
                    cs.source_event_seq = $source_event_seq,
                    cs.source_event_hash = $source_event_hash,
                    cs.source_event_type = $source_event_type
                MERGE (ev)-[ecs:CITES_SOURCE]->(src)
                ON CREATE SET ecs.created_at = datetime($observed_at)
                SET ecs.updated_at = datetime($observed_at),
                    ecs.source_start_line = $source_start_line,
                    ecs.source_end_line = $source_end_line,
                    ecs.source_sha256 = $source_sha256,
                    ecs.source_event_seq = $source_event_seq,
                    ecs.source_event_hash = $source_event_hash,
                    ecs.source_event_type = $source_event_type
                RETURN count(src) AS cited_sources
            }
            RETURN count(e) AS projected_entities, sum(cited_sources) AS cited_sources
            """,
            session_id=session_id,
            name=entity.name,
            entity_type=entity.entity_type,
            observed_at=entity.observed_at,
            summary=entity.summary,
            embedding=entity.embedding,
            source_event_seq=result.source_event_seq,
            source_event_hash=result.source_event_hash,
            source_event_type=result.source_event_type,
            source_thread=result.source_thread,
            properties=_neo4j_properties(entity.properties),
            **_source_citation_properties(entity.properties),
        )

    async def _upsert_relationship_projection(
        self,
        edge: ExtractedEdge,
        result: ExtractionResult,
        *,
        session_id: str,
        projection_steps: list[str],
    ) -> None:
        """Project one temporal relation plus typed and provenance edges."""
        typed_relationship_label = _typed_relationship_label(edge.relation_type)
        await self._execute_projection_statement(
            "relationship",
            projection_steps,
            f"""
            MATCH (s:Entity {{name: $source}})
            WHERE s.session_id = $session_id
              AND s.valid_from <= datetime($valid_from)
              AND (s.valid_to IS NULL OR s.valid_to > datetime($valid_from))
            MATCH (t:Entity {{name: $target}})
            WHERE t.session_id = $session_id
              AND t.valid_from <= datetime($valid_from)
              AND (t.valid_to IS NULL OR t.valid_to > datetime($valid_from))
            MERGE (s)-[r:RELATES {{relation_type: $relation_type, valid_from: datetime($valid_from)}}]->(t)
            ON CREATE SET r.created_at = datetime($valid_from)
            SET r.session_id = $session_id,
                r.valid_to = CASE
                    WHEN $valid_to IS NULL THEN null
                    ELSE datetime($valid_to)
                END,
                r.source_event_seq = $source_event_seq,
                r.source_event_hash = $source_event_hash,
                r.source_event_type = $source_event_type,
                r.source_thread = $source_thread,
                r.inferred = $inferred,
                r.confidence = $confidence,
                r.inference_method = $inference_method
            SET r += $edge_properties
            MERGE (s)-[typed:{typed_relationship_label} {{valid_from: datetime($valid_from)}}]->(t)
            ON CREATE SET typed.created_at = datetime($valid_from)
            SET typed.session_id = $session_id,
                typed.relation_type = $relation_type,
                typed.valid_to = CASE
                    WHEN $valid_to IS NULL THEN null
                    ELSE datetime($valid_to)
                END,
                typed.source_event_seq = $source_event_seq,
                typed.source_event_hash = $source_event_hash,
                typed.source_event_type = $source_event_type,
                typed.source_thread = $source_thread,
                typed.inferred = $inferred,
                typed.confidence = $confidence,
                typed.inference_method = $inference_method
            SET typed += $edge_properties
            WITH s, t, r, typed
            MATCH (ev:Event {{session_id: $session_id, seq: $source_event_seq}})
            MERGE (ev)-[pr:PROJECTED_RELATION {{
                source: $source,
                target: $target,
                relation_type: $relation_type,
                valid_from: datetime($valid_from)
            }}]->(t)
            ON CREATE SET pr.created_at = datetime($valid_from)
            SET pr.source_event_hash = $source_event_hash,
                pr.source_event_type = $source_event_type,
                pr.inferred = $inferred,
                pr.confidence = $confidence,
                pr.inference_method = $inference_method
            SET pr += $edge_properties
            """,
            session_id=session_id,
            source=edge.source,
            target=edge.target,
            relation_type=edge.relation_type,
            valid_from=edge.valid_from,
            valid_to=edge.valid_to,
            source_event_seq=result.source_event_seq,
            source_event_hash=result.source_event_hash,
            source_event_type=result.source_event_type,
            source_thread=result.source_thread,
            inferred=edge.inferred,
            confidence=edge.confidence,
            inference_method=edge.inference_method,
            edge_properties=_edge_evidence_properties(edge.evidence),
        )

    async def _execute_projection_statement(
        self,
        step: str,
        projection_steps: list[str],
        cypher: str,
        **parameters: Any,
    ) -> Any:
        """Execute one named projection phase and record completed step order."""
        assert self._driver is not None
        result = await self._driver.execute_query(cypher, **parameters)
        projection_steps.append(step)
        return result

    async def invalidate_entity(
        self,
        name: str,
        entity_type: str,
        invalid_at: str,
        session_id: str = "default",
    ) -> None:
        """Mark an entity as invalid after a given time (bi-temporal update)."""
        assert self._driver is not None
        safe_session_id = validate_session_id(session_id)
        await self._driver.execute_query(
            """
            MATCH (e:Entity {name: $name, entity_type: $entity_type})
            WHERE e.session_id = $session_id
              AND e.valid_to IS NULL
            SET e.valid_to = datetime($invalid_at)
            """,
            session_id=safe_session_id,
            name=name,
            entity_type=entity_type,
            invalid_at=invalid_at,
        )

    async def invalidate_edge(
        self,
        source: str,
        target: str,
        relation_type: str,
        valid_from: str,
        invalid_at: str,
        session_id: str = "default",
    ) -> None:
        """Mark an edge as invalid after a given time."""
        assert self._driver is not None
        safe_session_id = validate_session_id(session_id)
        await self._driver.execute_query(
            """
            MATCH (s:Entity {name: $source})-[r:RELATES {relation_type: $relation_type, valid_from: datetime($valid_from)}]->(t:Entity {name: $target})
            WHERE s.session_id = $session_id
              AND t.session_id = $session_id
              AND r.session_id = $session_id
              AND r.valid_to IS NULL
            SET r.valid_to = datetime($invalid_at)
            """,
            session_id=safe_session_id,
            source=source,
            target=target,
            relation_type=relation_type,
            valid_from=valid_from,
            invalid_at=invalid_at,
        )

    async def retire_source_projections(
        self,
        *,
        source_path: str,
        invalid_at: str,
        session_id: str = "default",
    ) -> None:
        """Expire active entities and relationships derived from a source path."""
        assert self._driver is not None
        safe_session_id = validate_session_id(session_id)
        await self._driver.execute_query(
            """
            MATCH (e:Entity)
            WHERE e.session_id = $session_id
              AND e.valid_to IS NULL
              AND (
                e.source_path = $source_path
                OR e.target_path = $source_path
                OR e.test_path = $source_path
                OR e.covered_path = $source_path
              )
            SET e.valid_to = datetime($invalid_at),
                e.updated_at = datetime($invalid_at)
            """,
            session_id=safe_session_id,
            source_path=source_path,
            invalid_at=invalid_at,
        )
        await self._driver.execute_query(
            """
            MATCH (source:Entity)-[r]->(target:Entity)
            WHERE r.session_id = $session_id
              AND r.valid_to IS NULL
              AND (
                source.source_path = $source_path
                OR source.target_path = $source_path
                OR source.test_path = $source_path
                OR source.covered_path = $source_path
                OR target.source_path = $source_path
                OR target.target_path = $source_path
                OR target.test_path = $source_path
                OR target.covered_path = $source_path
              )
            SET r.valid_to = datetime($invalid_at),
                r.updated_at = datetime($invalid_at)
            """,
            session_id=safe_session_id,
            source_path=source_path,
            invalid_at=invalid_at,
        )

    # ------------------------------------------------------------------
    # Retrieval
    # ------------------------------------------------------------------

    async def search_exact(
        self,
        name: str,
        entity_type: str | None = None,
        temporal_point: str | None = None,
        session_id: str = "default",
    ) -> list[GraphEntity]:
        """Exact-match lookup by name, optionally filtered by type and time."""
        assert self._driver is not None

        cypher = "MATCH (e:Entity {name: $name})"
        params: dict[str, Any] = {
            "name": name,
            "session_id": validate_session_id(session_id),
        }
        where_clauses: list[str] = ["e.session_id = $session_id"]

        if entity_type:
            where_clauses.append("e.entity_type = $entity_type")
            params["entity_type"] = entity_type

        if temporal_point:
            where_clauses.append(
                "e.valid_from <= datetime($t) AND (e.valid_to IS NULL OR e.valid_to > datetime($t))"
            )
            params["t"] = temporal_point
        else:
            where_clauses.append("e.valid_to IS NULL")

        if where_clauses:
            cypher += " WHERE " + " AND ".join(where_clauses)

        cypher += " RETURN e"

        records, _, _ = await self._driver.execute_query(cypher, **params)
        return [_record_to_entity(r["e"]) for r in records]

    async def search_traversal(
        self,
        start_name: str,
        relation_type: str | None = None,
        depth: int = 2,
        temporal_point: str | None = None,
        session_id: str = "default",
    ) -> list[GraphEntity]:
        """Graph traversal from a starting entity, optionally filtered."""
        assert self._driver is not None
        depth = validate_traversal_depth(depth)

        params: dict[str, Any] = {
            "start_name": start_name,
            "session_id": validate_session_id(session_id),
            "relation_type": relation_type,
        }

        temporal_checks = " AND rel.valid_to IS NULL AND ($relation_type IS NULL OR rel.relation_type = $relation_type)"
        entity_checks = (
            "start.session_id = $session_id"
            " AND neighbor.session_id = $session_id"
            " AND start.valid_to IS NULL"
            " AND neighbor.valid_to IS NULL"
        )
        if temporal_point:
            temporal_checks = (
                " AND rel.valid_from <= datetime($t)"
                " AND (rel.valid_to IS NULL OR rel.valid_to > datetime($t))"
                " AND ($relation_type IS NULL OR rel.relation_type = $relation_type)"
            )
            entity_checks = (
                "start.session_id = $session_id"
                " AND neighbor.session_id = $session_id"
                " AND start.valid_from <= datetime($t)"
                " AND (start.valid_to IS NULL OR start.valid_to > datetime($t))"
                " AND neighbor.valid_from <= datetime($t)"
                " AND (neighbor.valid_to IS NULL OR neighbor.valid_to > datetime($t))"
            )
            params["t"] = temporal_point

        cypher = f"""
        MATCH path = (start:Entity {{name: $start_name}})-[r:RELATES*1..{depth}]-(neighbor:Entity)
        WHERE {entity_checks}
          AND neighbor <> start
          AND ALL(rel IN relationships(path) WHERE true{temporal_checks})
        RETURN neighbor,
               [rel IN relationships(path) | rel.relation_type] AS path_relation_types,
               [rel IN relationships(path) | coalesce(rel.inferred, false)] AS path_inferred_flags,
               [rel IN relationships(path)
                    WHERE coalesce(rel.inferred, false)
                    | coalesce(rel.confidence, 0.0)] AS path_inferred_confidences,
               [rel IN relationships(path)
                    WHERE coalesce(rel.inferred, false)
                    | coalesce(properties(rel)['inference_method'], 'unknown')] AS path_inference_methods,
               [rel IN relationships(path)
                    WHERE coalesce(rel.inferred, false)
                      AND rel.source_event_seq IS NOT NULL
                      AND rel.source_event_hash IS NOT NULL
                    | rel.source_event_seq] AS path_inferred_source_event_refs,
               [rel IN relationships(path)
                    WHERE coalesce(rel.inferred, false)
                    | size([key IN keys(rel) WHERE key STARTS WITH 'evidence_'])] AS path_inferred_evidence_counts,
               length(path) AS path_length
        ORDER BY path_length ASC
        """

        records, _, _ = await self._driver.execute_query(cypher, **params)
        entities: list[GraphEntity] = []
        seen: set[tuple[str, str]] = set()
        for record in records:
            entity = _record_to_entity(record["neighbor"])
            key = (entity.name, entity.entity_type)
            if key in seen:
                continue
            seen.add(key)
            properties = {
                **entity.properties,
                "_path_relation_types": list(record.get("path_relation_types") or []),
                "_path_length": int(record.get("path_length") or 0),
                **_traversal_inferred_metadata(record),
            }
            entities.append(
                GraphEntity(
                    name=entity.name,
                    entity_type=entity.entity_type,
                    valid_from=entity.valid_from,
                    valid_to=entity.valid_to,
                    properties=properties,
                    session_id=entity.session_id,
                )
            )
        return entities

    async def search_causal_neighbors(
        self,
        entity_name: str,
        *,
        direction: Literal["successors", "predecessors"],
        relation_type: str | None = None,
        depth: int = 2,
        temporal_point: str | None = None,
        session_id: str = "default",
    ) -> list[GraphEntity]:
        """Search directed causal neighbors from a starting entity."""
        assert self._driver is not None
        if direction not in {"successors", "predecessors"}:
            raise ValueError("direction must be 'successors' or 'predecessors'")
        depth = validate_traversal_depth(depth)
        safe_session_id = validate_session_id(session_id)
        params: dict[str, Any] = {
            "entity_name": entity_name,
            "session_id": safe_session_id,
            "relation_type": relation_type,
        }
        relationship_checks = (
            "rel.session_id = $session_id"
            " AND rel.valid_to IS NULL"
            " AND rel.relation_type STARTS WITH 'causal_'"
            " AND ($relation_type IS NULL OR rel.relation_type = $relation_type)"
        )
        node_checks = "node.session_id = $session_id AND node.valid_to IS NULL"
        if temporal_point:
            params["t"] = temporal_point
            relationship_checks = (
                "rel.session_id = $session_id"
                " AND rel.valid_from <= datetime($t)"
                " AND (rel.valid_to IS NULL OR rel.valid_to > datetime($t))"
                " AND rel.relation_type STARTS WITH 'causal_'"
                " AND ($relation_type IS NULL OR rel.relation_type = $relation_type)"
            )
            node_checks = (
                "node.session_id = $session_id"
                " AND node.valid_from <= datetime($t)"
                " AND (node.valid_to IS NULL OR node.valid_to > datetime($t))"
            )

        pattern = (
            f"(start:Entity {{name: $entity_name}})-[r:RELATES*1..{depth}]->(neighbor:Entity)"
            if direction == "successors"
            else f"(neighbor:Entity)-[r:RELATES*1..{depth}]->(start:Entity {{name: $entity_name}})"
        )
        terminal_rel_expression = (
            "last(relationships(path))" if direction == "successors" else "first(relationships(path))"
        )
        cypher = f"""
        MATCH path = {pattern}
        WHERE neighbor <> start
          AND ALL(rel IN relationships(path) WHERE {relationship_checks})
          AND ALL(node IN nodes(path) WHERE {node_checks})
        WITH path, start, neighbor, {terminal_rel_expression} AS terminal_rel
        RETURN neighbor,
               startNode(terminal_rel) AS causal_source,
               endNode(terminal_rel) AS causal_target,
               terminal_rel.relation_type AS graph_relation_type,
               terminal_rel.confidence AS confidence,
               terminal_rel.inference_method AS inference_method,
               terminal_rel.source_event_seq AS source_event_seq,
               terminal_rel.source_event_hash AS source_event_hash,
               properties(terminal_rel) AS edge_properties,
               [rel IN relationships(path) | rel.relation_type] AS path_relation_types,
               [rel IN relationships(path)
                    WHERE rel.source_event_seq IS NOT NULL AND rel.source_event_hash IS NOT NULL
                    | {{seq: rel.source_event_seq, hash: rel.source_event_hash}}] AS path_citations,
               length(path) AS path_length
        ORDER BY path_length ASC
        """
        records, _, _ = await self._driver.execute_query(cypher, **params)
        entities: list[GraphEntity] = []
        seen: set[tuple[str, str]] = set()
        for record in records:
            entity = _record_to_entity(record["neighbor"])
            key = (entity.name, entity.entity_type)
            if key in seen:
                continue
            seen.add(key)
            edge_properties = dict(record.get("edge_properties") or {})
            properties = {
                **entity.properties,
                **_causal_edge_metadata_from_record(record, session_id=safe_session_id, edge_properties=edge_properties),
                "_path_relation_types": list(record.get("path_relation_types") or []),
                "_path_citations": _path_citations_from_record(record, safe_session_id),
                "_path_length": int(record.get("path_length") or 0),
            }
            entities.append(
                GraphEntity(
                    name=entity.name,
                    entity_type=entity.entity_type,
                    valid_from=entity.valid_from,
                    valid_to=entity.valid_to,
                    properties=properties,
                    session_id=entity.session_id,
                )
            )
        return entities

    async def search_keyword(
        self,
        query: str,
        limit: int = 10,
        temporal_point: str | None = None,
        session_id: str = "default",
    ) -> list[SearchResult]:
        """BM25 full-text search over entity names and summaries."""
        assert self._driver is not None
        if limit <= 0:
            return []
        limit = validate_limit(limit)

        cypher = "CALL db.index.fulltext.queryNodes('entity_fulltext', $query) YIELD node, score"
        params: dict[str, Any] = {
            "query": query,
            "limit": limit,
            "session_id": validate_session_id(session_id),
        }

        if temporal_point:
            cypher += (
                " WHERE node.session_id = $session_id"
                " AND (node.valid_from <= datetime($t) AND (node.valid_to IS NULL OR node.valid_to > datetime($t)))"
            )
            params["t"] = temporal_point
        else:
            cypher += " WHERE node.session_id = $session_id AND node.valid_to IS NULL"

        cypher += " RETURN node, score LIMIT $limit"

        records, _, _ = await self._driver.execute_query(cypher, **params)
        results = []
        for record in records:
            score = float(record["score"])
            results.append(SearchResult(entity=_record_to_entity(record["node"]), score=score, raw_score=score, source="keyword"))
        return results

    async def search_vector(
        self,
        embedding: list[float],
        limit: int = 10,
        temporal_point: str | None = None,
        session_id: str = "default",
    ) -> list[SearchResult]:
        """Vector similarity search over entity embeddings."""
        assert self._driver is not None
        if limit <= 0:
            return []
        if not vector_has_signal(embedding):
            return []
        limit = validate_limit(limit)

        cypher = """
        CALL db.index.vector.queryNodes('entity_vector', $limit, $embedding)
        YIELD node, score
        """
        params: dict[str, Any] = {
            "embedding": embedding,
            "limit": limit,
            "session_id": validate_session_id(session_id),
        }

        if temporal_point:
            cypher += (
                " WHERE node.session_id = $session_id"
                " AND (node.valid_from <= datetime($t) AND (node.valid_to IS NULL OR node.valid_to > datetime($t)))"
            )
            params["t"] = temporal_point
        else:
            cypher += " WHERE node.session_id = $session_id AND node.valid_to IS NULL"

        cypher += " RETURN node, score"

        records, _, _ = await self._driver.execute_query(cypher, **params)
        results = []
        for record in records:
            score = float(record["score"])
            results.append(SearchResult(entity=_record_to_entity(record["node"]), score=score, raw_score=score, source="vector"))
        return results


# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------

_ENTITY_STORAGE_PROPERTIES = frozenset(
    {
        "created_at",
        "embedding",
        "entity_type",
        "name",
        "session_id",
        "source_event_hash",
        "source_event_seq",
        "source_event_type",
        "source_thread",
        "summary",
        "updated_at",
        "valid_from",
        "valid_to",
    }
)


def _record_to_entity(node: Any) -> GraphEntity:
    """Convert a Neo4j node record to a GraphEntity."""
    props = dict(node)
    return GraphEntity(
        name=props.get("name", ""),
        entity_type=props.get("entity_type", ""),
        valid_from=str(props.get("valid_from", "")),
        valid_to=str(props.get("valid_to")) if props.get("valid_to") else None,
        session_id=props.get("session_id", "default"),
        properties={
            k: v
            for k, v in props.items()
            if k not in {"session_id", "name", "entity_type", "valid_from", "valid_to"}
        },
    )


def _record_to_inferred_edge_method(record: Any) -> GraphInferredEdgeMethodStatus:
    """Convert a Neo4j method aggregate into an inferred-edge status row."""
    relation_types = record.get("relation_types") or []
    safe_relation_types = sorted(str(value) for value in relation_types if value is not None)
    return GraphInferredEdgeMethodStatus(
        method=str(record.get("method") or "unknown"),
        edge_count=_int_record_value(record.get("edge_count")),
        relation_types=tuple(safe_relation_types),
        average_confidence=_optional_float_record_value(record.get("average_confidence")),
        minimum_confidence=_optional_float_record_value(record.get("minimum_confidence")),
        evidence_count=_int_record_value(record.get("evidence_count")),
        missing_evidence_count=_int_record_value(record.get("missing_evidence_count")),
        missing_source_event_count=_int_record_value(record.get("missing_source_event_count")),
    )


def _record_to_inferred_edge_sample(record: Any) -> GraphInferredEdgeSample:
    """Convert a Neo4j inferred relationship record into an audit sample."""
    evidence_keys = record.get("evidence_keys") or []
    safe_evidence_keys = sorted(str(value) for value in evidence_keys if value is not None)
    return GraphInferredEdgeSample(
        source=str(record.get("source") or ""),
        target=str(record.get("target") or ""),
        relation_type=str(record.get("relation_type") or ""),
        confidence=_optional_float_record_value(record.get("confidence")),
        method=str(record.get("method") or "unknown"),
        source_event_seq=_optional_int_record_value(record.get("source_event_seq")),
        source_event_hash=_optional_str_record_value(record.get("source_event_hash")),
        evidence_keys=tuple(safe_evidence_keys),
    )


def _traversal_inferred_metadata(record: Any) -> dict[str, Any]:
    """Return normalized inferred-edge trust metadata for a traversal path."""
    inferred_flags = [flag is True for flag in record.get("path_inferred_flags") or []]
    inferred_edge_count = sum(1 for flag in inferred_flags if flag)
    if inferred_edge_count == 0:
        return {}
    evidence_counts = [
        value
        for value in (_int_record_value(raw) for raw in record.get("path_inferred_evidence_counts") or [])
        if value > 0
    ]
    return {
        "_path_inferred_edge_count": inferred_edge_count,
        "_path_inferred_confidences": [
            value
            for value in (
                _optional_float_record_value(raw)
                for raw in record.get("path_inferred_confidences") or []
            )
            if value is not None
        ],
        "_path_inference_methods": [
            str(method)
            for method in record.get("path_inference_methods") or []
            if isinstance(method, str) and method
        ],
        "_path_inferred_source_event_count": len(record.get("path_inferred_source_event_refs") or []),
        "_path_inferred_evidence_count": sum(evidence_counts),
        "_path_inferred_evidenced_edge_count": len(evidence_counts),
    }


def _causal_edge_metadata_from_record(
    record: Any,
    *,
    session_id: str,
    edge_properties: dict[str, Any],
) -> dict[str, Any]:
    source = _record_to_entity(record["causal_source"])
    target = _record_to_entity(record["causal_target"])
    graph_relation_type = str(record.get("graph_relation_type") or "")
    evidence = _evidence_from_edge_properties(edge_properties)
    source_event_seq = record.get("source_event_seq")
    source_event_hash = _optional_str_record_value(record.get("source_event_hash"))
    return {
        "causal_source_name": source.name,
        "causal_source_type": source.entity_type,
        "causal_target_name": target.name,
        "causal_target_type": target.entity_type,
        "relation_type": graph_relation_type,
        "graph_relation_type": graph_relation_type,
        "causal_relation_type": evidence.get("causal_relation_type") or graph_relation_type.removeprefix("causal_"),
        "confidence": _optional_float_record_value(record.get("confidence")) or 1.0,
        "inference_method": _optional_str_record_value(record.get("inference_method")) or "unknown",
        "citation": _edge_citation(session_id, source_event_seq, source_event_hash),
        "review_status": evidence.get("review_status") or "proposed",
        "authority_status": evidence.get("authority_status") or "non_authoritative",
        "source_event_seq": _optional_int_record_value(source_event_seq),
        "source_event_hash": source_event_hash,
        "evidence": evidence,
        "session_id": session_id,
    }


def _evidence_from_edge_properties(properties: dict[str, Any]) -> dict[str, Any]:
    evidence: dict[str, Any] = {}
    for key, value in properties.items():
        if key.startswith("evidence_"):
            evidence[key.removeprefix("evidence_")] = value
    return evidence


def _path_citations_from_record(record: Any, session_id: str) -> list[str]:
    citations: list[str] = []
    for item in record.get("path_citations") or []:
        if not isinstance(item, dict):
            continue
        citation = _edge_citation(session_id, item.get("seq"), _optional_str_record_value(item.get("hash")))
        if citation.startswith("eventloom://unknown/"):
            continue
        citations.append(citation)
    return citations


def _edge_citation(session_id: str, source_event_seq: Any, source_event_hash: str | None) -> str:
    if source_event_seq is not None and source_event_hash:
        return f"eventloom://{session_id}/events/{source_event_seq}#{source_event_hash[:12]}"
    return "eventloom://unknown/events/unknown#unknown"


def _int_record_value(value: Any) -> int:
    if isinstance(value, int):
        return value
    return 0


def _optional_int_record_value(value: Any) -> int | None:
    if isinstance(value, int):
        return value
    return None


def _optional_str_record_value(value: Any) -> str | None:
    if isinstance(value, str):
        return value
    return None


def _optional_float_record_value(value: Any) -> float | None:
    if isinstance(value, bool):
        return None
    if isinstance(value, int | float):
        return float(value)
    return None


def _neo4j_properties(properties: dict[str, Any] | None) -> dict[str, _Neo4jPropertyValue]:
    """Filter extracted properties to values accepted by Neo4j node properties."""
    if not properties:
        return {}
    safe_properties: dict[str, _Neo4jPropertyValue] = {}
    for key, value in properties.items():
        if not _is_neo4j_property_value(value):
            continue
        safe_key = f"payload_{key}" if key in _ENTITY_STORAGE_PROPERTIES else key
        safe_properties[safe_key] = value
    return safe_properties


def _edge_evidence_properties(evidence: dict[str, Any] | None) -> dict[str, _Neo4jPropertyValue]:
    """Normalize flat inferred-edge evidence into namespaced relationship properties."""
    if not evidence:
        return {}
    safe_properties: dict[str, _Neo4jPropertyValue] = {}
    for key, value in evidence.items():
        if not _PROPERTY_KEY_RE.fullmatch(key):
            continue
        if not _is_neo4j_property_value(value):
            continue
        safe_properties[f"evidence_{key}"] = value
    return safe_properties


def _source_citation_properties(properties: dict[str, Any] | None) -> dict[str, Any]:
    """Normalize source citation fields for first-class graph projections."""
    if not properties:
        return {
            "source_path": None,
            "source_start_line": None,
            "source_end_line": None,
            "source_sha256": None,
        }
    source_path = properties.get("source_path")
    return {
        "source_path": source_path if isinstance(source_path, str) and source_path else None,
        "source_start_line": _optional_positive_int(properties.get("source_start_line")),
        "source_end_line": _optional_positive_int(properties.get("source_end_line")),
        "source_sha256": _optional_text_property(properties.get("source_sha256")),
    }


def _optional_positive_int(value: Any) -> int | None:
    """Return a positive integer property, otherwise None."""
    if isinstance(value, int) and not isinstance(value, bool) and value > 0:
        return value
    return None


def _optional_text_property(value: Any) -> str | None:
    """Return non-empty string property, otherwise None."""
    if not isinstance(value, str):
        return None
    text = value.strip()
    return text or None


def _extraction_observed_at(result: ExtractionResult) -> str | None:
    """Return the earliest available event timestamp carried by an extraction."""
    values = [entity.observed_at for entity in result.entities]
    values.extend(edge.valid_from for edge in result.edges)
    return min(values) if values else None


def _retired_source_path(result: ExtractionResult) -> str | None:
    """Return the source path targeted by a projection retirement event."""
    for entity in result.entities:
        source_path = entity.properties.get("source_path") if entity.properties else None
        if isinstance(source_path, str) and source_path:
            return source_path
    return None


def _typed_relationship_label(relation_type: str) -> str:
    """Convert a safe extractor relation type into a Neo4j relationship label."""
    if not _RELATION_TYPE_RE.fullmatch(relation_type):
        raise ValueError(f"Invalid relation_type for typed relationship label: {relation_type!r}")
    return relation_type.upper()


def _is_neo4j_property_value(value: Any) -> bool:
    """Return whether a value is a Neo4j scalar or scalar list."""
    if value is None:
        return False
    if isinstance(value, str | int | float | bool):
        return True
    if not isinstance(value, list):
        return False
    return all(isinstance(item, str | int | float | bool) for item in value)
