demo: stop claiming backup is the only other copy

The narration said the changed file's "only other copy is in last night's backup". That is false, and it gives away the argument the demo exists to make: Zerto has a copy from seconds before the change. That is the whole point. d5 now draws the contrast instead of conceding it. Backup has last night, hours old. Zerto has seconds before the change. d2 had the same problem in a quieter form, asserting there was "no other copy of that file anywhere" while Zerto was already protecting the machine. Cut, since d5 carries the comparison. Also fixes the pronunciation map that made the voice say "vem". The replace map takes a phrase and a pronunciation, so {"VM": "vee em"} is spoken as one run-together word. Expanded to "virtual machine". Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_016yVfC5nvZowoLFnEGWhLGn
demo: commit the recording harness
2026-09-21 21:18:18 -04:00 · 2026-09-21 21:17:24 -04:00
14 changed files with 21 additions and 575 deletions
@@ -64,22 +64,7 @@ Git never had the file. RPO is the journal, not last night's backup.
 vSphere ZVM 10.x and ZCA on AWS/Azure, same REST paths. HVM is out (separate swagger). Failover Live is not a tool.
-Tagged checkpoints **do** work when the protected site is Azure or AWS. The 9.0 API
+Tagged checkpoints cannot be inserted when the **protected** site is Azure or AWS (Zerto API). Point this server at the vSphere protected ZVM.
 reference says they cannot be inserted; that is wrong on 10.9.10, where both were
 accepted and the task reached `Completed`.
 What differs is latency and granularity, both set by the **protected** site:
 | protected at | journal gap | tag visible after |
 |---|---|---|
 | vSphere | 5s | ~4s |
 | Azure | 60s | ~34s |
 | AWS | 630s | ~128s |
 The tagged checkpoint is also stamped about 30s *after* the insert request, so on a
 cloud-protected VPG a prompt mutation can land *inside* the checkpoint meant to
 precede it. Recover from the newest checkpoint that already existed when the guard
 ran, not from the tag.
 ## Not this product
@@ -16,17 +16,6 @@ CYAN, GREEN, RED, YEL, MAG = "\033[36m", "\033[32m", "\033[31m", "\033[33m", "\0
 SPEED = float(os.environ.get("DEMO_SPEED", "1.0"))
 def recovered_bytes(rec: dict) -> bytes:
    """zerto_recover_file returns the content now, not a path on the MCP host.
    The server may be on another machine, so a path there is of no use here.
    """
    import base64
    if rec.get("encoding") == "base64":
        return base64.b64decode(rec["content"])
    return rec["content"].encode()
 def w(s=""):
    sys.stdout.write(s + "\n"); sys.stdout.flush()
@@ -223,8 +212,8 @@ async def main():
            if not rec.get("ok"):
                w(f"{RED}  FLR failed{RST}"); return 1
            beat(1.5)
-            w(f"{YEL}  recovered {rec['bytes']} bytes, sha256 {rec['sha256'][:16]}...{RST}")
+            w(f"{YEL}  $ cat {rec['path']}{RST}")
-            body = recovered_bytes(rec).decode()
+            body = Path(rec["path"]).read_text()
            for line in body.splitlines():
                w(f"{GREEN}{BOLD}  | {line}{RST}")
            beat(2.5)
@@ -233,13 +222,11 @@ async def main():
            step("Copy it back to the guest")
            note("This part is scp, not Zerto. Zerto got the bytes back.")
            target = f"{CFG['user']}@{CFG['host']}"
            local_copy = SP / rec["name"]
            local_copy.write_bytes(recovered_bytes(rec))
            scp = (["sshpass", "-p", CFG["password"]] if CFG.get("password") else []) + \
                  ["scp", "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null",
                   "-o", "LogLevel=ERROR"] + (["-i", CFG["key"]] if CFG.get("key") else []) + \
