From 04c5921687ddc413b05da6707821a964c3ec13ab Mon Sep 17 00:00:00 2001
From: Jack Colquitt <126733989+JackColquitt@users.noreply.github.com>
Date: Tue, 30 Jun 2026 19:39:33 -0700
Subject: [PATCH 1/6] Fix Discord link in README (#1006)
Updated Discord link in README.md
---
README.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/README.md b/README.md
index 38d1c39b..69fcfb92 100644
--- a/README.md
+++ b/README.md
@@ -7,7 +7,7 @@
[](https://discord.gg/sQMwkRz5GX) [](https://deepwiki.com/trustgraph-ai/trustgraph)
-[**Website**](https://trustgraph.ai) | [**Docs**](https://docs.trustgraph.ai) | [**YouTube**](https://www.youtube.com/@TrustGraphAI?sub_confirmation=1) | [**Configuration Terminal**](https://config-ui.demo.trustgraph.ai/) | [**Discord**](https://discord.gg/sQMwkRz5GX) | [**Blog**](https://blog.trustgraph.ai/subscribe)
+[**Website**](https://trustgraph.ai) | [**Docs**](https://docs.trustgraph.ai) | [**YouTube**](https://www.youtube.com/@TrustGraphAI?sub_confirmation=1) | [**Configuration Terminal**](https://config-ui.demo.trustgraph.ai/) | [**Discord**](https://discord.gg/yUWRkfbD) | [**Blog**](https://blog.trustgraph.ai/subscribe)
From d2a4abb7c5d9f2603e9689ff6f7bb27e7ca08828 Mon Sep 17 00:00:00 2001
From: Jack Colquitt <126733989+JackColquitt@users.noreply.github.com>
Date: Fri, 3 Jul 2026 11:40:52 -0700
Subject: [PATCH 2/6] Revise README with new TrustGraph context description
(#1022)
Updated the README to enhance the description of TrustGraph and its features.
---
README.md | 9 +++++++--
1 file changed, 7 insertions(+), 2 deletions(-)
diff --git a/README.md b/README.md
index 69fcfb92..60d9ec6e 100644
--- a/README.md
+++ b/README.md
@@ -9,12 +9,17 @@
[**Website**](https://trustgraph.ai) | [**Docs**](https://docs.trustgraph.ai) | [**YouTube**](https://www.youtube.com/@TrustGraphAI?sub_confirmation=1) | [**Configuration Terminal**](https://config-ui.demo.trustgraph.ai/) | [**Discord**](https://discord.gg/yUWRkfbD) | [**Blog**](https://blog.trustgraph.ai/subscribe)
+### The Anti-Palantir
+
-# Write context once. Run agents anywhere.
-
+**Open Source · Open Standards · Total Transparency**
+---
+
+Write context once. Run agents anywhere. Own your data and the models.
+
Stop rebuilding context from scratch. TrustGraph treats context as a holon — a modular, independent whole that naturally snaps into a larger domain-wide intelligence layer. By deploying context as holonic context graphs, TrustGraph powers multi-tenant agent workflows, dramatically reduces token consumption, and aligns with semantic web standards (RDF, OWL, SKOS, SHACL). Version your context, share it across teams, and scale with full provenance.
## What TrustGraph Does
From 8d630ef204fb5b8ab506798e418ce97137137151 Mon Sep 17 00:00:00 2001
From: Jack Colquitt <126733989+JackColquitt@users.noreply.github.com>
Date: Fri, 3 Jul 2026 11:43:56 -0700
Subject: [PATCH 3/6] Update README with access request link (#1023)
---
README.md | 2 ++
1 file changed, 2 insertions(+)
diff --git a/README.md b/README.md
index 60d9ec6e..1b81b7a0 100644
--- a/README.md
+++ b/README.md
@@ -14,6 +14,8 @@
**Open Source · Open Standards · Total Transparency**
+
+[**Request Access to the Playground Preview**](https://docs.google.com/forms/d/e/1FAIpQLSeTnF22ZjUP20FWV--VvS5606x-5cOvnKty6AqcPdtlnPuqbQ/viewform)
---
From 4ac6bc09860f1f0dec59e25fd614d9ad5ae357cc Mon Sep 17 00:00:00 2001
From: Jack Colquitt <126733989+JackColquitt@users.noreply.github.com>
Date: Sun, 5 Jul 2026 07:39:29 -0700
Subject: [PATCH 4/6] Update section titles and video links in README (#1025)
---
README.md | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/README.md b/README.md
index 1b81b7a0..af9d9a2e 100644
--- a/README.md
+++ b/README.md
@@ -140,9 +140,9 @@ For a browser based configuration, try the [Configuration Terminal](https://conf
[](https://www.youtube.com/watch?v=gZjlt5WcWB4)
-## Watch Holonic Context Graphs in Action
+## Watch Building Real Agents from a Context Graph
-[](https://www.youtube.com/watch?v=sWc7mkhITIo)
+[](https://www.youtube.com/watch?v=lmhmrJ7zRE0)
## Getting Started with TrustGraph
From cfbd5b90791d44bcb5d2bd587b484bed6b962900 Mon Sep 17 00:00:00 2001
From: Sunny Yang
Date: Mon, 6 Jul 2026 03:45:23 -0600
Subject: [PATCH 5/6] feat: export/import workspace knowledge in .tgx bundles
(#877 Phase 2) (#1024)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
* feat: streaming N-Quads serializer for wire-format triples
Groundwork for Phase 2 of #877 (knowledge export). Hand-rolled
N-Triples term encoding: rdflib's term.n3() emits Turtle-style forms
(numeric shorthand, unescaped newlines) that are invalid in
line-oriented N-Quads, so literals are escaped per the ECHAR grammar
and IRIs validated for representability. Round-trip tests parse the
output back with rdflib's nquads parser and compare term-for-term.
* feat: export/import workspace knowledge in .tgx bundles (#877)
Phase 2 of the workspace bundle commands: tg-export-workspace now
includes the workspace's knowledge by default — per-collection
knowledge-graph triples as N-Quads (the collection names the graph,
streamed through a tempfile so memory stays flat regardless of
knowledge-base size) and the document library (metadata plus content,
fetched one document at a time). --config-only skips knowledge on both
sides; --triples-limit bounds very large graphs; -f/--flow-id selects
the flow the triples services run through.
tg-import-workspace streams triples back through the bulk import per
collection and recreates library documents (children after parents).
Knowledge import is additive, unlike config's skip-existing semantics.
Embedding vectors are not carried in bundles: --process re-runs
imported documents through the flow, which regenerates extraction
output and embeddings; --process-collection targets it.
Round-trip covered by unit tests over real archives: export with a
mocked Api, re-import, and assert the bulk triples stream and library
add calls reproduce the original values (including datatyped literals
via the N-Quads path).
---
tests/unit/test_cli/conftest.py | 16 +-
tests/unit/test_cli/test_nquads.py | 97 +++++
.../test_workspace_bundle_commands.py | 405 ++++++++++++++----
.../trustgraph/cli/export_workspace.py | 260 ++++++++---
.../trustgraph/cli/import_workspace.py | 306 +++++++++++--
trustgraph-cli/trustgraph/cli/nquads.py | 137 ++++++
6 files changed, 1040 insertions(+), 181 deletions(-)
create mode 100644 tests/unit/test_cli/test_nquads.py
create mode 100644 trustgraph-cli/trustgraph/cli/nquads.py
diff --git a/tests/unit/test_cli/conftest.py b/tests/unit/test_cli/conftest.py
index b085345f..ff293844 100644
--- a/tests/unit/test_cli/conftest.py
+++ b/tests/unit/test_cli/conftest.py
@@ -45,4 +45,18 @@ def sample_metadata():
"metadata": [],
"user": "test-user",
"collection": "test-collection"
- }
\ No newline at end of file
+ }
+
+def iri(v):
+ """Wire-format IRI term dict, as triples_query_stream yields them."""
+ return {"t": "i", "i": v}
+
+
+def lit(v, d=None, lang=None):
+ """Wire-format literal term dict (optional datatype / language)."""
+ t = {"t": "l", "v": v}
+ if d:
+ t["d"] = d
+ if lang:
+ t["l"] = lang
+ return t
diff --git a/tests/unit/test_cli/test_nquads.py b/tests/unit/test_cli/test_nquads.py
new file mode 100644
index 00000000..d395b324
--- /dev/null
+++ b/tests/unit/test_cli/test_nquads.py
@@ -0,0 +1,97 @@
+"""
+Round-trip tests for the streaming N-Quads serializer: wire-format triples
+are serialized line-by-line, then parsed back with rdflib's nquads parser
+and compared term-for-term — proving the output is valid N-Quads and the
+encoding (escaping, datatypes, language tags, unicode) is lossless.
+"""
+
+import io
+
+import rdflib
+
+from trustgraph.cli.nquads import serialize_nquads, triple_to_nquad
+
+from tests.unit.test_cli.conftest import iri, lit
+
+GRAPH = "urn:trustgraph:collection:default"
+
+
+def roundtrip(batches):
+ """Serialize then parse back; return (parsed_dataset, written, skipped)."""
+ out = io.StringIO()
+ written, skipped = serialize_nquads(batches, GRAPH, out)
+ ds = rdflib.Dataset()
+ ds.parse(data=out.getvalue(), format="nquads")
+ return ds, written, skipped
+
+
+class TestNquadsRoundTrip:
+
+ def test_iri_and_literal_flavours_survive_roundtrip(self):
+ batches = [[
+ {"s": iri("http://example.com/s"), "p": iri("http://example.com/p"),
+ "o": iri("http://example.com/o")},
+ {"s": iri("http://example.com/s"), "p": iri("http://example.com/label"),
+ "o": lit("plain value")},
+ {"s": iri("http://example.com/s"), "p": iri("http://example.com/label"),
+ "o": lit("bonjour", lang="fr")},
+ {"s": iri("http://example.com/s"), "p": iri("http://example.com/count"),
+ "o": lit("42", d="http://www.w3.org/2001/XMLSchema#integer")},
+ ]]
+ ds, written, skipped = roundtrip(batches)
+
+ assert (written, skipped) == (4, 0)
+ quads = list(ds.quads((None, None, None, None)))
+ assert len(quads) == 4
+ g = rdflib.URIRef(GRAPH)
+ assert all(q[3] == g for q in quads)
+
+ objects = {q[2] for q in quads}
+ assert rdflib.URIRef("http://example.com/o") in objects
+ assert rdflib.Literal("plain value") in objects
+ assert rdflib.Literal("bonjour", lang="fr") in objects
+ assert rdflib.Literal(
+ "42", datatype=rdflib.URIRef("http://www.w3.org/2001/XMLSchema#integer")
+ ) in objects
+
+ def test_hostile_literal_content_is_escaped_losslessly(self):
+ nasty = 'line1\nline2\t"quoted" back\\slash 中文 emoji\U0001f680'
+ batches = [[{
+ "s": iri("http://example.com/s"),
+ "p": iri("http://example.com/note"),
+ "o": lit(nasty),
+ }]]
+ ds, written, skipped = roundtrip(batches)
+
+ assert (written, skipped) == (1, 0)
+ obj = next(iter(ds.quads((None, None, None, None))))[2]
+ assert str(obj) == nasty
+
+ def test_malformed_and_unrepresentable_terms_are_skipped_not_emitted(self):
+ batches = [[
+ # IRI with a space (matches graph_to_turtle's malformed skip)
+ {"s": iri("http://example.com/bad iri"), "p": iri("http://example.com/p"),
+ "o": lit("x")},
+ # RDF-star quoted triple: no N-Quads encoding
+ {"s": iri("http://example.com/s"), "p": iri("http://example.com/p"),
+ "o": {"t": "r", "r": {}}},
+ # literal in predicate position: invalid RDF
+ {"s": iri("http://example.com/s"), "p": lit("not-a-predicate"),
+ "o": lit("x")},
+ # one good triple to prove the stream continues past skips
+ {"s": iri("http://example.com/s"), "p": iri("http://example.com/p"),
+ "o": lit("good")},
+ ]]
+ ds, written, skipped = roundtrip(batches)
+
+ assert (written, skipped) == (1, 3)
+ assert len(list(ds.quads((None, None, None, None)))) == 1
+
+ def test_streaming_shape_one_line_per_triple(self):
+ line = triple_to_nquad(
+ {"s": iri("http://example.com/s"), "p": iri("http://example.com/p"),
+ "o": lit("v")},
+ f"<{GRAPH}>",
+ )
+ assert line.endswith(" .\n")
+ assert line.count("\n") == 1
diff --git a/tests/unit/test_cli/test_workspace_bundle_commands.py b/tests/unit/test_cli/test_workspace_bundle_commands.py
index e45c0ef4..915e0dee 100644
--- a/tests/unit/test_cli/test_workspace_bundle_commands.py
+++ b/tests/unit/test_cli/test_workspace_bundle_commands.py
@@ -3,19 +3,25 @@ Tests for tg-export-workspace / tg-import-workspace (.tgx bundle commands).
The Api class is mocked in each command module's namespace (same pattern as
test_config_commands.py); bundles are written to and read from tmp_path so
-the archive format itself is exercised end-to-end.
+the archive format itself is exercised end-to-end, including the Phase-2
+knowledge tree (per-collection N-Quads + library documents).
"""
+import datetime
+import io
import json
import tarfile
+from types import SimpleNamespace
from unittest.mock import Mock, patch
import pytest
-from trustgraph.api.types import ConfigValue
+from trustgraph.api.types import ConfigValue, Triple
from trustgraph.cli.export_workspace import export_workspace
from trustgraph.cli.import_workspace import import_workspace
+from tests.unit.test_cli.conftest import iri, lit
+
SAMPLE_CONFIG = {
"prompt": {
"extract-concepts": json.dumps({"template": "Extract {{q}}"}),
@@ -26,25 +32,110 @@ SAMPLE_CONFIG = {
},
}
+# Wire-format triples for one collection, incl. a datatyped literal.
+WIRE_BATCHES = [[
+ {"s": iri("http://ex.com/s"), "p": iri("http://ex.com/p"),
+ "o": iri("http://ex.com/o")},
+ {"s": iri("http://ex.com/s"), "p": iri("http://ex.com/count"),
+ "o": lit("42", d="http://www.w3.org/2001/XMLSchema#integer")},
+]]
-def make_mock_api():
- mock_config = Mock()
+DOC = SimpleNamespace(
+ id="doc-1",
+ time=datetime.datetime(2026, 7, 1, 12, 0, 0),
+ kind="text/plain",
+ title="Policy",
+ comments="returns policy",
+ metadata=[Triple(s="http://ex.com/doc-1", p="http://ex.com/about",
+ o="returns")],
+ tags=["policy"],
+ parent_id="",
+ document_type="source",
+)
+
+
+def make_mock_api(collections=(), batches=(), docs=(), contents=None):
+ """Full-surface Api mock; returns (mock_api, mock_config)."""
mock_api = Mock()
+
+ mock_config = Mock()
mock_api.config.return_value = mock_config
+ mock_config.all.return_value = (SAMPLE_CONFIG, "v42")
+
+ mock_api.collection.return_value.list_collections.return_value = \
+ list(collections)
+
+ flow = mock_api.socket.return_value.flow.return_value
+ flow.triples_query_stream.return_value = iter(batches)
+
+ library = mock_api.library.return_value
+ library.get_documents.return_value = list(docs)
+ library.get_document_content.side_effect = \
+ lambda id: (contents or {}).get(id, b"")
+
return mock_api, mock_config
+def export_bundle(path, collections=(), batches=(), docs=(), contents=None,
+ **kwargs):
+ """Export SAMPLE_CONFIG (+ optional knowledge mocks) to path."""
+ mock_api, mock_config = make_mock_api(
+ collections=collections, batches=batches, docs=docs,
+ contents=contents,
+ )
+ with patch("trustgraph.cli.export_workspace.Api") as api_cls:
+ api_cls.return_value = mock_api
+ export_workspace(
+ url="http://api/", workspace="source-ws", output=str(path),
+ **kwargs,
+ )
+ return mock_api, mock_config
+
+
+DEFAULT_MANIFEST = {
+ "format": "tgx", "format_version": 1, "workspace": "w",
+ "contents": {"config": True, "knowledge": True},
+}
+
+
+def write_bundle(path, members, manifest=DEFAULT_MANIFEST):
+ """Write a raw .tgx from name -> bytes (manifest added first)."""
+ entries = {"manifest.json": json.dumps(manifest).encode(), **members}
+ with tarfile.open(path, "w:gz") as tar:
+ for name, data in entries.items():
+ info = tarfile.TarInfo(name)
+ info.size = len(data)
+ tar.addfile(info, io.BytesIO(data))
+ return path
+
+
+def run_import(mock_api, input, **kwargs):
+ """Run import_workspace against a mocked Api."""
+ with patch("trustgraph.cli.import_workspace.Api") as api_cls:
+ api_cls.return_value = mock_api
+ import_workspace(url="http://api/", input=str(input), **kwargs)
+ return api_cls
+
+
@pytest.fixture
def bundle(tmp_path):
- """Export SAMPLE_CONFIG to a real .tgx and yield its path."""
+ """Config-only-shaped bundle (no collections/docs mocked)."""
path = tmp_path / "ws.tgx"
- with patch("trustgraph.cli.export_workspace.Api") as api_cls:
- mock_api, mock_config = make_mock_api()
- api_cls.return_value = mock_api
- mock_config.all.return_value = (SAMPLE_CONFIG, "v42")
- export_workspace(
- url="http://api/", workspace="source-ws", output=str(path),
- )
+ export_bundle(path)
+ return path
+
+
+@pytest.fixture
+def knowledge_bundle(tmp_path):
+ """Bundle with one collection (2 triples) and one document."""
+ path = tmp_path / "kws.tgx"
+ export_bundle(
+ path,
+ collections=[SimpleNamespace(collection="research")],
+ batches=WIRE_BATCHES,
+ docs=[DOC],
+ contents={"doc-1": b"Customers may return items."},
+ )
return path
@@ -58,7 +149,10 @@ class TestExportWorkspace:
assert manifest["format"] == "tgx"
assert manifest["workspace"] == "source-ws"
assert manifest["config_version"] == "v42"
- assert manifest["contents"] == {"config": True, "knowledge": False}
+ assert manifest["contents"] == {"config": True, "knowledge": True}
+ assert manifest["knowledge"] == {
+ "collections": [], "documents": 0, "triples": {},
+ }
assert "config/prompt/extract-concepts.json" in names
assert "config/prompt/answer.json" in names
@@ -79,12 +173,12 @@ class TestExportWorkspace:
def test_path_unsafe_keys_are_quoted_in_filenames(self, tmp_path):
path = tmp_path / "ws.tgx"
+ mock_api, mock_config = make_mock_api()
+ mock_config.all.return_value = (
+ {"prompt": {"a/b": json.dumps({"x": 1})}}, "v1",
+ )
with patch("trustgraph.cli.export_workspace.Api") as api_cls:
- mock_api, mock_config = make_mock_api()
api_cls.return_value = mock_api
- mock_config.all.return_value = (
- {"prompt": {"a/b": json.dumps({"x": 1})}}, "v1",
- )
export_workspace(
url="http://api/", workspace="ws", output=str(path),
)
@@ -94,16 +188,63 @@ class TestExportWorkspace:
assert "config/prompt/a%2Fb.json" in names
assert entry["key"] == "a/b"
+ def test_knowledge_tree_written_per_collection_and_document(
+ self, knowledge_bundle):
+ with tarfile.open(knowledge_bundle, "r:gz") as tar:
+ names = tar.getnames()
+ manifest = json.load(tar.extractfile("manifest.json"))
+ nq = tar.extractfile(
+ "knowledge/research/triples.nq").read().decode()
+ meta = json.load(
+ tar.extractfile("knowledge/library/doc-1.meta.json"))
+ content = tar.extractfile(
+ "knowledge/library/doc-1.content").read()
+
+ assert manifest["contents"]["knowledge"] is True
+ assert manifest["knowledge"] == {
+ "collections": ["research"], "documents": 1,
+ "triples": {"research": 2},
+ }
+ assert "knowledge/research/triples.nq" in names
+ # N-Quads: one line per triple, graph = the collection IRI, and
+ # the datatyped literal keeps its full quoted form.
+ lines = [ln for ln in nq.splitlines() if ln]
+ assert len(lines) == 2
+ assert all("" in ln
+ for ln in lines)
+ assert '"42"^^' in nq
+
+ assert meta["id"] == "doc-1"
+ assert meta["title"] == "Policy"
+ assert meta["metadata"] == [{
+ "s": "http://ex.com/doc-1", "p": "http://ex.com/about",
+ "o": "returns",
+ }]
+ assert content == b"Customers may return items."
+
+ def test_config_only_skips_knowledge(self, tmp_path):
+ path = tmp_path / "co.tgx"
+ mock_api, _ = export_bundle(
+ path,
+ collections=[SimpleNamespace(collection="research")],
+ config_only=True,
+ )
+ with tarfile.open(path, "r:gz") as tar:
+ names = tar.getnames()
+ manifest = json.load(tar.extractfile("manifest.json"))
+ assert manifest["contents"] == {"config": True, "knowledge": False}
+ assert "knowledge" not in manifest
+ assert not any(n.startswith("knowledge/") for n in names)
+ mock_api.collection.return_value.list_collections.assert_not_called()
+ mock_api.socket.assert_not_called()
+ mock_api.library.assert_not_called()
+
class TestImportWorkspace:
def test_roundtrip_puts_all_values_with_overwrite(self, bundle):
- with patch("trustgraph.cli.import_workspace.Api") as api_cls:
- mock_api, mock_config = make_mock_api()
- api_cls.return_value = mock_api
- import_workspace(
- url="http://api/", input=str(bundle), overwrite=True,
- )
+ mock_api, mock_config = make_mock_api()
+ api_cls = run_import(mock_api, bundle, overwrite=True)
# Target workspace defaults to the manifest's workspace.
api_cls.assert_called_once_with(
@@ -123,27 +264,22 @@ class TestImportWorkspace:
assert all(isinstance(v, ConfigValue) for v in values)
def test_workspace_flag_renames_target(self, bundle):
- with patch("trustgraph.cli.import_workspace.Api") as api_cls:
- mock_api, mock_config = make_mock_api()
- api_cls.return_value = mock_api
- import_workspace(
- url="http://api/", input=str(bundle), workspace="staging",
- overwrite=True,
- )
+ mock_api, mock_config = make_mock_api()
+ api_cls = run_import(
+ mock_api, bundle, workspace="staging", overwrite=True,
+ )
api_cls.assert_called_once_with(
"http://api/", token=None, workspace="staging",
)
def test_default_skips_existing_keys(self, bundle):
"""WorkspaceInit re-run semantics: only missing keys are written."""
- with patch("trustgraph.cli.import_workspace.Api") as api_cls:
- mock_api, mock_config = make_mock_api()
- api_cls.return_value = mock_api
- mock_config.list.side_effect = lambda t: {
- "prompt": ["extract-concepts"],
- "tool": [],
- }[t]
- import_workspace(url="http://api/", input=str(bundle))
+ mock_api, mock_config = make_mock_api()
+ mock_config.list.side_effect = lambda t: {
+ "prompt": ["extract-concepts"],
+ "tool": [],
+ }[t]
+ run_import(mock_api, bundle)
values = mock_config.put.call_args.args[0]
assert sorted((v.type, v.key) for v in values) == [
@@ -152,17 +288,11 @@ class TestImportWorkspace:
]
def test_dry_run_writes_nothing(self, bundle, capsys):
- with patch("trustgraph.cli.import_workspace.Api") as api_cls:
- mock_api, mock_config = make_mock_api()
- api_cls.return_value = mock_api
- import_workspace(
- url="http://api/", input=str(bundle), overwrite=True,
- dry_run=True,
- )
+ mock_api, mock_config = make_mock_api()
+ run_import(mock_api, bundle, overwrite=True, dry_run=True)
mock_config.put.assert_not_called()
out = capsys.readouterr().out
assert "would import prompt/extract-concepts" in out
- assert "3 item(s) would be imported" in out
def test_rejects_bundle_without_manifest(self, tmp_path):
path = tmp_path / "bad.tgx"
@@ -173,52 +303,147 @@ class TestImportWorkspace:
import_workspace(url="http://api/", input=str(path))
def test_rejects_newer_format_version(self, tmp_path):
- import io
- path = tmp_path / "future.tgx"
- manifest = json.dumps({
- "format": "tgx", "format_version": 99, "workspace": "w",
- "contents": {"config": True, "knowledge": False},
- }).encode()
- with tarfile.open(path, "w:gz") as tar:
- info = tarfile.TarInfo("manifest.json")
- info.size = len(manifest)
- tar.addfile(info, io.BytesIO(manifest))
+ path = write_bundle(
+ tmp_path / "future.tgx", {},
+ manifest={**DEFAULT_MANIFEST, "format_version": 99},
+ )
with patch("trustgraph.cli.import_workspace.Api"):
with pytest.raises(RuntimeError, match="newer than this tool"):
import_workspace(url="http://api/", input=str(path))
- def test_refuses_knowledge_bundle_without_config_only(self, tmp_path):
- import io
- path = tmp_path / "knowledge.tgx"
- manifest = json.dumps({
- "format": "tgx", "format_version": 1, "workspace": "w",
- "contents": {"config": True, "knowledge": True},
- }).encode()
- with tarfile.open(path, "w:gz") as tar:
- info = tarfile.TarInfo("manifest.json")
- info.size = len(manifest)
- tar.addfile(info, io.BytesIO(manifest))
- with patch("trustgraph.cli.import_workspace.Api"):
- with pytest.raises(RuntimeError, match="--config-only"):
- import_workspace(url="http://api/", input=str(path))
- def test_config_only_flag_allows_knowledge_bundle(self, tmp_path):
- import io
- path = tmp_path / "knowledge.tgx"
- manifest = json.dumps({
- "format": "tgx", "format_version": 1, "workspace": "w",
- "contents": {"config": True, "knowledge": True},
- }).encode()
- with tarfile.open(path, "w:gz") as tar:
- info = tarfile.TarInfo("manifest.json")
- info.size = len(manifest)
- tar.addfile(info, io.BytesIO(manifest))
- with patch("trustgraph.cli.import_workspace.Api") as api_cls:
- mock_api, mock_config = make_mock_api()
- api_cls.return_value = mock_api
- import_workspace(
- url="http://api/", input=str(path), config_only=True,
- overwrite=True,
- )
- # No config entries in this bundle; nothing written, no error.
- mock_config.put.assert_not_called()
+class TestImportKnowledge:
+
+ def test_roundtrip_imports_triples_and_documents(self, knowledge_bundle):
+ mock_api, mock_config = make_mock_api()
+ run_import(mock_api, knowledge_bundle, overwrite=True)
+
+ # Triples land in the bulk import stream for the right collection.
+ bulk = mock_api.bulk.return_value
+ call = bulk.import_triples.call_args
+ assert call.args[0] == "default" # flow id
+ triples = sorted(list(call.args[1]), key=lambda t: t.p)
+ assert triples == [
+ Triple(s="http://ex.com/s", p="http://ex.com/count", o="42"),
+ Triple(s="http://ex.com/s", p="http://ex.com/p",
+ o="http://ex.com/o"),
+ ]
+ assert call.kwargs["metadata"]["collection"] == "research"
+
+ # The document is recreated with its metadata and content.
+ add = mock_api.library.return_value.add_document.call_args
+ assert add.kwargs["id"] == "doc-1"
+ assert add.kwargs["document"] == b"Customers may return items."
+ assert add.kwargs["title"] == "Policy"
+ assert add.kwargs["kind"] == "text/plain"
+ assert add.kwargs["tags"] == ["policy"]
+ assert add.kwargs["metadata"] == [
+ Triple(s="http://ex.com/doc-1", p="http://ex.com/about",
+ o="returns"),
+ ]
+ # No processing unless asked: embeddings re-derivation is opt-in.
+ mock_api.library.return_value.start_processing.assert_not_called()
+
+ def test_config_only_skips_knowledge_on_import(self, knowledge_bundle):
+ mock_api, mock_config = make_mock_api()
+ run_import(mock_api, knowledge_bundle, overwrite=True,
+ config_only=True)
+ mock_api.bulk.assert_not_called()
+ mock_api.library.return_value.add_document.assert_not_called()
+ mock_config.put.assert_called_once()
+
+ def test_dry_run_covers_knowledge(self, knowledge_bundle, capsys):
+ mock_api, mock_config = make_mock_api()
+ run_import(mock_api, knowledge_bundle, overwrite=True, dry_run=True)
+ mock_api.bulk.assert_not_called()
+ mock_api.library.return_value.add_document.assert_not_called()
+ out = capsys.readouterr().out
+ assert "would import 2 triple(s) into collection 'research'" in out
+ assert "would import document doc-1" in out
+
+ def test_process_flag_reprocesses_documents(self, knowledge_bundle):
+ mock_api, mock_config = make_mock_api()
+ run_import(mock_api, knowledge_bundle, overwrite=True, process=True,
+ process_collection="research")
+ proc = mock_api.library.return_value.start_processing.call_args
+ assert proc.kwargs["document_id"] == "doc-1"
+ assert proc.kwargs["flow"] == "default"
+ assert proc.kwargs["collection"] == "research"
+
+
+class TestImportDocumentSkipOverwrite:
+ """Live-verified semantics: existing documents skip by default, replace
+ with --overwrite (remove + add; the library API has no content update)."""
+
+ def _bundle_with_doc(self, tmp_path):
+ return write_bundle(tmp_path / "doc.tgx", {
+ "knowledge/library/doc-1.meta.json": json.dumps(
+ {"id": "doc-1", "title": "T", "metadata": []}).encode(),
+ "knowledge/library/doc-1.content": b"hello",
+ })
+
+ def test_existing_document_is_skipped_not_fatal(self, tmp_path):
+ path = self._bundle_with_doc(tmp_path)
+ mock_api, mock_config = make_mock_api(
+ docs=[SimpleNamespace(id="doc-1")],
+ )
+ run_import(mock_api, path, overwrite=False)
+ lib = mock_api.library.return_value
+ lib.add_document.assert_not_called()
+ lib.remove_document.assert_not_called()
+
+ def test_overwrite_replaces_existing_document(self, tmp_path):
+ path = self._bundle_with_doc(tmp_path)
+ mock_api, mock_config = make_mock_api(
+ docs=[SimpleNamespace(id="doc-1")],
+ )
+ run_import(mock_api, path, overwrite=True)
+ lib = mock_api.library.return_value
+ lib.remove_document.assert_called_once_with("doc-1")
+ lib.add_document.assert_called_once()
+
+
+class TestCollectionDiscovery:
+ """Live-verified: implicitly-created collections (raw triple loads) are
+ queryable but unlisted, so export merges --collection extras and import
+ registers what it restores."""
+
+ def test_export_includes_extra_unregistered_collections(self, tmp_path):
+ path = tmp_path / "ws.tgx"
+ mock_api, _ = export_bundle(
+ path,
+ collections=[SimpleNamespace(collection="default")],
+ batches=[[]],
+ extra_collections=["research"],
+ )
+ with tarfile.open(path, "r:gz") as tar:
+ manifest = json.load(tar.extractfile("manifest.json"))
+ assert manifest["knowledge"]["collections"] == ["default", "research"]
+
+ def test_import_registers_each_restored_collection(self, tmp_path):
+ path = write_bundle(tmp_path / "kb.tgx", {
+ "knowledge/research/triples.nq": (
+ b' "v" '
+ b' .\n'
+ ),
+ })
+ mock_api, mock_config = make_mock_api()
+ run_import(mock_api, path)
+ mock_api.collection.return_value.update_collection.assert_called_once_with(
+ "research", name="research",
+ )
+
+ def test_registered_collections_are_not_reregistered(self, tmp_path):
+ """update_collection is an upsert that clears omitted fields, so a
+ collection already in the registry must be left untouched."""
+ path = write_bundle(tmp_path / "kb.tgx", {
+ "knowledge/research/triples.nq": (
+ b' "v" '
+ b' .\n'
+ ),
+ })
+ mock_api, mock_config = make_mock_api(
+ collections=[SimpleNamespace(collection="research")],
+ )
+ run_import(mock_api, path)
+ mock_api.collection.return_value.update_collection.assert_not_called()
diff --git a/trustgraph-cli/trustgraph/cli/export_workspace.py b/trustgraph-cli/trustgraph/cli/export_workspace.py
index 41d3403d..2474ce36 100644
--- a/trustgraph-cli/trustgraph/cli/export_workspace.py
+++ b/trustgraph-cli/trustgraph/cli/export_workspace.py
@@ -1,14 +1,14 @@
"""
-Exports a workspace's full configuration state as a portable .tgx bundle
-(a gzipped tar archive) for backup, migration between deployments, or
-sharing a pre-configured workspace.
+Exports a workspace's full state as a portable .tgx bundle (a gzipped tar
+archive) for backup, migration between deployments, or sharing a
+pre-configured workspace.
-The bundle is human-readable: a manifest.json plus one pretty-printed JSON
-file per config key under config//, so it can be inspected and
-hand-edited before import. Each entry file embeds its own type and key, so
-filenames are cosmetic. Knowledge export (triples, documents, embeddings)
-is not yet included; the manifest records that so future importers can
-distinguish config-only bundles.
+The bundle carries the workspace configuration (one pretty-printed JSON
+file per config key) and, by default, its knowledge: per-collection
+knowledge-graph triples as N-Quads (the collection names the graph) and
+the document library (metadata plus content). Pass --config-only to
+export just the configuration. Embedding vectors are not exported —
+re-processing imported documents through a flow regenerates them.
"""
import argparse
@@ -17,11 +17,14 @@ import json
import os
import sys
import tarfile
+import tempfile
import time
from urllib.parse import quote
from trustgraph.api import Api
+from . nquads import serialize_nquads
+
default_url = os.getenv("TRUSTGRAPH_URL", 'http://localhost:8088/')
default_token = os.getenv("TRUSTGRAPH_TOKEN", None)
default_workspace = os.getenv("TRUSTGRAPH_WORKSPACE", "default")
@@ -29,6 +32,10 @@ default_workspace = os.getenv("TRUSTGRAPH_WORKSPACE", "default")
MANIFEST_FORMAT = "tgx"
MANIFEST_FORMAT_VERSION = 1
+# triples_query_stream is bounded by a limit; exports want "everything", so
+# default high and let --triples-limit override for truly huge graphs.
+DEFAULT_TRIPLES_LIMIT = 1_000_000
+
def _add_bytes(tar, name, data):
info = tarfile.TarInfo(name=name)
@@ -37,56 +44,176 @@ def _add_bytes(tar, name, data):
tar.addfile(info, io.BytesIO(data))
-def export_workspace(url, workspace, output, token=None):
-
- api = Api(url, token=token, workspace=workspace).config()
-
- config, version = api.all()
-
- manifest = {
- "format": MANIFEST_FORMAT,
- "format_version": MANIFEST_FORMAT_VERSION,
- "workspace": workspace,
- "config_version": version,
- "exported_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
- "contents": {"config": True, "knowledge": False},
- }
-
+def _export_config(tar, config):
+ """Write one self-describing JSON file per config key; return count."""
count = 0
+ for type_, entries in sorted(config.items()):
+ for key, raw in sorted(entries.items()):
+
+ # Config values are stored as JSON strings; parse so the
+ # bundle is pretty-printed and hand-editable. A value that
+ # isn't valid JSON is preserved verbatim.
+ try:
+ value = json.loads(raw)
+ except (TypeError, json.JSONDecodeError):
+ value = raw
+
+ entry = {"type": type_, "key": key, "value": value}
+
+ # Keys may contain path-unsafe characters; the entry embeds
+ # the real key, so the quoted filename is cosmetic only.
+ name = f"config/{quote(type_, safe='')}/{quote(key, safe='')}.json"
+
+ _add_bytes(
+ tar, name,
+ json.dumps(entry, indent=2).encode("utf-8"),
+ )
+
+ count += 1
+ return count
+
+
+def _export_triples(tar, api, flow_id, collections, triples_limit):
+ """Stream each collection's triples into knowledge//triples.nq.
+
+ N-Quads are written to a tempfile first (tar members need their size
+ upfront), so memory stays flat regardless of knowledge-base size.
+ Returns {collection: written} and a total skipped count.
+ """
+ counts = {}
+ skipped_total = 0
+ socket = api.socket()
+ try:
+ flow = socket.flow(flow_id)
+ for c in collections:
+ graph_iri = f"urn:trustgraph:collection:{quote(c, safe='')}"
+ tmp = tempfile.NamedTemporaryFile(
+ "w", encoding="utf-8", suffix=".nq", delete=False,
+ )
+ try:
+ with tmp:
+ written, skipped = serialize_nquads(
+ flow.triples_query_stream(
+ s=None, p=None, o=None,
+ collection=c,
+ limit=triples_limit,
+ batch_size=100,
+ ),
+ graph_iri,
+ tmp,
+ )
+ if written:
+ tar.add(
+ tmp.name,
+ arcname=(
+ f"knowledge/{quote(c, safe='')}/triples.nq"
+ ),
+ )
+ counts[c] = written
+ skipped_total += skipped
+ finally:
+ os.unlink(tmp.name)
+ finally:
+ socket.close()
+ return counts, skipped_total
+
+
+def _export_library(tar, api):
+ """Write each library document's metadata + content; return count."""
+ library = api.library()
+ count = 0
+ for doc in library.get_documents(include_children=True):
+ # Content is fetched one document at a time so memory is bounded
+ # by the largest single document, not the whole library.
+ content = library.get_document_content(doc.id)
+ meta = {
+ "id": doc.id,
+ "time": doc.time.isoformat() if doc.time else None,
+ "kind": doc.kind,
+ "title": doc.title,
+ "comments": doc.comments,
+ "metadata": [
+ {"s": t.s, "p": t.p, "o": t.o} for t in (doc.metadata or [])
+ ],
+ "tags": list(doc.tags or []),
+ "parent_id": doc.parent_id or "",
+ "document_type": getattr(doc, "document_type", "") or "",
+ }
+ base = f"knowledge/library/{quote(doc.id, safe='')}"
+ _add_bytes(
+ tar, f"{base}.meta.json",
+ json.dumps(meta, indent=2).encode("utf-8"),
+ )
+ _add_bytes(tar, f"{base}.content", content or b"")
+ count += 1
+ return count
+
+
+def export_workspace(
+ url, workspace, output, token=None, config_only=False,
+ flow_id="default", triples_limit=DEFAULT_TRIPLES_LIMIT,
+ extra_collections=(),
+):
+
+ api = Api(url, token=token, workspace=workspace)
+
+ config, version = api.config().all()
+
+ # Collection discovery is registry-based: collections created implicitly
+ # by raw triple loads (e.g. tg-load-knowledge) are queryable but not
+ # listed, so they would silently drop out of the bundle. --collection
+ # names them explicitly; the enumeration is printed so what's included
+ # is never a guess.
+ collections = []
+ if not config_only:
+ registered = [c.collection for c in api.collection().list_collections()]
+ collections = sorted(set(registered) | set(extra_collections))
+ print(f"Exporting collections: {', '.join(collections)}", flush=True)
with tarfile.open(output, "w:gz") as tar:
+ config_count = _export_config(tar, config)
+
+ triple_counts = {}
+ skipped = 0
+ doc_count = 0
+ if not config_only:
+ triple_counts, skipped = _export_triples(
+ tar, api, flow_id, collections, triples_limit,
+ )
+ doc_count = _export_library(tar, api)
+
+ manifest = {
+ "format": MANIFEST_FORMAT,
+ "format_version": MANIFEST_FORMAT_VERSION,
+ "workspace": workspace,
+ "config_version": version,
+ "exported_at": time.strftime(
+ "%Y-%m-%dT%H:%M:%SZ", time.gmtime(),
+ ),
+ "contents": {"config": True, "knowledge": not config_only},
+ }
+ if not config_only:
+ manifest["knowledge"] = {
+ "collections": collections,
+ "documents": doc_count,
+ "triples": triple_counts,
+ }
+
_add_bytes(
tar, "manifest.json",
json.dumps(manifest, indent=2).encode("utf-8"),
)
- for type_, entries in sorted(config.items()):
- for key, raw in sorted(entries.items()):
-
- # Config values are stored as JSON strings; parse so the
- # bundle is pretty-printed and hand-editable. A value that
- # isn't valid JSON is preserved verbatim.
- try:
- value = json.loads(raw)
- except (TypeError, json.JSONDecodeError):
- value = raw
-
- entry = {"type": type_, "key": key, "value": value}
-
- # Keys may contain path-unsafe characters; the entry embeds
- # the real key, so the quoted filename is cosmetic only.
- name = f"config/{quote(type_, safe='')}/{quote(key, safe='')}.json"
-
- _add_bytes(
- tar, name,
- json.dumps(entry, indent=2).encode("utf-8"),
- )
-
- count += 1
-
- print(f"Exported {count} config item(s) from workspace "
- f"'{workspace}' to {output}", flush=True)
+ summary = f"Exported {config_count} config item(s)"
+ if not config_only:
+ summary += (
+ f", {sum(triple_counts.values())} triple(s) across "
+ f"{len(triple_counts)} collection(s), {doc_count} document(s)"
+ )
+ if skipped:
+ summary += f" ({skipped} triple(s) not representable, skipped)"
+ print(f"{summary} from workspace '{workspace}' to {output}", flush=True)
def main():
@@ -114,12 +241,41 @@ def main():
help=f'Workspace to export (default: {default_workspace})',
)
+ parser.add_argument(
+ '-c', '--collection',
+ action='append',
+ default=[],
+ help='Additionally export this collection even if it is not '
+ 'registered in collection management (repeatable)',
+ )
+
parser.add_argument(
'-o', '--output',
required=True,
help='Output bundle path, e.g. workspace-default.tgx',
)
+ parser.add_argument(
+ '-f', '--flow-id',
+ default="default",
+ help='Flow to query triples through (default: default)',
+ )
+
+ parser.add_argument(
+ '--config-only',
+ action='store_true',
+ help='Export only the configuration, skipping knowledge '
+ '(triples and library documents)',
+ )
+
+ parser.add_argument(
+ '--triples-limit',
+ type=int,
+ default=DEFAULT_TRIPLES_LIMIT,
+ help='Maximum triples to export per collection '
+ f'(default: {DEFAULT_TRIPLES_LIMIT})',
+ )
+
args = parser.parse_args()
try:
@@ -129,6 +285,10 @@ def main():
workspace=args.workspace,
output=args.output,
token=args.token,
+ config_only=args.config_only,
+ flow_id=args.flow_id,
+ triples_limit=args.triples_limit,
+ extra_collections=args.collection,
)
except Exception as e:
diff --git a/trustgraph-cli/trustgraph/cli/import_workspace.py b/trustgraph-cli/trustgraph/cli/import_workspace.py
index 71885bfb..13fabdd1 100644
--- a/trustgraph-cli/trustgraph/cli/import_workspace.py
+++ b/trustgraph-cli/trustgraph/cli/import_workspace.py
@@ -3,10 +3,16 @@ Imports a workspace bundle (.tgx, produced by tg-export-workspace) into a
TrustGraph deployment. The target workspace defaults to the name recorded
in the bundle's manifest and can be renamed with --workspace.
-By default existing (type, key) entries in the target workspace are left
-untouched and only missing keys are added, matching WorkspaceInit's
-re-run behaviour; pass --overwrite to replace every imported key. Use
---dry-run to show what would be written without changing anything.
+Configuration import follows WorkspaceInit's re-run behaviour: existing
+(type, key) entries are left untouched and only missing keys are added;
+pass --overwrite to replace every imported key. Knowledge import (triples
+and library documents) is additive — triples are streamed into the target
+collection and documents are added to the library; re-importing the same
+bundle twice will duplicate knowledge, not merge it. Use --dry-run to
+show what would be written without changing anything, --config-only to
+skip a bundle's knowledge, and --process to re-run imported documents
+through a flow (which regenerates embeddings, so bundles don't carry
+vectors).
"""
import argparse
@@ -14,9 +20,12 @@ import json
import os
import sys
import tarfile
+import uuid
+from urllib.parse import unquote
from trustgraph.api import Api
-from trustgraph.api.types import ConfigValue
+from trustgraph.api.types import ConfigValue, Triple
+from trustgraph.cli.nquads import parse_nquads
default_url = os.getenv("TRUSTGRAPH_URL", 'http://localhost:8088/')
default_token = os.getenv("TRUSTGRAPH_TOKEN", None)
@@ -26,10 +35,21 @@ SUPPORTED_FORMAT_VERSION = 1
def _read_bundle(path):
- """Read manifest and config entries from a .tgx bundle."""
+ """Read manifest, config entries, triples and documents from a .tgx.
+
+ Returns (manifest, config_entries, triples_by_collection, documents)
+ where triples_by_collection maps collection -> list[Triple] and
+ documents is a list of {meta: dict, content: bytes}.
+ """
manifest = None
- entries = []
+ config_entries = []
+ triples = {}
+ doc_meta = {}
+ doc_content = {}
+
+ def member_id(name, prefix, suffix):
+ return unquote(name[len(prefix):-len(suffix)])
with tarfile.open(path, "r:gz") as tar:
for member in tar.getmembers():
@@ -39,11 +59,28 @@ def _read_bundle(path):
if f is None:
continue
data = f.read()
- if member.name == "manifest.json":
+ name = member.name
+
+ if name == "manifest.json":
manifest = json.loads(data)
- elif member.name.startswith("config/") and \
- member.name.endswith(".json"):
- entries.append(json.loads(data))
+
+ elif name.startswith("config/") and name.endswith(".json"):
+ config_entries.append(json.loads(data))
+
+ elif name.startswith("knowledge/library/") and \
+ name.endswith(".meta.json"):
+ doc_id = member_id(name, "knowledge/library/", ".meta.json")
+ doc_meta[doc_id] = json.loads(data)
+
+ elif name.startswith("knowledge/library/") and \
+ name.endswith(".content"):
+ doc_id = member_id(name, "knowledge/library/", ".content")
+ doc_content[doc_id] = data
+
+ elif name.startswith("knowledge/") and \
+ name.endswith("/triples.nq"):
+ collection = member_id(name, "knowledge/", "/triples.nq")
+ triples[collection] = parse_nquads(data)
if manifest is None:
raise RuntimeError("not a workspace bundle: manifest.json missing")
@@ -60,27 +97,18 @@ def _read_bundle(path):
"upgrade trustgraph-cli"
)
- return manifest, entries
+ documents = [
+ {"meta": meta, "content": doc_content.get(doc_id, b"")}
+ for doc_id, meta in doc_meta.items()
+ ]
+
+ return manifest, config_entries, triples, documents
-def import_workspace(
- url, input, workspace=None, overwrite=False, config_only=False,
- dry_run=False, token=None,
-):
+def _import_config(api, entries, overwrite, dry_run):
+ """Import config entries; returns (imported, skipped) counts."""
- manifest, entries = _read_bundle(input)
-
- # Knowledge import (triples, documents, embeddings) is not implemented
- # yet; refuse to silently drop it from a bundle that carries it.
- if manifest.get("contents", {}).get("knowledge") and not config_only:
- raise RuntimeError(
- "bundle contains knowledge data, which this tool cannot import "
- "yet; re-run with --config-only to import just the configuration"
- )
-
- target = workspace or manifest.get("workspace") or "default"
-
- api = Api(url, token=token, workspace=target).config()
+ config = api.config()
# Mirror WorkspaceInit's re-run behaviour: without --overwrite, keys
# already present in the target workspace are skipped (per key, not per
@@ -88,7 +116,7 @@ def import_workspace(
existing = {}
if not overwrite:
for type_ in sorted({e["type"] for e in entries}):
- existing[type_] = set(api.list(type_))
+ existing[type_] = set(config.list(type_))
values = []
skipped = 0
@@ -106,16 +134,191 @@ def import_workspace(
if dry_run:
for v in values:
print(f"would import {v.type}/{v.key}", flush=True)
- print(f"Dry run: {len(values)} item(s) would be imported into "
- f"workspace '{target}', {skipped} skipped as existing",
- flush=True)
- return
+ elif values:
+ config.put(values)
- if values:
- api.put(values)
+ return len(values), skipped
- print(f"Imported {len(values)} config item(s) into workspace "
- f"'{target}', {skipped} skipped as existing", flush=True)
+
+def _import_triples(api, flow_id, triples_by_collection, dry_run):
+ """Stream each collection's triples into the flow; returns count."""
+
+ # Collections restored by the bulk triples path (unlike document
+ # processing) are not auto-registered, and an unregistered collection
+ # would silently drop out of a future export. Register only the missing
+ # ones: update_collection is an upsert whose omitted fields clear the
+ # description/tags that _import_config may have just restored.
+ # Best-effort — a registry hiccup shouldn't fail a completed triple
+ # import (tg-set-collection is the manual remedy).
+ registered = None
+ if triples_by_collection and not dry_run:
+ try:
+ registered = {
+ c.collection for c in api.collection().list_collections()
+ }
+ except Exception as e:
+ print(
+ f"warning: could not list collections for registration: "
+ f"{e} — register manually with tg-set-collection",
+ flush=True,
+ )
+
+ total = 0
+ for collection, triples in sorted(triples_by_collection.items()):
+ if dry_run:
+ print(
+ f"would import {len(triples)} triple(s) into "
+ f"collection '{collection}'",
+ flush=True,
+ )
+ else:
+ api.bulk().import_triples(
+ flow_id,
+ triples,
+ metadata={
+ "id": f"workspace-import-{uuid.uuid4()}",
+ "metadata": [],
+ "collection": collection,
+ },
+ )
+ if registered is not None and collection not in registered:
+ try:
+ api.collection().update_collection(
+ collection, name=collection,
+ )
+ except Exception as e:
+ print(
+ f"warning: could not register collection "
+ f"'{collection}': {e} — register manually with "
+ f"tg-set-collection",
+ flush=True,
+ )
+ total += len(triples)
+ return total
+
+
+def _import_documents(
+ api, documents, flow_id, process, process_collection, overwrite,
+ dry_run,
+):
+ """Add library documents back (children after parents).
+
+ Mirrors the config semantics: documents already present in the target
+ workspace are skipped unless --overwrite, which replaces them (the
+ library API has no in-place content update, so replace = remove + add).
+ Returns (imported, skipped).
+ """
+
+ library = api.library()
+
+ existing = set()
+ if documents:
+ existing = {d.id for d in library.get_documents(include_children=True)}
+
+ # Parents must exist before their children.
+ ordered = sorted(
+ documents, key=lambda d: bool(d["meta"].get("parent_id")),
+ )
+
+ count = 0
+ doc_skipped = 0
+ for doc in ordered:
+ meta = doc["meta"]
+
+ if meta["id"] in existing and not overwrite:
+ doc_skipped += 1
+ if dry_run:
+ print(f"would skip existing document {meta['id']}", flush=True)
+ continue
+
+ if dry_run:
+ print(f"would import document {meta['id']}", flush=True)
+ count += 1
+ continue
+
+ if meta["id"] in existing:
+ library.remove_document(meta["id"])
+
+ metadata = [
+ Triple(s=t["s"], p=t["p"], o=t["o"])
+ for t in meta.get("metadata", [])
+ ]
+ if meta.get("parent_id"):
+ library.add_child_document(
+ document=doc["content"],
+ id=meta["id"],
+ parent_id=meta["parent_id"],
+ title=meta.get("title", ""),
+ comments=meta.get("comments", ""),
+ kind=meta.get("kind", "text/plain"),
+ tags=meta.get("tags", []),
+ metadata=metadata,
+ )
+ else:
+ library.add_document(
+ document=doc["content"],
+ id=meta["id"],
+ metadata=metadata,
+ title=meta.get("title", ""),
+ comments=meta.get("comments", ""),
+ kind=meta.get("kind", "text/plain"),
+ tags=meta.get("tags", []),
+ )
+
+ # Re-processing regenerates extraction output and embeddings for
+ # the imported content (bundles carry no vectors).
+ if process:
+ library.start_processing(
+ id=f"proc-{uuid.uuid4()}",
+ document_id=meta["id"],
+ flow=flow_id,
+ collection=process_collection,
+ tags=meta.get("tags", []),
+ )
+
+ count += 1
+ return count, doc_skipped
+
+
+def import_workspace(
+ url, input, workspace=None, overwrite=False, config_only=False,
+ flow_id="default", process=False, process_collection="default",
+ dry_run=False, token=None,
+):
+
+ manifest, config_entries, triples, documents = _read_bundle(input)
+
+ target = workspace or manifest.get("workspace") or "default"
+
+ api = Api(url, token=token, workspace=target)
+
+ imported, skipped = _import_config(
+ api, config_entries, overwrite, dry_run,
+ )
+
+ triple_count = 0
+ doc_count = 0
+ has_knowledge = bool(triples or documents)
+ if has_knowledge and not config_only:
+ triple_count = _import_triples(api, flow_id, triples, dry_run)
+ doc_count, doc_skipped = _import_documents(
+ api, documents, flow_id, process, process_collection, overwrite,
+ dry_run,
+ )
+
+ verb = "Dry run:" if dry_run else "Imported"
+ summary = (
+ f"{verb} {imported} config item(s) into workspace '{target}', "
+ f"{skipped} skipped as existing"
+ )
+ if has_knowledge and not config_only:
+ summary += (
+ f"; {triple_count} triple(s), {doc_count} document(s)"
+ f" ({doc_skipped} skipped as existing)"
+ )
+ elif has_knowledge:
+ summary += "; knowledge skipped (--config-only)"
+ print(summary, flush=True)
def main():
@@ -150,11 +353,18 @@ def main():
'bundle manifest)',
)
+ parser.add_argument(
+ '-f', '--flow-id',
+ default="default",
+ help='Flow to import triples through and process documents with '
+ '(default: default)',
+ )
+
parser.add_argument(
'--overwrite',
action='store_true',
- help='Replace existing keys in the target workspace (default: '
- 'keep existing keys and only add missing ones)',
+ help='Replace existing config keys in the target workspace '
+ '(default: keep existing keys and only add missing ones)',
)
parser.add_argument(
@@ -164,6 +374,19 @@ def main():
'in the bundle',
)
+ parser.add_argument(
+ '--process',
+ action='store_true',
+ help='Re-process imported documents through the flow after import '
+ '(regenerates extraction output and embeddings)',
+ )
+
+ parser.add_argument(
+ '--process-collection',
+ default="default",
+ help='Collection that --process targets (default: default)',
+ )
+
parser.add_argument(
'--dry-run',
action='store_true',
@@ -180,6 +403,9 @@ def main():
workspace=args.workspace,
overwrite=args.overwrite,
config_only=args.config_only,
+ flow_id=args.flow_id,
+ process=args.process,
+ process_collection=args.process_collection,
dry_run=args.dry_run,
token=args.token,
)
diff --git a/trustgraph-cli/trustgraph/cli/nquads.py b/trustgraph-cli/trustgraph/cli/nquads.py
new file mode 100644
index 00000000..f38a7747
--- /dev/null
+++ b/trustgraph-cli/trustgraph/cli/nquads.py
@@ -0,0 +1,137 @@
+"""
+N-Quads serialization and parsing for workspace knowledge bundles: the
+wire-format triples yielded by triples_query_stream go out one line per
+triple (so an export never holds a whole graph in memory), and bundle
+members come back as api Triple values. Term encoding is hand-rolled to
+the N-Triples grammar: rdflib's term.n3() emits Turtle-style forms
+(numeric shorthand, unescaped newlines) that are not valid in
+line-oriented N-Quads.
+"""
+
+import re
+
+import rdflib
+
+from trustgraph.schema import IRI, LITERAL
+from trustgraph.api.types import Triple
+
+# RDF-star quoted triples have no standard N-Quads encoding; they are
+# skipped with a count so callers can surface the omission.
+
+# N-Triples string-literal escapes (ECHAR): backslash first, then the rest.
+_ESCAPES = [
+ ("\\", "\\\\"),
+ ('"', '\\"'),
+ ("\n", "\\n"),
+ ("\r", "\\r"),
+ ("\t", "\\t"),
+]
+
+# Characters the IRIREF production cannot carry: controls/space plus the
+# explicitly forbidden set. One compiled scan keeps this off the profile
+# for large exports (it runs per term).
+_BAD_IRI = re.compile(r'[\x00-\x20<>"{}|^\x60]')
+
+
+def _escape_literal(value):
+ for raw, esc in _ESCAPES:
+ value = value.replace(raw, esc)
+ return value
+
+
+def _encode_iri(iri):
+ """, or None for values the grammar cannot carry."""
+ if not iri or _BAD_IRI.search(iri):
+ return None
+ return f"<{iri}>"
+
+
+def encode_term(term, is_object=False):
+ """Encode one wire-format term dict for an N-Quads line.
+
+ :param is_object: literals are only valid in object position;
+ subjects and predicates must be IRIs (bnodes never appear on
+ the wire).
+ :returns: encoded string, or None when the term can't be represented.
+ """
+ if term is None:
+ return None
+
+ t = term.get("t", "")
+
+ if t == IRI:
+ return _encode_iri(term.get("i", ""))
+
+ if t == LITERAL and is_object:
+ value = _escape_literal(term.get("v", ""))
+ language = term.get("l")
+ datatype = term.get("d")
+ if language:
+ return f'"{value}"@{language}'
+ if datatype:
+ dt = _encode_iri(datatype)
+ if dt is None:
+ return None
+ return f'"{value}"^^{dt}'
+ return f'"{value}"'
+
+ # literals outside object position, RDF-star, unknown types
+ return None
+
+
+def triple_to_nquad(triple, graph_encoded):
+ """One wire-format triple dict -> an N-Quads line, or None to skip.
+
+ :param triple: {"s": term, "p": term, "o": term} wire dict
+ :param graph_encoded: pre-encoded string
+ """
+ s = encode_term(triple.get("s"))
+ p = encode_term(triple.get("p"))
+ o = encode_term(triple.get("o"), is_object=True)
+ if s is None or p is None or o is None:
+ return None
+ return f"{s} {p} {o} {graph_encoded} .\n"
+
+
+def serialize_nquads(batches, graph_iri, out):
+ """Write wire-format triple batches to a text file-like as N-Quads.
+
+ :param batches: iterable of lists of wire triple dicts
+ (e.g. triples_query_stream output)
+ :param graph_iri: graph name for every quad (str)
+ :param out: text-mode file-like with .write()
+ :returns: (written, skipped) counts
+ """
+ g = _encode_iri(graph_iri)
+ if g is None:
+ raise ValueError(f"graph IRI not representable in N-Quads: {graph_iri!r}")
+ written = 0
+ skipped = 0
+ for batch in batches:
+ for t in batch:
+ line = triple_to_nquad(t, g)
+ if line is None:
+ skipped += 1
+ else:
+ out.write(line)
+ written += 1
+ return written, skipped
+
+
+def parse_nquads(data):
+ """Parse N-Quads bytes back into api Triple values.
+
+ Terms are stringified with str(), the same convention tg-load-knowledge
+ uses, so values survive the store round trip unchanged. The whole
+ member is materialized in memory (bundles are bounded by
+ --triples-limit at export); line-streaming is a possible follow-up.
+
+ :param data: N-Quads bytes (one bundle member)
+ :returns: list of Triple
+ """
+ ds = rdflib.Dataset()
+ ds.parse(data=data.decode("utf-8"), format="nquads")
+ return [
+ Triple(s=str(s), p=str(p), o=str(o))
+ for s, p, o, _g in ds.quads((None, None, None, None))
+ ]
From 08f69007c91046e9a704c1fde734a0baa6ad2ab7 Mon Sep 17 00:00:00 2001
From: cybermaggedon
Date: Mon, 6 Jul 2026 10:47:24 +0100
Subject: [PATCH 6/6] fix: handle missing time field in library API
document/processing responses (#1028)
Documents and processing records without a stored time field caused a
KeyError crash in get_documents and list_children. Use optional access
so time resolves to None instead of failing.
---
trustgraph-base/trustgraph/api/library.py | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/trustgraph-base/trustgraph/api/library.py b/trustgraph-base/trustgraph/api/library.py
index b3506bb7..1e87ed6e 100644
--- a/trustgraph-base/trustgraph/api/library.py
+++ b/trustgraph-base/trustgraph/api/library.py
@@ -363,7 +363,7 @@ class Library:
return [
DocumentMetadata(
id = v["id"],
- time = datetime.datetime.fromtimestamp(v["time"]),
+ time = datetime.datetime.fromtimestamp(v["time"]) if "time" in v else None,
kind = v["kind"],
title = v.get("title", ""),
comments = v.get("comments", ""),
@@ -678,7 +678,7 @@ class Library:
ProcessingMetadata(
id = v["id"],
document_id = v["document-id"],
- time = datetime.datetime.fromtimestamp(v["time"]),
+ time = datetime.datetime.fromtimestamp(v["time"]) if "time" in v else None,
flow = v["flow"],
collection = v["collection"],
tags = v["tags"],
@@ -983,7 +983,7 @@ class Library:
return [
DocumentMetadata(
id=v["id"],
- time=datetime.datetime.fromtimestamp(v["time"]),
+ time=datetime.datetime.fromtimestamp(v["time"]) if "time" in v else None,
kind=v["kind"],
title=v["title"],
comments=v.get("comments", ""),