feat(context): add ContextGraph Markdown round-trip

docs/guides/context-graphs.md

@@ -436,6 +436,29 @@ d = graph.to_dict()
 # d["statistics"] → {"node_count": int, "edge_count": int}
 ```
 
+For a human-editable, version-control-friendly representation, save a Markdown
+directory instead:
+
+```python
+graph.save_to_file("context_graph/", format="markdown")
+
+restored = ContextGraph(advanced_analytics=True)
+restored.load_from_file("context_graph/", format="markdown")
+```
+
+The directory contains a versioned `graph.md` manifest for graph identity,
+relationships, and cross-graph link descriptors, plus one file per node under
+`nodes/`. A node's content is its Markdown body; its ID, type, properties,
+metadata, and temporal validity are YAML frontmatter. Node, edge, family, graph,
+and cross-graph link IDs are preserved across round trips.
+
+Markdown loading uses replacement semantics, like `from_dict()`: it parses and
+validates the complete directory before replacing the current graph. Invalid YAML,
+duplicate IDs, dangling edge endpoints, unsupported versions, and unsafe symbolic
+links fail without partially mutating the graph. Re-exporting to an existing managed
+directory atomically replaces it, removing stale node files. To avoid accidental data
+loss, a non-empty directory without the ContextGraph manifest is never replaced.
+
 If the graph had cross-graph links created with `link_graph()`, call `resolve_links()` after loading to restore live navigation — object references cannot be serialized, so they must be reconnected manually:
 
 ```python

docs/reference/context.md

@@ -435,8 +435,8 @@ print("Nodes: {}, Edges: {}".format(stats["node_count"], stats["edge_count"]))
 | `query(query, skip, limit)` | `List[Dict]` | Full-text search over node content |
 | `stats()` | `Dict` | Node/edge counts, type breakdowns, graph density |
 | `density()` | `float` | Graph density score |
-| `save_to_file(path)` | `None` | Persist graph to JSON |
-| `load_from_file(path)` | `None` | Load graph from JSON |
+| `save_to_file(path, format="json")` | `None` | Persist graph as JSON or a Markdown directory |
+| `load_from_file(path, format="json")` | `None` | Replace graph state from JSON or a Markdown directory |
 | `build_from_conversations(conversations, link_entities)` | `Dict` | Build graph from conversation data |
 | `link_graph(other_graph, source_node_id, target_node_id, link_type)` | `str` | Create cross-graph navigation link; returns `link_id` |
 | `navigate_to(link_id)` | `Tuple` | Follow a cross-graph link to `(target_graph, target_node_id)` |

semantica/context/context_graph.py

@@ -106,20 +106,68 @@ Production Use Cases:
 """
 
 from collections import defaultdict, deque
+import copy
 from dataclasses import dataclass, field
-from datetime import datetime, timezone
+from datetime import date, datetime, timezone
+import errno
+import hashlib
 import json
+import os
+from pathlib import Path
+import re
+import shutil
+import stat
+import tempfile
 import threading
 import itertools
 from typing import Any, Dict, List, Optional, Set, Tuple, Union
 import uuid
 
+import yaml
+
 from ..utils.logging import get_logger
 from ..utils.progress_tracker import get_progress_tracker
 from ..utils.helpers import classify_path_distance
 from ..utils.skos import is_skos_hierarchy_edge, validate_skos_hierarchy
 from .entity_linker import EntityLinker
 
+
+class _UniqueKeySafeLoader(yaml.SafeLoader):
+    """Safe YAML loader that rejects ambiguous duplicate mapping keys."""
+
+
+def _construct_unique_mapping(
+    loader: _UniqueKeySafeLoader, node: yaml.MappingNode, deep: bool = False
+) -> Dict[Any, Any]:
+    loader.flatten_mapping(node)
+    mapping = {}
+    for key_node, value_node in node.value:
+        key = loader.construct_object(key_node, deep=deep)
+        try:
+            duplicate = key in mapping
+        except TypeError as exc:
+            raise yaml.constructor.ConstructorError(
+                "while constructing a mapping",
+                node.start_mark,
+                "found an unhashable key",
+                key_node.start_mark,
+            ) from exc
+        if duplicate:
+            raise yaml.constructor.ConstructorError(
+                "while constructing a mapping",
+                node.start_mark,
+                f"found duplicate key {key!r}",
+                key_node.start_mark,
+            )
+        mapping[key] = loader.construct_object(value_node, deep=deep)
+    return mapping
+
+
+_UniqueKeySafeLoader.add_constructor(
+    yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, _construct_unique_mapping
+)
+
+
 # Optional imports for advanced features
 try:
     from ..kg import (
@@ -424,6 +472,12 @@ class ContextGraph:
     Perfect for building intelligent AI agents that can learn from decisions!
     """
 
+    _MARKDOWN_FORMAT = "semantica-context-graph"
+    _MARKDOWN_VERSION = 1
+    _MARKDOWN_MANIFEST = "graph.md"
+    _MARKDOWN_NODES_DIRECTORY = "nodes"
+    _MARKDOWN_EXTENSIONS = frozenset({".md", ".markdown"})
+
     def __init__(self, config: Optional[Dict[str, Any]] = None, **kwargs):
         """
         Initialize context graph with optional advanced features.
@@ -977,14 +1031,21 @@ class ContextGraph:
                 )
             )
 
-    def save_to_file(self, path: str) -> None:
+    def save_to_file(
+        self, path: Union[str, Path], format: str = "json"
+    ) -> None:
         """
-        Save context graph to file (JSON format).
+        Save the context graph in JSON or Markdown format.
 
         Args:
-            path: File path to save to
+            path: JSON file path or Markdown export directory
+            format: Persistence format (``json`` or ``markdown``)
         """
-        import json
+        normalized_format = self._normalize_persistence_format(format)
+        if normalized_format == "markdown":
+            self._save_markdown_directory(Path(path))
+            self.logger.info(f"Saved context graph Markdown to {path}")
+            return
 
         with self._lock:
     
@@ -1011,15 +1072,21 @@ class ContextGraph:
 
         self.logger.info(f"Saved context graph to {path}")
 
-    def load_from_file(self, path: str) -> None:
+    def load_from_file(
+        self, path: Union[str, Path], format: str = "json"
+    ) -> None:
         """