-                  [str(local_copy), f"{target}:{CFG['guest_path']}"]
+                  [rec["path"], f"{target}:{CFG['guest_path']}"]
-            w(f"{YEL}  $ scp {rec['name']} {target}:{CFG['guest_path']}{RST}")
+            w(f"{YEL}  $ scp {Path(rec['path']).name} {target}:{CFG['guest_path']}{RST}")
            subprocess.run(scp, capture_output=True, text=True, timeout=60)
            guest(f"cat {CFG['guest_path']}")
            beat(1.5)
@@ -22,17 +22,6 @@ T0 = time.time()
 STEP = [0]
 def recovered_bytes(rec: dict) -> bytes:
    """zerto_recover_file returns the content now, not a path on the MCP host.
    The server may be on another machine, so a path there is of no use here.
    """
    import base64
    if rec.get("encoding") == "base64":
        return base64.b64decode(rec["content"])
    return rec["content"].encode()
 def w(s=""):
    sys.stdout.write(s + "\n"); sys.stdout.flush()
@@ -166,7 +155,7 @@ async def main():
            show(rec, keep=["ok", "bytes", "flr_path", "unmount"])
            if not rec.get("ok"):
                w(f"{RED}  FLR failed{RST}"); return 1
-            body = recovered_bytes(rec).decode()
+            body = Path(rec["path"]).read_text()
            w(f"{YEL}  recovered file:{RST}")
            for line in body.splitlines():
                w(f"{GREEN}{BOLD}  | {line}{RST}")
@@ -177,7 +166,7 @@ async def main():
            # endings and appends one, which makes the restored file differ
            # from what Zerto handed back. WriteAllBytes does not.
            import base64
