fix(checkpoints): derive the tag wait from the VPG's cadence (#12)

README.md

@@ -64,7 +64,22 @@ 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 cannot be inserted when the **protected** site is Azure or AWS (Zerto API). Point this server at the vSphere protected ZVM.
+Tagged checkpoints **do** work when the protected site is Azure or AWS. The 9.0 API
+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
 

demo/driver.py

@@ -16,6 +16,17 @@ 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()
 
@@ -212,8 +223,8 @@ async def main():
             if not rec.get("ok"):
                 w(f"{RED}  FLR failed{RST}"); return 1
             beat(1.5)
-            w(f"{YEL}  $ cat {rec['path']}{RST}")
+            w(f"{YEL}  recovered {rec['bytes']} bytes, sha256 {rec['sha256'][:16]}...{RST}")
-            body = Path(rec["path"]).read_text()
+            body = recovered_bytes(rec).decode()
             for line in body.splitlines():
                 w(f"{GREEN}{BOLD}  | {line}{RST}")
             beat(2.5)
@@ -222,11 +233,13 @@ 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 []) + \
-                  [rec["path"], f"{target}:{CFG['guest_path']}"]
+                  [str(local_copy), f"{target}:{CFG['guest_path']}"]
-            w(f"{YEL}  $ scp {Path(rec['path']).name} {target}:{CFG['guest_path']}{RST}")
+            w(f"{YEL}  $ scp {rec['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)

demo/windows_driver.py

@@ -22,6 +22,17 @@ 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()
 
@@ -155,7 +166,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 = Path(rec["path"]).read_text()
+            body = recovered_bytes(rec).decode()
             w(f"{YEL}  recovered file:{RST}")
             for line in body.splitlines():
                 w(f"{GREEN}{BOLD}  | {line}{RST}")
@@ -166,7 +177,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 = Path(rec["path"]).read_bytes()
+            blob = recovered_bytes(rec)
             b64 = base64.b64encode(blob).decode()
             guest_ps(
                 f"[IO.File]::WriteAllBytes('{CFG['guest_path']}',"

hooks/README.md

@@ -0,0 +1,95 @@
+# 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. How long the guard takes is set by
+the VPG's checkpoint cadence, which is set in turn by its protected site:
+
+| protected at | cadence | guard takes |
+|---|---|---|
+| vSphere | 5s | ~7s |
+| Azure | 60s | ~40s |
+| AWS | 630s | ~111s |
+
+The tag wait is derived from that cadence and capped at 300s, so:
+
+- `timeout` in settings.json: 360 (seconds)
+- `ZERTO_HOOK_GUARD_TIMEOUT`: 330 (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.
+
+## Why the timeouts are derived, not fixed
+
+This hook used to deny every change to a cloud-protected VM.
+
+`wait_for_tag` gave up after a hardcoded 45s. That is generous on a
+vSphere-protected VPG, which checkpoints every 5s and surfaces a tag in about
+4s, and impossible on an AWS-protected one, where a tag takes ~128s because the
+journal only checkpoints every 630s.
+
+So the guard reported "no checkpoint, refusing the change" while Zerto was in the
+middle of creating one. The checkpoint landed a minute later, in the journal,
+after the agent had already been told there was no rewind point.
+
+That is a worse failure than the one this hook exists to prevent. It is silent,
+it looks correct in the log, and it blocks legitimate work on every
+cloud-protected VM in the estate.
+
+Both budgets are now derived from the VPG's measured cadence rather than
+guessed, which is why the numbers above differ by a factor of fifteen between
+platforms.

hooks/settings.example.json

@@ -0,0 +1,17 @@
+{
+  "_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": 360
+          }
+        ]
+      }
+    ]
+  }
+}

hooks/zerto_guard_hook.py

