From 20755e69e29b16dee50e4ccca2240b05a126baa5 Mon Sep 17 00:00:00 2001
From: Sameer Kadam <sskadam6305@gmail.com>
Date: Sun, 15 Feb 2026 15:57:07 +0530
Subject: [PATCH] feat(graph): add Apache AGE backend integration with
 configuration, registration, tests and documentation (#311)

---
 README.md                            |   17 +-
 docs/graph_stores/apache_age.md      |  243 +++++
 semantica/graph_store/__init__.py    |    3 +
 semantica/graph_store/age_store.py   | 1312 ++++++++++++++++++++++++++
 semantica/graph_store/config.py      |   18 +
 semantica/graph_store/graph_store.py |    7 +
 tests/graph_store/__init__.py        |    0
 tests/graph_store/test_age_store.py  |  772 +++++++++++++++
 8 files changed, 2369 insertions(+), 3 deletions(-)
 create mode 100644 docs/graph_stores/apache_age.md
 create mode 100644 semantica/graph_store/age_store.py
 create mode 100644 tests/graph_store/__init__.py
 create mode 100644 tests/graph_store/test_age_store.py

diff --git a/README.md b/README.md
index 6fce0ca0..ca507b48 100644
--- a/README.md
+++ b/README.md
@@ -137,6 +137,7 @@ print(f"Found {len(precedents)} precedents")
 
 - **Docling Support** — Document parsing with table extraction (PDF, DOCX, PPTX, XLSX)
 - **AWS Neptune** — Amazon Neptune graph database support with IAM authentication
+- **Apache AGE** — PostgreSQL graph extension backend (openCypher via SQL)
 - **Custom Ontology Import** — Import existing ontologies (OWL, RDF, Turtle, JSON-LD)
 
 > **Built for environments where every answer must be explainable and governed.**
@@ -284,7 +285,8 @@ The **semantic gap** is the fundamental disconnect between what AI systems can p
 - 📐 **OWL Ontologies** — HermiT/Pellet validated, custom ontology import support
 - 🔢 **Vector Embeddings** — FastEmbed by default
 - ☁️ **AWS Neptune** — Amazon Neptune graph database support
-- 🔍 **Provenance** — Every AI response links back to:
+- � **Apache AGE** — PostgreSQL graph extension with openCypher support
+- �🔍 **Provenance** — Every AI response links back to:
   - 📄 Source documents
   - 🏷️ Extracted entities & relations
   - 📐 Ontology rules applied
@@ -510,13 +512,13 @@ results = vector_store.search(query="supply chain", top_k=5)
 
 ### Graph Store & Triplet Store
 
-> **Neo4j, FalkorDB, Amazon Neptune** • **SPARQL queries** • **RDF triplets**
+> **Neo4j, FalkorDB, Amazon Neptune, Apache AGE** • **SPARQL queries** • **RDF triplets**
 
 ```python
 from semantica.graph_store import GraphStore
 from semantica.triplet_store import TripletStore
 
-# Graph Store (Neo4j, FalkorDB)
+# Graph Store (Neo4j, FalkorDB, Apache AGE)
 graph_store = GraphStore(backend="neo4j", uri="bolt://localhost:7687", user="neo4j", password="password")
 graph_store.add_nodes([{"id": "n1", "labels": ["Person"], "properties": {"name": "Alice"}}])
 
@@ -538,6 +540,15 @@ neptune_store.add_nodes([
 # Query Operations
 result = neptune_store.execute_query("MATCH (p:Person) RETURN p.name, p.age")
 
+# Apache AGE Graph Store (PostgreSQL + openCypher)
+age_store = GraphStore(
+    backend="age",
+    connection_string="host=localhost dbname=agedb user=postgres password=secret",
+    graph_name="semantica",
+)
+age_store.connect()
+age_store.create_node(labels=["Person"], properties={"name": "Alice", "age": 30})
+
 # Triplet Store (Blazegraph, Jena, RDF4J)
 triplet_store = TripletStore(backend="blazegraph", endpoint="http://localhost:9999/blazegraph")
 triplet_store.add_triplet({"subject": "Alice", "predicate": "knows", "object": "Bob"})
diff --git a/docs/graph_stores/apache_age.md b/docs/graph_stores/apache_age.md
new file mode 100644
index 00000000..27c16a9d
--- /dev/null
+++ b/docs/graph_stores/apache_age.md
@@ -0,0 +1,243 @@
+# Apache AGE Graph Store
+
+**Backend**: PostgreSQL + [Apache AGE](https://age.apache.org/)
+**Driver**: `psycopg2`
+
+Apache AGE is a PostgreSQL extension that adds graph database functionality, enabling you to run openCypher queries alongside traditional SQL. This backend lets Semantica use AGE as a property graph store with the same interface as Neo4j and FalkorDB.
+
+---
+
+## Prerequisites
+
+| Component | Version |
+|-----------|---------|
+| PostgreSQL | 12+ |
+| Apache AGE | 1.4+ (compiled and installed) |
+| psycopg2 | 2.9+ |
+
+```bash
+pip install psycopg2-binary
+```
+
+> **Note**: Apache AGE must be compiled and installed into your PostgreSQL instance. See the [AGE installation guide](https://age.apache.org/age-manual/master/intro/setup.html).
+
+---
+
+## Quick Start
+
+```python
+from semantica.graph_store import GraphStore
+
+# Using the unified GraphStore facade
+store = GraphStore(
+    backend="age",
+    connection_string="host=localhost dbname=agedb user=postgres password=secret",
+    graph_name="semantica",
+)
+store.connect()
+
+# Create nodes
+alice = store.create_node(labels=["Person"], properties={"name": "Alice", "age": 30})
+bob = store.create_node(labels=["Person"], properties={"name": "Bob", "age": 25})
+
+# Create relationship
+rel = store.create_relationship(alice["id"], bob["id"], "KNOWS", {"since": 2023})
+
+# Query
+result = store.execute_query("MATCH (p:Person) RETURN p", cols="p agtype")
+print(result["records"])
+
+store.close()
+```
+
+### Direct Usage (without facade)
+
+```python
+from semantica.graph_store.age_store import ApacheAgeStore
+
+store = ApacheAgeStore(
+    connection_string="host=localhost dbname=agedb user=postgres password=secret",
+    graph_name="my_graph",
+)
+store.connect()
+
+node = store.create_node(["Entity"], {"semantica_id": "ent-001", "value": "test"})
+print(node)
+# {"id": 844424930131969, "labels": ["Entity"], "properties": {"semantica_id": "ent-001", "value": "test"}}
+
+store.close()
+```
+
+---
+
+## Configuration
+
+### Environment Variables
+
+| Variable | Description | Default |
+|----------|-------------|---------|
+| `GRAPH_STORE_AGE_CONNECTION_STRING` | PostgreSQL connection string | `host=localhost dbname=agedb user=postgres password=postgres` |
+| `GRAPH_STORE_AGE_GRAPH_NAME` | AGE graph name | `semantica` |
+
+### Programmatic Configuration
+
+```python
+from semantica.graph_store.config import graph_store_config
+
+graph_store_config.set("age_connection_string", "host=db.example.com dbname=prod_age user=app")
+graph_store_config.set("age_graph_name", "production")
+```
+
+---
+
+## Connection & Initialization
+
+On `connect()`, the store performs idempotent setup:
+
+1. `CREATE EXTENSION IF NOT EXISTS age;`
+2. `LOAD 'age';`
+3. `SET search_path = ag_catalog, "$user", public;`
+4. Creates the named graph if it does not already exist.
+
+This is safe to call repeatedly.
+
+---
+
+## ID Handling
+
+Apache AGE auto-generates internal vertex/edge IDs (large integers). These are **not** the same as any semantic or application-level ID you may want to assign.
+
+| Concept | Description |
+|---------|-------------|
+| **AGE internal ID** | Auto-generated by AGE. Exposed as `"id"` in all returned dicts. Used in `delete_node()`, `get_node()`, etc. |
+| **Semantic ID** | Application-level identifier. Store it in the `semantica_id` property. |
+
+```python
+node = store.create_node(
+    labels=["Document"],
+    properties={"semantica_id": "doc-abc-123", "title": "My Doc"},
+)
+# node["id"] → AGE internal ID (e.g., 844424930131969)
+# node["properties"]["semantica_id"] → "doc-abc-123"
+```
+
+> **Important**: Never mix AGE internal IDs with semantic IDs. Use `node["id"]` for graph operations (delete, update, traverse) and `node["properties"]["semantica_id"]` for application-level lookups.
+
+---
+
+## Label Handling
+
+AGE supports exactly **one label per vertex**. Semantica handles this transparently:
+
+- `labels[0]` → used as the primary AGE vertex label.
+- `labels[1:]` → stored in a `labels` property array on the vertex.
+
+When reading nodes, the store reconstructs the full label list automatically.
+
+```python
+node = store.create_node(
+    labels=["Person", "Employee", "Admin"],
+    properties={"name": "Alice"},
+)
+# In AGE: vertex with label "Person" and property labels=["Employee", "Admin"]
+# Returned: {"id": ..., "labels": ["Person", "Employee", "Admin"], "properties": {"name": "Alice"}}
+```
+
+---
+
+## Cypher Query Execution
+
+All Cypher queries are executed via AGE's SQL wrapper:
+
+```sql
+SELECT * FROM cypher('graph_name', $$ <cypher_query> $$) AS (col1 agtype, ...);
+```
+
+### Parameter Substitution
+
+AGE does not support `$param` style binding inside `cypher()` calls. The store safely converts parameters to Cypher literals with proper escaping:
+
+```python
+result = store.execute_query(
+    "MATCH (p:Person) WHERE p.age > $min_age RETURN p",
+    parameters={"min_age": 25},
+    cols="p agtype",
+)
+```
+
+### Column Specification
+
+For custom queries, pass the `cols` option to specify the `AS` clause:
+
+```python
+result = store.execute_query(
+    "MATCH (a)-[r]->(b) RETURN a, r, b",
+    cols="a agtype, r agtype, b agtype",
+)
+```
+
+If omitted, the store attempts to infer columns from the `RETURN` clause.
+
+---
+
+## Transactions
+
+The store uses explicit PostgreSQL transactions:
+
+- **Success** → `COMMIT`
+- **Exception** → `ROLLBACK`, then re-raise as `ProcessingError`
+- No silent failures
+
+---
+
+## API Reference
+
+All methods match the standard Semantica graph store backend interface:
+
+| Method | Description |
+|--------|-------------|
+| `connect(**options)` | Connect and initialize AGE |
+| `close()` | Close the connection |
+| `create_node(labels, properties)` | Create a vertex |
+| `create_nodes(nodes)` | Batch create vertices |
+| `get_node(node_id)` | Get vertex by AGE ID |
+| `get_nodes(labels, properties, limit)` | Query vertices |
+| `update_node(node_id, properties, merge)` | Update vertex properties |
+| `delete_node(node_id, detach)` | Delete a vertex |
+| `create_relationship(start_id, end_id, type, properties)` | Create an edge |
+| `get_relationships(node_id, rel_type, direction, limit)` | Query edges |
+| `delete_relationship(rel_id)` | Delete an edge |
+| `execute_query(query, parameters)` | Run arbitrary Cypher |
+| `get_neighbors(node_id, rel_type, direction, depth)` | Graph traversal |
+| `shortest_path(start_id, end_id, rel_type, max_depth)` | Path finding |
+| `create_index(label, property_name, index_type)` | Create a PostgreSQL index |
+| `get_stats()` | Graph statistics |
+
+---
+
+## Docker Setup
+
+```yaml
+services:
+  age:
+    image: apache/age:latest
+    ports:
+      - "5432:5432"
+    environment:
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: secret
+      POSTGRES_DB: agedb
+```
+
+```bash
+docker compose up -d
+```
+
+Then connect:
+
+```python
+store = GraphStore(
+    backend="age",
+    connection_string="host=localhost port=5432 dbname=agedb user=postgres password=secret",
+)
+```
diff --git a/semantica/graph_store/__init__.py b/semantica/graph_store/__init__.py
index ea952b2e..51b02251 100644
--- a/semantica/graph_store/__init__.py
+++ b/semantica/graph_store/__init__.py
@@ -122,6 +122,7 @@ Author: Semantica Contributors
 License: MIT
 """
 
+from .age_store import ApacheAgeStore
 from .amazon_neptune import (
     AmazonNeptuneStore,
     NeptuneAuthTokenManager,
@@ -172,6 +173,8 @@ __all__ = [
     "Neo4jStore",
     "Neo4jDriver",
     "Neo4jTransaction",
+    # Apache AGE
+    "ApacheAgeStore",
     # Amazon Neptune
     "AmazonNeptuneStore",
     "NeptuneAuthTokenManager",
diff --git a/semantica/graph_store/age_store.py b/semantica/graph_store/age_store.py
new file mode 100644
index 00000000..1f4ea8b6
--- /dev/null
+++ b/semantica/graph_store/age_store.py
@@ -0,0 +1,1312 @@
+"""
+Apache AGE Store Module
+
+This module provides Apache AGE (PostgreSQL graph extension) integration for
+property graph storage and Cypher querying in the Semantica framework, supporting
+full CRUD operations, transactions, and graph analytics.
+
+Apache AGE extends PostgreSQL with graph database functionality, enabling
+hybrid relational + graph workloads using openCypher queries executed via SQL.
+
+Key Features:
+    - OpenCypher query language support via SQL wrapper
+    - Node and relationship CRUD operations
+    - Transaction support with explicit commit/rollback
+    - Parameterized queries to prevent SQL injection
+    - AGE internal ID / semantic ID separation
+    - Multi-label emulation (one AGE label + property array)
+    - Batch operations with progress tracking
+    - Optional dependency handling (psycopg2)
+
+Main Classes:
+    - ApacheAgeStore: Main AGE store for graph operations
+
+Example Usage:
+    >>> from semantica.graph_store.age_store import ApacheAgeStore
+    >>> store = ApacheAgeStore(
+    ...     connection_string="host=localhost dbname=agedb user=postgres password=secret",
+    ...     graph_name="semantica"
+    ... )
+    >>> store.connect()
+    >>> node = store.create_node(labels=["Person"], properties={"name": "Alice"})
+    >>> results = store.execute_query("MATCH (p:Person) RETURN p")
+    >>> store.close()
+
+Note:
+    - AGE auto-generates internal vertex/edge IDs.
+    - The ``node_id`` parameter in CRUD methods refers to the AGE internal ID.
+    - Semantic IDs can be stored in the ``semantica_id`` property.
+    - AGE supports exactly one label per vertex; additional labels are stored
+      in a ``labels`` property array.
+
+Author: Semantica Contributors
+License: MIT
+"""
+
+import json
+import re
+from typing import Any, Dict, List, Optional, Union
+
+from ..utils.exceptions import ProcessingError, ValidationError
+from ..utils.logging import get_logger
+from ..utils.progress_tracker import get_progress_tracker
+
+# Optional psycopg2 import
+try:
+    import psycopg2
+    import psycopg2.extras
+
+    PSYCOPG2_AVAILABLE = True
+except (ImportError, OSError):
+    PSYCOPG2_AVAILABLE = False
+    psycopg2 = None
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _sanitize_label(label: str) -> str:
+    """
+    Sanitize a Cypher label to prevent injection.
+
+    Only allows alphanumeric characters and underscores.
+
+    Args:
+        label: Raw label string.
+
+    Returns:
+        Sanitized label string.
+
+    Raises:
+        ValidationError: If the label contains invalid characters.
+    """
+    if not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", label):
+        raise ValidationError(
+            f"Invalid label '{label}': must start with a letter or underscore "
+            "and contain only alphanumeric characters and underscores."
+        )
+    return label
+
+
+def _sanitize_rel_type(rel_type: str) -> str:
+    """
+    Sanitize a relationship type string.
+
+    Args:
+        rel_type: Raw relationship type.
+
+    Returns:
+        Sanitized relationship type.
+
+    Raises:
+        ValidationError: If the type contains invalid characters.
+    """
+    if not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", rel_type):
+        raise ValidationError(
+            f"Invalid relationship type '{rel_type}': must start with a letter or "
+            "underscore and contain only alphanumeric characters and underscores."
+        )
+    return rel_type
+
+
+def _props_to_cypher_literal(properties: Dict[str, Any]) -> str:
+    """
+    Convert a Python dict to an AGE-compatible Cypher map literal.
+
+    AGE does not support ``$param`` style parameter binding inside
+    ``cypher()`` calls, so property values must be inlined as literals
+    with proper escaping.
+
+    Args:
+        properties: Dictionary of property key-value pairs.
+
+    Returns:
+        Cypher map literal string, e.g. ``{name: 'Alice', age: 30}``.
+    """
+    if not properties:
+        return "{}"
+    parts = []
+    for key, value in properties.items():
+        # Validate key
+        if not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", key):
+            raise ValidationError(f"Invalid property key: '{key}'")
+        parts.append(f"{key}: {_value_to_cypher_literal(value)}")
+    return "{" + ", ".join(parts) + "}"
+
+
+def _value_to_cypher_literal(value: Any) -> str:
+    """
+    Convert a single Python value to a Cypher literal string.
+
+    Args:
+        value: Python value.
+
+    Returns:
+        Cypher literal representation.
+    """
+    if value is None:
+        return "null"
+    if isinstance(value, bool):
+        return "true" if value else "false"
+    if isinstance(value, int):
+        return str(value)
+    if isinstance(value, float):
+        return repr(value)
+    if isinstance(value, str):
+        # Escape single quotes for Cypher strings
+        escaped = value.replace("\\", "\\\\").replace("'", "\\'")
+        return f"'{escaped}'"
+    if isinstance(value, (list, tuple)):
+        inner = ", ".join(_value_to_cypher_literal(v) for v in value)
+        return f"[{inner}]"
+    if isinstance(value, dict):
+        return _props_to_cypher_literal(value)
+    # Fallback: convert to string
+    escaped = str(value).replace("\\", "\\\\").replace("'", "\\'")
+    return f"'{escaped}'"
+
+
+def _parse_agtype(raw: Any) -> Any:
+    """
+    Parse an agtype value returned by AGE into a Python object.
+
+    AGE returns results as ``agtype`` which may be a JSON-like string
+    with an optional ``::vertex`` / ``::edge`` / ``::path`` suffix.
+
+    Args:
+        raw: Raw value from the cursor.
+
+    Returns:
+        Parsed Python object (dict, list, or scalar).
+    """
+    if raw is None:
+        return None
+    if not isinstance(raw, str):
+        return raw
+
+    text = raw.strip()
+
+    # Strip AGE type suffixes
+    for suffix in ("::vertex", "::edge", "::path", "::numeric",
+                    "::integer", "::float", "::boolean", "::text"):
+        if text.endswith(suffix):
+            text = text[: -len(suffix)].strip()
+            break
+
+    # Try JSON parse
+    try:
+        return json.loads(text)
+    except (json.JSONDecodeError, ValueError):
+        pass
+
+    # Boolean literals
+    if text.lower() == "true":
+        return True
+    if text.lower() == "false":
+        return False
+
+    # Numeric
+    try:
+        if "." in text:
+            return float(text)
+        return int(text)
+    except ValueError:
+        pass
+
+    return text
+
+
+def _vertex_to_node_dict(vertex: Any) -> Dict[str, Any]:
+    """
+    Convert a parsed AGE vertex dict to the standard node return format.
+
+    Expected vertex dict shape from AGE::
+
+        {"id": <int>, "label": "<label>", "properties": {…}}
+
+    The returned dict matches Neo4jStore output::
+
+        {"id": <int>, "labels": [<str>, …], "properties": {…}}
+
+    Args:
+        vertex: Parsed vertex dict.
+
+    Returns:
+        Standardised node dict.
+    """
+    if not isinstance(vertex, dict):
+        return {"id": None, "labels": [], "properties": {}}
+
+    props = dict(vertex.get("properties", {}))
+    primary_label = vertex.get("label", "")
+    labels = [primary_label] if primary_label else []
+
+    # Merge additional labels stored in the 'labels' property
+    extra_labels = props.pop("labels", None)
+    if extra_labels and isinstance(extra_labels, list):
+        labels.extend(extra_labels)
+
+    return {
+        "id": vertex.get("id"),
+        "labels": labels,
+        "properties": props,
+    }
+
+
+def _edge_to_rel_dict(edge: Any) -> Dict[str, Any]:
+    """
+    Convert a parsed AGE edge dict to the standard relationship return format.
+
+    Expected edge dict shape from AGE::
+
+        {"id": <int>, "label": "<type>", "end_id": <int>,
+         "start_id": <int>, "properties": {…}}
+
+    The returned dict matches Neo4jStore output::
+
+        {"id": <int>, "type": "<type>", "start_node_id": <int>,
+         "end_node_id": <int>, "properties": {…}}
+
+    Args:
+        edge: Parsed edge dict.
+
+    Returns:
+        Standardised relationship dict.
+    """
+    if not isinstance(edge, dict):
+        return {
+            "id": None,
+            "type": "",
+            "start_node_id": None,
+            "end_node_id": None,
+            "properties": {},
+        }
+    return {
+        "id": edge.get("id"),
+        "type": edge.get("label", ""),
+        "start_node_id": edge.get("start_id"),
+        "end_node_id": edge.get("end_id"),
+        "properties": dict(edge.get("properties", {})),
+    }
+
+
+# ---------------------------------------------------------------------------
+# ApacheAgeStore
+# ---------------------------------------------------------------------------
+
+class ApacheAgeStore:
+    """
+    Apache AGE store for property graph storage and Cypher querying.
+
+    Provides the same backend interface as Neo4jStore / FalkorDBStore so it
+    can be used transparently via the ``GraphStore`` facade.
+
+    Key capabilities:
+
+    - PostgreSQL + Apache AGE graph operations
+    - Node and relationship CRUD
+    - Cypher query execution wrapped in ``SELECT * FROM cypher(…)``
+    - Transaction support with explicit commit / rollback
+    - Multi-label emulation
+    - ID separation (AGE internal vs semantic ``semantica_id``)
+    """
+
+    def __init__(
+        self,
+        connection_string: Optional[str] = None,
+        graph_name: Optional[str] = None,
+        **config,
+    ):
+        """
+        Initialize Apache AGE store.
+
+        Args:
+            connection_string: PostgreSQL connection string
+                (e.g. ``"host=localhost dbname=agedb user=postgres password=secret"``).
+            graph_name: Name of the AGE graph to use.
+            **config: Additional configuration options.
+        """
+        self.logger = get_logger("age_store")
+        self.config = config
+        self.progress_tracker = get_progress_tracker()
+        if not self.progress_tracker.enabled:
+            self.progress_tracker.enabled = True
+
+        self.connection_string = connection_string or config.get(
+            "connection_string",
+            "host=localhost dbname=agedb user=postgres password=postgres",
+        )
+        self.graph_name = graph_name or config.get("graph_name", "semantica")
+
+        self._conn = None
+
+        if not PSYCOPG2_AVAILABLE:
+            self.logger.warning(
+                "psycopg2 not available. Install with: pip install psycopg2-binary"
+            )
+
+    # ------------------------------------------------------------------
+    # Connection management
+    # ------------------------------------------------------------------
+
+    def connect(self, **options) -> bool:
+        """
+        Connect to PostgreSQL and initialise the AGE extension & graph.
+
+        Performs idempotent setup:
+        1. ``CREATE EXTENSION IF NOT EXISTS age``
+        2. ``LOAD 'age'``
+        3. ``SET search_path = ag_catalog, "$user", public``
+        4. ``SELECT create_graph('<graph_name>')`` if not exists.
+
+        Args:
+            **options: Additional connection options (currently unused).
+
+        Returns:
+            True if connected and initialised successfully.
+        """
+        if not PSYCOPG2_AVAILABLE:
+            raise ProcessingError(
+                "psycopg2 is not available. Install with: pip install psycopg2-binary"
+            )
+
+        conn_str = options.get("connection_string", self.connection_string)
+
+        try:
+            self._conn = psycopg2.connect(conn_str)
+            self._conn.autocommit = False
+
+            # Idempotent AGE setup
+            with self._conn.cursor() as cur:
+                cur.execute("CREATE EXTENSION IF NOT EXISTS age;")
+                cur.execute("LOAD 'age';")
+                cur.execute(
+                    'SET search_path = ag_catalog, "$user", public;'
+                )
+                # Create graph if not exists
+                cur.execute(
+                    "SELECT count(*) FROM ag_catalog.ag_graph "
+                    "WHERE name = %s;",
+                    (self.graph_name,),
+                )
+                (count,) = cur.fetchone()
+                if count == 0:
+                    cur.execute(
+                        "SELECT create_graph(%s);", (self.graph_name,)
+                    )
+            self._conn.commit()
+            self.logger.info(
+                f"Connected to PostgreSQL/AGE, graph '{self.graph_name}'"
+            )
+            return True
+
+        except Exception as e:
+            if self._conn:
+                self._conn.rollback()
+            raise ProcessingError(f"Failed to connect to AGE: {str(e)}")
+
+    def close(self) -> None:
+        """Close the PostgreSQL connection."""
+        if self._conn:
+            try:
+                self._conn.close()
+            except Exception:
+                pass
+            self._conn = None
+        self.logger.info("Disconnected from PostgreSQL/AGE")
+
+    def _ensure_connection(self) -> None:
+        """Ensure the database connection is available."""
+        if self._conn is None or self._conn.closed:
+            self.connect()
+
+    # ------------------------------------------------------------------
+    # Low-level Cypher execution
+    # ------------------------------------------------------------------
+
+    def _execute_cypher(
+        self,
+        cypher: str,
+        cols: str = "result agtype",
+    ) -> List[Any]:
+        """
+        Execute a Cypher query via AGE's ``cypher()`` SQL function.
+
+        Wraps the query in::
+
+            SELECT * FROM cypher('{graph_name}', $$ {cypher} $$)
+            AS ({cols});
+
+        Args:
+            cypher: OpenCypher query string.
+            cols: Column definition for the ``AS`` clause. Defaults to
+                ``"result agtype"``.
+
+        Returns:
+            List of raw row tuples from the cursor.
+        """
+        self._ensure_connection()
+        sql = (
+            f"SELECT * FROM cypher('{self.graph_name}', $$ {cypher} $$) "
+            f"AS ({cols});"
+        )
+        try:
+            with self._conn.cursor() as cur:
+                cur.execute(sql)
+                rows = cur.fetchall()
+            self._conn.commit()
+            return rows
+        except Exception as e:
+            self._conn.rollback()
+            raise ProcessingError(f"Cypher execution failed: {str(e)}")
+
+    # ------------------------------------------------------------------
+    # Node CRUD
+    # ------------------------------------------------------------------
+
+    def create_node(
+        self,
+        labels: List[str],
+        properties: Dict[str, Any],
+        **options,
+    ) -> Dict[str, Any]:
+        """
+        Create a node in the graph.
+
+        AGE supports a single label per vertex.  ``labels[0]`` is used as the
+        primary AGE label; any additional labels are stored in a ``labels``
+        property array.
+
+        If ``semantica_id`` is present in *properties* it is preserved as-is;
+        otherwise no synthetic semantic ID is generated.
+
+        Args:
+            labels: Node labels (first is primary AGE label, rest stored as property).
+            properties: Node properties.
+            **options: Additional options.
+
+        Returns:
+            Created node information: ``{"id": <int>, "labels": [...], "properties": {…}}``.
+        """
+        tracking_id = self.progress_tracker.start_tracking(
+            module="graph_store",
+            submodule="ApacheAgeStore",
+            message=f"Creating node with labels {labels}",
+        )
+
+        try:
+            if not labels:
+                raise ValidationError("At least one label is required")
+
+            primary_label = _sanitize_label(labels[0])
+            props = dict(properties)
+
+            # Store additional labels in property
+            if len(labels) > 1:
+                props["labels"] = [_sanitize_label(l) for l in labels[1:]]
+
+            cypher = (
+                f"CREATE (n:{primary_label} {_props_to_cypher_literal(props)}) "
+                f"RETURN n"
+            )
+            rows = self._execute_cypher(cypher)
+
+            if rows:
+                vertex = _parse_agtype(rows[0][0])
+                node_data = _vertex_to_node_dict(vertex)
+                self.progress_tracker.stop_tracking(
+                    tracking_id,
+                    status="completed",
+                    message=f"Created node with ID {node_data['id']}",
+                )
+                return node_data
+
+            raise ProcessingError("Failed to create node - no result returned")
+
+        except (ProcessingError, ValidationError):
+            self.progress_tracker.stop_tracking(
+                tracking_id, status="failed", message="create_node failed"
+            )
+            raise
+        except Exception as e:
+            self.progress_tracker.stop_tracking(
+                tracking_id, status="failed", message=str(e)
+            )
+            raise ProcessingError(f"Failed to create node: {str(e)}")
+
+    def create_nodes(
+        self,
+        nodes: List[Dict[str, Any]],
+        **options,
+    ) -> List[Dict[str, Any]]:
+        """
+        Create multiple nodes in batch.
+
+        Args:
+            nodes: List of dicts with ``"labels"`` and ``"properties"`` keys.
+            **options: Additional options.
+
+        Returns:
+            List of created node dicts.
+        """
+        tracking_id = self.progress_tracker.start_tracking(
+            module="graph_store",
+            submodule="ApacheAgeStore",
+            message=f"Creating {len(nodes)} nodes in batch",
+        )
+
+        try:
+            created: List[Dict[str, Any]] = []
+            for node in nodes:
+                node_labels = node.get("labels", ["Node"])
+                node_props = node.get("properties", {})
+                created.append(self.create_node(node_labels, node_props))
+
+            self.progress_tracker.stop_tracking(
+                tracking_id,
+                status="completed",
+                message=f"Created {len(created)} nodes",
+            )
+            return created
+
+        except Exception as e:
+            self.progress_tracker.stop_tracking(
+                tracking_id, status="failed", message=str(e)
+            )
+            raise ProcessingError(f"Failed to create nodes: {str(e)}")
+
+    def get_node(
+        self,
+        node_id: int,
+        **options,
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Get a node by its AGE internal ID.
+
+        Args:
+            node_id: AGE internal vertex ID.
+            **options: Additional options.
+
+        Returns:
+            Node dict or ``None`` if not found.
+        """
+        try:
+            cypher = (
+                f"MATCH (n) WHERE id(n) = {int(node_id)} RETURN n"
+            )
+            rows = self._execute_cypher(cypher)
+
+            if rows:
+                vertex = _parse_agtype(rows[0][0])
+                return _vertex_to_node_dict(vertex)
+            return None
+
+        except Exception as e:
+            raise ProcessingError(f"Failed to get node: {str(e)}")
+
+    def get_nodes(
+        self,
+        labels: Optional[List[str]] = None,
+        properties: Optional[Dict[str, Any]] = None,
+        limit: int = 100,
+        **options,
+    ) -> List[Dict[str, Any]]:
+        """
+        Get nodes matching criteria.
+
+        Args:
+            labels: Filter by labels (matches against primary AGE label).
+            properties: Filter by property values.
+            limit: Maximum number of nodes to return.
+            **options: Additional options.
+
+        Returns:
+            List of matching node dicts.
+        """
+        try:
+            if labels:
+                primary = _sanitize_label(labels[0])
+                match = f"MATCH (n:{primary})"
+            else:
+                match = "MATCH (n)"
+
+            where_parts: List[str] = []
+            if properties:
+                for key, value in properties.items():
+                    if not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", key):
+                        raise ValidationError(f"Invalid property key: '{key}'")
+                    where_parts.append(
+                        f"n.{key} = {_value_to_cypher_literal(value)}"
+                    )
+
+            where_clause = ""
+            if where_parts:
+                where_clause = " WHERE " + " AND ".join(where_parts)
+
+            cypher = f"{match}{where_clause} RETURN n LIMIT {int(limit)}"
+            rows = self._execute_cypher(cypher)
+
+            nodes: List[Dict[str, Any]] = []
+            for row in rows:
+                vertex = _parse_agtype(row[0])
+                nodes.append(_vertex_to_node_dict(vertex))
+            return nodes
+
+        except (ProcessingError, ValidationError):
+            raise
+        except Exception as e:
+            raise ProcessingError(f"Failed to get nodes: {str(e)}")
+
+    def update_node(
+        self,
+        node_id: int,
+        properties: Dict[str, Any],
+        merge: bool = True,
+        **options,
+    ) -> Dict[str, Any]:
+        """
+        Update a node's properties.
+
+        Args:
+            node_id: AGE internal vertex ID.
+            properties: Properties to set.
+            merge: If ``True``, merge (``+=``); if ``False``, replace (``=``).
+            **options: Additional options.
+
+        Returns:
+            Updated node dict.
+        """
+        try:
+            props_lit = _props_to_cypher_literal(properties)
+            op = "+=" if merge else "="
+            cypher = (
+                f"MATCH (n) WHERE id(n) = {int(node_id)} "
+                f"SET n {op} {props_lit} "
+                f"RETURN n"
+            )
+            rows = self._execute_cypher(cypher)
+
+            if rows:
+                vertex = _parse_agtype(rows[0][0])
+                return _vertex_to_node_dict(vertex)
+
+            raise ProcessingError(f"Node with ID {node_id} not found")
+
+        except ProcessingError:
+            raise
+        except Exception as e:
+            raise ProcessingError(f"Failed to update node: {str(e)}")
+
+    def delete_node(
+        self,
+        node_id: int,
+        detach: bool = True,
+        **options,
+    ) -> bool:
+        """
+        Delete a node by its AGE internal ID.
+
+        Args:
+            node_id: AGE internal vertex ID.
+            detach: If ``True``, delete connected relationships as well.
+            **options: Additional options.
+
+        Returns:
+            ``True`` if deleted successfully.
+        """
+        try:
+            delete_kw = "DETACH DELETE" if detach else "DELETE"
+            cypher = (
+                f"MATCH (n) WHERE id(n) = {int(node_id)} "
+                f"{delete_kw} n"
+            )
+            self._execute_cypher(cypher, cols="v agtype")
+            return True
+
+        except Exception as e:
+            raise ProcessingError(f"Failed to delete node: {str(e)}")
+
+    # ------------------------------------------------------------------
+    # Relationship CRUD
+    # ------------------------------------------------------------------
+
+    def create_relationship(
+        self,
+        start_node_id: int,
+        end_node_id: int,
+        rel_type: str,
+        properties: Optional[Dict[str, Any]] = None,
+        **options,
+    ) -> Dict[str, Any]:
+        """
+        Create a relationship between two nodes.
+
+        Args:
+            start_node_id: Start vertex AGE internal ID.
+            end_node_id: End vertex AGE internal ID.
+            rel_type: Relationship type.
+            properties: Relationship properties.
+            **options: Additional options.
+
+        Returns:
+            Created relationship dict.
+        """
+        tracking_id = self.progress_tracker.start_tracking(
+            module="graph_store",
+            submodule="ApacheAgeStore",
+            message=f"Creating relationship [{rel_type}]",
+        )
+
+        try:
+            safe_type = _sanitize_rel_type(rel_type)
+            props = properties or {}
+            props_lit = _props_to_cypher_literal(props)
+
+            cypher = (
+                f"MATCH (a), (b) "
+                f"WHERE id(a) = {int(start_node_id)} AND id(b) = {int(end_node_id)} "
+                f"CREATE (a)-[r:{safe_type} {props_lit}]->(b) "
+                f"RETURN r"
+            )
+            rows = self._execute_cypher(cypher)
+
+            if rows:
+                edge = _parse_agtype(rows[0][0])
+                rel_data = _edge_to_rel_dict(edge)
+                self.progress_tracker.stop_tracking(
+                    tracking_id,
+                    status="completed",
+                    message=f"Created relationship with ID {rel_data['id']}",
+                )
+                return rel_data
+
+            raise ProcessingError(
+                "Failed to create relationship - nodes not found"
+            )
+
+        except (ProcessingError, ValidationError):
+            self.progress_tracker.stop_tracking(
+                tracking_id, status="failed", message="create_relationship failed"
+            )
+            raise
+        except Exception as e:
+            self.progress_tracker.stop_tracking(
+                tracking_id, status="failed", message=str(e)
+            )
+            raise ProcessingError(f"Failed to create relationship: {str(e)}")
+
+    def get_relationships(
+        self,
+        node_id: Optional[int] = None,
+        rel_type: Optional[str] = None,
+        direction: str = "both",
+        limit: int = 100,
+        **options,
+    ) -> List[Dict[str, Any]]:
+        """
+        Get relationships matching criteria.
+
+        Args:
+            node_id: Filter by vertex AGE internal ID.
+            rel_type: Filter by relationship type.
+            direction: ``"in"``, ``"out"``, or ``"both"``.
+            limit: Maximum number of results.
+            **options: Additional options.
+
+        Returns:
+            List of relationship dicts.
+        """
+        try:
+            type_filter = f":{_sanitize_rel_type(rel_type)}" if rel_type else ""
+
+            if node_id is not None:
+                nid = int(node_id)
+                if direction == "out":
+                    cypher = (
+                        f"MATCH (a)-[r{type_filter}]->(b) "
+                        f"WHERE id(a) = {nid} "
+                        f"RETURN r LIMIT {int(limit)}"
+                    )
+                elif direction == "in":
+                    cypher = (
+                        f"MATCH (a)<-[r{type_filter}]-(b) "
+                        f"WHERE id(a) = {nid} "
+                        f"RETURN r LIMIT {int(limit)}"
+                    )
+                else:
+                    cypher = (
+                        f"MATCH (a)-[r{type_filter}]-(b) "
+                        f"WHERE id(a) = {nid} "
+                        f"RETURN r LIMIT {int(limit)}"
+                    )
+            else:
+                cypher = (
+                    f"MATCH (a)-[r{type_filter}]->(b) "
+                    f"RETURN r LIMIT {int(limit)}"
+                )
+
+            rows = self._execute_cypher(cypher)
+
+            relationships: List[Dict[str, Any]] = []
+            for row in rows:
+                edge = _parse_agtype(row[0])
+                relationships.append(_edge_to_rel_dict(edge))
+            return relationships
+
+        except (ProcessingError, ValidationError):
+            raise
+        except Exception as e:
+            raise ProcessingError(f"Failed to get relationships: {str(e)}")
+
+    def delete_relationship(
+        self,
+        rel_id: int,
+        **options,
+    ) -> bool:
+        """
+        Delete a relationship by its AGE internal ID.
+
+        Args:
+            rel_id: Relationship AGE internal ID.
+            **options: Additional options.
+
+        Returns:
+            ``True`` if deleted successfully.
+        """
+        try:
+            cypher = (
+                f"MATCH ()-[r]->() WHERE id(r) = {int(rel_id)} DELETE r"
+            )
+            self._execute_cypher(cypher, cols="v agtype")
+            return True
+
+        except Exception as e:
+            raise ProcessingError(f"Failed to delete relationship: {str(e)}")
+
+    # ------------------------------------------------------------------
+    # Query execution
+    # ------------------------------------------------------------------
+
+    def execute_query(
+        self,
+        query: str,
+        parameters: Optional[Dict[str, Any]] = None,
+        **options,
+    ) -> Dict[str, Any]:
+        """
+        Execute an arbitrary Cypher query.
+
+        If ``parameters`` are provided they are substituted into the query
+        string using safe Cypher literal escaping (AGE does not support
+        ``$param`` style binding inside ``cypher()`` calls).
+
+        Args:
+            query: OpenCypher query string.
+            parameters: Optional parameter mapping.
+            **options: Additional options. Use ``cols`` to override the
+                default ``AS`` clause columns.
+
+        Returns:
+            Query result dict matching Neo4jStore format::
+
+                {
+                    "success": True,
+                    "records": [<row dicts>],
+                    "keys": [<column names>],
+                    "metadata": {"query": <str>},
+                }
+        """
+        tracking_id = self.progress_tracker.start_tracking(
+            module="graph_store",
+            submodule="ApacheAgeStore",
+            message="Executing Cypher query",
+        )
+
+        try:
+            final_query = query
+            if parameters:
+                for param_key, param_value in parameters.items():
+                    placeholder = f"${param_key}"
+                    literal = _value_to_cypher_literal(param_value)
+                    final_query = final_query.replace(placeholder, literal)
+
+            # Determine column spec
+            cols = options.get("cols", None)
+            if cols is None:
+                # Attempt to infer columns from RETURN clause
+                cols = self._infer_return_cols(final_query)
+
+            self._ensure_connection()
+            sql = (
+                f"SELECT * FROM cypher('{self.graph_name}', $$ {final_query} $$) "
+                f"AS ({cols});"
+            )
+
+            with self._conn.cursor() as cur:
+                cur.execute(sql)
+                if cur.description:
+                    keys = [desc[0] for desc in cur.description]
+                else:
+                    keys = []
+                raw_rows = cur.fetchall()
+            self._conn.commit()
+
+            records = []
+            for raw_row in raw_rows:
+                row: Dict[str, Any] = {}
+                for i, key in enumerate(keys):
+                    parsed = _parse_agtype(raw_row[i])
+                    # Convert vertex / edge dicts to standard shape
+                    if isinstance(parsed, dict):
+                        if "label" in parsed and "properties" in parsed:
+                            if "start_id" in parsed:
+                                parsed = _edge_to_rel_dict(parsed)
+                            else:
+                                parsed = _vertex_to_node_dict(parsed)
+                    row[key] = parsed
+                records.append(row)
+
+            self.progress_tracker.stop_tracking(
+                tracking_id,
+                status="completed",
+                message=f"Query returned {len(records)} records",
+            )
+
+            return {
+                "success": True,
+                "records": records,
+                "keys": keys,
+                "metadata": {"query": query},
+            }
+
+        except ProcessingError:
+            self.progress_tracker.stop_tracking(
+                tracking_id, status="failed", message="query failed"
+            )
+            raise
+        except Exception as e:
+            if self._conn and not self._conn.closed:
+                self._conn.rollback()
+            self.progress_tracker.stop_tracking(
+                tracking_id, status="failed", message=str(e)
+            )
+            raise ProcessingError(f"Query execution failed: {str(e)}")
+
+    @staticmethod
+    def _infer_return_cols(cypher: str) -> str:
+        """
+        Attempt to infer the ``AS (...)`` column definitions from the
+        ``RETURN`` clause of a Cypher query.
+
+        Falls back to ``"result agtype"`` if inference is not possible.
+
+        Args:
+            cypher: Cypher query string.
+
+        Returns:
+            Column definition string for the SQL ``AS`` clause.
+        """
+        match = re.search(r"\bRETURN\b\s+(.+?)(?:\s+ORDER\s+|\s+LIMIT\s+|\s+SKIP\s+|$)",
+                          cypher, re.IGNORECASE | re.DOTALL)
+        if not match:
+            return "result agtype"
+
+        return_body = match.group(1).strip()
+        # Split on commas not inside parentheses
+        depth = 0
+        parts: List[str] = []
+        current: List[str] = []
+        for ch in return_body:
+            if ch == "(":
+                depth += 1
+            elif ch == ")":
+                depth -= 1
+            elif ch == "," and depth == 0:
+                parts.append("".join(current).strip())
+                current = []
+                continue
+            current.append(ch)
+        if current:
+            parts.append("".join(current).strip())
+
+        col_defs: List[str] = []
+        for part in parts:
+            # Check for alias via AS
+            alias_match = re.search(r"\bAS\s+(\w+)\s*$", part, re.IGNORECASE)
+            if alias_match:
+                col_defs.append(f"{alias_match.group(1)} agtype")
+            else:
+                # Use sanitised expression name
+                clean = re.sub(r"[^A-Za-z0-9_]", "_", part.strip())
+                clean = re.sub(r"_+", "_", clean).strip("_") or "col"
+                col_defs.append(f"{clean} agtype")
+
+        return ", ".join(col_defs) if col_defs else "result agtype"
+
+    # ------------------------------------------------------------------
+    # Graph traversal / analytics
+    # ------------------------------------------------------------------
+
+    def get_neighbors(
+        self,
+        node_id: int,
+        rel_type: Optional[str] = None,
+        direction: str = "both",
+        depth: int = 1,
+        **options,
+    ) -> List[Dict[str, Any]]:
+        """
+        Get neighboring nodes.
+
+        Args:
+            node_id: Starting vertex AGE internal ID.
+            rel_type: Filter by relationship type.
+            direction: ``"in"``, ``"out"``, or ``"both"``.
+            depth: Traversal depth.
+            **options: Additional options.
+
+        Returns:
+            List of neighbouring node dicts.
+        """
+        try:
+            type_filter = f":{_sanitize_rel_type(rel_type)}" if rel_type else ""
+
+            if direction == "out":
+                pattern = f"-[r{type_filter}*1..{int(depth)}]->"
+            elif direction == "in":
+                pattern = f"<-[r{type_filter}*1..{int(depth)}]-"
+            else:
+                pattern = f"-[r{type_filter}*1..{int(depth)}]-"
+
+            cypher = (
+                f"MATCH (start){pattern}(neighbor) "
+                f"WHERE id(start) = {int(node_id)} "
+                f"RETURN DISTINCT neighbor"
+            )
+
+            rows = self._execute_cypher(cypher, cols="neighbor agtype")
+
+            neighbors: List[Dict[str, Any]] = []
+            for row in rows:
+                vertex = _parse_agtype(row[0])
+                neighbors.append(_vertex_to_node_dict(vertex))
+            return neighbors
+
+        except Exception as e:
+            raise ProcessingError(f"Failed to get neighbors: {str(e)}")
+
+    def shortest_path(
+        self,
+        start_node_id: int,
+        end_node_id: int,
+        rel_type: Optional[str] = None,
+        max_depth: int = 10,
+        **options,
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Find the shortest path between two nodes.
+
+        Args:
+            start_node_id: Start vertex AGE internal ID.
+            end_node_id: End vertex AGE internal ID.
+            rel_type: Filter by relationship type.
+            max_depth: Maximum path length.
+            **options: Additional options.
+
+        Returns:
+            Path dict with ``length``, ``nodes``, ``relationships``
+            or ``None`` if no path found.
+        """
+        try:
+            type_filter = f":{_sanitize_rel_type(rel_type)}" if rel_type else ""
+
+            cypher = (
+                f"MATCH path = shortestPath("
+                f"(s)-[r{type_filter}*..{int(max_depth)}]-(e)) "
+                f"WHERE id(s) = {int(start_node_id)} AND id(e) = {int(end_node_id)} "
+                f"RETURN path"
+            )
+
+            rows = self._execute_cypher(cypher, cols="path agtype")
+
+            if not rows:
+                return None
+
+            path_data = _parse_agtype(rows[0][0])
+
+            # AGE returns a path as a list [vertex, edge, vertex, edge, …]
+            nodes: List[Dict[str, Any]] = []
+            relationships: List[Dict[str, Any]] = []
+
+            if isinstance(path_data, list):
+                for i, element in enumerate(path_data):
+                    if not isinstance(element, dict):
+                        continue
+                    if "start_id" in element:
+                        relationships.append(_edge_to_rel_dict(element))
+                    elif "label" in element and "properties" in element:
+                        nodes.append(_vertex_to_node_dict(element))
+            elif isinstance(path_data, dict):
+                # Single hop — may be returned as dict with vertices/edges
+                for v in path_data.get("vertices", []):
+                    nodes.append(_vertex_to_node_dict(v))
+                for e in path_data.get("edges", []):
+                    relationships.append(_edge_to_rel_dict(e))
+
+            return {
+                "length": len(relationships),
+                "nodes": nodes,
+                "relationships": relationships,
+            }
+
+        except Exception as e:
+            raise ProcessingError(f"Failed to find shortest path: {str(e)}")
+
+    # ------------------------------------------------------------------
+    # Index and stats
+    # ------------------------------------------------------------------
+
+    def create_index(
+        self,
+        label: str,
+        property_name: str,
+        index_type: str = "btree",
+        **options,
+    ) -> bool:
+        """
+        Create a PostgreSQL index on a vertex property.
+
+        Since AGE stores vertex properties in a JSONB column, this creates
+        a PostgreSQL index on the underlying table using a JSONB expression.
+
+        Args:
+            label: Vertex label (used for table name resolution).
+            property_name: Property to index.
+            index_type: Index type (``"btree"``, ``"gin"``, ``"hash"``).
+            **options: Additional options (e.g. ``index_name``).
+
+        Returns:
+            ``True`` if the index was created successfully.
+        """
+        try:
+            self._ensure_connection()
+            safe_label = _sanitize_label(label)
+            if not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", property_name):
+                raise ValidationError(f"Invalid property name: '{property_name}'")
+
+            index_name = options.get(
+                "index_name", f"idx_{self.graph_name}_{safe_label}_{property_name}"
+            )
+            # Sanitise index name
+            index_name = re.sub(r"[^A-Za-z0-9_]", "_", index_name)
+
+            # AGE stores vertex data in schema-qualified tables
+            table_name = f'"{self.graph_name}"."{safe_label}"'
+
+            sql = (
+                f"CREATE INDEX IF NOT EXISTS {index_name} ON {table_name} "
+                f"USING {index_type} ((properties->>'{property_name}'));"
+            )
+
+            with self._conn.cursor() as cur:
+                cur.execute(sql)
+            self._conn.commit()
+            self.logger.info(
+                f"Created index {index_name} on {safe_label}.{property_name}"
+            )
+            return True
+
+        except (ProcessingError, ValidationError):
+            raise
+        except Exception as e:
+            if self._conn and not self._conn.closed:
+                self._conn.rollback()
+            raise ProcessingError(f"Failed to create index: {str(e)}")
+
+    def get_stats(self) -> Dict[str, Any]:
+        """
+        Get graph statistics.
+
+        Returns:
+            Dict with ``node_count``, ``relationship_count``,
+            ``label_counts``, and ``relationship_type_counts``.
+        """
+        try:
+            stats: Dict[str, Any] = {}
+
+            # Node count
+            rows = self._execute_cypher(
+                "MATCH (n) RETURN count(n)",
+                cols="count agtype",
+            )
+            stats["node_count"] = _parse_agtype(rows[0][0]) if rows else 0
+
+            # Relationship count
+            rows = self._execute_cypher(
+                "MATCH ()-[r]->() RETURN count(r)",
+                cols="count agtype",
+            )
+            stats["relationship_count"] = _parse_agtype(rows[0][0]) if rows else 0
+
+            # Label counts — query the AGE catalog
+            stats["label_counts"] = {}
+            try:
+                self._ensure_connection()
+                with self._conn.cursor() as cur:
+                    cur.execute(
+                        "SELECT name FROM ag_catalog.ag_label "
+                        "WHERE graph = (SELECT graphid FROM ag_catalog.ag_graph WHERE name = %s) "
+                        "AND kind = 'v' AND name != '_ag_label_vertex';",
+                        (self.graph_name,),
+                    )
+                    label_rows = cur.fetchall()
+                self._conn.commit()
+
+                for (lbl_name,) in label_rows:
+                    try:
+                        cnt_rows = self._execute_cypher(
+                            f"MATCH (n:{lbl_name}) RETURN count(n)",
+                            cols="count agtype",
+                        )
+                        stats["label_counts"][lbl_name] = (
+                            _parse_agtype(cnt_rows[0][0]) if cnt_rows else 0
+                        )
+                    except Exception:
+                        stats["label_counts"][lbl_name] = 0
+            except Exception:
+                pass
+
+            # Relationship type counts
+            stats["relationship_type_counts"] = {}
+            try:
+                self._ensure_connection()
+                with self._conn.cursor() as cur:
+                    cur.execute(
+                        "SELECT name FROM ag_catalog.ag_label "
+                        "WHERE graph = (SELECT graphid FROM ag_catalog.ag_graph WHERE name = %s) "
+                        "AND kind = 'e' AND name != '_ag_label_edge';",
+                        (self.graph_name,),
+                    )
+                    edge_label_rows = cur.fetchall()
+                self._conn.commit()
+
+                for (etype,) in edge_label_rows:
+                    try:
+                        cnt_rows = self._execute_cypher(
+                            f"MATCH ()-[r:{etype}]->() RETURN count(r)",
+                            cols="count agtype",
+                        )
+                        stats["relationship_type_counts"][etype] = (
+                            _parse_agtype(cnt_rows[0][0]) if cnt_rows else 0
+                        )
+                    except Exception:
+                        stats["relationship_type_counts"][etype] = 0
+            except Exception:
+                pass
+
+            return stats
+
+        except Exception as e:
+            self.logger.warning(f"Failed to get stats: {str(e)}")
+            return {"status": "error", "message": str(e)}
diff --git a/semantica/graph_store/config.py b/semantica/graph_store/config.py
index 387584cd..20b36aa8 100644
--- a/semantica/graph_store/config.py
+++ b/semantica/graph_store/config.py
@@ -138,6 +138,9 @@ class GraphStoreConfig:
             "AWS_ACCESS_KEY_ID": "neptune_access_key",
             "AWS_SECRET_ACCESS_KEY": "neptune_secret_key",
             "AWS_SESSION_TOKEN": "neptune_session_token",
+            # Apache AGE settings
+            "GRAPH_STORE_AGE_CONNECTION_STRING": "age_connection_string",
+            "GRAPH_STORE_AGE_GRAPH_NAME": "age_graph_name",
         }
 
         for env_var, config_key in env_mappings.items():
@@ -199,6 +202,9 @@ class GraphStoreConfig:
             "neptune_access_key": None,
             "neptune_secret_key": None,
             "neptune_session_token": None,
+            # Apache AGE defaults
+            "age_connection_string": "host=localhost dbname=agedb user=postgres password=postgres",
+            "age_graph_name": "semantica",
         }
 
         for key, default_value in defaults.items():
@@ -315,6 +321,18 @@ class GraphStoreConfig:
             "session_token": self._config.get("neptune_session_token"),
         }
 
+    def get_age_config(self) -> Dict[str, Any]:
+        """
+        Get Apache AGE-specific configuration.
+
+        Returns:
+            Apache AGE configuration dictionary
+        """
+        return {
+            "connection_string": self._config.get("age_connection_string"),
+            "graph_name": self._config.get("age_graph_name"),
+        }
+
     def reset(self) -> None:
         """Reset configuration to defaults."""
         self._config.clear()
diff --git a/semantica/graph_store/graph_store.py b/semantica/graph_store/graph_store.py
index 933b167a..2d185c19 100644
--- a/semantica/graph_store/graph_store.py
+++ b/semantica/graph_store/graph_store.py
@@ -583,6 +583,13 @@ class GraphStore:
             neptune_config.update(self.config)
             self._store_backend = AmazonNeptuneStore(**neptune_config)
 
+        elif self.backend == "age" or self.backend == "apache_age":
+            from .age_store import ApacheAgeStore
+
+            age_config = graph_store_config.get_age_config()
+            age_config.update(self.config)
+            self._store_backend = ApacheAgeStore(**age_config)
+
         else:
             raise ValidationError(f"Unknown backend: {self.backend}")
 
diff --git a/tests/graph_store/__init__.py b/tests/graph_store/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/graph_store/test_age_store.py b/tests/graph_store/test_age_store.py
new file mode 100644
index 00000000..16fda47b
--- /dev/null
+++ b/tests/graph_store/test_age_store.py
@@ -0,0 +1,772 @@
+"""
+Tests for Apache AGE Store Module
+
+Tests cover:
+    - Node CRUD (create, read, update, delete)
+    - Relationship CRUD
+    - Query execution
+    - Graph traversal (get_neighbors, shortest_path)
+    - Transaction rollback on error
+    - Multi-label handling
+    - ID separation (AGE internal vs semantic)
+    - Input validation / sanitisation
+    - Stats retrieval
+    - Index creation
+
+The psycopg2 database layer is fully mocked to enable offline testing.
+"""
+
+import json
+import unittest
+from typing import Any, Dict, List, Optional
+from unittest.mock import MagicMock, call, patch
+
+# ---------------------------------------------------------------------------
+# Mock psycopg2 before importing the module under test so that ``PSYCOPG2_AVAILABLE``
+# is ``True`` inside age_store.
+# ---------------------------------------------------------------------------
+import sys
+
+_mock_psycopg2 = MagicMock()
+_mock_psycopg2_extras = MagicMock()
+sys.modules["psycopg2"] = _mock_psycopg2
+sys.modules["psycopg2.extras"] = _mock_psycopg2_extras
+
+from semantica.graph_store.age_store import (
+    ApacheAgeStore,
+    _edge_to_rel_dict,
+    _parse_agtype,
+    _props_to_cypher_literal,
+    _sanitize_label,
+    _sanitize_rel_type,
+    _value_to_cypher_literal,
+    _vertex_to_node_dict,
+)
+from semantica.utils.exceptions import ProcessingError, ValidationError
+
+
+# ---------------------------------------------------------------------------
+# Helper fixtures
+# ---------------------------------------------------------------------------
+
+def _make_vertex_agtype(vid: int, label: str, props: Dict[str, Any]) -> str:
+    """Return a string mimicking AGE agtype vertex output."""
+    obj = {"id": vid, "label": label, "properties": props}
+    return json.dumps(obj) + "::vertex"
+
+
+def _make_edge_agtype(
+    eid: int, label: str, start_id: int, end_id: int, props: Dict[str, Any]
+) -> str:
+    """Return a string mimicking AGE agtype edge output."""
+    obj = {
+        "id": eid,
+        "label": label,
+        "start_id": start_id,
+        "end_id": end_id,
+        "properties": props,
+    }
+    return json.dumps(obj) + "::edge"
+
+
+# ---------------------------------------------------------------------------
+# Unit tests — helpers
+# ---------------------------------------------------------------------------
+
+class TestHelpers(unittest.TestCase):
+    """Tests for module-level helper functions."""
+
+    # -- _sanitize_label --------------------------------------------------
+
+    def test_sanitize_label_valid(self):
+        self.assertEqual(_sanitize_label("Person"), "Person")
+        self.assertEqual(_sanitize_label("_hidden"), "_hidden")
+        self.assertEqual(_sanitize_label("Rel_Type2"), "Rel_Type2")
+
+    def test_sanitize_label_invalid(self):
+        with self.assertRaises(ValidationError):
+            _sanitize_label("123bad")
+        with self.assertRaises(ValidationError):
+            _sanitize_label("no spaces")
+        with self.assertRaises(ValidationError):
+            _sanitize_label("no-dashes")
+
+    # -- _sanitize_rel_type -----------------------------------------------
+
+    def test_sanitize_rel_type_valid(self):
+        self.assertEqual(_sanitize_rel_type("KNOWS"), "KNOWS")
+
+    def test_sanitize_rel_type_invalid(self):
+        with self.assertRaises(ValidationError):
+            _sanitize_rel_type("bad type!")
+
+    # -- _value_to_cypher_literal -----------------------------------------
+
+    def test_literal_none(self):
+        self.assertEqual(_value_to_cypher_literal(None), "null")
+
+    def test_literal_bool(self):
+        self.assertEqual(_value_to_cypher_literal(True), "true")
+        self.assertEqual(_value_to_cypher_literal(False), "false")
+
+    def test_literal_int(self):
+        self.assertEqual(_value_to_cypher_literal(42), "42")
+
+    def test_literal_float(self):
+        self.assertIn("3.14", _value_to_cypher_literal(3.14))
+
+    def test_literal_string(self):
+        self.assertEqual(_value_to_cypher_literal("hello"), "'hello'")
+
+    def test_literal_string_escape(self):
+        result = _value_to_cypher_literal("it's a \"test\"")
+        self.assertIn("\\'", result)
+
+    def test_literal_list(self):
+        result = _value_to_cypher_literal([1, "a"])
+        self.assertEqual(result, "[1, 'a']")
+
+    def test_literal_dict(self):
+        result = _value_to_cypher_literal({"x": 1})
+        self.assertEqual(result, "{x: 1}")
+
+    # -- _props_to_cypher_literal -----------------------------------------
+
+    def test_props_empty(self):
+        self.assertEqual(_props_to_cypher_literal({}), "{}")
+
+    def test_props_simple(self):
+        result = _props_to_cypher_literal({"name": "Alice", "age": 30})
+        self.assertIn("name: 'Alice'", result)
+        self.assertIn("age: 30", result)
+
+    def test_props_invalid_key(self):
+        with self.assertRaises(ValidationError):
+            _props_to_cypher_literal({"bad key!": 1})
+
+    # -- _parse_agtype ----------------------------------------------------
+
+    def test_parse_agtype_none(self):
+        self.assertIsNone(_parse_agtype(None))
+
+    def test_parse_agtype_vertex(self):
+        text = '{"id": 1, "label": "Person", "properties": {"name": "Alice"}}::vertex'
+        result = _parse_agtype(text)
+        self.assertEqual(result["id"], 1)
+        self.assertEqual(result["label"], "Person")
+
+    def test_parse_agtype_edge(self):
+        text = '{"id": 10, "label": "KNOWS", "start_id": 1, "end_id": 2, "properties": {}}::edge'
+        result = _parse_agtype(text)
+        self.assertEqual(result["id"], 10)
+        self.assertEqual(result["start_id"], 1)
+
+    def test_parse_agtype_numeric(self):
+        self.assertEqual(_parse_agtype("42::numeric"), 42)
+        self.assertEqual(_parse_agtype("3.14::float"), 3.14)
+
+    def test_parse_agtype_boolean(self):
+        self.assertTrue(_parse_agtype("true::boolean"))
+        self.assertFalse(_parse_agtype("false::boolean"))
+
+    def test_parse_agtype_plain_json(self):
+        self.assertEqual(_parse_agtype('{"a": 1}'), {"a": 1})
+
+    def test_parse_agtype_non_string(self):
+        self.assertEqual(_parse_agtype(99), 99)
+
+    # -- _vertex_to_node_dict ---------------------------------------------
+
+    def test_vertex_to_node_dict_basic(self):
+        vertex = {"id": 5, "label": "Person", "properties": {"name": "Alice"}}
+        result = _vertex_to_node_dict(vertex)
+        self.assertEqual(result["id"], 5)
+        self.assertEqual(result["labels"], ["Person"])
+        self.assertEqual(result["properties"]["name"], "Alice")
+
+    def test_vertex_to_node_dict_extra_labels(self):
+        vertex = {
+            "id": 7,
+            "label": "Person",
+            "properties": {"name": "Bob", "labels": ["Employee", "Admin"]},
+        }
+        result = _vertex_to_node_dict(vertex)
+        self.assertEqual(result["labels"], ["Person", "Employee", "Admin"])
+        # 'labels' property should be removed from properties
+        self.assertNotIn("labels", result["properties"])
+
+    def test_vertex_to_node_dict_non_dict(self):
+        result = _vertex_to_node_dict("not a dict")
+        self.assertIsNone(result["id"])
+
+    # -- _edge_to_rel_dict ------------------------------------------------
+
+    def test_edge_to_rel_dict_basic(self):
+        edge = {
+            "id": 10,
+            "label": "KNOWS",
+            "start_id": 1,
+            "end_id": 2,
+            "properties": {"since": 2020},
+        }
+        result = _edge_to_rel_dict(edge)
+        self.assertEqual(result["id"], 10)
+        self.assertEqual(result["type"], "KNOWS")
+        self.assertEqual(result["start_node_id"], 1)
+        self.assertEqual(result["end_node_id"], 2)
+        self.assertEqual(result["properties"]["since"], 2020)
+
+    def test_edge_to_rel_dict_non_dict(self):
+        result = _edge_to_rel_dict(42)
+        self.assertIsNone(result["id"])
+
+
+# ---------------------------------------------------------------------------
+# Unit tests — ApacheAgeStore with mocked DB
+# ---------------------------------------------------------------------------
+
+class TestApacheAgeStore(unittest.TestCase):
+    """Tests for ApacheAgeStore with a fully mocked psycopg2 connection."""
+
+    def setUp(self):
+        """Set up a store with a mocked PostgreSQL connection."""
+        self.mock_conn = MagicMock()
+        self.mock_conn.closed = False
+        self.mock_cursor = MagicMock()
+        self.mock_conn.cursor.return_value.__enter__ = MagicMock(
+            return_value=self.mock_cursor
+        )
+        self.mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
+
+        # Patch psycopg2.connect to return the mock connection
+        _mock_psycopg2.connect.return_value = self.mock_conn
+
+        self.store = ApacheAgeStore(
+            connection_string="host=localhost dbname=testdb user=test password=test",
+            graph_name="test_graph",
+        )
+        # Simulate a successful connect (graph already exists)
+        self.mock_cursor.fetchone.return_value = (1,)  # graph exists
+        self.store.connect()
+        # Reset mock call history after connect
+        self.mock_cursor.reset_mock()
+        self.mock_conn.reset_mock()
+        self.mock_conn.closed = False
+
+    def tearDown(self):
+        self.store.close()
+
+    # -- connect ----------------------------------------------------------
+
+    def test_connect_creates_extension_and_graph(self):
+        """connect() should run idempotent setup commands."""
+        store = ApacheAgeStore(
+            connection_string="host=localhost dbname=agedb user=test",
+            graph_name="new_graph",
+        )
+        # Graph does NOT exist yet
+        self.mock_cursor.fetchone.return_value = (0,)
+
+        result = store.connect()
+        self.assertTrue(result)
+
+        # Verify setup SQL was executed
+        executed = [
+            str(c) for c in self.mock_cursor.execute.call_args_list
+        ]
+        setup_text = " ".join(executed)
+        self.assertIn("CREATE EXTENSION IF NOT EXISTS age", setup_text)
+        self.assertIn("LOAD 'age'", setup_text)
+        self.assertIn("search_path", setup_text)
+        self.assertIn("create_graph", setup_text)
+
+    def test_connect_idempotent_existing_graph(self):
+        """connect() should skip create_graph if graph already exists."""
+        store = ApacheAgeStore(
+            connection_string="host=localhost dbname=agedb user=test",
+            graph_name="existing_graph",
+        )
+        self.mock_cursor.fetchone.return_value = (1,)  # graph exists
+
+        result = store.connect()
+        self.assertTrue(result)
+
+        executed = [
+            str(c) for c in self.mock_cursor.execute.call_args_list
+        ]
+        setup_text = " ".join(executed)
+        # create_graph should NOT be called (graph count=1)
+        self.assertNotIn("create_graph", setup_text.split("ag_graph")[1] if "ag_graph" in setup_text else "")
+
+    # -- create_node ------------------------------------------------------
+
+    def test_create_node_single_label(self):
+        """create_node with one label should use it as AGE label."""
+        vertex_str = _make_vertex_agtype(100, "Person", {"name": "Alice"})
+        self.mock_cursor.fetchall.return_value = [(vertex_str,)]
+
+        node = self.store.create_node(
+            labels=["Person"], properties={"name": "Alice"}
+        )
+
+        self.assertEqual(node["id"], 100)
+        self.assertEqual(node["labels"], ["Person"])
+        self.assertEqual(node["properties"]["name"], "Alice")
+
+    def test_create_node_multiple_labels(self):
+        """Additional labels beyond the first are stored as property array."""
+        vertex_str = _make_vertex_agtype(
+            101, "Person", {"name": "Bob", "labels": ["Employee", "Admin"]}
+        )
+        self.mock_cursor.fetchall.return_value = [(vertex_str,)]
+
+        node = self.store.create_node(
+            labels=["Person", "Employee", "Admin"],
+            properties={"name": "Bob"},
+        )
+
+        self.assertEqual(node["id"], 101)
+        self.assertIn("Person", node["labels"])
+        self.assertIn("Employee", node["labels"])
+        self.assertIn("Admin", node["labels"])
+        # 'labels' property should be moved out of properties
+        self.assertNotIn("labels", node["properties"])
+
+    def test_create_node_with_semantica_id(self):
+        """semantica_id in properties should be preserved."""
+        vertex_str = _make_vertex_agtype(
+            102, "Entity", {"semantica_id": "abc-123", "value": "test"}
+        )
+        self.mock_cursor.fetchall.return_value = [(vertex_str,)]
+
+        node = self.store.create_node(
+            labels=["Entity"],
+            properties={"semantica_id": "abc-123", "value": "test"},
+        )
+
+        self.assertEqual(node["id"], 102)  # AGE internal ID
+        self.assertEqual(node["properties"]["semantica_id"], "abc-123")
+
+    def test_create_node_empty_labels_raises(self):
+        """create_node with empty labels should raise ValidationError."""
+        with self.assertRaises(ValidationError):
+            self.store.create_node(labels=[], properties={"name": "X"})
+
+    def test_create_node_invalid_label_raises(self):
+        """create_node with invalid label should raise ValidationError."""
+        with self.assertRaises(ValidationError):
+            self.store.create_node(labels=["bad label!"], properties={})
+
+    # -- create_nodes -----------------------------------------------------
+
+    def test_create_nodes_batch(self):
+        """create_nodes should create multiple nodes."""
+        responses = [
+            [(_make_vertex_agtype(200, "Person", {"name": "A"}),)],
+            [(_make_vertex_agtype(201, "Person", {"name": "B"}),)],
+        ]
+        self.mock_cursor.fetchall.side_effect = responses
+
+        nodes_data = [
+            {"labels": ["Person"], "properties": {"name": "A"}},
+            {"labels": ["Person"], "properties": {"name": "B"}},
+        ]
+        result = self.store.create_nodes(nodes_data)
+
+        self.assertEqual(len(result), 2)
+        self.assertEqual(result[0]["id"], 200)
+        self.assertEqual(result[1]["id"], 201)
+
+    # -- get_node ---------------------------------------------------------
+
+    def test_get_node_found(self):
+        vertex_str = _make_vertex_agtype(100, "Person", {"name": "Alice"})
+        self.mock_cursor.fetchall.return_value = [(vertex_str,)]
+
+        node = self.store.get_node(100)
+
+        self.assertIsNotNone(node)
+        self.assertEqual(node["id"], 100)
+        self.assertEqual(node["properties"]["name"], "Alice")
+
+    def test_get_node_not_found(self):
+        self.mock_cursor.fetchall.return_value = []
+
+        node = self.store.get_node(999)
+        self.assertIsNone(node)
+
+    # -- get_nodes --------------------------------------------------------
+
+    def test_get_nodes_with_label_filter(self):
+        vertex_str = _make_vertex_agtype(100, "Person", {"name": "Alice"})
+        self.mock_cursor.fetchall.return_value = [(vertex_str,)]
+
+        nodes = self.store.get_nodes(labels=["Person"])
+
+        self.assertEqual(len(nodes), 1)
+        self.assertEqual(nodes[0]["labels"], ["Person"])
+
+    def test_get_nodes_with_property_filter(self):
+        vertex_str = _make_vertex_agtype(100, "Person", {"name": "Alice"})
+        self.mock_cursor.fetchall.return_value = [(vertex_str,)]
+
+        nodes = self.store.get_nodes(properties={"name": "Alice"})
+        self.assertEqual(len(nodes), 1)
+
+    def test_get_nodes_empty(self):
+        self.mock_cursor.fetchall.return_value = []
+        nodes = self.store.get_nodes()
+        self.assertEqual(nodes, [])
+
+    # -- update_node ------------------------------------------------------
+
+    def test_update_node_merge(self):
+        vertex_str = _make_vertex_agtype(100, "Person", {"name": "Alice", "age": 31})
+        self.mock_cursor.fetchall.return_value = [(vertex_str,)]
+
+        node = self.store.update_node(100, {"age": 31}, merge=True)
+
+        self.assertEqual(node["id"], 100)
+        self.assertEqual(node["properties"]["age"], 31)
+        self.assertEqual(node["properties"]["name"], "Alice")
+
+        # Verify the Cypher used += for merge
+        executed_sql = self.mock_cursor.execute.call_args[0][0]
+        self.assertIn("+=", executed_sql)
+
+    def test_update_node_replace(self):
+        vertex_str = _make_vertex_agtype(100, "Person", {"age": 31})
+        self.mock_cursor.fetchall.return_value = [(vertex_str,)]
+
+        node = self.store.update_node(100, {"age": 31}, merge=False)
+        self.assertEqual(node["properties"]["age"], 31)
+
+        # Verify SET n = (not +=) for replace
+        executed_sql = self.mock_cursor.execute.call_args[0][0]
+        self.assertIn("SET n =", executed_sql)
+        self.assertNotIn("+=", executed_sql)
+
+    def test_update_node_not_found(self):
+        self.mock_cursor.fetchall.return_value = []
+
+        with self.assertRaises(ProcessingError):
+            self.store.update_node(999, {"age": 31})
+
+    # -- delete_node ------------------------------------------------------
+
+    def test_delete_node_detach(self):
+        self.mock_cursor.fetchall.return_value = []
+
+        result = self.store.delete_node(100, detach=True)
+        self.assertTrue(result)
+
+        executed_sql = self.mock_cursor.execute.call_args[0][0]
+        self.assertIn("DETACH DELETE", executed_sql)
+
+    def test_delete_node_no_detach(self):
+        self.mock_cursor.fetchall.return_value = []
+
+        result = self.store.delete_node(100, detach=False)
+        self.assertTrue(result)
+
+        executed_sql = self.mock_cursor.execute.call_args[0][0]
+        self.assertIn("DELETE", executed_sql)
+        self.assertNotIn("DETACH", executed_sql)
+
+    # -- create_relationship ----------------------------------------------
+
+    def test_create_relationship(self):
+        edge_str = _make_edge_agtype(500, "KNOWS", 100, 200, {"since": 2023})
+        self.mock_cursor.fetchall.return_value = [(edge_str,)]
+
+        rel = self.store.create_relationship(100, 200, "KNOWS", {"since": 2023})
+
+        self.assertEqual(rel["id"], 500)
+        self.assertEqual(rel["type"], "KNOWS")
+        self.assertEqual(rel["start_node_id"], 100)
+        self.assertEqual(rel["end_node_id"], 200)
+        self.assertEqual(rel["properties"]["since"], 2023)
+
+    def test_create_relationship_no_properties(self):
+        edge_str = _make_edge_agtype(501, "FOLLOWS", 100, 200, {})
+        self.mock_cursor.fetchall.return_value = [(edge_str,)]
+
+        rel = self.store.create_relationship(100, 200, "FOLLOWS")
+        self.assertEqual(rel["type"], "FOLLOWS")
+        self.assertEqual(rel["properties"], {})
+
+    def test_create_relationship_invalid_type_raises(self):
+        with self.assertRaises(ValidationError):
+            self.store.create_relationship(100, 200, "BAD TYPE!")
+
+    # -- get_relationships ------------------------------------------------
+
+    def test_get_relationships_outgoing(self):
+        edge_str = _make_edge_agtype(500, "KNOWS", 100, 200, {})
+        self.mock_cursor.fetchall.return_value = [(edge_str,)]
+
+        rels = self.store.get_relationships(
+            node_id=100, rel_type="KNOWS", direction="out"
+        )
+
+        self.assertEqual(len(rels), 1)
+        self.assertEqual(rels[0]["type"], "KNOWS")
+
+    def test_get_relationships_incoming(self):
+        edge_str = _make_edge_agtype(501, "KNOWS", 200, 100, {})
+        self.mock_cursor.fetchall.return_value = [(edge_str,)]
+
+        rels = self.store.get_relationships(
+            node_id=100, direction="in"
+        )
+        self.assertEqual(len(rels), 1)
+
+    def test_get_relationships_all(self):
+        self.mock_cursor.fetchall.return_value = []
+        rels = self.store.get_relationships()
+        self.assertEqual(rels, [])
+
+    # -- delete_relationship ----------------------------------------------
+
+    def test_delete_relationship(self):
+        self.mock_cursor.fetchall.return_value = []
+
+        result = self.store.delete_relationship(500)
+        self.assertTrue(result)
+
+    # -- execute_query ----------------------------------------------------
+
+    def test_execute_query_basic(self):
+        """execute_query should return Neo4jStore-compatible result dict."""
+        self.mock_cursor.description = [("n",)]
+        vertex_str = _make_vertex_agtype(100, "Person", {"name": "Alice"})
+        self.mock_cursor.fetchall.return_value = [(vertex_str,)]
+
+        result = self.store.execute_query(
+            "MATCH (n:Person) RETURN n", cols="n agtype"
+        )
+
+        self.assertTrue(result["success"])
+        self.assertEqual(len(result["records"]), 1)
+        self.assertIn("keys", result)
+        self.assertIn("metadata", result)
+        self.assertEqual(result["metadata"]["query"], "MATCH (n:Person) RETURN n")
+
+    def test_execute_query_with_parameters(self):
+        """Parameters should be substituted as safe literals."""
+        self.mock_cursor.description = [("count",)]
+        self.mock_cursor.fetchall.return_value = [("5::numeric",)]
+
+        result = self.store.execute_query(
+            "MATCH (n) WHERE n.age > $min_age RETURN count(n) AS count",
+            parameters={"min_age": 25},
+            cols="count agtype",
+        )
+
+        self.assertTrue(result["success"])
+        # Check the SQL that was executed contained the literal, not $min_age
+        executed_sql = self.mock_cursor.execute.call_args[0][0]
+        self.assertIn("25", executed_sql)
+        self.assertNotIn("$min_age", executed_sql)
+
+    def test_execute_query_empty_result(self):
+        self.mock_cursor.description = []
+        self.mock_cursor.fetchall.return_value = []
+
+        result = self.store.execute_query("MATCH (n) RETURN n", cols="n agtype")
+        self.assertTrue(result["success"])
+        self.assertEqual(result["records"], [])
+
+    # -- get_neighbors ----------------------------------------------------
+
+    def test_get_neighbors_out(self):
+        vertex_str = _make_vertex_agtype(200, "Person", {"name": "Bob"})
+        self.mock_cursor.fetchall.return_value = [(vertex_str,)]
+
+        neighbors = self.store.get_neighbors(
+            node_id=100, rel_type="KNOWS", direction="out", depth=2
+        )
+
+        self.assertEqual(len(neighbors), 1)
+        self.assertEqual(neighbors[0]["id"], 200)
+
+    def test_get_neighbors_both(self):
+        self.mock_cursor.fetchall.return_value = []
+        neighbors = self.store.get_neighbors(node_id=100)
+        self.assertEqual(neighbors, [])
+
+    # -- shortest_path ----------------------------------------------------
+
+    def test_shortest_path_found(self):
+        """When a path is found, it should be returned as dict with nodes/relationships."""
+        path_list = [
+            {"id": 1, "label": "Person", "properties": {"name": "A"}},
+            {"id": 10, "label": "KNOWS", "start_id": 1, "end_id": 2, "properties": {}},
+            {"id": 2, "label": "Person", "properties": {"name": "B"}},
+        ]
+        path_str = json.dumps(path_list) + "::path"
+        self.mock_cursor.fetchall.return_value = [(path_str,)]
+
+        result = self.store.shortest_path(1, 2)
+
+        self.assertIsNotNone(result)
+        self.assertEqual(result["length"], 1)
+        self.assertEqual(len(result["nodes"]), 2)
+        self.assertEqual(len(result["relationships"]), 1)
+
+    def test_shortest_path_not_found(self):
+        self.mock_cursor.fetchall.return_value = []
+
+        result = self.store.shortest_path(1, 999)
+        self.assertIsNone(result)
+
+    # -- create_index -----------------------------------------------------
+
+    def test_create_index(self):
+        result = self.store.create_index("Person", "name", "btree")
+        self.assertTrue(result)
+
+        executed_sql = self.mock_cursor.execute.call_args[0][0]
+        self.assertIn("CREATE INDEX IF NOT EXISTS", executed_sql)
+        self.assertIn("Person", executed_sql)
+        self.assertIn("name", executed_sql)
+
+    def test_create_index_invalid_property_raises(self):
+        with self.assertRaises(ValidationError):
+            self.store.create_index("Person", "bad name!", "btree")
+
+    # -- get_stats --------------------------------------------------------
+
+    def test_get_stats(self):
+        """get_stats should return structured dict."""
+        # Mock call sequence:
+        # 1. node count cypher → fetchall
+        # 2. relationship count cypher → fetchall
+        # 3. label catalog query → fetchall (pg cursor)
+        #    then per-label cypher → fetchall
+        # 4. edge type catalog query → fetchall (pg cursor)
+        #    then per-type cypher → fetchall
+
+        call_count = [0]
+        fetch_responses = [
+            [("42::numeric",)],   # node count
+            [("10::numeric",)],   # relationship count
+            [("5::numeric",)],    # label count for Person
+            [("10::numeric",)],   # edge count for KNOWS
+        ]
+        cursor_fetch_responses = [
+            (1,),                  # ensure_connection: ag_graph check (not used here)
+            [("Person",)],        # label catalog
+            [("KNOWS",)],         # edge type catalog
+        ]
+
+        def mock_fetchall():
+            idx = call_count[0]
+            call_count[0] += 1
+            if idx < len(fetch_responses):
+                return fetch_responses[idx]
+            return []
+
+        def mock_cursor_fetchall():
+            # Returns for the catalog queries
+            if not hasattr(mock_cursor_fetchall, "_idx"):
+                mock_cursor_fetchall._idx = 0
+            idx = mock_cursor_fetchall._idx
+            mock_cursor_fetchall._idx += 1
+            if idx < len(cursor_fetch_responses):
+                return cursor_fetch_responses[idx]
+            return []
+
+        self.mock_cursor.fetchall.side_effect = mock_fetchall
+        self.mock_cursor.fetchone.return_value = (1,)
+
+        stats = self.store.get_stats()
+
+        self.assertIn("node_count", stats)
+        self.assertIn("relationship_count", stats)
+        self.assertIn("label_counts", stats)
+        self.assertIn("relationship_type_counts", stats)
+
+    # -- Transaction rollback ---------------------------------------------
+
+    def test_cypher_execution_rollback_on_error(self):
+        """If a query fails, the connection should be rolled back."""
+        self.mock_cursor.execute.side_effect = Exception("SQL error")
+
+        with self.assertRaises(ProcessingError):
+            self.store.get_node(100)
+
+        self.mock_conn.rollback.assert_called()
+
+    def test_create_node_db_error_raises(self):
+        """Database errors during create_node should raise ProcessingError."""
+        self.mock_cursor.execute.side_effect = Exception("Disk full")
+
+        with self.assertRaises(ProcessingError):
+            self.store.create_node(["Test"], {"key": "val"})
+
+    # -- close ------------------------------------------------------------
+
+    def test_close(self):
+        self.store.close()
+        self.assertIsNone(self.store._conn)
+
+    def test_close_idempotent(self):
+        """Calling close() twice should not raise."""
+        self.store.close()
+        self.store.close()  # Should not raise
+
+
+# ---------------------------------------------------------------------------
+# Integration-style test with GraphStore facade
+# ---------------------------------------------------------------------------
+
+class TestGraphStoreFacadeAge(unittest.TestCase):
+    """Test that GraphStore(backend='age') initialises ApacheAgeStore."""
+
+    @patch("semantica.graph_store.age_store.ApacheAgeStore", autospec=True)
+    def test_age_backend_initialisation(self, MockAgeStore):
+        """GraphStore should instantiate ApacheAgeStore for 'age' backend."""
+        from semantica.graph_store.graph_store import GraphStore
+
+        mock_instance = MagicMock()
+        MockAgeStore.return_value = mock_instance
+
+        store = GraphStore(backend="age")
+        self.assertIs(store._store_backend, mock_instance)
+
+    @patch("semantica.graph_store.age_store.ApacheAgeStore", autospec=True)
+    def test_apache_age_backend_alias(self, MockAgeStore):
+        """GraphStore should accept 'apache_age' as backend alias."""
+        from semantica.graph_store.graph_store import GraphStore
+
+        mock_instance = MagicMock()
+        MockAgeStore.return_value = mock_instance
+
+        store = GraphStore(backend="apache_age")
+        self.assertIs(store._store_backend, mock_instance)
+
+
+# ---------------------------------------------------------------------------
+# Return format conformance tests
+# ---------------------------------------------------------------------------
+
+class TestReturnFormatConformance(unittest.TestCase):
+    """Verify that returned dicts match Neo4jStore structure exactly."""
+
+    def test_node_return_keys(self):
+        vertex = {"id": 1, "label": "X", "properties": {"a": 1}}
+        result = _vertex_to_node_dict(vertex)
+        self.assertSetEqual(set(result.keys()), {"id", "labels", "properties"})
+
+    def test_relationship_return_keys(self):
+        edge = {"id": 1, "label": "R", "start_id": 2, "end_id": 3, "properties": {}}
+        result = _edge_to_rel_dict(edge)
+        self.assertSetEqual(
+            set(result.keys()),
+            {"id", "type", "start_node_id", "end_node_id", "properties"},
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()