-            blob = recovered_bytes(rec)
+            blob = Path(rec["path"]).read_bytes()
            b64 = base64.b64encode(blob).decode()
            guest_ps(
                f"[IO.File]::WriteAllBytes('{CFG['guest_path']}',"
@@ -1,93 +0,0 @@
 # PreToolUse guard hook
 Makes the rewind guard enforceable instead of advisory.
 The MCP server cannot see another server's tool calls, so `zerto_check_tool` and
 `zerto_guard_before_mutate` only work if the model chooses to call them. A model
 that skips the step is not stopped by anything. A `PreToolUse` hook runs in the
 host, where the tool call genuinely pauses, so a failed checkpoint stops the
 change.
 ## Decisions
 | situation | decision | effect |
 |---|---|---|
 | tool in `read_only_tools` | none | runs, no checkpoint |
 | mutating, checkpoint confirmed | none, plus `additionalContext` | runs, and the model is told which checkpoint to recover from |
 | mutating, checkpoint failed | `deny` | **the call never happens** |
 | mutating, VM unknown to Zerto | `prompt` | human decides; nothing to rewind to |
 | mutating, no VM in the arguments | `prompt` | catalog's `vm_arg` did not match |
 | unlisted tool | `prompt` | nobody said it was read-only |
 `ZERTO_HOOK_UNKNOWN` switches the unlisted case to `allow` or `deny`.
 ## Install
 ```bash
 cp hooks/settings.example.json /tmp/x        # then merge the hooks block into
                                             # .claude/settings.json
 export ZERTO_REWIND_CONFIG=/abs/path/config.json
 ```
 Both paths in the command must be absolute, and the interpreter must be the
 venv that has this package installed.
 ## Timeouts, which matter here
 The hook is synchronous: the host waits. That is the point, because the
 checkpoint has to exist before the change does.
 Budget for the slow path, not the fast one. A successful tag took about 4s
 against a healthy vSphere-protected VPG, but a **refusal took 63s**, because
 `wait_for_tag` spends 45s before giving up.
 So:
 - `timeout` in settings.json: 180 (seconds)
 - `ZERTO_HOOK_GUARD_TIMEOUT`: 150 (seconds), kept under it
 If the host's timeout fires first it cancels the hook and **discards its
 output**, and the tool call carries on through the normal permission flow. A
 timeout is therefore a silent failure of the guard, which is why the hook's own
 budget is the smaller of the two: it would rather deny than be cancelled.
 ## Verified behaviour
 Against a live ZVM 10.9.10, all six rows of the table above. Two worth naming:
 - `jp-ubuntu` (healthy, local VPG) tagged checkpoint 7180 and allowed the call,
  passing the tag back through `additionalContext`.
 - `win2019-1` (VPG `CMH-AWS-1`, protected site AWS) was **denied**. The deny path
  works, but see the known issue below: that particular denial was wrong.
 Both decisions held with `permission_mode: bypassPermissions`. A hook still
 blocks when the user has turned permissions off, which is when an agent is most
 likely to be running unattended.
 ## Log
 `~/.zerto-guard-hook.log`, or `ZERTO_HOOK_LOG`. One line per decision.
 ## Known issue: false denials on cloud-protected VPGs
 The `win2019-1` denial above was a **false negative**, and it is worth
 understanding before relying on this hook in an estate with cloud-protected
 workloads.
 `wait_for_tag` gives up after a hardcoded 45s. That is generous for a
 vSphere-protected VPG, which checkpoints every 5s and surfaces a tag in about
 4s. It is far too short elsewhere: a tag takes ~34s to appear on an
 Azure-protected VPG and ~128s on an AWS-protected one, because journal cadence
 is set by the protected site (5s vSphere, 60s Azure, 630s AWS).
 So the hook denied the change, and the checkpoint landed anyway. It is in the
 journal as `cp 56`, timestamped a minute after the hook reported failure. The
 guard told the agent there was no rewind point while Zerto was in the middle of
 creating one.
 That failure mode is worse than the one the hook guards against, because it is
 silent and looks correct: legitimate work is refused on every cloud-protected VM
 while the log reads like the guard is doing its job.
 Until `wait_for_tag` becomes cadence-aware, scope this hook's matcher to
 vSphere-protected workloads.
@@ -1,17 +0,0 @@
 {
  "_comment": "Copy the hooks block into .claude/settings.json (project) or ~/.claude/settings.json (all projects). Use absolute paths.",
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "mcp__.*",
        "hooks": [
          {
            "type": "command",
            "command": "/home/you/zerto-ai-rewind/.venv/bin/python /home/you/zerto-ai-rewind/hooks/zerto_guard_hook.py",
            "timeout": 180
          }
        ]
      }
    ]
  }
 }
