diff --git a/README.md b/README.md index 720df7d..1307ecc 100644 --- a/README.md +++ b/README.md @@ -246,11 +246,34 @@ Commit multiple records as one batch intent: ```bash cfg commit --bulk-from batch.json -m "switch planner routing" +cfg commit --bulk-from batch.json --dry-run # preview every record's delta, write nothing ``` Bulk commit preflights the whole batch before writing. If any target has un-adopted drift, is missing, duplicates another target, or trips the secret -policy, cfgit applies none of the batch. +policy, cfgit applies none of the batch. `--dry-run` previews the per-record delta +(`would_commit` / `noop` / `changed_outside_cfgit`) for the whole batch without writing — +run it before any collection-scale replace. + +### Snapshot and restore a whole collection + +For "back it up, replace many rows, roll the whole thing back," `cfg export` writes a portable, +re-importable snapshot of the current live documents (each stamped with its cfgit head seq/oid), +and `cfg import --from` writes those documents back through the drift-guarded bulk-commit path — +so the restore is recorded in history like any other change: + +```bash +cfg export --out backup.json # snapshot every collection to a file +cfg export modelgarden_models:openai/gpt-4o-mini # or one record, to stdout +# ... do a risky bulk change ... +cfg import --from backup.json --dry-run # preview the restore +cfg import --from backup.json -m "restore from backup" +``` + +This is the file-based backup artifact (portable, out-of-tool, diffable). For an in-tool +rollback with no file, use `cfg tag` + `cfg restore --tag` / `cfg restore --as-of `, which +version the whole system inside cfgit. Both record the restore in history; export/import also +gives you a standalone file to keep or share. Draft changes on a branch before mutating runtime: @@ -324,12 +347,15 @@ cfg init cfg doctor [record] cfg doctor --status cfg import --all -m "initial import" +cfg export [record] [--out file.json] +cfg import --from [--dry-run] -m "message" cfg status [record] cfg diff [from] [to] cfg impact [from] [to] cfg commit --from -m "message" cfg commit --from --dry-run cfg commit --bulk-from -m "message" +cfg commit --bulk-from --dry-run cfg set field=value nested.field=value -m "message" cfg edit -m "message" cfg branch list @@ -464,6 +490,7 @@ Tools include: - `cfg_commit` - `cfg_bulk_commit` - `cfg_set` +- `cfg_export` - `cfg_branch_list` - `cfg_branch_create` - `cfg_branch_delete` diff --git a/docs/USAGE.md b/docs/USAGE.md index 08cb497..d2e1baa 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -109,7 +109,39 @@ cfg commit --bulk-from batch.json -m "switch planner routing" Bulk commit preflights every target before writing. If any record has un-adopted drift, is missing, duplicates another target, or trips the secret policy, no -record in the batch is applied. +record in the batch is applied. Add `--dry-run` to preview the per-record delta for the whole +batch (`would_commit` / `noop` / `changed_outside_cfgit`) without writing — run it before any +collection-scale replace: + +```bash +cfg commit --bulk-from batch.json --dry-run +``` + +## Snapshot and restore a collection + +`cfg export` writes a portable, re-importable snapshot of the current live documents, each +stamped with its cfgit head seq/oid. `cfg import --from` writes those documents back through the +drift-guarded bulk-commit path, so the restore is recorded in history like any other change. +Together they make "back it up, do the bulk change, roll it back" a fully in-cfgit workflow: + +```bash +cfg export --out backup.json # snapshot every collection to a file +cfg export modelgarden_models:openai/gpt-4o-mini # or one record, to stdout +# ... risky bulk change ... +cfg import --from backup.json --dry-run # preview the restore +cfg import --from backup.json -m "restore from backup" +``` + +For an in-tool rollback without a file, use `cfg tag` + `cfg restore --tag` (or +`cfg restore --as-of `) instead — see [Tags](#tags) and [Restore](#restore). Both record +the restore in history; export/import additionally gives you a standalone file to keep or share. + +`cfg export` warns (without stopping) when a snapshot is large by control-plane standards — a +strong hint that the config points at a data-plane collection (events, user content, jobs), +which cfgit is not designed to version. If you cancel an `import --from` partway, cfgit reports +`partial` with `cancelled: true` listing what committed and what is still pending; each record +is applied atomically, so rerunning the same command resumes (already-applied records are +no-ops). If the live record changed after cfgit last recorded it, commit returns `changed_outside_cfgit` and does not apply your document. Inspect the drift first: diff --git a/skills/cfgit/SKILL.md b/skills/cfgit/SKILL.md index 4e332f0..5bd6c78 100644 --- a/skills/cfgit/SKILL.md +++ b/skills/cfgit/SKILL.md @@ -52,7 +52,8 @@ Use cfgit as the safety layer around a live datastore. The app still reads and w (returns `would_commit` with the delta, or `changed_outside_cfgit`/`noop`; writes nothing). - Every outcome now carries a top-level `next` block (why + remedy + copy-paste `commands`); on a refusal, follow `next.commands` rather than guessing. - - For a coupled multi-record change, write a batch JSON file and run `cfg commit --bulk-from -m "" --json`. The batch file can be `[{"record":"collection:id","doc":{...}}]` or `{"collection:id": {...}}`. + - For a coupled multi-record change, write a batch JSON file and run `cfg commit --bulk-from -m "" --json`. The batch file can be `[{"record":"collection:id","doc":{...}}]` or `{"collection:id": {...}}`. Preview a collection-scale batch first with `cfg commit --bulk-from --dry-run --json` — it reports each record's `would_commit`/`noop`/`changed_outside_cfgit` and writes nothing. + - Before a risky bulk change, snapshot the current live docs with `cfg export --out backup.json --json` (a portable, re-importable artifact stamped with head seq/oid). To roll back, `cfg import --from backup.json -m "" --json` writes them back through the drift-guarded path (preview with `--dry-run`). For an in-tool rollback with no file, use `cfg tag` + `cfg restore --tag` instead. - For a draft branch, use `cfg --branch commit --from -m "" --json` or `cfg --branch commit --bulk-from -m "" --json`. This writes cfgit refs only; runtime is unchanged. - If commit returns `changed_outside_cfgit`, stop and inspect drift. - If bulk commit returns `blocked`, no record was applied; inspect `failed`. If it returns `partial`, some records were applied before a race/failure; inspect `results`, `failed`, and `pending` before continuing. @@ -61,6 +62,7 @@ Use cfgit as the safety layer around a live datastore. The app still reads and w 4. Reconcile drift. - `cfg diff =HEAD =live --json` - `cfg adopt -m "" --json` + - If `cfg doctor --status` reports a high drift ratio (many records `changed_outside_cfgit`), baseline the whole collection once with `cfg adopt --all -m "" --json` so later edits and restore work without a per-record adopt. 5. Restore. - Single record: `cfg restore -m "" --json` @@ -85,6 +87,7 @@ If the cfgit MCP server is available, prefer its tools over shelling out: - `cfg_impact` - `cfg_commit` - `cfg_bulk_commit` +- `cfg_export` - `cfg_branch_list` - `cfg_branch_create` - `cfg_branch_delete` diff --git a/src/cfg/cli/main.py b/src/cfg/cli/main.py index dd8846a..6cc0285 100644 --- a/src/cfg/cli/main.py +++ b/src/cfg/cli/main.py @@ -145,8 +145,15 @@ def _parser() -> argparse.ArgumentParser: p_import = sub.add_parser("import") p_import.add_argument("record", nargs="?") p_import.add_argument("--all", action="store_true") + p_import.add_argument("--from", dest="from_file", + help="restore documents from a `cfg export` file (writes via the drift-guarded bulk-commit path)") p_import.add_argument("-m", "--message", default="initial import") p_import.add_argument("--allow-secret", action="store_true") + p_import.add_argument("--dry-run", action="store_true", help="with --from: preview the bulk write without applying") + + p_export = sub.add_parser("export", help="dump live documents to a portable, re-importable JSON snapshot") + p_export.add_argument("record", nargs="?", help="one collection:id; omit to export every configured collection") + p_export.add_argument("--out", help="write the snapshot to this file instead of stdout") p_doctor = sub.add_parser("doctor") p_doctor.add_argument("record", nargs="?") @@ -191,7 +198,7 @@ def _parser() -> argparse.ArgumentParser: p_commit.add_argument( "--dry-run", action="store_true", - help="preview the field-level delta vs live and exit without writing (main-branch, single record)", + help="preview the delta and exit without writing; works with --from (single) and --bulk-from (per-record)", ) p_set = sub.add_parser("set", help="edit scalar fields in place; routes through commit (drift-guarded)") @@ -321,8 +328,18 @@ def _dispatch(engine: Engine, args: argparse.Namespace) -> tuple[Any, int]: raise ValueError(f"unknown PR command: {args.pr_cmd}") if args.cmd == "import": + if getattr(args, "from_file", None): + from cfg.interfaces.actions import import_from_file + + return import_from_file( + engine, + _load_json_any(args.from_file), + message=args.message if args.message != "initial import" else "import from export file", + allow_secret=args.allow_secret, + dry_run=args.dry_run, + ) if not args.all and not args.record: - raise ValueError("import needs --all or a record") + raise ValueError("import needs --all, a record, or --from ") result = engine.import_records( _parse_record(args.record) if args.record else None, message=args.message, @@ -330,6 +347,18 @@ def _dispatch(engine: Engine, args: argparse.Namespace) -> tuple[Any, int]: ) return result, EXIT_OK + if args.cmd == "export": + from cfg.interfaces.actions import export_records as _export_action + + report, _ = _export_action(engine, args.record if args.record else None) + warning = report.pop("warning", None) + if warning: + print(f"⚠ {warning}", file=sys.stderr) + if getattr(args, "out", None): + Path(args.out).write_text(json.dumps(_to_json(report), indent=2), encoding="utf-8") + return {"state": "exported", "out": args.out, "count": report["count"]}, EXIT_OK + return report, EXIT_OK + if args.cmd == "status": rows = engine.status(_parse_record(args.record) if args.record else None) code = EXIT_DIRTY if any(r.state == "changed_outside_cfgit" for r in rows) else EXIT_OK @@ -383,11 +412,15 @@ def _dispatch(engine: Engine, args: argparse.Namespace) -> tuple[Any, int]: if args.bulk_from_file: if args.record or args.from_file: raise ValueError("bulk commit uses --bulk-from without record or --from") - if not args.message: - raise ValueError("bulk commit needs -m/--message") - from cfg.interfaces.actions import bulk_commit_exit_code, parse_bulk_commit_items + from cfg.interfaces.actions import bulk_commit_exit_code, bulk_commit_preview, parse_bulk_commit_items items = parse_bulk_commit_items(_load_json_any(args.bulk_from_file)) + if args.dry_run: + if branch != engine.config.branches.default_branch: + raise ValueError("--dry-run is only supported on the main-branch commit path") + return bulk_commit_preview(engine, items), EXIT_OK + if not args.message: + raise ValueError("bulk commit needs -m/--message") if branch != engine.config.branches.default_branch: result = engine.branch_commit_many( branch, @@ -674,6 +707,8 @@ def _format_status_report(r: dict[str, Any]) -> str: ] for warning in r.get("warnings") or []: lines.append(f"⚠ {warning}") + for nudge in r.get("nudges") or []: + lines.append(f"→ {nudge}") return "\n".join(lines) diff --git a/src/cfg/core/engine.py b/src/cfg/core/engine.py index badcf71..04ee430 100644 --- a/src/cfg/core/engine.py +++ b/src/cfg/core/engine.py @@ -493,6 +493,32 @@ def import_records( results.append({"collection": item.collection, "record_id": item.record_id, "state": "imported", "seq": result.seq, "oid": result.oid}) return results + def export_records(self, ref: RecordRef | None = None) -> dict[str, Any]: + """Read-only snapshot of live documents into a portable, re-importable artifact. + + With a ref, exports that one record; without, exports every configured collection's live + records. Each item carries its live doc plus the current cfgit head seq/oid (or null if + untracked), so the snapshot records exactly which version it represents. Writes nothing. + """ + refs = [ref] if ref else self._all_refs(include_history=False) + items: list[dict[str, Any]] = [] + for item in sorted(refs, key=lambda r: (r.collection, r.record_id)): + live = self.adapter.get_record(item.collection, item.record_id) + if live is None: + continue + head = self.adapter.get_head(item.collection, item.record_id) + items.append( + { + "record": f"{item.collection}:{item.record_id}", + "collection": item.collection, + "record_id": item.record_id, + "head_seq": head.get("seq") if head else None, + "head_oid": head.get("oid") if head else None, + "doc": live, + } + ) + return {"version": 1, "kind": "cfgit-export", "count": len(items), "items": items} + def doctor(self, ref: RecordRef | None = None, *, large_field_bytes: int = 20000) -> dict[str, Any]: """Read-only preflight. Walks live records and reports what would trip an import/commit BEFORE anything is written: secret-deny matches (grouped by @@ -748,6 +774,18 @@ def commit_many( expected_live_oid=plan["expected_live"], make_head=True, ) + except KeyboardInterrupt: + # A cancel between per-record writes: each apply() is atomic, so already-committed + # records are durable. Return a clean partial report (what landed, what did not) + # instead of dying silently and leaving the operator guessing about DB state. + pending = [_plan_result(p) for p in plans[offset:]] + return { + "state": "partial", + "cancelled": True, + "results": results, + "failed": [], + "pending": pending, + } except Exception as exc: failed = [ { diff --git a/src/cfg/core/remedy.py b/src/cfg/core/remedy.py index a867df4..f1ffcef 100644 --- a/src/cfg/core/remedy.py +++ b/src/cfg/core/remedy.py @@ -102,7 +102,7 @@ def to_json(self) -> dict[str, Any]: commands=("cfg commit {record} --from -m ''",), ), "dry_run": Next( - why="Dry run: this is what a restore would change. Nothing was written.", + why="Dry run: this is what would change. Nothing was written.", remedy="Rerun without --dry-run to apply.", commands=(), ), diff --git a/src/cfg/interfaces/actions.py b/src/cfg/interfaces/actions.py index 0a2c01e..6625923 100644 --- a/src/cfg/interfaces/actions.py +++ b/src/cfg/interfaces/actions.py @@ -138,10 +138,23 @@ def status_report(engine: Engine) -> tuple[dict[str, Any], int]: "tracked": tracked, "drift": drift, "warnings": _env_warnings(engine), + "nudges": _drift_nudges(tracked, drift), } return report, EXIT_OK +def _drift_nudges(tracked: int, drift: int) -> list[str]: + """Turn a high drift ratio into an actionable suggestion. When a large share of tracked + records drifted, every edit needs a manual adopt first and restore has no clean baseline — + so nudge the operator to baseline the collection once with `cfg adopt --all`.""" + if drift >= 5 and tracked and drift / tracked >= 0.25: + return [ + f"{drift}/{tracked} tracked records drifted — run `cfg adopt --all` to baseline the " + "collection, then edits and restore work without a per-record adopt." + ] + return [] + + def _env_warnings(engine: Engine) -> list[str]: env = engine.config.envs[engine.env] warnings: list[str] = [] @@ -183,6 +196,78 @@ def import_records( return result, EXIT_OK +EXPORT_WARN_RECORDS = 2000 +EXPORT_WARN_BYTES = 50 * 1024 * 1024 # 50 MB serialized + + +def export_records(engine: Engine, record: str | None = None) -> tuple[dict[str, Any], int]: + report = engine.export_records(parse_record(record) if record else None) + warning = _export_size_warning(report) + if warning: + report["warning"] = warning + return report, EXIT_OK + + +def _export_size_warning(report: dict[str, Any]) -> str | None: + """Soft warning (never a hard cap) when a snapshot is large by control-plane standards — + it usually means the config points at a data-plane collection (events, user content, jobs), + which cfgit is not designed to version. Triggers on record count OR serialized size.""" + count = report.get("count", 0) + try: + size = len(json.dumps(to_json(report))) + except (TypeError, ValueError): + size = 0 + reasons = [] + if count >= EXPORT_WARN_RECORDS: + reasons.append(f"{count} records") + if size >= EXPORT_WARN_BYTES: + reasons.append(f"~{size // (1024 * 1024)}MB") + if not reasons: + return None + return ( + f"large export ({', '.join(reasons)}). cfgit is built for control-plane collections " + "(hundreds–low thousands of hand-curated records). If this is a data-plane collection " + "(events, user content, jobs), it is the wrong fit — use a backup or warehouse instead." + ) + + +def import_from_file( + engine: Engine, + export_obj: Any, + *, + message: str, + allow_secret: bool = False, + dry_run: bool = False, +) -> tuple[dict[str, Any], int]: + """Restore documents from a `cfg export` artifact by writing them back through the + drift-guarded bulk-commit path (preflights the whole batch; applies none on any drift).""" + items = _export_items_to_commit(export_obj) + if dry_run: + return bulk_commit_preview(engine, items), EXIT_OK + result = engine.commit_many(items, message=message, allow_secret=allow_secret) + return result, bulk_commit_exit_code(result) + + +def _export_items_to_commit(export_obj: Any) -> list[tuple[RecordRef, dict[str, Any]]]: + if isinstance(export_obj, dict) and export_obj.get("kind") == "cfgit-export": + rows = export_obj.get("items") or [] + elif isinstance(export_obj, list): + rows = export_obj + else: + raise ValueError("import --from expects a cfg export artifact or a list of {record, doc}") + out: list[tuple[RecordRef, dict[str, Any]]] = [] + for i, row in enumerate(rows, start=1): + if not isinstance(row, dict) or "doc" not in row: + raise ValueError(f"export item {i} needs a doc") + rec = row.get("record") or ( + f"{row.get('collection')}:{row.get('record_id')}" if row.get("collection") else None + ) + if not rec: + raise ValueError(f"export item {i} needs record or collection+record_id") + out.append((parse_record(rec), row["doc"])) + return out + + def diff(engine: Engine, record: str, a: str = "=HEAD", b: str = "=live") -> tuple[dict[str, Any], int]: changes = engine.diff(parse_record(record), a, b) return {"changes": changes, "text": format_diff(changes)}, EXIT_OK @@ -287,8 +372,13 @@ def bulk_commit( message: str, allow_secret: bool = False, branch: str | None = None, + dry_run: bool = False, ) -> tuple[dict[str, Any], int]: parsed = _bulk_commit_items(items) + if dry_run: + if branch and branch != engine.config.branches.default_branch: + raise ValueError("--dry-run is only supported on the main-branch commit path") + return bulk_commit_preview(engine, parsed), EXIT_OK if branch and branch != engine.config.branches.default_branch: result = engine.branch_commit_many(branch, parsed, message=message, allow_secret=allow_secret) else: @@ -296,6 +386,35 @@ def bulk_commit( return result, bulk_commit_exit_code(result) +def bulk_commit_preview( + engine: Engine, items: list[tuple[RecordRef, dict[str, Any]]] +) -> dict[str, Any]: + """Dry-run a bulk commit: preview each record without writing. Aggregates per-record + would_commit / noop / changed_outside_cfgit so the operator sees the whole blast radius of a + collection-scale replace before applying it. Mirrors commit_many's per-record semantics.""" + results: list[dict[str, Any]] = [] + would_change = drift = noop = 0 + for ref, doc in items: + try: + preview = engine.commit_preview(ref, doc) + except NoSuchConfig: + results.append({"record": f"{ref.collection}:{ref.record_id}", "state": "missing"}) + continue + state = preview.get("state") + if state == "would_commit": + would_change += 1 + elif state == "changed_outside_cfgit": + drift += 1 + elif state == "noop": + noop += 1 + results.append({"record": f"{ref.collection}:{ref.record_id}", **preview}) + return { + "state": "dry_run", + "summary": {"total": len(items), "would_commit": would_change, "drift": drift, "noop": noop}, + "results": results, + } + + def log(engine: Engine, record: str, *, limit: int | None = 20) -> tuple[list[dict[str, Any]], int]: return engine.log(parse_record(record), limit=limit), EXIT_OK @@ -461,6 +580,14 @@ def run_named_action(name: str, engine: Engine, payload: dict[str, Any] | None = large_field_bytes=int(payload.get("large_field_bytes") or 20000), ) if name == "import": + if payload.get("export") is not None or payload.get("from") is not None: + return import_from_file( + engine, + payload.get("export") if payload.get("export") is not None else payload.get("from"), + message=str(payload.get("message") or "import from export file"), + allow_secret=bool(payload.get("allow_secret")), + dry_run=bool(payload.get("dry_run")), + ) return import_records( engine, _blank_to_none(payload.get("record")), @@ -468,6 +595,8 @@ def run_named_action(name: str, engine: Engine, payload: dict[str, Any] | None = message=str(payload.get("message") or "initial import"), allow_secret=bool(payload.get("allow_secret")), ) + if name == "export": + return export_records(engine, _blank_to_none(payload.get("record"))) if name == "diff": return diff( engine, @@ -483,6 +612,7 @@ def run_named_action(name: str, engine: Engine, payload: dict[str, Any] | None = message=str(payload.get("message") or "commit"), allow_secret=bool(payload.get("allow_secret")), branch=_blank_to_none(payload.get("branch")), + dry_run=bool(payload.get("dry_run")), ) return commit( engine, @@ -509,6 +639,7 @@ def run_named_action(name: str, engine: Engine, payload: dict[str, Any] | None = message=str(payload.get("message") or "bulk commit"), allow_secret=bool(payload.get("allow_secret")), branch=_blank_to_none(payload.get("branch")), + dry_run=bool(payload.get("dry_run")), ) if name == "log": return log(engine, _required(payload, "record"), limit=int(payload.get("limit") or 20)) diff --git a/src/cfg/mcp/server.py b/src/cfg/mcp/server.py index c12abc5..d9e62aa 100644 --- a/src/cfg/mcp/server.py +++ b/src/cfg/mcp/server.py @@ -67,17 +67,24 @@ def cfg_doctor( def cfg_import( record: str | None = None, all_records: bool = False, + from_export: dict[str, Any] | list[dict[str, Any]] | None = None, + dry_run: bool = False, message: str = "initial import", allow_secret: bool = False, config_file: str | None = None, env: str = "dev", author: str | None = None, ) -> dict[str, Any]: + """Start tracking live records (record or all_records), or restore documents from a + `cfg_export` artifact via from_export (writes them back through the drift-guarded bulk-commit + path; set dry_run=true to preview).""" return _call( "import", { "record": record, "all_records": all_records, + "export": from_export, + "dry_run": dry_run, "message": message, "allow_secret": allow_secret, }, @@ -87,6 +94,19 @@ def cfg_import( ) +@mcp.tool() +def cfg_export( + record: str | None = None, + config_file: str | None = None, + env: str = "dev", + author: str | None = None, +) -> dict[str, Any]: + """Read-only snapshot of live documents into a portable, re-importable artifact. Pass a + record for one, or omit it to export every configured collection. Each item is stamped with + its cfgit head seq/oid. Feed the result back to cfg_import(from_export=...) to restore.""" + return _call("export", {"record": record}, config_file=config_file, env=env, author=author) + + @mcp.tool() def cfg_diff( record: str, @@ -168,9 +188,10 @@ def cfg_commit( @mcp.tool() def cfg_bulk_commit( items: list[dict[str, Any]] | dict[str, Any] | str, - message: str, + message: str = "", allow_secret: bool = False, branch: str | None = None, + dry_run: bool = False, config_file: str | None = None, env: str = "dev", author: str | None = None, @@ -180,10 +201,13 @@ def cfg_bulk_commit( `items` may be either: [{"record":"collection:id","doc":{...}}, ...] {"collection:id": {...}, ...}, or a JSON string in either shape. + + Set dry_run=true to preview every record's delta (would_commit / noop / changed_outside_cfgit) + without writing — recommended before a collection-scale replace. """ return _call( "bulk_commit", - {"items": items, "message": message, "allow_secret": allow_secret, "branch": branch}, + {"items": items, "message": message, "allow_secret": allow_secret, "branch": branch, "dry_run": dry_run}, config_file=config_file, env=env, author=author, diff --git a/tests/test_export_import.py b/tests/test_export_import.py new file mode 100644 index 0000000..5ffc635 --- /dev/null +++ b/tests/test_export_import.py @@ -0,0 +1,214 @@ +# Copyright 2026 Mohammad Ausaf. Licensed under the Apache License, Version 2.0. +"""cfg export snapshot + import --from round-trip + --bulk-from --dry-run + drift nudge.""" +from __future__ import annotations + +import pytest + +from cfg.core.engine import RecordRef +from cfg.interfaces import actions + +from test_engine_safety import _engine + + +# --- export --------------------------------------------------------------------------------- + + +def test_export_dumps_live_docs_stamped_with_head(): + engine, adapter = _engine(records={ + ("demo", "a"): {"id": "a", "value": 1}, + ("demo", "b"): {"id": "b", "value": 2}, + }) + engine.commit(RecordRef("demo", "a"), {"id": "a", "value": 1}, message="seed a") + + snap = engine.export_records() + assert snap["kind"] == "cfgit-export" + assert snap["count"] == 2 + by_rec = {i["record"]: i for i in snap["items"]} + # tracked record carries its head seq/oid; untracked record has null head + assert by_rec["demo:a"]["head_seq"] is not None + assert by_rec["demo:a"]["doc"] == {"id": "a", "value": 1} + assert by_rec["demo:b"]["head_seq"] is None + + +def test_export_single_record(): + engine, _ = _engine(records={("demo", "a"): {"id": "a", "value": 1}, ("demo", "b"): {"id": "b"}}) + snap = engine.export_records(RecordRef("demo", "a")) + assert snap["count"] == 1 + assert snap["items"][0]["record"] == "demo:a" + + +def test_export_writes_nothing(): + engine, adapter = _engine(records={("demo", "a"): {"id": "a", "value": 1}}) + before = len(adapter.history) + engine.export_records() + assert len(adapter.history) == before + + +# --- import --from (round-trip via drift-guarded commit) ------------------------------------ + + +def test_export_then_import_round_trip_restores_docs(): + engine, adapter = _engine(records={("demo", "a"): {"id": "a", "value": 1}}) + engine.commit(RecordRef("demo", "a"), {"id": "a", "value": 1}, message="seed") + snap = engine.export_records() # capture value=1 + + # mutate live away from the snapshot, adopt so there's no drift blocking the restore + adapter.records[("demo", "a")] = {"id": "a", "value": 999} + engine.adopt(RecordRef("demo", "a"), message="adopt drift") + + result, code = actions.import_from_file(engine, snap, message="restore from backup") + assert result["state"] == "committed" + # live is back to the snapshot value + assert adapter.records[("demo", "a")]["value"] == 1 + + +def test_import_from_accepts_bare_list_shape(): + engine, adapter = _engine(records={("demo", "a"): {"id": "a", "value": 1}}) + engine.commit(RecordRef("demo", "a"), {"id": "a", "value": 1}, message="seed") + items = [{"record": "demo:a", "doc": {"id": "a", "value": 7}}] + result, _ = actions.import_from_file(engine, items, message="restore list") + assert result["state"] == "committed" + assert adapter.records[("demo", "a")]["value"] == 7 + + +def test_import_from_dry_run_writes_nothing(): + engine, adapter = _engine(records={("demo", "a"): {"id": "a", "value": 1}}) + engine.commit(RecordRef("demo", "a"), {"id": "a", "value": 1}, message="seed") + snap = engine.export_records() + adapter.records[("demo", "a")] = {"id": "a", "value": 5} + engine.adopt(RecordRef("demo", "a"), message="adopt") + before = len(adapter.history) + + result, _ = actions.import_from_file(engine, snap, message="preview", dry_run=True) + assert result["state"] == "dry_run" + assert len(adapter.history) == before + + +def test_import_from_rejects_garbage(): + engine, _ = _engine(records={("demo", "a"): {"id": "a"}}) + with pytest.raises(ValueError): + actions.import_from_file(engine, {"not": "an export"}, message="x") + + +# --- bulk commit --dry-run ------------------------------------------------------------------ + + +def test_bulk_commit_preview_reports_per_record_and_writes_nothing(): + engine, adapter = _engine(records={ + ("demo", "a"): {"id": "a", "value": 1}, + ("demo", "b"): {"id": "b", "value": 1}, + }) + engine.commit(RecordRef("demo", "a"), {"id": "a", "value": 1}, message="seed a") + engine.commit(RecordRef("demo", "b"), {"id": "b", "value": 1}, message="seed b") + before = len(adapter.history) + + items = [ + (RecordRef("demo", "a"), {"id": "a", "value": 2}), # would_commit + (RecordRef("demo", "b"), {"id": "b", "value": 1}), # noop + ] + preview = actions.bulk_commit_preview(engine, items) + assert preview["state"] == "dry_run" + assert preview["summary"] == {"total": 2, "would_commit": 1, "drift": 0, "noop": 1} + assert len(adapter.history) == before # nothing written + + +def test_bulk_commit_dry_run_via_action_writes_nothing(): + engine, adapter = _engine(records={("demo", "a"): {"id": "a", "value": 1}}) + engine.commit(RecordRef("demo", "a"), {"id": "a", "value": 1}, message="seed") + before = len(adapter.history) + result, code = actions.bulk_commit( + engine, [{"record": "demo:a", "doc": {"id": "a", "value": 9}}], message="x", dry_run=True + ) + assert result["state"] == "dry_run" + assert len(adapter.history) == before + + +def test_bulk_commit_preview_flags_drift_without_writing(): + engine, adapter = _engine(records={("demo", "a"): {"id": "a", "value": 1}}) + engine.commit(RecordRef("demo", "a"), {"id": "a", "value": 1}, message="seed") + adapter.records[("demo", "a")] = {"id": "a", "value": 42} # out-of-band drift + before = len(adapter.history) + preview = actions.bulk_commit_preview(engine, [(RecordRef("demo", "a"), {"id": "a", "value": 2})]) + assert preview["summary"]["drift"] == 1 + assert len(adapter.history) == before + + +# --- doctor adopt-all nudge ----------------------------------------------------------------- + + +def test_drift_nudge_fires_only_above_threshold(): + # below ratio / below count → no nudge + assert actions._drift_nudges(tracked=100, drift=4) == [] + assert actions._drift_nudges(tracked=100, drift=10) == [] # 10% < 25% + # high drift → nudge mentioning adopt --all + nudge = actions._drift_nudges(tracked=288, drift=201) + assert nudge and "adopt --all" in nudge[0] + assert "201/288" in nudge[0] + + +# --- export size warning -------------------------------------------------------------------- + + +def test_export_size_warning_absent_for_small_collection(): + engine, _ = _engine(records={("demo", "a"): {"id": "a", "value": 1}}) + report, _ = actions.export_records(engine) + assert "warning" not in report + + +def test_export_size_warning_fires_on_high_record_count(monkeypatch): + monkeypatch.setattr(actions, "EXPORT_WARN_RECORDS", 3) + engine, _ = _engine(records={ + ("demo", "a"): {"id": "a"}, ("demo", "b"): {"id": "b"}, + ("demo", "c"): {"id": "c"}, ("demo", "d"): {"id": "d"}, + }) + report, _ = actions.export_records(engine) + assert "warning" in report + assert "control-plane" in report["warning"] + assert "4 records" in report["warning"] + + +def test_export_size_warning_fires_on_byte_size(monkeypatch): + monkeypatch.setattr(actions, "EXPORT_WARN_BYTES", 500) + big = {"id": "a", "blob": "x" * 2000} + engine, _ = _engine(records={("demo", "a"): big}) + report, _ = actions.export_records(engine) + assert "warning" in report and "MB" in report["warning"] + + +# --- cancellation safety on the import/bulk write path -------------------------------------- + + +def test_commit_many_cancel_returns_clean_partial(monkeypatch): + engine, adapter = _engine(records={ + ("demo", "a"): {"id": "a", "value": 0}, + ("demo", "b"): {"id": "b", "value": 0}, + ("demo", "c"): {"id": "c", "value": 0}, + }) + for r in ("a", "b", "c"): + engine.commit(RecordRef("demo", r), {"id": r, "value": 0}, message="seed") + + # simulate a Ctrl-C during the 2nd record's write; the 1st is already durable + real_apply = adapter.apply + calls = {"n": 0} + + def flaky_apply(*a, **k): + calls["n"] += 1 + if calls["n"] == 2: + raise KeyboardInterrupt + return real_apply(*a, **k) + + monkeypatch.setattr(adapter, "apply", flaky_apply) + + result = engine.commit_many( + [ + (RecordRef("demo", "a"), {"id": "a", "value": 1}), + (RecordRef("demo", "b"), {"id": "b", "value": 1}), + (RecordRef("demo", "c"), {"id": "c", "value": 1}), + ], + message="bulk restore", + ) + assert result["state"] == "partial" + assert result["cancelled"] is True + assert len(result["results"]) == 1 # 'a' committed before the cancel + assert result["results"][0]["record_id"] == "a" + assert len(result["pending"]) == 2 # 'b' and 'c' not applied — rerun resumes