@@ -0,0 +1,251 @@
+#!/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": 360}]}]}}
+"""
+
+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.
+# Must exceed the largest tag wait the guard can take. That is now derived from
+# the VPG's checkpoint cadence and capped at 300s (MAX_TAG_TIMEOUT_S), because a
+# tag takes ~128s to surface on an AWS-protected VPG. Too small a budget here
+# just moves the false denial from the guard into the hook.
+GUARD_TIMEOUT_S = float(os.environ.get("ZERTO_HOOK_GUARD_TIMEOUT", "330"))
+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()

pyproject.toml

@@ -41,3 +41,8 @@ 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"]

skills/zerto-rewind/SKILL.md

@@ -70,7 +70,10 @@ 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 are not supported when the **protected** site is Azure or AWS. Talk to the vSphere protected ZVM.
+- Tagged checkpoints work on Azure and AWS protected VPGs, but they appear late:
+  ~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.

src/zerto_rewind_mcp/checkpoints.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import re
+import statistics
 from datetime import UTC, datetime
 from typing import Any
 
@@ -23,6 +24,15 @@ _WS = re.compile(r"\s+")
 # so the name stays readable in the Zerto UI checkpoint list.
 TAG_MAX_LEN = 250
 
+# Tag-wait budget. Checkpoint cadence is set by the VPG's protected site and
+# measured 5s (vSphere), 60s (Azure) and 630s (AWS) in one estate, so the wait
+# has to be derived rather than fixed.
+DEFAULT_TAG_TIMEOUT_S = 45.0  # cadence unmeasurable; the old constant
+MIN_TAG_TIMEOUT_S = 45.0
+MAX_TAG_TIMEOUT_S = 300.0  # AWS needed 128s; this leaves real headroom
+MIN_POLL_INTERVAL_S = 1.5
+MAX_POLL_INTERVAL_S = 15.0
+
 
 def _clean(value: Any, limit: int) -> str:
     """One field of a checkpoint name: single-line, no separator collisions."""
@@ -74,26 +84,81 @@ def checkpoint_id(row: dict[str, Any]) -> str:
     return str(value) if value is not None else ""
 
 
+def checkpoint_gaps(rows: list[dict[str, Any]], sample: int = 20) -> list[float]:
+    """Seconds between consecutive checkpoints, newest `sample` of them."""
+    stamps: list[datetime] = []
+    for row in rows[-sample:]:
+        raw = str(pick(row, "TimeStamp", "Timestamp", "timestamp") or "")
+        try:
+            stamps.append(datetime.fromisoformat(raw))
+        except ValueError:
+            continue
+    return [
+        (stamps[i + 1] - stamps[i]).total_seconds()
+        for i in range(len(stamps) - 1)
+        if stamps[i + 1] >= stamps[i]
+    ]
+
+
+def cadence_seconds(rows: list[dict[str, Any]], sample: int = 20) -> float | None:
+    """How often this VPG writes a checkpoint. None when it cannot be measured."""
+    gaps = checkpoint_gaps(rows, sample)
+    return statistics.median(gaps) if gaps else None
+
+
+def tag_wait_budget(cadence: float | None) -> tuple[float, float]:
+    """How long to wait for a tag, and how often to look, given the cadence.
+
+    Cadence is set by the VPG's PROTECTED site, and the spread is enormous:
+    measured 5s on vSphere, 60s on Azure, 630s on AWS. A single constant cannot
+    serve all three. The old fixed 45s was generous for vSphere and impossible
+    for AWS, where a tag took 128s to surface, so the guard reported "no
+    checkpoint" while Zerto was still creating one and the change was refused
+    for no reason.
+
+    Visibility does not scale linearly with cadence (the insert makes its own
+    off-cadence checkpoint), so this is 2x cadence plus headroom, clamped.
+    """
+    if cadence is None or cadence <= 0:
+        return DEFAULT_TAG_TIMEOUT_S, MIN_POLL_INTERVAL_S
+    timeout = min(max(2 * cadence + 30, MIN_TAG_TIMEOUT_S), MAX_TAG_TIMEOUT_S)
+    interval = min(max(cadence / 10, MIN_POLL_INTERVAL_S), MAX_POLL_INTERVAL_S)
+    return timeout, interval
+
+
 async def wait_for_tag(
     client: ZertoClient,
     vpg_identifier: str,
     tag: str,
     *,
-    timeout_s: float = 45.0,
+    timeout_s: float | None = None,
-    interval_s: float = 1.5,
+    interval_s: float | None = None,
 ) -> dict[str, Any]:
+    """Wait until the tag is listed. Budget derived from the VPG's own cadence."""
+    rows = await client.list_checkpoints(vpg_identifier)
+    for row in rows:
+        if checkpoint_tag(row) == tag:
+            return row
+
+    cadence = cadence_seconds(rows)
+    budget, poll = tag_wait_budget(cadence)
+    timeout_s = budget if timeout_s is None else timeout_s
+    interval_s = poll if interval_s is None else interval_s
+
     deadline = asyncio.get_event_loop().time() + timeout_s