@@ -1,247 +0,0 @@
 #!/usr/bin/env python3
 """Claude Code PreToolUse hook: capture before execute, enforced by the host.
 The MCP server cannot see another server's tool calls, so the guard it exposes
 is advice the model may skip. This hook sits in the host, where the call really
 does pause, so a failed checkpoint stops the change instead of merely
 suggesting it should.
 Decisions:
  read-only tool            allow, no checkpoint
  mutating, guard ok        allow, and tell the model which checkpoint to use
  mutating, guard failed    DENY. No checkpoint means no rewind, so no change.
  mutating, VM unprotected  prompt. Zerto cannot rewind it; a human decides.
  unknown tool              prompt. Nobody said it was read-only.
 Exit 0 always, with the decision in stdout JSON. Exit 2 would block
 unconditionally and ignore the JSON, which loses the reason text.
 Install (project .claude/settings.json). Give it room: the guard inserts a
 tagged checkpoint and waits for the Zerto task to reach Completed.
  {"hooks": {"PreToolUse": [{
     "matcher": "mcp__.*",
     "hooks": [{"type": "command",
                "command": "/path/to/.venv/bin/python /path/to/hooks/zerto_guard_hook.py",
                "timeout": 180}]}]}}
 """
 from __future__ import annotations
 import asyncio
 import json
 import os
 import re
 import sys
 import time
 LOG = os.environ.get("ZERTO_HOOK_LOG", os.path.expanduser("~/.zerto-guard-hook.log"))
 # Seconds the hook will wait for the checkpoint. Must stay under the hook
 # timeout configured in settings.json, or the host cancels us and the tool
 # call proceeds unguarded through the normal permission flow.
 GUARD_TIMEOUT_S = float(os.environ.get("ZERTO_HOOK_GUARD_TIMEOUT", "150"))
 UNKNOWN_DECISION = os.environ.get("ZERTO_HOOK_UNKNOWN", "prompt")  # prompt | allow | deny
 def log(msg: str) -> None:
    try:
        with open(LOG, "a") as fh:
            fh.write(f"{time.strftime('%Y-%m-%dT%H:%M:%S')} {msg}\n")
    except Exception:
        pass
 def emit(decision: str | None, reason: str = "", context: str = "") -> None:
    """Write the PreToolUse decision and leave. Never raises."""
    if decision is None and not context:
        sys.exit(0)  # no opinion; normal permission flow applies
    out: dict = {"hookEventName": "PreToolUse"}
    if decision:
        out["permissionDecision"] = decision
        out["permissionDecisionReason"] = reason
    if context:
        out["additionalContext"] = context
    print(json.dumps({"hookSpecificOutput": out}))
    sys.exit(0)
 def split_mcp_name(tool_name: str) -> tuple[str, str] | None:
    """mcp__<server>__<tool> -> (server, tool). Server names may contain _."""
    m = re.match(r"^mcp__(.+?)__(.+)$", tool_name or "")
    return (m.group(1), m.group(2)) if m else None
 def resolve_vm(tool_input: dict, vm_arg: str) -> str:
    """Pull the VM identifier out of the pending call's arguments."""
    if vm_arg and tool_input.get(vm_arg):
        return str(tool_input[vm_arg])
    # vm_arg is the catalog's answer, but fall back to the usual suspects so a
    # slightly-wrong catalog entry degrades to a prompt rather than a crash.
    for key in (
        "host",
        "hostname",
        "computer_name",
        "computerName",
        "vm",
        "vm_name",
        "target",
        "limit",
        "address",
        "server",
    ):
        if tool_input.get(key):
            return str(tool_input[key])
    return ""
 async def run_guard(query: str, change_id: str, action: str) -> dict:
    from zerto_rewind_mcp.checkpoints import make_tag, tag_vpgs
    from zerto_rewind_mcp.client import ZertoClient, ZertoError
    from zerto_rewind_mcp.config import load_config
    from zerto_rewind_mcp.protection import find_from_rows
    settings = load_config()
    client = ZertoClient(
        base_url=str(settings["zerto_url"]),
        username=str(settings["username"]),
        password=str(settings.get("password") or ""),
        client_id=str(settings.get("client_id") or "zerto-client"),
        verify_tls=bool(settings.get("verify_tls")),
    )
    try:
        # Two different failures hide here. "Zerto has never heard of this VM"
        # is not the same as "Zerto refused to tag it", and they get different
        # answers, so a lookup miss must not surface as a refusal. Looking up a
        # plain hostname by vmIdentifier returns HTTP 400, which is a miss.
        rows = []
        for kwargs in ({"vm_name": query}, {"vm_identifier": query}):
            try:
                rows = await client.get_vms(**kwargs)
            except ZertoError:
                rows = []
            if rows:
                break
        if not rows:
            return {
                "ok": False,
                "protected": False,
                "message": f"Zerto has no VM matching {query!r}.",
            }
        result = find_from_rows(query, rows)
        if result.outcome != "ok" or not result.taggable_vpgs:
            return {"ok": False, "protected": False, "message": result.message}
        tag = make_tag(
            "claude-code",
            change_id,
            action=action,
            vm_name=result.vm.vm_name if result.vm else None,
        )
        out = await tag_vpgs(client, result, tag)
        out["protected"] = True
        return out
    except ZertoError as exc:
        # Reached Zerto and it refused, or never reached it at all. Either way
        # there is no checkpoint, so this is not a pass.
        return {"ok": False, "protected": True, "message": str(exc)}
    finally:
        try:
            await client.aclose()
        except Exception:
            pass
 def main() -> None:
    try:
        payload = json.loads(sys.stdin.read() or "{}")
    except Exception as exc:
        log(f"unparseable stdin: {exc}")
        emit(None)
    tool_name = payload.get("tool_name") or ""
    tool_input = payload.get("tool_input") or {}
    parts = split_mcp_name(tool_name)
    if not parts:
        emit(None)  # not an MCP tool; this hook has nothing to say
    server, tool = parts
    try:
        from zerto_rewind_mcp.catalog import MUTATING, READ_ONLY
        from zerto_rewind_mcp.config import load_catalog, load_config
        catalog = load_catalog(load_config())
        verdict = catalog.classify(server, tool)
        entry = catalog.get(server, tool)
    except Exception as exc:
        log(f"catalog unavailable ({exc}); staying out of the way")
        emit(None)
    if verdict == READ_ONLY:
        log(f"{tool_name}: read-only, allowed")
        emit(None)
    if verdict != MUTATING:
        log(f"{tool_name}: unknown -> {UNKNOWN_DECISION}")
        emit(
            UNKNOWN_DECISION,
            f"{server}/{tool} is not a known read-only tool and is not in the "
            "Zerto mutating catalog, so no checkpoint was taken. If it changes a "
            "protected VM, that change cannot be rewound. Allow it?",
        )
    vm = resolve_vm(tool_input, entry.vm_arg if entry else "")
    if not vm:
        log(f"{tool_name}: mutating but no VM in args {sorted(tool_input)}")
        emit(
            "prompt",
            f"{server}/{tool} is a guest-mutating tool, but no VM could be read "
            f"from its arguments (expected {entry.vm_arg if entry else '?'}). "
            "No checkpoint was taken. Allow it?",
        )
    change_id = f"{payload.get('session_id', 'session')[:8]}-{payload.get('tool_use_id', '')[-8:]}"
    action = f"{server}/{tool} on {vm}"
    try:
        guard = asyncio.run(asyncio.wait_for(run_guard(vm, change_id, action), GUARD_TIMEOUT_S))
    except TimeoutError:
        log(f"{tool_name}: guard timed out after {GUARD_TIMEOUT_S}s -> deny")
        emit(
            "deny",
            f"Zerto did not confirm a tagged checkpoint for {vm} within "
            f"{GUARD_TIMEOUT_S:.0f}s, so the change cannot be rewound. Refusing.",
        )
    except Exception as exc:
        log(f"{tool_name}: guard raised {type(exc).__name__}: {exc} -> deny")
        emit("deny", f"The Zerto guard failed for {vm}: {exc}. No checkpoint, so refusing.")
    if guard.get("ok"):
        tagged = (guard.get("tagged") or [{}])[0]
        log(f"{tool_name}: tagged cp {tagged.get('checkpoint_id')} on {vm}, allowed")
        emit(
            None,
            context=(
                f"Zerto tagged checkpoint {tagged.get('checkpoint_id')} on VPG "
                f"{tagged.get('vpg_name')} for {vm} before this call, tag "
                f"{guard.get('tag')!r}. If this change breaks the guest, recover from "
                "that checkpoint with zerto_recover_file."
            ),
        )
    if not guard.get("protected", True):
        log(f"{tool_name}: {vm} not protected -> prompt")
        emit(
            "prompt",
            f"{vm} is not protected by Zerto, so this change cannot be rewound. "
            f"({guard.get('message', '')}) Allow it anyway?",
        )
    log(f"{tool_name}: guard failed on {vm} -> deny :: {guard.get('message')}")
    emit(
        "deny",
        f"Zerto could not tag a checkpoint for {vm}, so this change would not be "
        f"recoverable: {guard.get('message')}",
    )
 if __name__ == "__main__":
    main()
