fix(cli): align export dispatch with registry contract

Fix the export runtime mismatch where get_export_method expected the existing (task, name) registry contract but the CLI passed only the format argument.

semantica/cli.py

@@ -2089,30 +2089,73 @@ def export(
                  format=fmt, output=output)
             return
         try:
+            import tempfile
+
             from .export import get_export_method
-            fn = get_export_method(fmt)
+            from .graph_store import get_nodes, get_relationships
+            from .graph_store.config import graph_store_config
+
+            fn = get_export_method("export", "knowledge_graph")
+            if fn is None:
+                raise click.ClickException("Export method not available: export/knowledge_graph")
+
+            graph_db = dict(cli_ctx.config.to_dict().get("graph_db", {}))
+            backend = cli_ctx.store_backend or graph_db.pop("backend", None)
+            previous_graph_config = graph_store_config.get_all()
+            graph_store_config.update(graph_db)
+            if backend:
+                graph_store_config.set("default_backend", backend)
+            try:
+                entities = get_nodes(limit=sys.maxsize)
+                relationships = get_relationships(limit=sys.maxsize)
+            finally:
+                graph_store_config.update(previous_graph_config)
+
+            knowledge_graph = {
+                "entities": entities,
+                "relationships": relationships,
+                "nodes": entities,
+                "edges": relationships,
+            }
+
             kwargs: Dict[str, Any] = {"format": fmt}
             if with_provenance:
-                kwargs["with_provenance"] = True
+                kwargs["include_provenance"] = True
             if filter_str:
                 kwargs["filter"] = filter_str
-            result = fn(config=cli_ctx.config.to_dict(), **kwargs)
+
+            temp_output = None
+            target_output = output
+            if target_output is None or compress:
+                temp_handle = tempfile.NamedTemporaryFile(delete=False, suffix=".tmp")
+                temp_output = temp_handle.name
+                temp_handle.close()
+                target_output = temp_output
+
+            fn(knowledge_graph, target_output, **kwargs)
         except ImportError as exc:
             raise click.ClickException(f"Export module not available: {exc}") from exc
-        data = result if isinstance(result, (str, bytes)) else json.dumps(result, default=str)
         if compress:
             import gzip
-            compressed = gzip.compress(data.encode() if isinstance(data, str) else data)
+
+            assert temp_output is not None
+            compressed = gzip.compress(Path(temp_output).read_bytes())
             if output:
                 Path(output).write_bytes(compressed)
                 _ok(cli_ctx, f"Wrote compressed {output}")
             else:
                 sys.stdout.buffer.write(compressed)
         elif output:
-            Path(output).write_text(str(data), encoding="utf-8")
             _ok(cli_ctx, f"Wrote {output}")
         else:
-            click.echo(data)
+            assert temp_output is not None
+            try:
+                click.echo(Path(temp_output).read_text(encoding="utf-8"))
+            except UnicodeDecodeError:
+                sys.stdout.buffer.write(Path(temp_output).read_bytes())
+
+        if temp_output:
+            Path(temp_output).unlink(missing_ok=True)
 
     _run_with_error_handling(_action)
 

tests/test_cli_commands.py

@@ -947,14 +947,96 @@ class TestExport:
         data = _json_output(result)
         assert data["dry_run"] is True
 
+    def test_real_export_runtime_path(self, runner, tmp_path, monkeypatch):
+        class FakeGraphStore:
+            def get_nodes(self, labels=None, properties=None, limit=100, **options):
+                return [
+                    {
+                        "id": "n1",
+                        "type": "Person",
+                        "name": "Alice",
+                        "properties": {"name": "Alice"},
+                    }
+                ]
+
+            def get_relationships(self, node_id=None, rel_type=None, direction="both", limit=100, **options):
+                return [
+                    {
+                        "id": "r1",
+                        "source": "n1",
+                        "target": "n1",
+                        "type": "KNOWS",
+                        "properties": {},
+                    }
+                ]
+
+        monkeypatch.setattr(
+            "semantica.graph_store.methods._get_store",
+            lambda: FakeGraphStore(),
+        )
+        monkeypatch.setattr(
+            "semantica.graph_store.get_nodes",
+            lambda **kwargs: [
+                {
+                    "id": "n1",
+                    "type": "Person",
+                    "name": "Alice",
+                    "properties": {"name": "Alice"},
+                }
+            ],
+        )
+        monkeypatch.setattr(
+            "semantica.graph_store.methods.get_nodes",
+            lambda **kwargs: [
+                {
+                    "id": "n1",
+                    "type": "Person",
+                    "name": "Alice",
+                    "properties": {"name": "Alice"},
+                }
+            ],
+        )
+        monkeypatch.setattr(
+            "semantica.graph_store.get_relationships",
+            lambda **kwargs: [
+                {
+                    "id": "r1",
+                    "source": "n1",
+                    "target": "n1",
+                    "type": "KNOWS",
+                    "properties": {},
+                }
+            ],
+        )
+        monkeypatch.setattr(
+            "semantica.graph_store.methods.get_relationships",
+            lambda **kwargs: [
+                {
+                    "id": "r1",
+                    "source": "n1",
+                    "target": "n1",
+                    "type": "KNOWS",
+                    "properties": {},
+                }
+            ],
+        )
+
+        output_path = tmp_path / "export.json"
+        result = runner.invoke(main, ["export", "--format", "json", "--output", str(output_path)])
+        _ok(result)
+        exported = output_path.read_text(encoding="utf-8")
+        assert "Alice" in exported
+        assert "KNOWS" in exported
+
     def test_invalid_format_fails(self, runner):
         result = runner.invoke(main, ["export", "--format", "magic"])
         assert result.exit_code != 0
 
     def test_import_error_is_clean(self, runner):
+        original_import = __import__
         with patch("builtins.__import__", side_effect=lambda n, *a, **k: (
             (_ for _ in ()).throw(ImportError(n))
-            if "semantica.export" in n else __import__(n, *a, **k)
+            if "semantica.export" in n else original_import(n, *a, **k)
         )):
             result = runner.invoke(main, ["export", "--format", "json"])
         assert result.exit_code != 0

tests/test_export_module.py

@@ -60,6 +60,17 @@ class TestExportModule(unittest.TestCase):
         entities_path = Path(self.test_dir) / "entities.json"
         exporter.export_entities(self.entities, str(entities_path))
         self.assertTrue(entities_path.exists())
+
+    def test_export_knowledge_graph_smoke(self):
+        from semantica.export.methods import export_knowledge_graph
+
+        output_path = Path(self.test_dir) / "smoke.json"
+        export_knowledge_graph(self.kg, output_path, format="json")
+
+        self.assertTrue(output_path.exists())
+        exported = output_path.read_text(encoding="utf-8")
+        self.assertIn("Alice", exported)
+        self.assertIn("Acme Corp", exported)
         
     def test_csv_exporter(self):
         exporter = CSVExporter()