-    last: list[dict[str, Any]] = []
     while asyncio.get_event_loop().time() < deadline:
-        last = await client.list_checkpoints(vpg_identifier)
+        await asyncio.sleep(interval_s)
-        for row in last:
+        rows = await client.list_checkpoints(vpg_identifier)
+        for row in rows:
             if checkpoint_tag(row) == tag:
                 return row
-        await asyncio.sleep(interval_s)
+    measured = f"{cadence:.0f}s" if cadence else "unknown"
     raise ZertoError(
         f"Tagged checkpoint {tag!r} did not appear on VPG {vpg_identifier} "
-        f"within {timeout_s:.0f}s. Do not mutate. "
+        f"within {timeout_s:.0f}s (this VPG checkpoints about every {measured}). "
-        "If the protected site is Azure or AWS, tagged checkpoints are not supported."
+        "Do not mutate. Check the Zerto task before assuming the insert failed: "
+        "a completed task with no visible checkpoint means the wait was short, "
+        "not that the insert was rejected."
     )
 
 

src/zerto_rewind_mcp/protection.py

@@ -137,8 +137,7 @@ 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. "
+                f"VM {vm.vm_name} ({vm.vm_identifier}) has no VPG. Unprotected: refuse the change."
-                "Unprotected: refuse the change."
             ),
             vm=vm,
         )

src/zerto_rewind_mcp/server.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import base64
+import hashlib
 import json
 from pathlib import Path
 from typing import Any
@@ -32,6 +34,20 @@ _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)
 