@@ -41,8 +41,3 @@ ignore = ["ASYNC240"]
 asyncio_mode = "auto"
 testpaths = ["tests"]
 pythonpath = ["src"]
 [tool.ruff.lint.per-file-ignores]
 # A PreToolUse hook runs inside the host's tool call. If it raises, it breaks
 # the call it was meant to protect, so it swallows everything and logs instead.
 "hooks/*.py" = ["BLE001", "S110"]
@@ -70,10 +70,7 @@ so a stuck session blocks the next recovery. Find it with
 ## Facts that bite
 - A tagged checkpoint is crash-consistent, not app-quiesced.
- Tagged checkpoints work on Azure and AWS protected VPGs, but they appear late:
+- Tagged checkpoints are not supported when the **protected** site is Azure or AWS. Talk to the vSphere protected ZVM.
  ~34s (Azure) and ~128s (AWS) versus ~4s on vSphere, and the checkpoint is stamped
  about 30s after you ask for it. On those VPGs the tag can end up *after* your
  change, so treat the newest checkpoint that already existed as the rewind point.
 - 10.9 FLR Operator RBAC fails; Administrator is the documented workaround.
 - FLR cannot run during clone, test, live failover, or EJC.
 - Linux FLR: files >1.5GB are a bad idea; some characters in names are refused.
