Merge pull request #852 from SaurabhScripts/codex/context-graph-markdown-round-trip

feat(context): add ContextGraph Markdown round-trip

CHANGELOG.md

@@ -400,6 +400,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - Entities and relationships round-trip as memory-local provenance only — Markdown import intentionally does not write into `ContextGraph`, matching the MVP scope agreed on in #765
   - Documented the file contract and workflow in `docs/reference/context.md`; 43 new tests in `tests/context/test_agent_memory_markdown.py` cover round-trip losslessness, idempotency, validation errors, rollback on failure, and vector-store sync ordering
 
+- **Markdown directory round trips for `ContextGraph`** (#852) by @SaurabhScripts
+  - `ContextGraph.save_to_file(..., format="markdown")` and `load_from_file(..., format="markdown")` persist a deterministic `graph.md` relationship manifest plus one human-editable Markdown file per node, preserving graph, node, edge, family, temporal, and cross-graph link identities
+  - Imports validate the complete directory before replacing graph state, rebuild indexes and analytics state atomically, create JSON-compatible stub nodes for dangling edge endpoints, and emit the same granular node/edge audit events as JSON loading
+  - Existing exports are replaced atomically only after their complete canonical layout is validated; untracked files, renamed node files, symlinks, Windows directory junctions, and other reparse points cause a fail-closed error instead of authorizing directory deletion
+  - Added 30 focused tests covering deterministic round trips, manual edits, validation rollback, managed-directory identity, publish rollback, audit-manager compatibility, stale-cache clearing, mocked and real Windows junctions, and missing-path behavior
+
 ### Fixed
 
 - **Markdown import followed filesystem links even though Markdown export already refused to overwrite them** (#851, follow-up to #765, #786) by @SaurabhScripts

docs/guides/context-graphs.md

@@ -436,6 +436,35 @@ 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, unsupported versions, and unsafe filesystem links fail without
+partially mutating the graph. As with JSON loading, an edge endpoint without a node
+file creates an `entity` stub node. Symlinks, Windows directory junctions, and other
+Windows reparse points are rejected.
+
+Re-exporting to an existing managed directory atomically replaces it, removing stale
+node files. Before replacement, Semantica validates the complete canonical export
+layout, not just the manifest header. Untracked files, assets, extra directories, or
+renamed node files therefore cause the export to fail closed instead of being deleted.
+Keep attachments and hand-written indexes outside the managed export directory.
+
 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

@@ -105,27 +105,82 @@ Production Use Cases:
     - Business: Workflow decisions, policy compliance, audit trails
 """
 
+import copy
+import errno
+import hashlib
+import itertools
+import json
+import os
+import re
+import shutil
+import stat
+import tempfile
+import threading
+import uuid
 from collections import defaultdict, deque
 from dataclasses import dataclass, field
-from datetime import datetime, timezone
-import json
-import threading
-import itertools
+from datetime import date, datetime, timezone
+from pathlib import Path
 from typing import Any, Dict, List, Optional, Set, Tuple, Union
-import uuid
 
+import yaml
+
+from ..utils.helpers import classify_path_distance
 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 ._markdown_filesystem import find_filesystem_link
 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 (
-        GraphBuilder, GraphAnalyzer, CentralityCalculator, CommunityDetector,
-        PathFinder, NodeEmbedder, SimilarityCalculator, LinkPredictor,
-        ConnectivityAnalyzer
+        CentralityCalculator,
+        CommunityDetector,
+        ConnectivityAnalyzer,
+        GraphAnalyzer,
+        GraphBuilder,
+        LinkPredictor,
+        NodeEmbedder,
+        PathFinder,
+        SimilarityCalculator,
     )
     KG_AVAILABLE = True
 except ImportError:
@@ -470,6 +525,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.
@@ -1101,14 +1162,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:
     
@@ -1135,15 +1203,31 @@ 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":
+            markdown_path = Path(path)
+            linked_component = find_filesystem_link(markdown_path)
+            if linked_component is not None:
+                raise ValueError(
+                    "Refusing to import Markdown symbolic link or junction: "
+                    f"{linked_component}"
+                )
+            if not markdown_path.exists():
+                self.logger.warning(f"File not found: {path}")
+                return
+            self._load_markdown_directory(markdown_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}")
@@ -1173,6 +1257,7 @@ class ContextGraph:
             self.edge_type_index.clear()
             self._linked_graphs.clear()
             self._unresolved_links.clear()
+            self._analytics_cache.clear()
             # Deletion metadata belongs to the graph being replaced; keeping it
             # would make entities in the loaded graph read as already retracted.
             self._retractions.clear()
@@ -1208,6 +1293,798 @@ 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.")
+        linked_component = find_filesystem_link(destination)
+        if linked_component is not None:
+            raise ValueError(
+                "Refusing to replace Markdown symbolic link or junction: "
+                f"{linked_component}"
+            )
+        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()):
+            try:
+                self._parse_markdown_directory(
+                    destination, require_canonical_layout=True
+                )
+            except (FileNotFoundError, ValueError):
+                is_managed = False
+            else:
+                is_managed = True
+            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)
+        linked_component = find_filesystem_link(destination.parent)
+        if linked_component is not None:
+            raise ValueError(
+                "Refusing to export through Markdown symbolic link or junction: "
+                f"{linked_component}"
+            )
+        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 as publish_error:
+                if backup_path is not None and not destination.exists():
+                    try:
+                        os.replace(backup_path, destination)
+                    except BaseException as restore_error:
+                        self.logger.error(
+                            "Failed to restore previous ContextGraph Markdown "
+                            "export from %s after publish failure; preserving "
+                            "the original publish error",
+                            backup_path,
+                            exc_info=(
+                                type(restore_error),
+                                restore_error,
+                                restore_error.__traceback__,
+                            ),
+                        )
+                        add_note = getattr(publish_error, "add_note", None)
+                        if add_note is not None:
+                            add_note(
+                                "Restoring the previous ContextGraph Markdown "
+                                f"export also failed: {restore_error}"
+                            )
+                    else:
+                        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"]),
+            )
+        )
+        seen_edge_ids: Set[str] = set()
+        duplicate_edge_ids: Set[str] = set()
+        for edge in edges:
+            edge_id = edge["id"]
+            if not isinstance(edge_id, str) or not edge_id.strip():
+                raise ValueError(
+                    "Cannot export ContextGraph: every edge must have a string ID."
+                )
+            if edge_id in seen_edge_ids:
+                duplicate_edge_ids.add(edge_id)
+            seen_edge_ids.add(edge_id)
+        if duplicate_edge_ids:
+            duplicates = ", ".join(
+                repr(edge_id) for edge_id in sorted(duplicate_edge_ids)
+            )
+            raise ValueError(
+                f"Cannot export ContextGraph: duplicate edge ID(s): {duplicates}."
+            )
+        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:
+        parsed_state = self._parse_markdown_directory(source)
+        graph_id, nodes_by_id, edges, unresolved_links = parsed_state
+
+        adjacency: Dict[str, List[ContextEdge]] = defaultdict(list)
+        edge_index: Dict[str, ContextEdge] = {}
+        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:
+            edge_index[edge.edge_id] = edge
+            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._edge_index.clear()
+            self._edge_index.update(edge_index)
+            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()
+            self._retractions.clear()
+            self._tombstones.clear()
+
+        if self.mutation_callback and not self._suspend_mutation_callback:
+            mutation_events = [
+                ("ADD_NODE", node.node_id, node.to_dict())
+                for node in nodes_by_id.values()
+            ]
+            mutation_events.extend(
+                ("ADD_EDGE", edge.edge_id, edge.to_dict()) for edge in edges
+            )
+            for operation, entity_id, payload in mutation_events:
+                try:
+                    self.mutation_callback(operation, entity_id, payload)
+                except Exception as exc:
+                    self.logger.warning(
+                        "Audit trail callback failed for Markdown graph load "
+                        "%s %s: %s",
+                        operation,
+                        entity_id,
+                        exc,
+                    )
+
+    def _parse_markdown_directory(
+        self, source: Path, require_canonical_layout: bool = False
+    ) -> Tuple[
+        str,
+        Dict[str, ContextNode],
+        List[ContextEdge],
+        Dict[str, Dict[str, str]],
+    ]:
+        manifest_document, node_documents = self._read_markdown_directory(
+            source, require_canonical_layout=require_canonical_layout
+        )
+        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}."
+                )
+            node_filename = Path(node_source).name
+            if (
+                require_canonical_layout
+                and node_filename != self._node_markdown_filename(node.node_id)
+            ):
+                raise ValueError(
+                    "Invalid managed ContextGraph export: node file "
+                    f"{node_filename!r} is not the canonical filename "
+                    f"for node {node.node_id!r}."
+                )
+            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 and require_canonical_layout:
+            missing = ", ".join(repr(endpoint) for endpoint in missing_endpoints)
+            raise ValueError(
+                "Invalid managed ContextGraph export: edge endpoint(s) "
+                f"{missing} do not have node files."
+            )
+        for endpoint in missing_endpoints:
+            nodes_by_id[endpoint] = ContextNode(endpoint, "entity", endpoint)
+
+        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
+        return graph_id, nodes_by_id, edges, unresolved_links
+
+    def _read_markdown_directory(
+        self, source: Path, require_canonical_layout: bool = False
+    ) -> Tuple[str, List[Tuple[str, str]]]:
+        linked_component = find_filesystem_link(source)
+        if linked_component is not None:
+            raise ValueError(
+                "Refusing to import Markdown symbolic link or junction: "
+                f"{linked_component}"
+            )
+        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}"
+            )
+
+        if require_canonical_layout:
+            expected_entries = {
+                self._MARKDOWN_MANIFEST,
+                self._MARKDOWN_NODES_DIRECTORY,
+            }
+            actual_entries = {path.name for path in source.iterdir()}
+            if actual_entries != expected_entries:
+                unexpected = sorted(actual_entries - expected_entries)
+                missing = sorted(expected_entries - actual_entries)
+                details = []
+                if unexpected:
+                    details.append(f"unexpected entries: {unexpected!r}")
+                if missing:
+                    details.append(f"missing entries: {missing!r}")
+                raise ValueError(
+                    "Invalid managed ContextGraph export layout ("
+                    + "; ".join(details)
+                    + ")."
+                )
+
+        manifest_path = source / self._MARKDOWN_MANIFEST
+        manifest_document = self._read_markdown_file(manifest_path)
+        nodes_path = source / self._MARKDOWN_NODES_DIRECTORY
+        linked_component = find_filesystem_link(nodes_path)
+        if linked_component is not None:
+            raise ValueError(
+                "Refusing to import Markdown symbolic link or junction: "
+                f"{linked_component}"
+            )
+        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():
+            linked_component = find_filesystem_link(path)
+            if linked_component is not None:
+                raise ValueError(
+                    "Refusing to import Markdown symbolic link or junction: "
+                    f"{linked_component}"
+                )
+            if path.suffix.lower() not in self._MARKDOWN_EXTENSIONS:
+                if require_canonical_layout:
+                    raise ValueError(
+                        "Invalid managed ContextGraph export: unexpected node "
+                        f"entry {path.name!r}."
+                    )
+                continue
+            if not path.is_file():
+                raise ValueError(f"Markdown node path is not a regular file: {path}")
+            node_paths.append(path)
+        linked_component = find_filesystem_link(nodes_path)
+        if linked_component is not None:
+            raise ValueError(
+                "Refusing to import Markdown symbolic link or junction: "
+                f"{linked_component}"
+            )
+        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:
+        linked_component = find_filesystem_link(path)
+        if linked_component is not None:
+            raise ValueError(
+                "Refusing to import Markdown symbolic link or junction: "
+                f"{linked_component}"
+            )
+        flags = os.O_RDONLY
+        if hasattr(os, "O_NOFOLLOW"):
+            flags |= os.O_NOFOLLOW
+        try:
+            descriptor = os.open(path, flags)
+        except OSError as exc:
+            linked_component = find_filesystem_link(path)
+            if exc.errno == errno.ELOOP or linked_component is not None:
+                raise ValueError(
+                    "Refusing to import Markdown symbolic link or junction: "
+                    f"{linked_component or 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:
+            linked_component = find_filesystem_link(path)
+            if linked_component is not None:
+                raise ValueError(
+                    "Refusing to import Markdown symbolic link or junction: "
+                    f"{linked_component}"
+                )
+            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, datetime):
+            return _normalize_temporal_input(value)
+        if isinstance(value, date):
+            return value.isoformat()
+        if isinstance(value, str) and value.strip():
+            try:
+                _normalize_temporal_input(value)
+            except ValueError as exc:
+                raise ValueError(
+                    f"Invalid Markdown frontmatter in {source}: "
+                    f"'{field_name}' must be a valid ISO-8601 string."
+                ) from exc
+            return value
+        raise ValueError(
+            f"Invalid Markdown frontmatter in {source}: "
+            f"'{field_name}' must be a valid ISO-8601 string."
+        )
+
+
     def find_node(self, node_id: str) -> Optional[Dict[str, Any]]:
         """Return a dict representation of the node identified by *node_id*.
 
@@ -2416,9 +3293,10 @@ class ContextGraph:
 
     def _load_conversation(self, file_path: str) -> Dict[str, Any]:
         """Load conversation from file."""
-        from ..utils.helpers import read_json_file
         from pathlib import Path
 
+        from ..utils.helpers import read_json_file
+
         return read_json_file(Path(file_path))
 
     def to_dict(self) -> Dict[str, Any]:
@@ -3230,7 +4108,7 @@ class ContextGraph:
         """
         import uuid
         from datetime import datetime
-        
+
         # Input validation
         if not isinstance(category, str) or not category.strip():
             raise ValueError("Category must be a non-empty string")

tests/context/test_context_graph_markdown.py

@@ -0,0 +1,713 @@
+import os
+import stat
+import subprocess
+from pathlib import Path
+
+import pytest
+import yaml
+
+import semantica.context.context_graph as context_graph_module
+from semantica.change_management.managers import TemporalVersionManager
+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 == []
+
+
+def test_markdown_export_rejects_duplicate_edge_ids_before_writing(tmp_path):
+    graph = ContextGraph(advanced_analytics=False)
+    graph.add_node("source", "Note", "Source")
+    graph.add_node("target", "Note", "Target")
+    graph.add_edge("source", "target", "REFERENCES")
+    graph.edges.append(graph.edges[0])
+    export_path = tmp_path / "graph"
+
+    assert graph.edges[0].edge_id == graph.edges[1].edge_id
+    with pytest.raises(ValueError, match=r"duplicate edge ID.*[0-9a-f-]+"):
+        graph.save_to_file(export_path, format="markdown")
+
+    assert not export_path.exists()
+
+
+@pytest.mark.parametrize(
+    "corruption, expected_error",
+    [
+        ("unsupported-version", "Unsupported ContextGraph Markdown version"),
+        ("duplicate-edge", "Duplicate Markdown edge ID"),
+        ("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 == "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_markdown_import_creates_json_compatible_stub_nodes_for_dangling_edges(
+    tmp_path,
+):
+    source, _, _ = _sample_graph()
+    export_path = tmp_path / "dangling-edge"
+    source.save_to_file(export_path, format="markdown")
+    manifest_path = export_path / "graph.md"
+    manifest, manifest_body = _read_markdown(manifest_path)
+    manifest["edges"][0]["target"] = "missing-node"
+    _write_markdown(manifest_path, manifest, manifest_body)
+
+    restored = ContextGraph(advanced_analytics=False)
+    restored.load_from_file(export_path, format="markdown")
+
+    stub = restored.nodes["missing-node"]
+    assert stub.node_type == "entity"
+    assert stub.content == "missing-node"
+    assert any(edge.target_id == "missing-node" for edge in restored.edges)
+
+
+@pytest.mark.parametrize(
+    ("location", "field_name"),
+    [("node", "valid_from"), ("edge", "valid_until")],
+)
+def test_invalid_markdown_temporal_value_does_not_mutate_existing_graph(
+    tmp_path, location, field_name
+):
+    source, _, _ = _sample_graph()
+    export_path = tmp_path / location
+    source.save_to_file(export_path, format="markdown")
+
+    if location == "node":
+        document_path = _node_file(export_path, "policy/\u6771\u4eac")
+    else:
+        document_path = export_path / "graph.md"
+    frontmatter, body = _read_markdown(document_path)
+    if location == "node":
+        frontmatter[field_name] = "not-a-date"
+    else:
+        frontmatter["edges"][0][field_name] = "not-a-date"
+    _write_markdown(document_path, frontmatter, body)
+
+    target = ContextGraph(advanced_analytics=False)
+    target.add_node("sentinel", "Existing", "Do not replace")
+    before = _normalized_state(target)
+
+    with pytest.raises(ValueError, match=rf"'{field_name}'.*valid ISO-8601"):
+        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()
+
+
+@pytest.mark.parametrize("extra_location", ["root", "nodes"])
+def test_markdown_export_refuses_managed_directory_with_untracked_files(
+    tmp_path, extra_location
+):
+    graph = ContextGraph(advanced_analytics=False)
+    graph.add_node("node-1", "Note", "Body")
+    destination = tmp_path / "existing"
+    graph.save_to_file(destination, format="markdown")
+    parent = destination if extra_location == "root" else destination / "nodes"
+    human_file = parent / "human-notes.txt"
+    human_file.write_text("do not delete", encoding="utf-8")
+    original_contents = _directory_contents(destination)
+
+    with pytest.raises(ValueError, match="not a managed ContextGraph export"):
+        graph.save_to_file(destination, format="markdown")
+
+    assert _directory_contents(destination) == original_contents
+
+
+def test_markdown_export_refuses_noncanonical_node_layout(tmp_path):
+    graph = ContextGraph(advanced_analytics=False)
+    graph.add_node("node-1", "Note", "Body")
+    destination = tmp_path / "existing"
+    graph.save_to_file(destination, format="markdown")
+    canonical = _node_file(destination, "node-1")
+    renamed = canonical.with_name("human-name.md")
+    canonical.rename(renamed)
+    original_contents = _directory_contents(destination)
+
+    with pytest.raises(ValueError, match="not a managed ContextGraph export"):
+        graph.save_to_file(destination, format="markdown")
+
+    assert _directory_contents(destination) == original_contents
+
+
+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_export_preserves_publish_error_when_restore_fails(
+    tmp_path, monkeypatch, caplog
+):
+    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_publish_and_restore(source, target):
+        source_path = Path(source)
+        if ".staging-" in source_path.name and Path(target) == destination:
+            raise OSError("simulated publish failure")
+        if ".backup-" in source_path.name and Path(target) == destination:
+            raise PermissionError("simulated restore failure")
+        return real_replace(source, target)
+
+    monkeypatch.setattr(context_graph_module.os, "replace", fail_publish_and_restore)
+    caplog.set_level("ERROR")
+
+    with pytest.raises(OSError, match="simulated publish failure"):
+        graph.save_to_file(destination, format="markdown")
+
+    assert "preserving the original publish error" in caplog.text
+    assert "simulated restore failure" in caplog.text
+    assert not destination.exists()
+    backup_paths = list(tmp_path.glob(".graph.backup-*"))
+    assert len(backup_paths) == 1
+    assert _directory_contents(backup_paths[0]) == original_contents
+    assert not list(tmp_path.glob(".graph.staging-*"))
+
+
+def test_markdown_load_rebuilds_indexes_and_emits_json_compatible_events(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._retractions[("node", "stale")] = {"entity_id": "stale"}
+    target._tombstones[("edge", "stale")] = {"entity_id": "stale"}
+
+    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._edge_index["edge-supports"].edge_type == "SUPPORTS"
+    assert target._adjacency["evidence-1"][0].target_id == "policy/\u6771\u4eac"
+    assert target._retractions == {}
+    assert target._tombstones == {}
+    assert [event[0] for event in events] == ["ADD_NODE"] * len(target.nodes) + [
+        "ADD_EDGE"
+    ] * len(target.edges)
+    assert {event[1] for event in events if event[0] == "ADD_NODE"} == set(target.nodes)
+    assert {event[1] for event in events if event[0] == "ADD_EDGE"} == {
+        edge.edge_id for edge in target.edges
+    }
+
+
+def test_markdown_load_records_granular_change_manager_history(tmp_path):
+    source, _, _ = _sample_graph()
+    destination = tmp_path / "graph"
+    source.save_to_file(destination, format="markdown")
+    target = ContextGraph(advanced_analytics=False)
+    manager = TemporalVersionManager()
+    manager.attach_to_graph(target)
+
+    target.load_from_file(destination, format="markdown")
+
+    assert manager.get_node_history("evidence-1")[0]["operation"] == "ADD_NODE"
+    assert manager.get_node_history("edge-supports")[0]["operation"] == "ADD_EDGE"
+
+
+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"
+        )
+
+
+def test_markdown_import_and_export_reject_windows_junctions(tmp_path, monkeypatch):
+    graph = ContextGraph(advanced_analytics=False)
+    graph.add_node("node-1", "Note", "Body")
+    export_path = tmp_path / "junction"
+    graph.save_to_file(export_path, format="markdown")
+
+    monkeypatch.setattr(
+        os.path,
+        "isjunction",
+        lambda candidate: Path(candidate) == export_path,
+        raising=False,
+    )
+
+    with pytest.raises(ValueError, match="junction"):
+        ContextGraph(advanced_analytics=False).load_from_file(
+            export_path, format="markdown"
+        )
+    with pytest.raises(ValueError, match="junction"):
+        graph.save_to_file(export_path, format="markdown")
+
+
+def test_markdown_import_rejects_windows_reparse_point_fallback(tmp_path, monkeypatch):
+    source = tmp_path / "reparse-point"
+    source.mkdir()
+    real_lstat = os.lstat
+
+    class ReparseStat:
+        st_file_attributes = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400)
+
+    monkeypatch.delattr(os.path, "isjunction", raising=False)
+    monkeypatch.setattr(Path, "is_symlink", lambda self: False)
+    monkeypatch.setattr(
+        os,
+        "lstat",
+        lambda candidate: (
+            ReparseStat() if Path(candidate) == source else real_lstat(candidate)
+        ),
+    )
+
+    with pytest.raises(ValueError, match="junction"):
+        ContextGraph(advanced_analytics=False).load_from_file(source, format="markdown")
+
+
+@pytest.mark.skipif(os.name != "nt", reason="requires Windows junctions")
+def test_markdown_import_rejects_real_windows_junction(tmp_path):
+    graph = ContextGraph(advanced_analytics=False)
+    graph.add_node("node-1", "Note", "Body")
+    outside = tmp_path / "outside"
+    graph.save_to_file(outside, format="markdown")
+    source = tmp_path / "junction"
+    result = subprocess.run(
+        ["cmd.exe", "/c", "mklink", "/J", str(source), str(outside)],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        pytest.skip(f"could not create Windows junction: {result.stderr}")
+
+    try:
+        with pytest.raises(ValueError, match="junction"):
+            ContextGraph(advanced_analytics=False).load_from_file(
+                source, format="markdown"
+            )
+    finally:
+        os.rmdir(source)
+
+
+@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._analytics_cache["stale"] = {"value": True}
+    restored.load_from_file(json_path)
+    assert "node-1" in restored.nodes
+    assert restored._analytics_cache == {}
+
+    before = _normalized_state(restored)
+    restored.load_from_file(tmp_path / "missing", format="markdown")
+    assert _normalized_state(restored) == before
+
+    with pytest.raises(ValueError, match="Unsupported context graph"):
+        graph.save_to_file(tmp_path / "graph", format="html")