-        Load context graph from file (JSON format).
+        Load the context graph from JSON or Markdown.
 
         Args:
-            path: File path to load from
+            path: JSON file path or Markdown export directory
+            format: Persistence format (``json`` or ``markdown``)
         """
-        import json
-        import os
+        normalized_format = self._normalize_persistence_format(format)
+        if normalized_format == "markdown":
+            self._load_markdown_directory(Path(path))
+            self.logger.info(f"Loaded context graph Markdown from {path}")
+            return
 
         if not os.path.exists(path):
             self.logger.warning(f"File not found: {path}")
@@ -1079,6 +1146,653 @@ class ContextGraph:
 
         self.logger.info(f"Loaded context graph from {path}")
 
+    @staticmethod
+    def _normalize_persistence_format(format: str) -> str:
+        if not isinstance(format, str) or not format.strip():
+            raise ValueError("Context graph persistence format must be a string.")
+        normalized = format.strip().lower()
+        if normalized not in {"json", "markdown"}:
+            raise ValueError(
+                f"Unsupported context graph persistence format: {format!r}. "
+                "Expected 'json' or 'markdown'."
+            )
+        return normalized
+
+    def _save_markdown_directory(self, destination: Path) -> None:
+        if not destination.name:
+            raise ValueError("Markdown export destination cannot be a filesystem root.")
+        if destination.is_symlink():
+            raise ValueError(
+                f"Refusing to replace Markdown symbolic link: {destination}"
+            )
+        if destination.exists() and not destination.is_dir():
+            raise ValueError(
+                f"Markdown export destination is not a directory: {destination}"
+            )
+        if destination.exists() and any(destination.iterdir()):
+            manifest = destination / self._MARKDOWN_MANIFEST
+            try:
+                document = self._read_markdown_file(manifest)
+                frontmatter, _ = self._parse_markdown_document(
+                    document, str(manifest)
+                )
+                is_managed = (
+                    frontmatter.get("format") == self._MARKDOWN_FORMAT
+                    and not isinstance(frontmatter.get("version"), bool)
+                    and frontmatter.get("version") == self._MARKDOWN_VERSION
+                )
+            except (FileNotFoundError, ValueError):
+                is_managed = False
+            if not is_managed:
+                raise ValueError(
+                    "Refusing to replace a non-empty directory that is not a "
+                    f"managed ContextGraph export: {destination}"
+                )
+
+        manifest_document, node_documents = self._markdown_documents()
+        destination.parent.mkdir(parents=True, exist_ok=True)
+        staging_path = Path(
+            tempfile.mkdtemp(
+                dir=str(destination.parent), prefix=f".{destination.name}.staging-"
+            )
+        )
+        backup_path: Optional[Path] = None
+        try:
+            nodes_path = staging_path / self._MARKDOWN_NODES_DIRECTORY
+            nodes_path.mkdir()
+            self._write_staged_markdown(
+                staging_path / self._MARKDOWN_MANIFEST, manifest_document
+            )
+            for filename, document in node_documents:
+                self._write_staged_markdown(nodes_path / filename, document)
+
+            if destination.exists():
+                backup_path = destination.parent / (
+                    f".{destination.name}.backup-{uuid.uuid4().hex}"
+                )
+                os.replace(destination, backup_path)
+            try:
+                os.replace(staging_path, destination)
+                staging_path = None
+            except BaseException:
+                if backup_path is not None and not destination.exists():
+                    os.replace(backup_path, destination)
+                    backup_path = None
+                raise
+
+            if backup_path is not None:
+                shutil.rmtree(backup_path)
+                backup_path = None
+        finally:
+            if staging_path is not None:
+                shutil.rmtree(staging_path, ignore_errors=True)
+            if backup_path is not None and backup_path.exists():
+                self.logger.warning(
+                    "ContextGraph export left backup directory %s", backup_path
+                )
+
+    @staticmethod
+    def _write_staged_markdown(path: Path, document: str) -> None:
+        with path.open("x", encoding="utf-8") as output:
+            output.write(document)
+            output.flush()
+            os.fsync(output.fileno())
+
+    def _markdown_documents(self) -> Tuple[str, List[Tuple[str, str]]]:
+        with self._lock:
+            nodes = [
+                {
+                    "id": node.node_id,
+                    "type": node.node_type,
+                    "properties": copy.deepcopy(node.properties),
+                    "metadata": copy.deepcopy(node.metadata),
+                    "valid_from": node.valid_from,
+                    "valid_until": node.valid_until,
+                    "content": node.content,
+                }
+                for node in self.nodes.values()
+            ]
+            edges = [
+                {
+                    "id": edge.edge_id,
+                    "family_id": edge.family_id or edge.edge_id,
+                    "source": edge.source_id,
+                    "target": edge.target_id,
+                    "type": edge.edge_type,
+                    "weight": edge.weight,
+                    "metadata": copy.deepcopy(edge.metadata),
+                    "valid_from": edge.valid_from,
+                    "valid_until": edge.valid_until,
+                }
+                for edge in self.edges
+            ]
+            graph_id = self.graph_id
+            links_by_id = {
+                link_id: copy.deepcopy(link)
+                for link_id, link in self._unresolved_links.items()
+            }
+            for link_id, (
+                other_graph,
+                source_node_id,
+                target_node_id,
+            ) in self._linked_graphs.items():
+                links_by_id[link_id] = {
+                    "link_id": link_id,
+                    "source_node_id": source_node_id,
+                    "target_node_id": target_node_id,
+                    "other_graph_id": other_graph.graph_id,
+                }
+
+        edges.sort(
+            key=lambda edge: (
+                str(edge["id"]),
+                str(edge["source"]),
+                str(edge["target"]),
+                str(edge["type"]),
+            )
+        )
+        links = sorted(
+            links_by_id.values(), key=lambda link: str(link.get("link_id", ""))
+        )
+        manifest = {
+            "format": self._MARKDOWN_FORMAT,
+            "version": self._MARKDOWN_VERSION,
+            "graph_id": graph_id,
+            "edges": edges,
+            "links": links,
+        }
+        manifest_body = (
+            "# Context Graph\n\n"
+            "Graph relationships are stored in frontmatter. Node content is in "
+            "the `nodes` directory.\n"
+        )
+        manifest_document = self._render_markdown_document(
+            manifest, manifest_body, "graph manifest"
+        )
+
+        node_documents = []
+        filenames = set()
+        for node in sorted(nodes, key=lambda item: str(item["id"])):
+            filename = self._node_markdown_filename(node["id"])
+            normalized_filename = filename.casefold()
+            if normalized_filename in filenames:
+                raise ValueError(
+                    f"Cannot export ContextGraph: duplicate filename {filename!r}."
+                )
+            filenames.add(normalized_filename)
+            content = node.pop("content")
+            node_documents.append(
+                (
+                    filename,
+                    self._render_markdown_document(
+                        node, content, f"node {node['id']!r}"
+                    ),
+                )
+            )
+        return manifest_document, node_documents
+
+    @staticmethod
+    def _node_markdown_filename(node_id: Any) -> str:
+        if not isinstance(node_id, str) or not node_id.strip():
+            raise ValueError("Cannot export a ContextGraph node without a string ID.")
+        slug = re.sub(r"[^A-Za-z0-9._-]+", "-", node_id)
+        slug = re.sub(r"-+", "-", slug).strip("._-")[:80].rstrip("._-")
+        slug = slug or "node"
+        digest = hashlib.sha256(node_id.encode("utf-8")).hexdigest()[:12]
+        return f"{slug}--{digest}.md"
+
+    @classmethod
+    def _render_markdown_document(
+        cls, frontmatter: Dict[str, Any], body: str, source: str
+    ) -> str:
+        if not isinstance(body, str):
+            raise ValueError(f"Cannot export {source}: Markdown body must be a string.")
+        canonical = cls._canonical_markdown_value(frontmatter, source)
+        try:
+            yaml_text = yaml.safe_dump(
+                canonical,
+                sort_keys=False,
+                allow_unicode=True,
+                default_flow_style=False,
+            )
+        except yaml.YAMLError as exc:
+            raise ValueError(
+                f"Cannot export {source}: metadata is not YAML serializable."
+            ) from exc
+        return f"---\n{yaml_text}---\n\n{body}"
+
+    @classmethod
+    def _canonical_markdown_value(
+        cls, value: Any, source: str, ancestors: Optional[Set[int]] = None
+    ) -> Any:
+        ancestors = set() if ancestors is None else ancestors
+        if isinstance(value, dict):
+            if any(not isinstance(key, str) for key in value):
+                raise ValueError(
+                    f"Invalid Markdown metadata in {source}: "
+                    "mapping keys must be strings."
+                )
+            identity = id(value)
+            if identity in ancestors:
+                raise ValueError(
+                    f"Invalid Markdown metadata in {source}: "
+                    "values cannot contain cycles."
+                )
+            ancestors.add(identity)
+            try:
+                return {
+                    key: cls._canonical_markdown_value(
+                        value[key], source, ancestors
+                    )
+                    for key in sorted(value)
+                }
+            finally:
+                ancestors.remove(identity)
+        if isinstance(value, list):
+            identity = id(value)
+            if identity in ancestors:
+                raise ValueError(
+                    f"Invalid Markdown metadata in {source}: "
+                    "values cannot contain cycles."
+                )
+            ancestors.add(identity)
+            try:
+                return [
+                    cls._canonical_markdown_value(item, source, ancestors)
+                    for item in value
+                ]
+            finally:
+                ancestors.remove(identity)
+        if isinstance(value, tuple) or isinstance(value, set):
+            raise ValueError(
+                f"Invalid Markdown metadata in {source}: "
+                "tuples and sets are not supported."
+            )
+        return value
+
+    def _load_markdown_directory(self, source: Path) -> None:
+        manifest_document, node_documents = self._read_markdown_directory(source)
+        manifest, _ = self._parse_markdown_document(
+            manifest_document, str(source / self._MARKDOWN_MANIFEST)
+        )
+        graph_id, edges, links = self._parse_markdown_manifest(manifest, source)
+
+        nodes_by_id: Dict[str, ContextNode] = {}
+        for node_source, document in node_documents:
+            frontmatter, body = self._parse_markdown_document(document, node_source)
+            node = self._parse_markdown_node(frontmatter, body, node_source)
+            if node.node_id in nodes_by_id:
+                raise ValueError(
+                    f"Duplicate Markdown node ID {node.node_id!r} in {node_source}."
+                )
+            nodes_by_id[node.node_id] = node
+
+        missing_endpoints = sorted(
+            {
+                endpoint
+                for edge in edges
+                for endpoint in (edge.source_id, edge.target_id)
+                if endpoint not in nodes_by_id
+            }
+        )
+        if missing_endpoints:
+            missing = ", ".join(repr(endpoint) for endpoint in missing_endpoints)
+            raise ValueError(
+                f"Invalid ContextGraph Markdown: edge endpoint(s) {missing} "
+                "do not have node files."
+            )
+
+        hierarchy_edges = [
+            {
+                "source": edge.source_id,
+                "target": edge.target_id,
+                "type": edge.edge_type,
+            }
+            for edge in edges
+            if is_skos_hierarchy_edge(edge.to_dict())
+        ]
+        if hierarchy_edges:
+            validate_skos_hierarchy(hierarchy_edges, [])
+
+        unresolved_links = {}
+        for link in links:
+            link_id = link["link_id"]
+            if link_id in unresolved_links:
+                raise ValueError(
+                    f"Duplicate cross-graph link ID {link_id!r} in graph manifest."
+                )
+            if link["source_node_id"] not in nodes_by_id:
+                raise ValueError(
+                    f"Cross-graph link {link_id!r} references missing source node "
+                    f"{link['source_node_id']!r}."
+                )
+            unresolved_links[link_id] = link
+
+        adjacency: Dict[str, List[ContextEdge]] = defaultdict(list)
+        node_type_index: Dict[str, Set[str]] = defaultdict(set)
+        edge_type_index: Dict[str, List[ContextEdge]] = defaultdict(list)
+        for node in nodes_by_id.values():
+            node_type_index[node.node_type].add(node.node_id)
+        for edge in edges:
+            adjacency[edge.source_id].append(edge)
+            edge_type_index[edge.edge_type].append(edge)
+
+        with self._lock:
+            self.graph_id = graph_id
+            self.nodes.clear()
+            self.nodes.update(nodes_by_id)
+            self.edges.clear()
+            self.edges.extend(edges)
+            self._adjacency.clear()
+            self._adjacency.update(adjacency)
+            self.node_type_index.clear()
+            self.node_type_index.update(node_type_index)
+            self.edge_type_index.clear()
+            self.edge_type_index.update(edge_type_index)
+            self._linked_graphs.clear()
+            self._unresolved_links.clear()
+            self._unresolved_links.update(unresolved_links)
+            self._analytics_cache.clear()
+
+        if self.mutation_callback and not self._suspend_mutation_callback:
+            try:
+                self.mutation_callback(
+                    "RELOAD_GRAPH",
+                    graph_id,
+                    {"node_count": len(nodes_by_id), "edge_count": len(edges)},
+                )
+            except Exception as exc:
+                self.logger.warning(
+                    "Audit trail callback failed for Markdown graph load: %s", exc
+                )
+
+    def _read_markdown_directory(
+        self, source: Path
+    ) -> Tuple[str, List[Tuple[str, str]]]:
+        if source.is_symlink():
+            raise ValueError(f"Refusing to import Markdown symbolic link: {source}")
+        if not source.exists():
+            raise FileNotFoundError(
+                f"ContextGraph Markdown import path does not exist: {source}"
+            )
+        if not source.is_dir():
+            raise ValueError(
+                f"ContextGraph Markdown import path is not a directory: {source}"
+            )
+
+        manifest_path = source / self._MARKDOWN_MANIFEST
+        manifest_document = self._read_markdown_file(manifest_path)
+        nodes_path = source / self._MARKDOWN_NODES_DIRECTORY
+        if nodes_path.is_symlink():
+            raise ValueError(f"Refusing to import Markdown symbolic link: {nodes_path}")
+        if not nodes_path.is_dir():
+            raise ValueError(
+                f"ContextGraph Markdown nodes directory is missing: {nodes_path}"
+            )
+
+        node_paths = []
+        for path in nodes_path.iterdir():
+            if path.is_symlink():
+                raise ValueError(f"Refusing to import Markdown symbolic link: {path}")
+            if path.suffix.lower() not in self._MARKDOWN_EXTENSIONS:
+                continue
+            if not path.is_file():
+                raise ValueError(f"Markdown node path is not a regular file: {path}")
+            node_paths.append(path)
+        node_paths.sort(key=lambda path: (path.name.casefold(), path.name))
+        return manifest_document, [
+            (str(path), self._read_markdown_file(path)) for path in node_paths
+        ]
+
+    @staticmethod
+    def _read_markdown_file(path: Path) -> str:
+        if path.is_symlink():
+            raise ValueError(f"Refusing to import Markdown symbolic link: {path}")
+        flags = os.O_RDONLY
+        if hasattr(os, "O_NOFOLLOW"):
+            flags |= os.O_NOFOLLOW
+        try:
+            descriptor = os.open(path, flags)
+        except OSError as exc:
+            if exc.errno == errno.ELOOP or path.is_symlink():
+                raise ValueError(
+                    f"Refusing to import Markdown symbolic link: {path}"
+                ) from exc
+            if exc.errno == errno.ENOENT:
+                raise FileNotFoundError(f"Markdown file is missing: {path}") from exc
+            raise OSError(
+                exc.errno,
+                f"Failed to read Markdown file {path}: {exc.strerror or str(exc)}",
+                exc.filename or str(path),
+            ) from exc
+
+        try:
+            if not stat.S_ISREG(os.fstat(descriptor).st_mode):
+                raise ValueError(f"Markdown path is not a regular file: {path}")
+            with os.fdopen(descriptor, "r", encoding="utf-8") as input_file:
+                descriptor = None
+                return input_file.read()
+        finally:
+            if descriptor is not None:
+                os.close(descriptor)
+
+    @staticmethod
+    def _parse_markdown_document(
+        document: str, source: str
+    ) -> Tuple[Dict[str, Any], str]:
+        lines = document.splitlines(keepends=True)
+        if not lines or lines[0].rstrip("\r\n") != "---":
+            raise ValueError(
+                f"Invalid Markdown frontmatter in {source}: "
+                "document must start with '---'."
+            )
+        closing_index = next(
+            (
+                index
+                for index, line in enumerate(lines[1:], start=1)
+                if line.rstrip("\r\n") == "---"
+            ),
+            None,
+        )
+        if closing_index is None:
+            raise ValueError(
+                f"Invalid Markdown frontmatter in {source}: missing closing '---'."
+            )
+        try:
+            loaded = yaml.load(
+                "".join(lines[1:closing_index]), Loader=_UniqueKeySafeLoader
+            )
+        except yaml.YAMLError as exc:
+            raise ValueError(
+                f"Invalid Markdown frontmatter in {source}: {exc}"
+            ) from exc
+        frontmatter = {} if loaded is None else loaded
+        if not isinstance(frontmatter, dict):
+            raise ValueError(
+                f"Invalid Markdown frontmatter in {source}: expected a YAML mapping."
+            )
+        if any(not isinstance(key, str) for key in frontmatter):
+            raise ValueError(
+                f"Invalid Markdown frontmatter in {source}: "
+                "field names must be strings."
+            )
+
+        body = "".join(lines[closing_index + 1 :])
+        if body.startswith("\r\n"):
+            body = body[2:]
+        elif body.startswith("\n"):
+            body = body[1:]
+        return frontmatter, body
+
+    def _parse_markdown_manifest(
+        self, manifest: Dict[str, Any], source: Path
+    ) -> Tuple[str, List[ContextEdge], List[Dict[str, str]]]:
+        manifest_source = str(source / self._MARKDOWN_MANIFEST)
+        if manifest.get("format") != self._MARKDOWN_FORMAT:
+            raise ValueError(
+                f"Invalid ContextGraph Markdown manifest in {manifest_source}: "
+                f"'format' must be {self._MARKDOWN_FORMAT!r}."
+            )
+        version = manifest.get("version")
+        if isinstance(version, bool) or version != self._MARKDOWN_VERSION:
+            raise ValueError(
+                f"Unsupported ContextGraph Markdown version {version!r} in "
+                f"{manifest_source}; expected {self._MARKDOWN_VERSION}."
+            )
+        graph_id = self._required_markdown_string(
+            manifest.get("graph_id"), "graph_id", manifest_source
+        )
+
+        raw_edges = manifest.get("edges", [])
+        if not isinstance(raw_edges, list):
+            raise ValueError(
+                f"Invalid ContextGraph Markdown manifest in {manifest_source}: "
+                "'edges' must be a list."
+            )
+        edges = []
+        edge_ids = set()
+        for index, raw_edge in enumerate(raw_edges):
+            edge_source = f"{manifest_source} edge[{index}]"
+            edge = self._parse_markdown_edge(raw_edge, edge_source)
+            if edge.edge_id in edge_ids:
+                raise ValueError(
+                    f"Duplicate Markdown edge ID {edge.edge_id!r} in {edge_source}."
+                )
+            edge_ids.add(edge.edge_id)
+            edges.append(edge)
+
+        raw_links = manifest.get("links", [])
+        if not isinstance(raw_links, list):
+            raise ValueError(
+                f"Invalid ContextGraph Markdown manifest in {manifest_source}: "
+                "'links' must be a list."
+            )
+        links = [
+            self._parse_markdown_link(link, f"{manifest_source} link[{index}]")
+            for index, link in enumerate(raw_links)
+        ]
+        return graph_id, edges, links
+
+    def _parse_markdown_node(
+        self, frontmatter: Dict[str, Any], body: str, source: str
+    ) -> ContextNode:
+        node_id = self._required_markdown_string(frontmatter.get("id"), "id", source)
+        node_type = self._required_markdown_string(
+            frontmatter.get("type"), "type", source
+        )
+        properties = self._markdown_mapping(
+            frontmatter.get("properties", {}), "properties", source
+        )
+        metadata = self._markdown_mapping(
+            frontmatter.get("metadata", {}), "metadata", source
+        )
+        return ContextNode(
+            node_id=node_id,
+            node_type=node_type,
+            content=body,
+            properties=properties,
+            metadata=metadata,
+            valid_from=self._markdown_temporal_value(
+                frontmatter.get("valid_from"), "valid_from", source
+            ),
+            valid_until=self._markdown_temporal_value(
+                frontmatter.get("valid_until"), "valid_until", source
+            ),
+        )
+
+    def _parse_markdown_edge(self, raw_edge: Any, source: str) -> ContextEdge:
+        if not isinstance(raw_edge, dict):
+            raise ValueError(f"Invalid Markdown edge in {source}: expected a mapping.")
+        edge_id = self._required_markdown_string(raw_edge.get("id"), "id", source)
+        family_value = raw_edge.get("family_id", raw_edge.get("familyId", edge_id))
+        family_id = self._required_markdown_string(family_value, "family_id", source)
+        source_id = self._required_markdown_string(
+            raw_edge.get("source"), "source", source
+        )
+        target_id = self._required_markdown_string(
+            raw_edge.get("target"), "target", source
+        )
+        edge_type = self._required_markdown_string(raw_edge.get("type"), "type", source)
+        weight = raw_edge.get("weight", 1.0)
+        if isinstance(weight, bool) or not isinstance(weight, (int, float)):
+            raise ValueError(
+                f"Invalid Markdown edge in {source}: 'weight' must be a number."
+            )
+        metadata = self._markdown_mapping(
+            raw_edge.get("metadata", {}), "metadata", source
+        )
+        return ContextEdge(
+            edge_id=edge_id,
+            family_id=family_id,
+            source_id=source_id,
+            target_id=target_id,
+            edge_type=edge_type,
+            weight=float(weight),
+            metadata=metadata,
+            valid_from=self._markdown_temporal_value(
+                raw_edge.get("valid_from"), "valid_from", source
+            ),
+            valid_until=self._markdown_temporal_value(
+                raw_edge.get("valid_until"), "valid_until", source
+            ),
+        )
+
+    def _parse_markdown_link(self, raw_link: Any, source: str) -> Dict[str, str]:
+        if not isinstance(raw_link, dict):
+            raise ValueError(
+                f"Invalid cross-graph link in {source}: expected a mapping."
+            )
+        return {
+            field_name: self._required_markdown_string(
+                raw_link.get(field_name), field_name, source
+            )
+            for field_name in (
+                "link_id",
+                "source_node_id",
+                "target_node_id",
+                "other_graph_id",
+            )
+        }
+
+    @staticmethod
+    def _required_markdown_string(value: Any, field_name: str, source: str) -> str:
+        if not isinstance(value, str) or not value.strip():
+            raise ValueError(
+                f"Invalid Markdown frontmatter in {source}: "
+                f"'{field_name}' must be a non-empty string."
+            )
+        return value
+
+    @classmethod
+    def _markdown_mapping(
+        cls, value: Any, field_name: str, source: str
+    ) -> Dict[str, Any]:
+        if value is None:
+            return {}
+        if not isinstance(value, dict):
+            raise ValueError(
+                f"Invalid Markdown frontmatter in {source}: "
+                f"'{field_name}' must be a mapping."
+            )
+        canonical = cls._canonical_markdown_value(value, source)
+        return dict(canonical)
+
+    @staticmethod
+    def _markdown_temporal_value(
+        value: Any, field_name: str, source: str
+    ) -> Optional[str]:
+        if value is None:
+            return None
+        if isinstance(value, (date, datetime)):
+            return value.isoformat()
+        if isinstance(value, str) and value.strip():
+            return value
+        raise ValueError(
+            f"Invalid Markdown frontmatter in {source}: "
+            f"'{field_name}' must be an ISO-8601 string."
+        )
+
+
     def find_node(self, node_id: str) -> Optional[Dict[str, Any]]:
         """Find a node by ID."""
         with self._lock:

tests/context/test_context_graph_markdown.py

@@ -0,0 +1,482 @@
+from pathlib import Path
+
+import pytest
+import yaml
+
+import semantica.context.context_graph as context_graph_module
+from semantica.context.context_graph import ContextEdge, ContextGraph, ContextNode
+
+
+def _read_markdown(path: Path):
+    lines = path.read_text(encoding="utf-8").splitlines(keepends=True)
+    closing_index = next(
+        index
+        for index, line in enumerate(lines[1:], start=1)
+        if line.rstrip("\r\n") == "---"
+    )
+    frontmatter = yaml.safe_load("".join(lines[1:closing_index])) or {}
+    body = "".join(lines[closing_index + 1 :])
+    if body.startswith("\r\n"):
+        body = body[2:]
+    elif body.startswith("\n"):
+        body = body[1:]
+    return frontmatter, body
+
+
+def _write_markdown(path: Path, frontmatter, body: str) -> None:
+    yaml_text = yaml.safe_dump(
+        frontmatter,
+        sort_keys=False,
+        allow_unicode=True,
+        default_flow_style=False,
+    )
+    path.write_text(f"---\n{yaml_text}---\n\n{body}", encoding="utf-8")
+
+
+def _node_file(export_path: Path, node_id: str) -> Path:
+    for path in (export_path / "nodes").glob("*.md"):
+        frontmatter, _ = _read_markdown(path)
+        if frontmatter.get("id") == node_id:
+            return path
+    raise AssertionError(f"No Markdown file found for node {node_id!r}")
+
+
+def _normalized_state(graph: ContextGraph):
+    links = {link_id: dict(link) for link_id, link in graph._unresolved_links.items()}
+    for link_id, (
+        other_graph,
+        source_node_id,
+        target_node_id,
+    ) in graph._linked_graphs.items():
+        links[link_id] = {
+            "link_id": link_id,
+            "source_node_id": source_node_id,
+            "target_node_id": target_node_id,
+            "other_graph_id": other_graph.graph_id,
+        }
+    return {
+        "graph_id": graph.graph_id,
+        "nodes": {
+            node_id: {
+                "type": node.node_type,
+                "content": node.content,
+                "properties": node.properties,
+                "metadata": node.metadata,
+                "valid_from": node.valid_from,
+                "valid_until": node.valid_until,
+            }
+            for node_id, node in graph.nodes.items()
+        },
+        "edges": sorted(
+            (
+                {
+                    "id": edge.edge_id,
+                    "family_id": edge.family_id,
+                    "source": edge.source_id,
+                    "target": edge.target_id,
+                    "type": edge.edge_type,
+                    "weight": edge.weight,
+                    "metadata": edge.metadata,
+                    "valid_from": edge.valid_from,
+                    "valid_until": edge.valid_until,
+                }
+                for edge in graph.edges
+            ),
+            key=lambda edge: edge["id"],
+        ),
+        "links": links,
+    }
+
+
+def _sample_graph():
+    graph = ContextGraph(advanced_analytics=False)
+    graph.graph_id = "graph-primary"
+    graph._add_internal_node(
+        ContextNode(
+            node_id="policy/\u6771\u4eac",
+            node_type="Policy",
+            content="# Retention\n\nKeep evidence.\n---\n",
+            properties={"priority": 2, "nested": {"owner": "governance"}},
+            metadata={"source": "manual", "tags": ["retention", "legal"]},
+            valid_from="2026-01-01T00:00:00+00:00",
+            valid_until="2027-01-01T00:00:00+00:00",
+        )
+    )
+    graph._add_internal_node(
+        ContextNode(
+            node_id="evidence-1",
+            node_type="Evidence",
+            content="Original source",
+            properties={"checksum": "abc123"},
+            metadata={"classification": "internal"},
+        )
+    )
+    graph._add_internal_edge(
+        ContextEdge(
+            edge_id="edge-supports",
+            family_id="family-supports",
+            source_id="evidence-1",
+            target_id="policy/\u6771\u4eac",
+            edge_type="SUPPORTS",
+            weight=0.75,
+            metadata={"confidence": 0.9},
+            valid_from="2026-02-01",
+            valid_until="2026-12-31",
+        )
+    )
+
+    other = ContextGraph(advanced_analytics=False)
+    other.graph_id = "graph-secondary"
+    other.add_node("source-page", "Page", "Source page")
+    link_id = graph.link_graph(other, "policy/\u6771\u4eac", "source-page")
+    return graph, other, link_id
+
+
+def _directory_contents(path: Path):
+    return {
+        str(file_path.relative_to(path)): file_path.read_bytes()
+        for file_path in path.rglob("*")
+        if file_path.is_file()
+    }
+
+
+def test_markdown_round_trip_preserves_complete_graph_state(tmp_path):
+    graph, other, link_id = _sample_graph()
+    export_path = tmp_path / "context-graph"
+
+    graph.save_to_file(export_path, format="markdown")
+
+    restored = ContextGraph(advanced_analytics=False)
+    restored.load_from_file(export_path, format="markdown")
+
+    assert _normalized_state(restored) == _normalized_state(graph)
+    assert restored.resolve_links({other.graph_id: other}) == 1
+    linked_graph, entry_node = restored.navigate_to(link_id)
+    assert linked_graph is other
+    assert entry_node == "source-page"
+
+
+def test_markdown_manual_node_and_edge_edits_are_imported(tmp_path):
+    graph, _, _ = _sample_graph()
+    export_path = tmp_path / "context-graph"
+    graph.save_to_file(export_path, format="markdown")
+
+    node_path = _node_file(export_path, "policy/\u6771\u4eac")
+    node_frontmatter, _ = _read_markdown(node_path)
+    node_frontmatter["metadata"]["reviewed"] = True
+    _write_markdown(node_path, node_frontmatter, "# Updated policy\n")
+
+    manifest_path = export_path / "graph.md"
+    manifest, manifest_body = _read_markdown(manifest_path)
+    manifest["edges"][0]["type"] = "VERIFIES"
+    manifest["edges"][0]["weight"] = 1.0
+    manifest["edges"][0]["metadata"]["reviewed_by"] = "human"
+    _write_markdown(manifest_path, manifest, manifest_body)
+
+    restored = ContextGraph(advanced_analytics=False)
+    restored.load_from_file(export_path, format="markdown")
+
+    node = restored.nodes["policy/\u6771\u4eac"]
+    assert node.content == "# Updated policy\n"
+    assert node.metadata["reviewed"] is True
+    edge = restored.edges[0]
+    assert edge.edge_type == "VERIFIES"
+    assert edge.weight == 1.0
+    assert edge.metadata["reviewed_by"] == "human"
+
+
+def test_markdown_export_is_deterministic_and_removes_stale_nodes(tmp_path):
+    graph = ContextGraph(advanced_analytics=False)
+    graph.graph_id = "deterministic-graph"
+    graph.add_node("kept", "Note", "Keep")
+    graph.add_node("removed", "Note", "Remove")
+    graph.add_edge("kept", "removed", "REFERENCES")
+    first = tmp_path / "first"
+    second = tmp_path / "second"
+
+    graph.save_to_file(first, format="markdown")
+    graph.save_to_file(second, format="markdown")
+    assert _directory_contents(first) == _directory_contents(second)
+
+    stale_path = _node_file(first, "removed")
+    graph.nodes.pop("removed")
+    graph.edges.clear()
+    graph.save_to_file(first, format="markdown")
+
+    assert not stale_path.exists()
+    restored = ContextGraph(advanced_analytics=False)
+    restored.load_from_file(first, format="markdown")
+    assert set(restored.nodes) == {"kept"}
+    assert restored.edges == []
+
+
+def test_markdown_empty_graph_round_trip(tmp_path):
+    graph = ContextGraph(advanced_analytics=False)
+    graph.graph_id = "empty-graph"
+    export_path = tmp_path / "empty"
+
+    graph.save_to_file(export_path, format="markdown")
+    restored = ContextGraph(advanced_analytics=False)
+    restored.load_from_file(export_path, format="markdown")
+
+    assert restored.graph_id == "empty-graph"
+    assert restored.nodes == {}
+    assert restored.edges == []
+
+
+@pytest.mark.parametrize(
+    "corruption, expected_error",
+    [
+        ("unsupported-version", "Unsupported ContextGraph Markdown version"),
+        ("duplicate-edge", "Duplicate Markdown edge ID"),
+        ("dangling-edge", "do not have node files"),
+        ("duplicate-node", "Duplicate Markdown node ID"),
+        ("cyclic-skos", "SKOS hierarchy contains a cycle"),
+    ],
+)
+def test_invalid_markdown_does_not_mutate_existing_graph(
+    tmp_path, corruption, expected_error
+):
+    source, _, _ = _sample_graph()
+    export_path = tmp_path / corruption
+    source.save_to_file(export_path, format="markdown")
+
+    manifest_path = export_path / "graph.md"
+    manifest, manifest_body = _read_markdown(manifest_path)
+    if corruption == "unsupported-version":
+        manifest["version"] = 2
+    elif corruption == "duplicate-edge":
+        manifest["edges"].append(dict(manifest["edges"][0]))
+    elif corruption == "dangling-edge":
+        manifest["edges"][0]["target"] = "missing-node"
+    elif corruption == "duplicate-node":
+        original = _node_file(export_path, "evidence-1")
+        (original.parent / "duplicate.md").write_bytes(original.read_bytes())
+    elif corruption == "cyclic-skos":
+        manifest["edges"] = [
+            {
+                "id": "broader-1",
+                "family_id": "broader-1",
+                "source": "evidence-1",
+                "target": "policy/\u6771\u4eac",
+                "type": "skos:broader",
+                "weight": 1.0,
+                "metadata": {},
+            },
+            {
+                "id": "broader-2",
+                "family_id": "broader-2",
+                "source": "policy/\u6771\u4eac",
+                "target": "evidence-1",
+                "type": "skos:broader",
+                "weight": 1.0,
+                "metadata": {},
+            },
+        ]
+    _write_markdown(manifest_path, manifest, manifest_body)
+
+    target = ContextGraph(advanced_analytics=False)
+    target.add_node("sentinel", "Existing", "Do not replace")
+    before = _normalized_state(target)
+
+    with pytest.raises(ValueError, match=expected_error):
+        target.load_from_file(export_path, format="markdown")
+
+    assert _normalized_state(target) == before
+
+
+def test_duplicate_yaml_keys_are_rejected(tmp_path):
+    graph = ContextGraph(advanced_analytics=False)
+    graph.add_node("node-1", "Note", "Body")
+    export_path = tmp_path / "graph"
+    graph.save_to_file(export_path, format="markdown")
+    node_path = _node_file(export_path, "node-1")
+    document = node_path.read_text(encoding="utf-8")
+    node_path.write_text(document.replace("id: node-1", "id: node-1\nid: duplicate"))
+
+    with pytest.raises(ValueError, match="duplicate key 'id'"):
+        ContextGraph(advanced_analytics=False).load_from_file(
+            export_path, format="markdown"
+        )
+
+
+def test_markdown_export_refuses_unmanaged_nonempty_directory(tmp_path):
+    destination = tmp_path / "existing"
+    destination.mkdir()
+    marker = destination / "keep.txt"
+    marker.write_text("keep", encoding="utf-8")
+
+    with pytest.raises(ValueError, match="not a managed ContextGraph export"):
+        ContextGraph(advanced_analytics=False).save_to_file(
+            destination, format="markdown"
+        )
+
+    assert marker.read_text(encoding="utf-8") == "keep"
+
+
+def test_markdown_export_rejects_unrelated_graph_markdown_file(tmp_path):
+    destination = tmp_path / "existing"
+    destination.mkdir()
+    unrelated = destination / "graph.md"
+    _write_markdown(unrelated, {"title": "Unrelated notes"}, "Keep me")
+
+    with pytest.raises(ValueError, match="not a managed ContextGraph export"):
+        ContextGraph(advanced_analytics=False).save_to_file(
+            destination, format="markdown"
+        )
+
+    assert unrelated.exists()
+
+
+def test_markdown_export_preserves_manifest_inspection_errors(tmp_path, monkeypatch):
+    graph = ContextGraph(advanced_analytics=False)
+    destination = tmp_path / "existing"
+    graph.save_to_file(destination, format="markdown")
+    real_reader = ContextGraph._read_markdown_file
+
+    def fail_manifest_read(path):
+        if path == destination / "graph.md":
+            raise PermissionError("permission denied")
+        return real_reader(path)
+
+    monkeypatch.setattr(
+        ContextGraph, "_read_markdown_file", staticmethod(fail_manifest_read)
+    )
+
+    with pytest.raises(PermissionError, match="permission denied"):
+        graph.save_to_file(destination, format="markdown")
+
+
+def test_markdown_export_restores_previous_directory_when_publish_fails(
+    tmp_path, monkeypatch
+):
+    graph = ContextGraph(advanced_analytics=False)
+    graph.add_node("original", "Note", "Original")
+    destination = tmp_path / "graph"
+    graph.save_to_file(destination, format="markdown")
+    original_contents = _directory_contents(destination)
+    graph.add_node("new", "Note", "New")
+
+    real_replace = context_graph_module.os.replace
+
+    def fail_staged_publish(source, target):
+        if ".staging-" in Path(source).name and Path(target) == destination:
+            raise OSError("simulated publish failure")
+        return real_replace(source, target)
+
+    monkeypatch.setattr(context_graph_module.os, "replace", fail_staged_publish)
+
+    with pytest.raises(OSError, match="simulated publish failure"):
+        graph.save_to_file(destination, format="markdown")
+
+    assert _directory_contents(destination) == original_contents
+    assert not list(tmp_path.glob(".graph.staging-*"))
+    assert not list(tmp_path.glob(".graph.backup-*"))
+
+
+def test_markdown_load_rebuilds_indexes_and_emits_one_reload_event(tmp_path):
+    source, _, _ = _sample_graph()
+    destination = tmp_path / "graph"
+    source.save_to_file(destination, format="markdown")
+    events = []
+    target = ContextGraph(
+        advanced_analytics=False,
+        mutation_callback=lambda *event: events.append(event),
+    )
+
+    target.load_from_file(destination, format="markdown")
+
+    assert target.node_type_index["Policy"] == {"policy/\u6771\u4eac"}
+    assert target.edge_type_index["SUPPORTS"][0].edge_id == "edge-supports"
+    assert target._adjacency["evidence-1"][0].target_id == "policy/\u6771\u4eac"
+    assert events == [
+        (
+            "RELOAD_GRAPH",
+            "graph-primary",
+            {"node_count": len(target.nodes), "edge_count": len(target.edges)},
+        )
+    ]
+
+
+def test_markdown_export_rejects_recursive_metadata(tmp_path):
+    recursive = {}
+    recursive["self"] = recursive
+    graph = ContextGraph(advanced_analytics=False)
+    graph._add_internal_node(ContextNode("node-1", "Note", "Body", metadata=recursive))
+
+    with pytest.raises(ValueError, match="values cannot contain cycles"):
+        graph.save_to_file(tmp_path / "graph", format="markdown")
+
+
+def test_markdown_import_and_export_reject_symlinks(tmp_path):
+    graph = ContextGraph(advanced_analytics=False)
+    graph.add_node("node-1", "Note", "Body")
+    real_export = tmp_path / "real"
+    graph.save_to_file(real_export, format="markdown")
+
+    directory_link = tmp_path / "directory-link"
+    try:
+        directory_link.symlink_to(real_export, target_is_directory=True)
+    except (NotImplementedError, OSError):
+        pytest.skip("Symbolic links are not available on this platform")
+
+    with pytest.raises(ValueError, match="symbolic link"):
+        ContextGraph(advanced_analytics=False).load_from_file(
+            directory_link, format="markdown"
+        )
+    with pytest.raises(ValueError, match="symbolic link"):
+        graph.save_to_file(directory_link, format="markdown")
+
+    manifest_path = real_export / "graph.md"
+    manifest_target = tmp_path / "manifest.md"
+    manifest_target.write_bytes(manifest_path.read_bytes())
+    manifest_path.unlink()
+    manifest_path.symlink_to(manifest_target)
+    with pytest.raises(ValueError, match="symbolic link"):
+        ContextGraph(advanced_analytics=False).load_from_file(
+            real_export, format="markdown"
+        )
+
+
+@pytest.mark.skipif(
+    not hasattr(context_graph_module.os, "O_NOFOLLOW"),
+    reason="O_NOFOLLOW is unavailable on this platform",
+)
+def test_markdown_import_nofollow_check_closes_symlink_race(tmp_path, monkeypatch):
+    graph = ContextGraph(advanced_analytics=False)
+    graph.add_node("node-1", "Note", "Body")
+    export_path = tmp_path / "graph"
+    graph.save_to_file(export_path, format="markdown")
+    node_path = _node_file(export_path, "node-1")
+    target = tmp_path / "target.md"
+    target.write_bytes(node_path.read_bytes())
+    node_path.unlink()
+    node_path.symlink_to(target)
+
+    real_is_symlink = Path.is_symlink
+
+    def miss_precheck(path):
+        if path == node_path:
+            return False
+        return real_is_symlink(path)
+
+    monkeypatch.setattr(Path, "is_symlink", miss_precheck)
+
+    with pytest.raises(ValueError, match="symbolic link"):
+        ContextGraph(advanced_analytics=False).load_from_file(
+            export_path, format="markdown"
+        )
+
+
+def test_json_remains_default_and_unknown_format_is_rejected(tmp_path):
+    graph = ContextGraph(advanced_analytics=False)
+    graph.add_node("node-1", "Note", "Body")
+    json_path = tmp_path / "graph.json"
+    graph.save_to_file(json_path)
+
+    restored = ContextGraph(advanced_analytics=False)
+    restored.load_from_file(json_path)
+    assert "node-1" in restored.nodes
+
+    with pytest.raises(ValueError, match="Unsupported context graph"):
+        graph.save_to_file(tmp_path / "graph", format="html")