@@ -93,10 +93,7 @@ async def wait_for_tag(
    raise ZertoError(
        f"Tagged checkpoint {tag!r} did not appear on VPG {vpg_identifier} "
        f"within {timeout_s:.0f}s. Do not mutate. "
-        "On a cloud-protected VPG the tag routinely takes longer than this to appear "
+        "If the protected site is Azure or AWS, tagged checkpoints are not supported."
        "(measured ~34s on Azure, ~128s on AWS), so this timeout may simply be too "
        "short rather than the insert having failed. Check the Zerto task before "
        "assuming it did not land."
    )
@@ -137,7 +137,8 @@ def find_from_rows(query: str, rows: list[dict[str, Any]]) -> FindResult:
            outcome="none",
            query=query,
            message=(
-                f"VM {vm.vm_name} ({vm.vm_identifier}) has no VPG. Unprotected: refuse the change."
+                f"VM {vm.vm_name} ({vm.vm_identifier}) has no VPG. "
                "Unprotected: refuse the change."
            ),
            vm=vm,
        )
@@ -2,8 +2,6 @@
 from __future__ import annotations
 import base64
 import hashlib
 import json
 from pathlib import Path
 from typing import Any
@@ -34,20 +32,6 @@ _catalog: MutatingCatalog | None = None
 _settings: dict[str, Any] = {}
 # Files above this are refused rather than streamed through a tool result.
 # FLR is for a config file or a dropped directory; a disk image belongs on the
 # whole-VM ladder in docs/recover-ladder.md.
 MAX_RECOVER_BYTES = 1_048_576
 def _encode_recovered(blob: bytes) -> dict[str, Any]:
    """Text where it is text, base64 otherwise, so the caller can just use it."""
    try:
        return {"encoding": "text", "content": blob.decode("utf-8")}
    except UnicodeDecodeError:
        return {"encoding": "base64", "content": base64.b64encode(blob).decode("ascii")}
 def _dump(payload: Any) -> str:
    return json.dumps(payload, indent=2, default=str)