@@ -133,8 +149,9 @@ 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>
-    Docs: tagged checkpoints are not supported when the protected site is Azure or AWS;
+    Works on Azure and AWS protected VPGs despite what the 9.0 API reference says,
-    run this against the vSphere protected ZVM.
+    but the tag appears late there (~34s Azure, ~128s AWS) and is stamped after the
+    request, so it may sit after a prompt mutation.
     """
     try:
         result = await _find(query)
@@ -446,16 +463,20 @@ 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(
@@ -471,7 +492,10 @@ async def zerto_recover_file(
     gate = await _flr_site_gate(client, vpg_identifier)
     if not gate.get("ok"):
         return _dump(gate)
-    dest = Path(dest_dir or _settings.get("recovery_dir") or "./recovered")
+    # Server-owned, from config only. This string is also handed to the
+    # 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)
@@ -490,16 +514,30 @@ 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"
-        out_path = dest / name
+        limit = int(_settings.get("max_recover_bytes") or MAX_RECOVER_BYTES)
-        out_path.write_bytes(blob)
+        if len(blob) > limit:
-        payload = {
+            payload = {
-            "ok": True,
+                "ok": False,
-            "path": str(out_path.resolve()),
+                "too_large": True,
-            "bytes": len(blob),
+                "bytes": len(blob),
-            "session_id": session_id,
+                "limit": limit,
-            "flr_path": flr_path,
+                "message": (
-            "message": f"Wrote {len(blob)} bytes to {out_path}",
+                    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,
+                "bytes": len(blob),
+                "sha256": hashlib.sha256(blob).hexdigest(),
+                "session_id": session_id,
+                "flr_path": flr_path,
+                **_encode_recovered(blob),
+                "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)}
     finally:

src/zerto_rewind_mcp/status.py

@@ -127,7 +127,6 @@ 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)}; "
+            f"VPG is {substatus_name(substatus)}; checkpoints are not durable until sync ends"
-            "checkpoints are not durable until sync ends"
         )
     return True, None

tests/test_checkpoints.py

@@ -1,5 +1,7 @@
 from datetime import UTC, datetime
 
+import pytest
+
 from zerto_rewind_mcp.checkpoints import (
     TAG_MAX_LEN,
     checkpoint_id,
@@ -50,3 +52,93 @@ def test_checkpoint_row_keys():
     row2 = {"checkpointId": "cp-2", "tag": "t"}
     assert checkpoint_id(row2) == "cp-2"
     assert checkpoint_tag(row2) == "t"
+
+
+def _rows(*offsets_seconds):
+    from datetime import timedelta
+
+    base = datetime(2026, 9, 22, 12, 0, 0, tzinfo=UTC)
+    return [
+        {"TimeStamp": (base + timedelta(seconds=o)).isoformat().replace("+00:00", "Z")}
+        for o in offsets_seconds
+    ]
+
+
+def test_cadence_measures_the_median_gap():
+    from zerto_rewind_mcp.checkpoints import cadence_seconds
+
+    assert cadence_seconds(_rows(0, 5, 10, 15, 20)) == 5.0
+    assert cadence_seconds(_rows(0, 60, 120, 180)) == 60.0
+    # one irregular gap must not drag the answer around
+    assert cadence_seconds(_rows(0, 5, 10, 400, 405, 410)) == 5.0
+
+
+def test_cadence_is_none_when_unmeasurable():
+    from zerto_rewind_mcp.checkpoints import cadence_seconds
+
+    assert cadence_seconds([]) is None
+    assert cadence_seconds([{"TimeStamp": "not-a-date"}]) is None
+    assert cadence_seconds(_rows(0)) is None  # one checkpoint gives no gap
+
+
+def test_tag_wait_budget_covers_every_measured_platform():
+    """The three cadences measured in one estate, and what each actually needed."""
+    from zerto_rewind_mcp.checkpoints import tag_wait_budget
+
+    for cadence, observed_visibility in ((5.0, 4.0), (60.0, 34.0), (630.0, 128.0)):
+        budget, interval = tag_wait_budget(cadence)
+        assert budget > observed_visibility, (
+            f"cadence {cadence}s budgets {budget}s but the tag took {observed_visibility}s"
+        )
+        assert interval >= 1.5
+
+
+def test_tag_wait_budget_is_clamped_at_both_ends():
+    from zerto_rewind_mcp.checkpoints import (
+        DEFAULT_TAG_TIMEOUT_S,
+        MAX_TAG_TIMEOUT_S,
+        MIN_TAG_TIMEOUT_S,
+        tag_wait_budget,
+    )
+
+    assert tag_wait_budget(0.1)[0] == MIN_TAG_TIMEOUT_S  # absurdly fast VPG
+    assert tag_wait_budget(100_000)[0] == MAX_TAG_TIMEOUT_S  # absurdly slow one
+    assert tag_wait_budget(None)[0] == DEFAULT_TAG_TIMEOUT_S
+    assert tag_wait_budget(None)[1] == 1.5
+
+
+def test_wait_for_tag_returns_immediately_when_already_present():
+    import asyncio
+
+    from zerto_rewind_mcp.checkpoints import wait_for_tag
+
+    class Client:
+        def __init__(self):
+            self.calls = 0
+
+        async def list_checkpoints(self, vpg):
+            self.calls += 1
+            return [{"Tag": "ai:x", "CheckpointId": "7"}]
+
+    c = Client()
+    row = asyncio.run(wait_for_tag(c, "vpg", "ai:x"))
+    assert row["CheckpointId"] == "7"
+    assert c.calls == 1  # no sleep, no second poll
+
+
+def test_wait_for_tag_error_names_the_measured_cadence():
+    import asyncio
+
+    from zerto_rewind_mcp.checkpoints import wait_for_tag
+    from zerto_rewind_mcp.client import ZertoError
+
+    class Client:
+        async def list_checkpoints(self, vpg):
+            return _rows(0, 60, 120, 180)  # 60s cadence, tag never appears
+
+    with pytest.raises(ZertoError) as err:
+        # explicit tiny timeout so the test does not actually wait 150s
+        asyncio.run(wait_for_tag(Client(), "vpg", "ai:missing", timeout_s=0.01, interval_s=0.01))
+    msg = str(err.value)
+    assert "about every 60s" in msg
+    assert "was short, not that the insert was rejected" in msg

tests/test_guard_hook.py

@@ -0,0 +1,70 @@
+"""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"

tests/test_recover.py

@@ -256,3 +256,44 @@ 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