From 2f855faa853402c079979210db930bcd1538bd0c Mon Sep 17 00:00:00 2001 From: Devon Stewart Date: Wed, 16 Sep 2026 15:47:46 -0700 Subject: [PATCH 1/5] fix: stabilize graph export contract --- src/kindex/cli.py | 44 +++++++++++++++++++++++--- src/kindex/graph_transfer.py | 48 +++++++++++++++++++++++++++-- tests/test_import_export.py | 60 ++++++++++++++++++++++++++++++++++++ 3 files changed, 146 insertions(+), 6 deletions(-) diff --git a/src/kindex/cli.py b/src/kindex/cli.py index e544c8e..969f4f3 100644 --- a/src/kindex/cli.py +++ b/src/kindex/cli.py @@ -2009,17 +2009,51 @@ def cmd_export(args): # A snapshot must not silently truncate at the query helper's display limit. nodes = [n for audience in audiences for n in store.all_nodes(audience=audience, limit=-1)] + from .graph_transfer import canonical_status, export_record + superseded_ids = { + edge["to_id"] + for node in nodes + for edge in store.edges_from(node["id"]) + if edge["type"] == "supersedes" + } + if getattr(args, "active_only", False): + # Status is canonicalized for old rows, while both the persisted marker + # and supersedes edge retire a predecessor. + nodes = [node for node in nodes if canonical_status(node) == "active" + and node["id"] not in superseded_ids] + # Apply PII stripping for public/org exports strip_pii = target_audience in ("public", "org") - # Strip edges that cross audience boundaries + # Canonicalize outgoing rows and infer only missing reciprocals at the same + # 0.8 ratio Store.add_edge has always used. This makes old one-sided graph + # rows and new bidirectional rows export identically. output = [] node_ids = {n["id"] for n in nodes} - from .graph_transfer import export_record - for n in nodes: + edges_by_node = {node_id: [] for node_id in node_ids} + edge_keys = set() + for node_id in sorted(node_ids): + for edge in store.edges_from(node_id): + if edge["to_id"] not in node_ids: + continue + edge_keys.add((node_id, edge["to_id"], edge["type"])) + edges_by_node[node_id].append(edge) + for from_id, to_id, edge_type in sorted(edge_keys): + if (to_id, from_id, edge_type) not in edge_keys: + primary = next(edge for edge in edges_by_node[from_id] + if edge["to_id"] == to_id and edge["type"] == edge_type) + edges_by_node[to_id].append({ + "to_id": from_id, + "type": edge_type, + "weight": primary["weight"] * 0.8, + "provenance": primary.get("provenance", ""), + }) + for edges in edges_by_node.values(): + edges.sort(key=lambda edge: (edge["to_id"], edge["type"], edge["weight"])) + for n in sorted(nodes, key=lambda node: node["id"]): if strip_pii: n = _strip_pii(n) - output.append(export_record(n, store.edges_from(n["id"]), node_ids, public=strip_pii)) + output.append(export_record(n, edges_by_node[n["id"]], node_ids, public=strip_pii)) if args.format == "jsonl": for item in output: @@ -7394,6 +7428,8 @@ def build_parser() -> argparse.ArgumentParser: help="Export graph (default) or a UA-compatible code map") s.add_argument("--audience", choices=["private", "team", "org", "public"], default="team") s.add_argument("--format", choices=["json", "jsonl", "understand-anything"], default="json") + s.add_argument("--active-only", action="store_true", + help="Export only canonical active nodes, excluding superseded predecessors") s.add_argument("--directory", help="Repository root for code-map metadata") s.add_argument("--project-name", help="Project name for code-map export") s.add_argument("--output", help="Write export to this file instead of stdout") diff --git a/src/kindex/graph_transfer.py b/src/kindex/graph_transfer.py index 79aae2a..a73d8ef 100644 --- a/src/kindex/graph_transfer.py +++ b/src/kindex/graph_transfer.py @@ -28,6 +28,21 @@ ) +def canonical_status(node: dict) -> str: + """Normalize legacy lifecycle omissions for the portable graph contract.""" + extra = node.get("extra") + if isinstance(extra, dict) and extra.get("superseded_by"): + return "superseded" + status = node.get("status") + return status.strip() if isinstance(status, str) and status.strip() else "active" + + +def canonical_standing(node: dict) -> str: + """Normalize pre-v13 records, which did not persist an explicit standing.""" + standing = node.get("standing") + return standing.strip() if isinstance(standing, str) and standing.strip() else "unruled" + + def scrub_shared_text(value: str) -> str: """Remove contact and machine-path prose while retaining evidence URLs.""" value = re.sub(r'\S+@\S+\.\S+', '[email]', value) @@ -90,8 +105,37 @@ def scrub(value): def export_record(node: dict, edges: list[dict], visible_ids: set[str], *, public: bool) -> dict: - record = {key: node[key] for key in (*TEXT_FIELDS, *LIST_FIELDS, *CLOCK_FIELDS, - *VERIFICATION_FIELDS, "id", "weight", "referent") if key in node} + # This is a public interchange schema, not a projection of whichever + # columns an older writer happened to load. Keep every field present so + # unchanged JSONL snapshots remain byte-stable across CLI versions. + record = { + "id": node["id"], + "type": node.get("type") or "concept", + "title": node.get("title") or "", + "content": node.get("content") or "", + "intent": node.get("intent") or "", + "status": canonical_status(node), + "audience": node.get("audience") or "private", + "prov_when": node.get("prov_when") or "", + "prov_activity": node.get("prov_activity") or "", + "prov_why": node.get("prov_why") or "", + "prov_source": node.get("prov_source") or "", + "created_at": node.get("created_at") or "", + "updated_at": node.get("updated_at") or "", + "standing": canonical_standing(node), + "aka": node.get("aka") or [], + "domains": node.get("domains") or [], + "prov_who": node.get("prov_who") or [], + "valid_at": node.get("valid_at") or None, + "invalid_at": node.get("invalid_at") or None, + "asserted_at": node.get("asserted_at") or None, + "true_of": node.get("true_of") or None, + "verified_at": node.get("verified_at") or None, + "verified_by": node.get("verified_by") or None, + "prov_method": node.get("prov_method") or None, + "weight": node.get("weight", 0.5), + "referent": node.get("referent") if isinstance(node.get("referent"), dict) else None, + } for key in LIST_FIELDS: record[key] = node.get(key) or [] # Older SQLite defaults used empty text. record["extra"] = {k: v for k, v in (node.get("extra") or {}).items() if k in LIFECYCLE_KEYS} diff --git a/tests/test_import_export.py b/tests/test_import_export.py index 64c17cc..c73f517 100644 --- a/tests/test_import_export.py +++ b/tests/test_import_export.py @@ -122,6 +122,66 @@ def test_export_jsonl_multiple_nodes(self, tmp_path): lines = [l for l in r.stdout.strip().split("\n") if l.strip()] assert len(lines) >= 3 + def test_export_uses_complete_stable_active_contract(self, tmp_path): + """JSONL rows retain their shape across stores with legacy omissions.""" + d = str(tmp_path) + run("init", data_dir=d) + s = Store(Config(data_dir=d)) + s.add_node("Legacy active", node_id="legacy", audience="private") + s.add_node("Missing status", node_id="missing", audience="private") + s.add_node("Replacement", node_id="replacement", audience="private") + s.add_node("Archived", node_id="archived", audience="private", + status="archived") + # Simulate an older persisted row: no lifecycle value, but a successor + # marker that must still win when filtering the active projection. + s.conn.execute("UPDATE nodes SET status = '', extra = ? WHERE id = 'legacy'", + (json.dumps({"superseded_by": "replacement"}),)) + s.conn.execute("UPDATE nodes SET status = '' WHERE id = 'missing'") + s.add_edge("replacement", "archived", edge_type="relates_to", weight=0.75, + bidirectional=False) + s.close() + + r = run("export", "--audience", "private", "--format", "jsonl", + "--active-only", data_dir=d) + assert r.returncode == 0, r.stderr + rows = [json.loads(line) for line in r.stdout.splitlines() if line] + assert [row["id"] for row in rows] == ["missing", "replacement"] + assert tuple(rows[0]) == ( + "id", "type", "title", "content", "intent", "status", "audience", + "prov_when", "prov_activity", "prov_why", "prov_source", "created_at", + "updated_at", "standing", "aka", "domains", "prov_who", "valid_at", + "invalid_at", "asserted_at", "true_of", "verified_at", "verified_by", + "prov_method", "weight", "referent", "extra", "edges", + ) + for row in rows: + assert row["status"] == "active" + assert row["standing"] == "unruled" + assert row["referent"] is None + assert row["asserted_at"] is None + assert row["true_of"] is None + + def test_export_inferrs_missing_reciprocal_edge_weight(self, tmp_path): + d = str(tmp_path) + run("init", data_dir=d) + s = Store(Config(data_dir=d)) + s.add_node("A", node_id="a", audience="private") + s.add_node("B", node_id="b", audience="private") + s.add_edge("a", "b", edge_type="implements", weight=0.75, + bidirectional=False) + s.close() + + r = run("export", "--audience", "private", "--format", "json", data_dir=d) + assert r.returncode == 0, r.stderr + rows = {row["id"]: row for row in json.loads(r.stdout)} + assert rows["a"]["edges"] == [ + {"to": "b", "type": "implements", "weight": 0.75, + "bidirectional": False, "provenance": ""}, + ] + assert rows["b"]["edges"] == [ + {"to": "a", "type": "implements", "weight": 0.75 * 0.8, + "bidirectional": False, "provenance": ""}, + ] + class TestImportJSON: def test_import_json(self, tmp_path): From ace95e14aee1d5e5e1bcd8ac94a556211acb7502 Mon Sep 17 00:00:00 2001 From: Devon Stewart Date: Wed, 23 Sep 2026 14:21:26 -0700 Subject: [PATCH 2/5] fix(export): preserve directed graph arcs --- src/kindex/cli.py | 28 ++++------------------------ tests/test_import_export.py | 30 +++++++++++++++++++++++++----- 2 files changed, 29 insertions(+), 29 deletions(-) diff --git a/src/kindex/cli.py b/src/kindex/cli.py index 969f4f3..3023026 100644 --- a/src/kindex/cli.py +++ b/src/kindex/cli.py @@ -2025,35 +2025,15 @@ def cmd_export(args): # Apply PII stripping for public/org exports strip_pii = target_audience in ("public", "org") - # Canonicalize outgoing rows and infer only missing reciprocals at the same - # 0.8 ratio Store.add_edge has always used. This makes old one-sided graph - # rows and new bidirectional rows export identically. + # Export the stored arcs verbatim. Store.add_edge persists a reciprocal + # itself when an edge is bidirectional; inferring one here would invert + # directed relationships during an export/import round trip. output = [] node_ids = {n["id"] for n in nodes} - edges_by_node = {node_id: [] for node_id in node_ids} - edge_keys = set() - for node_id in sorted(node_ids): - for edge in store.edges_from(node_id): - if edge["to_id"] not in node_ids: - continue - edge_keys.add((node_id, edge["to_id"], edge["type"])) - edges_by_node[node_id].append(edge) - for from_id, to_id, edge_type in sorted(edge_keys): - if (to_id, from_id, edge_type) not in edge_keys: - primary = next(edge for edge in edges_by_node[from_id] - if edge["to_id"] == to_id and edge["type"] == edge_type) - edges_by_node[to_id].append({ - "to_id": from_id, - "type": edge_type, - "weight": primary["weight"] * 0.8, - "provenance": primary.get("provenance", ""), - }) - for edges in edges_by_node.values(): - edges.sort(key=lambda edge: (edge["to_id"], edge["type"], edge["weight"])) for n in sorted(nodes, key=lambda node: node["id"]): if strip_pii: n = _strip_pii(n) - output.append(export_record(n, edges_by_node[n["id"]], node_ids, public=strip_pii)) + output.append(export_record(n, store.edges_from(n["id"]), node_ids, public=strip_pii)) if args.format == "jsonl": for item in output: diff --git a/tests/test_import_export.py b/tests/test_import_export.py index c73f517..3be6224 100644 --- a/tests/test_import_export.py +++ b/tests/test_import_export.py @@ -160,7 +160,7 @@ def test_export_uses_complete_stable_active_contract(self, tmp_path): assert row["asserted_at"] is None assert row["true_of"] is None - def test_export_inferrs_missing_reciprocal_edge_weight(self, tmp_path): + def test_export_preserves_stored_directed_edges(self, tmp_path): d = str(tmp_path) run("init", data_dir=d) s = Store(Config(data_dir=d)) @@ -177,10 +177,7 @@ def test_export_inferrs_missing_reciprocal_edge_weight(self, tmp_path): {"to": "b", "type": "implements", "weight": 0.75, "bidirectional": False, "provenance": ""}, ] - assert rows["b"]["edges"] == [ - {"to": "a", "type": "implements", "weight": 0.75 * 0.8, - "bidirectional": False, "provenance": ""}, - ] + assert rows["b"]["edges"] == [] class TestImportJSON: @@ -275,6 +272,29 @@ def test_import_merge(self, tmp_path): class TestRoundtrip: + def test_cli_roundtrip_active_only_preserves_superseding_successor(self, tmp_path): + """A one-way supersedes arc must not retire its successor on replay.""" + source_dir, dest_dir = str(tmp_path / "source"), str(tmp_path / "dest") + run("init", data_dir=source_dir) + source = Store(Config(data_dir=source_dir)) + source.add_node("Prior decision", node_id="prior", node_type="decision") + source.add_node("Current decision", node_id="current", node_type="decision") + source.add_edge("current", "prior", edge_type="supersedes", bidirectional=False) + source.close() + + exported = run("export", "--audience", "private", "--format", "json", + data_dir=source_dir) + assert exported.returncode == 0, exported.stderr + transfer = tmp_path / "graph.json" + transfer.write_text(exported.stdout) + + imported = run("import", str(transfer), data_dir=dest_dir) + assert imported.returncode == 0, imported.stderr + active_only = run("export", "--audience", "private", "--format", "json", + "--active-only", data_dir=dest_dir) + assert active_only.returncode == 0, active_only.stderr + assert [node["id"] for node in json.loads(active_only.stdout)] == ["current"] + def test_roundtrip(self, tmp_path): """Export then import, verify lossless.""" d = str(tmp_path) From d519c1d3491c7463001a9f0517c413139eb23bc3 Mon Sep 17 00:00:00 2001 From: Devon Stewart Date: Wed, 23 Sep 2026 15:05:24 -0700 Subject: [PATCH 3/5] test(export): cover JSONL directed-edge roundtrip --- tests/test_import_export.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/tests/test_import_export.py b/tests/test_import_export.py index 3be6224..5b10579 100644 --- a/tests/test_import_export.py +++ b/tests/test_import_export.py @@ -272,7 +272,13 @@ def test_import_merge(self, tmp_path): class TestRoundtrip: - def test_cli_roundtrip_active_only_preserves_superseding_successor(self, tmp_path): + @pytest.mark.parametrize( + ("export_format", "extension"), + [("json", "json"), ("jsonl", "jsonl")], + ) + def test_cli_roundtrip_active_only_preserves_superseding_successor( + self, tmp_path, export_format, extension, + ): """A one-way supersedes arc must not retire its successor on replay.""" source_dir, dest_dir = str(tmp_path / "source"), str(tmp_path / "dest") run("init", data_dir=source_dir) @@ -282,10 +288,10 @@ def test_cli_roundtrip_active_only_preserves_superseding_successor(self, tmp_pat source.add_edge("current", "prior", edge_type="supersedes", bidirectional=False) source.close() - exported = run("export", "--audience", "private", "--format", "json", + exported = run("export", "--audience", "private", "--format", export_format, data_dir=source_dir) assert exported.returncode == 0, exported.stderr - transfer = tmp_path / "graph.json" + transfer = tmp_path / f"graph.{extension}" transfer.write_text(exported.stdout) imported = run("import", str(transfer), data_dir=dest_dir) From f2a92af88d5b8d0a2f9f1884b4916b31ce3cadb6 Mon Sep 17 00:00:00 2001 From: Devon Stewart Date: Thu, 24 Sep 2026 14:13:17 -0700 Subject: [PATCH 4/5] fix: enforce active export contract --- docs/human-guide.md | 4 ++++ src/kindex/cli.py | 17 +++++++++-------- src/kindex/graph_transfer.py | 6 +++++- 3 files changed, 18 insertions(+), 9 deletions(-) diff --git a/docs/human-guide.md b/docs/human-guide.md index ce27565..421c232 100644 --- a/docs/human-guide.md +++ b/docs/human-guide.md @@ -398,6 +398,10 @@ kin import graph.jsonl --data-dir /path/to/isolated/graph --dry-run kin import graph.jsonl --data-dir /path/to/isolated/graph ``` +Add `--active-only` to export only canonically active nodes. Superseding edges +are evaluated across the source graph even when the successor is outside the +selected audience; out-of-scope node IDs remain omitted from the export. + JSON arrays, single JSON records and JSONL remain supported. These are knowledge snapshots, **not full backups** of tasks, locks, reminders, policy or runtime state. Only lifecycle keys from `extra` travel: expiry, stale-referent and supersession diff --git a/src/kindex/cli.py b/src/kindex/cli.py index 3023026..e3f2e99 100644 --- a/src/kindex/cli.py +++ b/src/kindex/cli.py @@ -2010,15 +2010,16 @@ def cmd_export(args): nodes = [n for audience in audiences for n in store.all_nodes(audience=audience, limit=-1)] from .graph_transfer import canonical_status, export_record - superseded_ids = { - edge["to_id"] - for node in nodes - for edge in store.edges_from(node["id"]) - if edge["type"] == "supersedes" - } if getattr(args, "active_only", False): - # Status is canonicalized for old rows, while both the persisted marker - # and supersedes edge retire a predecessor. + # Decide retirement against the full graph before audience projection: + # a visible predecessor can be superseded by a hidden successor. Only + # target IDs are used for filtering; edge serialization remains scoped. + superseded_ids = { + row["to_id"] + for row in store.conn.execute( + "SELECT DISTINCT to_id FROM edges WHERE type = 'supersedes'" + ).fetchall() + } nodes = [node for node in nodes if canonical_status(node) == "active" and node["id"] not in superseded_ids] diff --git a/src/kindex/graph_transfer.py b/src/kindex/graph_transfer.py index a73d8ef..83625b2 100644 --- a/src/kindex/graph_transfer.py +++ b/src/kindex/graph_transfer.py @@ -143,10 +143,14 @@ def export_record(node: dict, edges: list[dict], visible_ids: set[str], *, publi for key in ("supersedes", "superseded_by"): if record["extra"].get(key) not in visible_ids: record["extra"].pop(key, None) + visible_edges = [e for e in edges if e["to_id"] in visible_ids] + # Store.edges_from sorts by weight only. Add a total tie-break order here + # so equivalent graphs serialize identically across insertion histories. + visible_edges.sort(key=lambda edge: (-edge["weight"], edge["to_id"], edge["type"])) record["edges"] = [ {"to": e["to_id"], "type": e["type"], "weight": e["weight"], "bidirectional": False, "provenance": "" if public else e.get("provenance", "")} - for e in edges if e["to_id"] in visible_ids + for e in visible_edges ] if public: for key in ("title", "content"): From 79f2c0aa2631e152c0f8d847a68f8645a37fd05d Mon Sep 17 00:00:00 2001 From: jmcentire Date: Thu, 1 Oct 2026 13:33:37 -0500 Subject: [PATCH 5/5] test(export): cover audience contracts and edge ordering Document the portable export contract and verify shared-audience privacy projections, hidden successor retirement, and stable edge order after storage rewrites. Co-Authored-By: Codex --- CHANGELOG.md | 7 ++++ docs/human-guide.md | 3 ++ tests/test_import_export.py | 70 +++++++++++++++++++++++++++++++++++++ 3 files changed, 80 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index ae1f01e..dd590cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,13 @@ All notable changes to Kindex are documented here. Format follows [Keep a Change `KIN_LOG_TIMESTAMPS=1`, and `kin` prefixes its stdout/stderr lines when that variable is set. Interactive and piped output is unchanged. Re-run `kin setup-cron` to pick it up on an existing install. +- `kin export --active-only` excludes retired and superseded nodes, including + predecessors replaced by a successor outside the selected audience. + +### Fixed +- Graph exports use a stable schema per audience and deterministic node/edge + ordering. Directed relationships retain their stored direction on import; + public/org exports preserve existing privacy omissions. ## [0.45.0] - 2026-09-28 diff --git a/docs/human-guide.md b/docs/human-guide.md index 421c232..af501be 100644 --- a/docs/human-guide.md +++ b/docs/human-guide.md @@ -401,6 +401,9 @@ kin import graph.jsonl --data-dir /path/to/isolated/graph Add `--active-only` to export only canonically active nodes. Superseding edges are evaluated across the source graph even when the successor is outside the selected audience; out-of-scope node IDs remain omitted from the export. +Records have a stable field order for each audience, and nodes and stored arcs +have deterministic ordering. Public/org exports retain their privacy projection: +they omit intent, activity, reasoning and verification-method prose. JSON arrays, single JSON records and JSONL remain supported. These are knowledge snapshots, **not full backups** of tasks, locks, reminders, policy or runtime state. diff --git a/tests/test_import_export.py b/tests/test_import_export.py index 5b10579..b78993b 100644 --- a/tests/test_import_export.py +++ b/tests/test_import_export.py @@ -85,6 +85,76 @@ def test_export_json_audience_filter(self, tmp_path): class TestExportJSONL: + @pytest.mark.parametrize("audience", ["private", "team", "org", "public"]) + def test_schema_is_stable_for_each_audience(self, tmp_path, audience): + d = str(tmp_path) + s = Store(Config(data_dir=d)) + s.add_node("Minimal", node_id="minimal", audience="public") + s.add_node("Detailed", node_id="detailed", audience="public", + content="Public evidence", intent="Internal intent") + s.close() + result = run("export", "--audience", audience, "--format", "jsonl", data_dir=d) + assert result.returncode == 0, result.stderr + rows = [json.loads(line) for line in result.stdout.splitlines()] + expected = ( + "id", "type", "title", "content", "intent", "status", "audience", + "prov_when", "prov_activity", "prov_why", "prov_source", "created_at", + "updated_at", "standing", "aka", "domains", "prov_who", "valid_at", + "invalid_at", "asserted_at", "true_of", "verified_at", "verified_by", + "prov_method", "weight", "referent", "extra", "edges", + ) + if audience in ("org", "public"): + expected = tuple(key for key in expected + if key not in ("intent", "prov_activity", "prov_why", "prov_method")) + assert "Internal intent" not in result.stdout + assert len(rows) == 2 + assert all(tuple(row) == expected for row in rows) + + @pytest.mark.parametrize("export_format", ["json", "jsonl"]) + def test_export_edge_order_survives_storage_reordering(self, tmp_path, export_format): + d = str(tmp_path) + s = Store(Config(data_dir=d)) + for node_id in ("hub", "a", "b", "c"): + s.add_node(node_id, node_id=node_id) + arcs = [("b", "relates_to"), ("a", "relates_to"), ("a", "implements")] + for target, edge_type in arcs: + s.add_edge("hub", target, edge_type=edge_type, weight=0.5, bidirectional=False) + s.add_edge("hub", "c", weight=0.75, bidirectional=False) + before = run("export", "--audience", "private", "--format", export_format, data_dir=d) + assert before.returncode == 0, before.stderr + s.conn.execute("DELETE FROM edges WHERE from_id = 'hub' AND weight = 0.5") + for target, edge_type in reversed(arcs): + s.add_edge("hub", target, edge_type=edge_type, weight=0.5, bidirectional=False) + s.close() + after = run("export", "--audience", "private", "--format", export_format, data_dir=d) + assert after.returncode == 0, after.stderr + assert after.stdout == before.stdout + rows = json.loads(after.stdout) if export_format == "json" else [ + json.loads(line) for line in after.stdout.splitlines()] + hub = next(row for row in rows if row["id"] == "hub") + assert [(edge["to"], edge["type"]) for edge in hub["edges"]] == [ + ("c", "relates_to"), ("a", "implements"), ("a", "relates_to"), ("b", "relates_to")] + + @pytest.mark.parametrize("audience", ["team", "org", "public"]) + def test_active_only_honors_hidden_successor_without_exposing_it(self, tmp_path, audience): + d = str(tmp_path) + s = Store(Config(data_dir=d)) + s.add_node("Visible predecessor", node_id="prior", audience="public") + s.add_node("Visible active", node_id="active", audience="public") + s.add_node("Hidden successor", node_id="secret-successor", audience="private") + s.add_edge("secret-successor", "prior", edge_type="supersedes", bidirectional=False) + s.add_edge("active", "secret-successor", bidirectional=False) + s.close() + result = run("export", "--audience", audience, "--active-only", data_dir=d) + assert result.returncode == 0, result.stderr + rows = json.loads(result.stdout) + assert [row["id"] for row in rows] == ["active"] + assert rows[0]["edges"] == [] + assert "secret-successor" not in result.stdout + full = run("export", "--audience", audience, data_dir=d) + assert full.returncode == 0, full.stderr + assert [row["id"] for row in json.loads(full.stdout)] == ["active", "prior"] + def test_export_jsonl(self, tmp_path): """Export JSONL format.""" d = str(tmp_path)