@@ -149,9 +133,8 @@ async def zerto_create_tagged_checkpoint(
    the Zerto API accepts, so this is the only place that context can live.
    Name format: ai:<agent> | <action> | vm=<vm> | change=<change_id> | <utc>
-    Works on Azure and AWS protected VPGs despite what the 9.0 API reference says,
+    Docs: tagged checkpoints are not supported when the protected site is Azure or AWS;
-    but the tag appears late there (~34s Azure, ~128s AWS) and is stamped after the
+    run this against the vSphere protected ZVM.
    request, so it may sit after a prompt mutation.
    """
    try:
        result = await _find(query)
@@ -463,20 +446,16 @@ async def zerto_recover_file(
    checkpoint_identifier: str,
    guest_path: str,
    confirmed: bool = False,
    dest_dir: str | None = None,
 ) -> str:
    """File-level recovery from a journal checkpoint. The VM stays up.
    Returns the file CONTENT, not a path. The caller may be on another machine,
    so a path on this host is of no use to them, and letting a caller choose
    where bytes land is an arbitrary write once this server is shared.
    Requires confirmed=true (human yes). Cannot run during clone/test/live/EJC.
    10.9 FLR Operator role fails; use an Administrator account.
    Locally replicated VPGs only. FLR is performed at the VPG's recovery site,
    so a VPG replicating to a cloud ZCA must be recovered from that ZCA's API.
    guest_path is the path on the guest: /home/x/f.conf or C:\\Users\\x\\f.txt.
    Files larger than max_recover_bytes are refused; use the whole-VM ladder.
    """
    if not confirmed:
        return _dump(
@@ -492,10 +471,7 @@ async def zerto_recover_file(
    gate = await _flr_site_gate(client, vpg_identifier)
    if not gate.get("ok"):
        return _dump(gate)
-    # Server-owned, from config only. This string is also handed to the
+    dest = Path(dest_dir or _settings.get("recovery_dir") or "./recovered")
    # appliance as initialDownloadPath, so a caller-supplied value would let a
    # caller point the ZVM at a path of their choosing.
    dest = Path(_settings.get("recovery_dir") or "./recovered")
    dest.mkdir(parents=True, exist_ok=True)
    session_id: str | None = None
    before = await _live_session_ids(client)
@@ -514,29 +490,15 @@ async def zerto_recover_file(
        token = download_token_from(token_payload)
        blob = await client.fetch_download(token)
        name = Path(guest_path.replace("\\", "/")).name or "recovered.bin"
-        limit = int(_settings.get("max_recover_bytes") or MAX_RECOVER_BYTES)
+        out_path = dest / name
-        if len(blob) > limit:
+        out_path.write_bytes(blob)
            payload = {
                "ok": False,
                "too_large": True,
                "bytes": len(blob),
                "limit": limit,
                "message": (
                    f"{name} is {len(blob)} bytes, over the {limit} byte limit for "
                    "file level recovery. Use a bounded whole-VM operation instead: "
                    "offsite clone or failover test."
                ),
            }
        else:
        payload = {
            "ok": True,
-                "name": name,
+            "path": str(out_path.resolve()),
            "bytes": len(blob),
                "sha256": hashlib.sha256(blob).hexdigest(),
            "session_id": session_id,
            "flr_path": flr_path,
-                **_encode_recovered(blob),
+            "message": f"Wrote {len(blob)} bytes to {out_path}",
                "message": f"Recovered {len(blob)} bytes of {name} from the journal.",
        }
    except ZertoError as exc:
        payload = {"ok": False, "session_id": session_id, "message": str(exc)}
@@ -127,6 +127,7 @@ def can_tag(status: int | None, substatus: int | None) -> tuple[bool, str | None
        return False, f"VPG status is {status_name(status)}; not MeetingSLA"
    if substatus in SYNCING:
        return False, (
-            f"VPG is {substatus_name(substatus)}; checkpoints are not durable until sync ends"
+            f"VPG is {substatus_name(substatus)}; "
            "checkpoints are not durable until sync ends"
        )
    return True, None
@@ -1,70 +0,0 @@
 """The PreToolUse hook's pure helpers. The decision paths are exercised live."""
 import importlib.util
 from pathlib import Path
 import pytest
 spec = importlib.util.spec_from_file_location(
    "zerto_guard_hook", Path(__file__).resolve().parent.parent / "hooks" / "zerto_guard_hook.py"
 )
 hook = importlib.util.module_from_spec(spec)
 spec.loader.exec_module(hook)
@pytest.mark.parametrize(
    ("tool_name", "expected"),
    [
        ("mcp__ssh__exec", ("ssh", "exec")),
        ("mcp__winrm__run_ps", ("winrm", "run_ps")),
        # server names may themselves contain underscores; split on the FIRST __
        ("mcp__plugin_my_thing__write_file", ("plugin_my_thing", "write_file")),
        ("mcp__ob1__OB1__search_thoughts", ("ob1", "OB1__search_thoughts")),
        ("Bash", None),
        ("", None),
    ],
 )
 def test_split_mcp_name(tool_name, expected):
    assert hook.split_mcp_name(tool_name) == expected
 def test_resolve_vm_prefers_the_catalog_arg():
    assert hook.resolve_vm({"host": "a", "computer_name": "b"}, "computer_name") == "b"
 def test_resolve_vm_falls_back_when_catalog_arg_is_wrong():
    # a slightly wrong catalog entry should degrade to a prompt, not a crash
    assert hook.resolve_vm({"computer_name": "web01"}, "host") == "web01"
 def test_resolve_vm_returns_empty_when_no_vm_present():
    assert hook.resolve_vm({"command": "ls"}, "host") == ""
 def test_emit_with_no_decision_and_no_context_exits_silently(capsys):
    with pytest.raises(SystemExit) as exc:
        hook.emit(None)
    assert exc.value.code == 0
    assert capsys.readouterr().out == ""
 def test_emit_deny_shape(capsys):
    import json
    with pytest.raises(SystemExit) as exc:
        hook.emit("deny", "no checkpoint")
    assert exc.value.code == 0  # exit 2 would discard the JSON and lose the reason
    out = json.loads(capsys.readouterr().out)["hookSpecificOutput"]
    assert out["hookEventName"] == "PreToolUse"
    assert out["permissionDecision"] == "deny"
    assert out["permissionDecisionReason"] == "no checkpoint"
 def test_emit_context_without_decision(capsys):
    import json
    with pytest.raises(SystemExit):
        hook.emit(None, context="checkpoint 7180")
    out = json.loads(capsys.readouterr().out)["hookSpecificOutput"]
    assert "permissionDecision" not in out
    assert out["additionalContext"] == "checkpoint 7180"
@@ -256,44 +256,3 @@ def test_flr_gate_refuses_remote_recovery_site_and_names_it():
    # must tell the operator where the operation actually lives
    assert out["recovery_site"] == "aws-zca"
    assert "aws-zca" in out["message"]
 def test_encode_recovered_text_and_binary():
    from zerto_rewind_mcp.server import _encode_recovered
    text = _encode_recovered(b"listen: 0.0.0.0:8443\n")
    assert text["encoding"] == "text"
    assert text["content"] == "listen: 0.0.0.0:8443\n"
    binary = _encode_recovered(b"\x89PNG\r\n\x1a\n\xff\xfe")
    assert binary["encoding"] == "base64"
    import base64 as b64
    assert b64.b64decode(binary["content"]) == b"\x89PNG\r\n\x1a\n\xff\xfe"
 def test_recover_file_takes_no_caller_destination():
    """The caller must not choose where bytes land, nor where the ZVM mounts.
    dest_dir used to be a tool parameter whose value was also passed to the
    appliance as initialDownloadPath.
    """
    import inspect
    from zerto_rewind_mcp.server import zerto_recover_file
    params = set(inspect.signature(zerto_recover_file).parameters)
    assert "dest_dir" not in params
    assert params == {
        "vpg_identifier",
        "vm_identifier",
        "checkpoint_identifier",
        "guest_path",
        "confirmed",
    }
 def test_recover_byte_cap_is_configurable_and_has_a_default():
    from zerto_rewind_mcp.server import MAX_RECOVER_BYTES
    assert MAX_RECOVER_BYTES == 1_048_576