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

.gitignore

@@ -16,3 +16,13 @@ catalog.json
 recovered/
 .claude/*
 !.claude/gitea-ship.json
+
+# demo harness: live credentials and generated media never go in git
+demo/demo_win.json
+demo/vo/*.wav
+demo/vo/narration.json
+demo/vo/durations*.json
+demo/*.cast
+demo/*.gif
+demo/*.mp4
+demo/marks.jsonl

CONTEXT.md

@@ -22,6 +22,7 @@ _Avoid_: recover (unqualified), fail back, restore the VPG
 
 **File-level recovery (FLR)**:
 Mount a VM from a journal checkpoint and pull files. The VM stays up. 10.9 FLR Operator RBAC is broken; Administrator is the documented workaround.
+Runs at the VPG's **recovery** site, so this server supports it only for locally replicated VPGs (protected site == recovery site). Paths are partition-rooted; on Windows the drive letter is the partition.
 _Avoid_: file restore (unqualified), instant restore (local-journal VMs only, not v1)
 
 **find_protection**:
@@ -33,8 +34,12 @@ A VPG whose status is MeetingSLA or a NotMeetingSLA variant, and whose substatus
 _Avoid_: healthy, in sync, Protecting (as status 0)
 
 **Mutating catalog**:
-The opt-in list of MCP tools that must call `zerto_guard_before_mutate` first. Unlisted tools pass through. Users add entries; the starter list is not the whole world.
-_Avoid_: denylist, hold-everything
+The opt-in list of MCP tools that must call `zerto_guard_before_mutate` first. Paired with `read_only_tools`, the list known not to change a guest. A tool in neither is **unknown**, which is the normal case: the agent asks the human whether to checkpoint rather than assuming either way.
+_Avoid_: denylist, hold-everything, treating unknown as safe
+
+**Zerto task**:
+Write operations return a task id, not a result. A 200 means queued. `GET /v1/tasks/{id}` carries the real outcome in `Status.State`: 1 InProgress, 4 Failed, 5 Stopped, 6 Completed (terminal is 4/5/6). A second tagged-checkpoint insert fired at a VPG while the first runs comes back Failed, which is only visible if the task is read.
+_Avoid_: treating HTTP 200 as success
 
 **Official ZVM MCP**:
 HPE Zerto 10.9 MCP (`ZVM.MCP`): inventory, VPG settings, failover test. Not in the demo path. This PoC is one server.

README.md

@@ -6,10 +6,11 @@ If the loop works, these tools are the delta to add to official ZVM MCP (`ZVM.MC
 
 ## What it does
 
+0. `zerto_check_tool` — before running anything against a VM: `read_only` (go), `mutating` (guard first), or `unknown` (**ask the human** whether to checkpoint).
 1. `zerto_find_protection` — VM name, hostname, or vmIdentifier to exactly one VM and every VPG. Zero or two-plus VMs: stop.
 2. `zerto_create_tagged_checkpoint` / `zerto_guard_before_mutate` — same tag on every protecting VPG, wait until listed. The name records which agent and what it is doing: `ai:<agent> | <action> | vm=<vm> | change=<id> | <utc>`.
-3. `zerto_recover_file` — FLR after a human sets `confirmed=true`.
-4. Mutating catalog — opt-in list of MCP tools that must be guarded. Unlisted tools pass through. Users add entries.
+3. `zerto_recover_file` — FLR after a human sets `confirmed=true`. Linux and Windows guest paths. Locally replicated VPGs only: FLR runs at the VPG's recovery site. Reports its own unmount; `zerto_list_flr_sessions` / `zerto_end_flr_session` find and reap a mount orphaned by a crashed recovery.
+4. Two catalogs — `mutating_tools` (guard first) and `read_only_tools` (safe). Both cover Linux (`ssh`, `ansible`) and Windows (`winrm`, `powershell`, `smb`), and both are illustrative, not exhaustive. A tool in neither is **unknown, not safe**: ask the human.
 
 Official ZVM MCP already has inventory and failover test. It does not insert tagged checkpoints or run FLR.
 
@@ -63,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
 

config.example.json

@@ -17,6 +17,52 @@
       "tool": "run_playbook",
       "vm_arg": "limit",
       "notes": "Playbook target host/group. Resolve to a single VM before guard."
+    },
+    {
+      "server": "winrm",
+      "tool": "run_command",
+      "vm_arg": "host",
+      "notes": "Windows remote shell over WinRM. vm_arg is the hostname."
+    },
+    {
+      "server": "winrm",
+      "tool": "run_ps",
+      "vm_arg": "host",
+      "notes": "PowerShell over WinRM. Same blast radius as run_command."
+    },
+    {
+      "server": "powershell",
+      "tool": "invoke_command",
+      "vm_arg": "computer_name",
+      "notes": "Invoke-Command against a remote Windows guest."
+    },
+    {
+      "server": "smb",
+      "tool": "write_file",
+      "vm_arg": "host",
+      "notes": "Writes a file into a Windows share. Changes the guest without a shell."
+    }
+  ],
+  "read_only_tools": [
+    {
+      "server": "ssh",
+      "tool": "read_file",
+      "notes": "Reads a file from the guest. Does not change it."
+    },
+    {
+      "server": "ssh",
+      "tool": "stat",
+      "notes": "File metadata only."
+    },
+    {
+      "server": "winrm",
+      "tool": "get_item",
+      "notes": "Windows read of a file or registry value."
+    },
+    {
+      "server": "powershell",
+      "tool": "get_content",
+      "notes": "Reads file contents on a Windows guest."
     }
   ]
 }

demo/README.md

@@ -0,0 +1,75 @@
+# Demo recording harness
+
+Records the rewind loop running against a live ZVM as a narrated MP4. Terminal
+only: no screen capture, no video editor.
+
+```
+tmux ──▶ asciinema ──▶ agg ──▶ ffmpeg ──▶ mp4
+  │                                        ▲
+  └── left pane: driver, right pane: journal │
+                                             │
+      xAI /v1/tts ──▶ wav per beat ──────────┘  (mux_vo.py)
+```
+
+## Pieces
+
+| file | what it does |
+|---|---|
+| `windows_driver.py` | Windows demo. Intro over the diagram, then the live loop via WinRM. |
+| `driver.py` | The Linux equivalent, over SSH. |
+| `diagram.py` | Architecture diagram, revealed in four chunks against intro beats i1..i4. |
+| `journal.py` | Right-hand pane. Polls a VPG's checkpoints so the tag appears on camera. |
+| `record_win.sh` / `record.sh` | Drive tmux + asciinema, then agg and ffmpeg. |
+| `narration.md` | The script. One `[beat]` per block; the source of truth. |
+| `vo/build_narration.py` | Parses `narration.md`, synthesises a wav per beat, records durations. |
+| `mux_vo.py` | Aligns the wavs to the recorded beat marks and muxes the audio. |
+
+## Running it
+
+```bash
+cp demo/demo_win.example.json demo/demo_win.json   # then fill in the guest creds
+export XAI_KEY_FILE=~/xai-api.key XAI_VOICE_ID=<id from /v1/custom-voices>
+python3 demo/vo/build_narration.py 1.0             # synthesise, writes durations.json
+demo/record_win.sh win1                            # record; writes marks.jsonl
+python3 demo/mux_vo.py win1                        # -> win1_narrated.mp4
+```
+
+`demo_win.json` holds live guest credentials and is gitignored. So are the
+generated `.wav`, `.cast`, `.gif` and `.mp4` files.
+
+## How the audio stays in sync
+
+The driver writes `marks.jsonl` as it runs: one line per beat with the real
+elapsed time it started. `mux_vo.py` delays each wav to its recorded mark, so
+sync survives a slow API call or an FLR mount that takes longer than usual.
+Nothing is predicted.
+
+Two things this depends on:
+
+- **`agg --idle-time-limit` must be larger than the longest pause** (the scripts
+  pass 3600). The default is 5 seconds, which compresses idle time, and that
+  silently breaks the mapping between wall clock and video time.
+- **Each beat holds for its narration length.** `hold()` sleeps out whatever is
+  left after the work finishes, so a line is never cut off mid-sentence.
+
+Check alignment after a mux: the FLR wait should measure near silence.
+
+```bash
+ffmpeg -v error -i out.mp4 -vn -ac 1 /tmp/a.wav
+ffmpeg -hide_banner -ss 150 -t 6 -i /tmp/a.wav -af volumedetect -f null /dev/null 2>&1 | grep mean_volume
+```
+
+Speech sits around -22 dB; a correctly aligned gap reads about -91 dB.
+
+## Narration gotchas
+
+- **Do not map acronyms to run-together phonetics.** The `replace` map takes
+  `{"phrase": "pronunciation"}`, and `{"VM": "vee em"}` gets spoken as one word,
+  "vem". Either leave the acronym alone or expand it: `{"VM": "virtual machine"}`.
+- **Write for speech, not for the page.** Short declaratives and fragment stacks
+  read well and sound robotic out loud. Commas and full stops are what the engine
+  uses for pacing, so clauses joined with commas breathe; a wall of four-word
+  sentences marches.
+- `volumedetect` reports `n_samples: 0` when pointed at a file whose first stream
+  is video. Extract the audio first, then measure, or you will think a working
+  track is silent.

demo/demo_win.example.json

@@ -0,0 +1,12 @@
+{
+  "vpg_id": "1816d7a4-3316-44dc-bafd-6d84f63d4ba9",
+  "vpg_name": "local",
+  "vm_query": "ad1(1)",
+  "vm_identifier": "f7f0835d-a5e4-46fa-b710-4bc32076e820.vm-2033",
+  "change_id": "win-demo-rewind",
+  "guest_path": "C:\\demo\\app-config.yaml",
+  "bad_content": "upstream: http://0.0.0.0:1   # BROKEN-BY-AGENT",
+  "host": "192.0.2.10",
+  "user": "EXAMPLE\\administrator",
+  "password": "change-me"
+}

demo/diagram.py

@@ -0,0 +1,44 @@
+"""Architecture diagram, revealed in four chunks aligned to intro beats i1..i4."""
+C = "\033[36m"; B = "\033[1m"; D = "\033[2m"; G = "\033[32m"; Y = "\033[33m"; R = "\033[0m"
+
+CHUNKS = [
+[
+ f"  {B}{C}┌────────────────────────────────┐{R}",
+ f"  {B}{C}│{R} Developer / process owner      {B}{C}│{R}   {D}\"change this config\"{R}",
+ f"  {B}{C}└───────────────┬────────────────┘{R}",
+ f"  {B}{C}                │{R}",
+ f"  {B}{C}                ▼{R}",
+ f"  {B}{C}┌────────────────────────────────┐{R}",
+ f"  {B}{C}│{R} Business application on a VM   {B}{C}│{R}",
+ f"  {B}{C}└───────────────┬────────────────┘{R}",
+],
+[
+ f"  {D}                │ protected by{R}",
+ f"  {D}                ▼{R}",
+ f"  {B}{G}┌────────────────────────────────┐{R}        {D}┌───────────────────────────┐{R}",
+ f"  {B}{G}│{R} Zerto continuous protection    {B}{G}│{R}   vs   {D}│ Nightly backup            │{R}",
+ f"  {B}{G}│{R} {G}recovery point every few secs{R}  {B}{G}│{R}        {D}│ recovery point: hours old │{R}",
+ f"  {B}{G}└────────────────────────────────┘{R}        {D}└───────────────────────────┘{R}",
+],
+[
+ "",
+ f"  {B}{Y}┌────────────────────────────────┐{R}",
+ f"  {B}{Y}│{R} AI agent                       {B}{Y}│{R}  1. ask Zerto: is this protected?",
+ f"  {B}{Y}│{R} authorised to make the change  {B}{Y}│{R}  2. tag a checkpoint in the journal",
+ f"  {B}{Y}└────────────────────────────────┘{R}  3. make the change, and test it",
+],
+[
+ "",
+ f"                                       4. {B}broken, and the agent cannot fix it{R}",
+ f"                                          recover the file, or the whole VM,",
+ f"                                          {G}from the checkpoint taken at step 2{R}",
+],
+]
+
+
+def frame(upto: int) -> str:
+    """Everything revealed through chunk `upto` (1-based)."""
+    out = ["\033[H\033[2J", f"  {B}{C}ZERTO AI REWIND{R}  {D}how it fits together{R}", ""]
+    for chunk in CHUNKS[:upto]:
+        out.extend(chunk)
+    return "\n".join(out)

demo/driver.py

@@ -0,0 +1,254 @@
+"""Left pane: drive zerto_rewind_mcp over real MCP stdio. Full rewind loop."""
+from __future__ import annotations
+import asyncio, json, os, shlex, subprocess, sys, time
+from pathlib import Path
+
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+SP = Path(__file__).parent
+CFG = json.loads((SP / "demo.json").read_text())
+REPO = "/home/justin/github/zerto-ai-rewind"
+
+RST, BOLD, DIM = "\033[0m", "\033[1m", "\033[2m"
+CYAN, GREEN, RED, YEL, MAG = "\033[36m", "\033[32m", "\033[31m", "\033[33m", "\033[35m"
+
+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()
+
+
+def type_out(s, delay=0.012, color=""):
+    sys.stdout.write(color)
+    for ch in s:
+        sys.stdout.write(ch); sys.stdout.flush()
+        time.sleep(delay * SPEED)
+    sys.stdout.write(RST + "\n"); sys.stdout.flush()
+
+
+def beat(n=1.2):
+    time.sleep(n * SPEED)
+
+
+STEP = [0]
+
+
+def step(title):
+    STEP[0] += 1
+    w()
+    w(f"{BOLD}{CYAN}{'='*66}{RST}")
+    type_out(f"  STEP {STEP[0]}  {title}", 0.008, BOLD + CYAN)
+    w(f"{BOLD}{CYAN}{'='*66}{RST}")
+    beat(0.5)
+
+
+def note(s):
+    type_out(f"  {s}", 0.010, DIM)
+
+
+def call_banner(tool, args):
+    w(f"{MAG}  -> MCP call{RST} {BOLD}{tool}{RST}")
+    for k, v in args.items():
+        w(f"{DIM}       {k} = {v}{RST}")
+
+
+def show(payload, keep=None, limit=22):
+    if keep:
+        payload = {k: payload[k] for k in keep if k in payload}
+    text = json.dumps(payload, indent=2, default=str)
+    lines = text.splitlines()
+    for line in lines[:limit]:
+        w(f"{DIM}  |{RST} {line}")
+    if len(lines) > limit:
+        w(f"{DIM}  | ... {len(lines)-limit} more lines{RST}")
+
+
+def ssh_base():
+    if CFG.get("password"):
+        return ["sshpass", "-p", CFG["password"], "ssh", "-o", "StrictHostKeyChecking=no",
+                "-o", "UserKnownHostsFile=/dev/null", "-o", "LogLevel=ERROR"]
+    base = ["ssh", "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null",
+            "-o", "LogLevel=ERROR"]
+    if CFG.get("key"):
+        base += ["-i", CFG["key"]]
+    return base
+
+
+def guest(cmd, show_cmd=True):
+    target = f"{CFG['user']}@{CFG['host']}"
+    full = ssh_base() + [target, cmd]
+    if show_cmd:
+        w(f"{YEL}  $ ssh {target} {shlex.quote(cmd)}{RST}")
+    r = subprocess.run(full, capture_output=True, text=True, timeout=60)
+    out = (r.stdout + r.stderr).rstrip()
+    for line in out.splitlines():
+        w(f"{DIM}  |{RST} {line}")
+    return r.returncode, out
+
+
+async def main():
+    w(f"\n{BOLD}{CYAN}  ZERTO AI REWIND{RST}  {DIM}live end-to-end{RST}\n")
+    type_out("  An agent is about to change a file on a protected VM.", 0.014)
+    type_out("  Zerto already journals that VM. Make the agent use it.", 0.014)
+    beat(1.5)
+
+    params = StdioServerParameters(
+        command=f"{REPO}/.venv/bin/zerto-rewind-mcp",
+        env={**os.environ, "ZERTO_REWIND_CONFIG": f"{REPO}/config.json"},
+    )
+    devnull = open(os.devnull, 'w')
+    async with stdio_client(params, errlog=devnull) as (r, s):
+        async with ClientSession(r, s) as sess:
+            step("Connect to the rewind MCP")
+            await sess.initialize()
+            tools = await sess.list_tools()
+            note(f"{len(tools.tools)} tools from zerto_rewind_mcp")
+            for t in tools.tools:
+                w(f"{DIM}    - {t.name}{RST}")
+            beat(2.0)
+
+            async def _ticker(label):
+                t0 = time.time()
+                try:
+                    while True:
+                        await asyncio.sleep(5)
+                        el = int(time.time() - t0)
+                        sys.stdout.write(f"\r{DIM}       {label} ... {el}s{RST}")
+                        sys.stdout.flush()
+                except asyncio.CancelledError:
+                    el = int(time.time() - t0)
+                    if el >= 5:
+                        sys.stdout.write(f"\r{DIM}       {label} ... {el}s done{RST}\n")
+                    else:
+                        sys.stdout.write("\r" + " " * 50 + "\r")
+                    sys.stdout.flush()
+                    raise
+
+            async def call(name, **args):
+                call_banner(name, args)
+                tick = asyncio.create_task(_ticker("working"))
+                try:
+                    res = await sess.call_tool(name, args)
+                finally:
+                    tick.cancel()
+                    try:
+                        await tick
+                    except asyncio.CancelledError:
+                        pass
+                return json.loads(res.content[0].text)
+
+            # ---- 1. the good state on the guest
+            step("The file the agent is about to break")
+            guest(f"cat {CFG['guest_path']}")
+            beat(1.8)
+
+            # ---- 2. find protection
+            step("zerto_find_protection - is this VM protected?")
+            note("Reads skip the guard. This is the read.")
+            found = await call("zerto_find_protection", query=CFG["vm_query"])
+            show(found)
+            if not found.get("ok"):
+                w(f"{RED}  find_protection not ok - stopping{RST}"); return 1
+            vm = found["vm"]
+            beat(2.5)
+
+            # ---- 3. guard
+            step("zerto_guard_before_mutate - pin the journal FIRST")
+            note("Capture-before-execute. No tag, no change.")
+            note("The checkpoint name records which agent, and what it is about to do.")
+            note("Watch the journal pane on the right.")
+            guard = await call("zerto_guard_before_mutate",
+                               query=CFG["vm_query"], change_id=CFG["change_id"], agent="claude",
+                               action=f"edit {CFG['guest_path']}")
+            show(guard, keep=["ok", "tag", "tagged", "skipped", "errors", "message"])
+            if not guard.get("ok"):
+                w(f"{RED}  GUARD FAILED - refusing the change{RST}"); return 1
+            tag = guard["tag"]
+            tagged = guard["tagged"][0]
+            t_tag = time.strftime("%H:%M:%S")
+            w(f"{GREEN}{BOLD}  tagged checkpoint {tagged['checkpoint_id']} on {tagged['vpg_name']}{RST}")
+            w(f"{GREEN}  inserted at {t_tag}{RST}")
+            w(f"{GREEN}  name: {tag}{RST}")
+            w(f"{DIM}  this checkpoint is the rewind point. remember cp {tagged['checkpoint_id']}.{RST}")
+            beat(4.0)
+
+            # ---- 4. the bad change
+            step("Now the agent makes the bad change")
+            guest(f"printf '%s\\n' {shlex.quote(CFG['bad_content'])} > {CFG['guest_path']}")
+            guest(f"cat {CFG['guest_path']}")
+            t_bad = time.strftime("%H:%M:%S")
+            w(f"{RED}{BOLD}  the good content is gone from the guest ({t_bad}){RST}")
+            note("Git never had this file. Last night's backup is hours stale.")
+            beat(2.5)
+
+            # ---- 5. refusal without a human
+            step("zerto_recover_file without a human yes")
+            deny = await call("zerto_recover_file",
+                              vpg_identifier=tagged["vpg_identifier"],
+                              vm_identifier=vm["vm_identifier"],
+                              checkpoint_identifier=tagged["checkpoint_id"],
+                              guest_path=CFG["guest_path"], confirmed=False)
+            show(deny)
+            w(f"{YEL}  refused. FLR mounts a disk; a human says yes.{RST}")
+            beat(2.5)
+
+            # ---- 6. recover
+            rec_cp = tagged["checkpoint_id"]
+            step("Human says yes - FLR from the pre-mutation checkpoint")
+            w(f"{BOLD}  provenance{RST}")
+            w(f"    {GREEN}cp {tagged['checkpoint_id']} inserted {t_tag}{RST}  {DIM}<- guard, BEFORE the change{RST}")
+            w(f"    {RED}bad write        {t_bad}{RST}  {DIM}<- the mutation, AFTER{RST}")
+            assert rec_cp == tagged["checkpoint_id"], "recovery must use the guard checkpoint"
+            w(f"{DIM}  recovering from cp {rec_cp}, not the newest checkpoint.{RST}")
+            note(f"name: {tag}")
+            rec = await call("zerto_recover_file",
+                             vpg_identifier=tagged["vpg_identifier"],
+                             vm_identifier=vm["vm_identifier"],
+                             checkpoint_identifier=tagged["checkpoint_id"],
+                             guest_path=CFG["guest_path"], confirmed=True)
+            show(rec)
+            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}")
+            body = recovered_bytes(rec).decode()
+            for line in body.splitlines():
+                w(f"{GREEN}{BOLD}  | {line}{RST}")
+            beat(2.5)
+
+            # ---- 7. put it back
+            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']}"]
+            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)
+
+            w()
+            w(f"{BOLD}{GREEN}{'='*66}{RST}")
+            type_out("  REWOUND. RPO was the journal, not last night.", 0.016, BOLD + GREEN)
+            w(f"{BOLD}{GREEN}{'='*66}{RST}")
+            beat(4.0)
+    return 0
+
+sys.exit(asyncio.run(main()))

demo/journal.py

@@ -0,0 +1,78 @@
+"""Right pane: live view of a VPG journal, tagged checkpoints highlighted."""
+from __future__ import annotations
+import asyncio, os, sys, time
+os.environ.setdefault("ZERTO_REWIND_CONFIG", "/home/justin/github/zerto-ai-rewind/config.json")
+from zerto_rewind_mcp.config import load_config
+from zerto_rewind_mcp.client import ZertoClient
+from zerto_rewind_mcp.checkpoints import checkpoint_tag, checkpoint_id
+from zerto_rewind_mcp.util import pick
+
+VPG_ID = sys.argv[1]
+VPG_NAME = sys.argv[2] if len(sys.argv) > 2 else VPG_ID
+ROWS = int(os.environ.get("JOURNAL_ROWS", "10"))
+
+DIM, RST, BOLD = "\033[2m", "\033[0m", "\033[1m"
+CYAN, GREEN, YEL = "\033[36m", "\033[32m", "\033[33m"
+
+
+def _wrap_tag(tag, width=52, maxlines=3):
+    tag = tag.replace("; Used for File Level Restore", " [FLR]")
+    out = []
+    while tag and len(out) < maxlines:
+        out.append(tag[:width]); tag = tag[width:]
+    return out
+
+
+def ts_of(row):
+    return str(pick(row, "TimeStamp", "Timestamp", "timestamp") or "")
+
+
+def render(rows, spin, tagged):
+    out = ["\033[H\033[2J"]
+    out.append(f"{BOLD}{CYAN}  ZERTO JOURNAL  {RST}{BOLD}{VPG_NAME}{RST}\n")
+    out.append(f"{DIM}  vpg {VPG_ID}{RST}\n")
+    out.append(f"{DIM}  {len(rows)} checkpoints{RST}   {YEL}{len(tagged)} ai-tagged{RST}  {DIM}{spin}{RST}\n\n")
+
+    out.append(f"{BOLD}  AI TAGS{RST}\n")
+    if not tagged:
+        out.append(f"{DIM}    (none yet){RST}\n")
+    for row in tagged[-4:]:
+        t = ts_of(row).replace("T", " ").replace(".000Z", "")[11:]
+        out.append(f"{GREEN}{BOLD}  * {t}  cp {checkpoint_id(row)}{RST}\n")
+        for seg in _wrap_tag(checkpoint_tag(row)):
+            out.append(f"{GREEN}    {seg}{RST}\n")
+
+    out.append(f"\n{BOLD}  NEWEST{RST}\n")
+    for row in rows[-ROWS:]:
+        tag = checkpoint_tag(row)
+        t = ts_of(row).replace("T", " ").replace(".000Z", "")
+        cid = checkpoint_id(row)
+        t = t[11:]
+        if tag.startswith("ai:"):
+            out.append(f"{GREEN}{BOLD}  * {t}  cp {cid}  <-- new{RST}\n")
+        elif tag:
+            out.append(f"{DIM}    {t}  cp {cid}  {tag[:20]}{RST}\n")
+        else:
+            out.append(f"{DIM}    {t}  cp {cid}{RST}\n")
+    sys.stdout.write("".join(out))
+    sys.stdout.flush()
+
+
+async def main():
+    s = load_config()
+    c = ZertoClient(base_url=s["zerto_url"], username=s["username"], password=s["password"],
+                    client_id=s.get("client_id", "zerto-client"), verify_tls=bool(s.get("verify_tls")))
+    spinner = "|/-\\"
+    i = 0
+    while True:
+        try:
+            rows = await c.list_checkpoints(VPG_ID)
+            tagged = [r for r in rows if checkpoint_tag(r).startswith("ai:")]
+            render(rows, spinner[i % 4], tagged)
+        except Exception as exc:
+            sys.stdout.write(f"\n  journal poll error: {exc}\n")
+            sys.stdout.flush()
+        i += 1
+        await asyncio.sleep(3)
+
+asyncio.run(main())

demo/mux_vo.py

@@ -0,0 +1,39 @@
+"""Build the narration track from recorded beat marks and mux it onto the video."""
+import json, subprocess, sys
+from pathlib import Path
+
+SP = Path(__file__).parent
+VO = SP / "vo"
+take = sys.argv[1] if len(sys.argv) > 1 else "win1"
+video = SP / f"{take}.mp4"
+out = SP / f"{take}_narrated.mp4"
+
+marks = [json.loads(l) for l in (SP / "marks.jsonl").read_text().splitlines() if l.strip()]
+dur = json.loads((VO / "durations.json").read_text())
+vlen = float(subprocess.run(["ffprobe", "-v", "error", "-show_entries", "format=duration",
+                             "-of", "default=nw=1:nk=1", str(video)],
+                            capture_output=True, text=True).stdout.strip())
+
+inputs, filters, labels = [], [], []
+for i, m in enumerate(marks):
+    wav = VO / f"{m['key']}.wav"
+    if not wav.exists():
+        print(f"  missing {wav.name}, skipping"); continue
+    delay_ms = int(m["t"] * 1000)
+    inputs += ["-i", str(wav)]
+    filters.append(f"[{i}:a]adelay={delay_ms}|{delay_ms},apad[a{i}]")
+    labels.append(f"[a{i}]")
+    end = m["t"] + dur.get(m["key"], 0)
+    flag = "" if end <= vlen + 0.5 else "  <-- OVERRUNS VIDEO"
+    print(f"  {m['key']:9} start {m['t']:7.2f}s  len {dur.get(m['key'],0):5.2f}s  end {end:7.2f}s{flag}")
+
+mix = "".join(labels) + f"amix=inputs={len(labels)}:normalize=0,atrim=0:{vlen},asetpts=N/SR/TB[a]"
+cmd = (["ffmpeg", "-y", "-loglevel", "error"] + inputs + ["-i", str(video),
+       "-filter_complex", ";".join(filters) + ";" + mix,
+       "-map", f"{len(labels)}:v", "-map", "[a]",
+       "-c:v", "copy", "-c:a", "aac", "-b:a", "160k", "-shortest", str(out)])
+r = subprocess.run(cmd, capture_output=True, text=True)
+if r.returncode != 0:
+    print("ffmpeg failed:\n", r.stderr[-1500:]); sys.exit(1)
+print(f"\n  video {vlen:.1f}s | narration ends {max(m['t']+dur.get(m['key'],0) for m in marks):.1f}s")
+print(f"  -> {out}")

demo/narration.md

@@ -0,0 +1,52 @@
+# Zerto AI Rewind - demo narration  (v3, rewritten against the jp-voice profile)
+# Edit any line, then tell Claude to re-read this file.
+# Blank line separates beats. Lines starting with # are ignored.
+
+## PART 1 - INTRO   (architecture diagram on screen, builds as each beat lands)
+
+[i1] Hello, and welcome to the Zerto AI Rewind demo.
+
+[i2] AI agents are acting on production systems faster than ever, and a lot faster
+than backup was ever designed to keep up with. Zerto's continuous data
+protection is a good fit for that, because the journal is always running.
+
+[i3] In this demo an AI agent makes a change to a production system. But before it
+does, it checks whether that VM is protected by Zerto, and if it is, it inserts
+a tagged checkpoint into the journal first.
+
+[i4] That gives the agent something to fall back on. If the change breaks the
+application, and the agent can't fix it on its own, it can use file level or
+full system recovery to get back to the moment before it touched anything.
+
+## PART 2 - THE DEMO   (audience: a customer, not an engineer)
+
+[d1] This is the agent connecting to Zerto. Everything from here runs against a live
+Zerto environment.
+
+[d2] The machine is a production Windows server, and Zerto is already protecting it.
+The agent has been asked to change a configuration file on that server.
+
+[d3] Before it changes anything, it asks Zerto a simple question. Is this machine
+protected, and which protection group is it in?
+
+[d4] It is, so Zerto puts a tagged checkpoint into the journal, and the agent waits
+until Zerto confirms that checkpoint is really there before it goes any further.
+
+[d5] Now the change goes in, and it breaks the file. The version that was there is
+gone from the server. Backup has a copy from last night, so that's hours old
+already. Zerto has one from seconds before the change.
+
+[d6] The agent can't just recover on its own. Recovery has to be approved by a
+person, so at this point it stops and asks.
+
+[d7] With approval, Zerto goes back to the checkpoint from just before the change,
+and hands back the original file exactly as it was.
+
+[d8] Putting that file back onto the server is an ordinary copy. Zerto's job was
+keeping the data in the first place.
+
+[d9] This was a single file, so file level recovery was enough. That same checkpoint
+would let you bring back the entire machine if the damage were bigger.
+
+[d10] And that's the point. The recovery point is the journal, seconds before the
+change, instead of last night's backup window.

demo/record.sh

@@ -0,0 +1,52 @@
+#!/usr/bin/env bash
+# Record the split-pane rewind demo to an MP4.
+set -uo pipefail
+SP="$(cd "$(dirname "$0")" && pwd)"
+REPO=/home/justin/github/zerto-ai-rewind
+PY="$REPO/.venv/bin/python"
+VPG_ID="$(python3 -c "import json;print(json.load(open('$SP/demo.json'))['vpg_id'])")"
+VPG_NAME="$(python3 -c "import json;print(json.load(open('$SP/demo.json'))['vpg_name'])")"
+TAKE="${1:-take1}"
+CAST="$SP/$TAKE.cast"; GIF="$SP/$TAKE.gif"; MP4="$SP/$TAKE.mp4"
+
+tmux -L demo kill-server 2>/dev/null
+tmux -L rec  kill-server 2>/dev/null
+sleep 1
+
+# demo session: driver left, journal right
+tmux -L demo new-session -d -s demo -x 200 -y 50 -c "$REPO" \
+  "$PY $SP/driver.py; echo; echo '  [take complete]'; sleep 3; tmux -L demo kill-server"
+tmux -L demo set-option -g status off
+tmux -L demo split-window -h -l 60 -t demo -c "$REPO" \
+  "JOURNAL_ROWS=10 $PY $SP/journal.py $VPG_ID $VPG_NAME"
+tmux -L demo select-pane -t 0
+
+# recorder session gives asciinema a pty
+tmux -L rec new-session -d -s rec -x 200 -y 50 \
+  "$SP/recvenv/bin/asciinema rec '$CAST' --overwrite --quiet -c 'tmux -L demo attach -t demo'"
+
+echo "recording -> $CAST"
+for i in $(seq 1 600); do
+  tmux -L rec has-session -t rec 2>/dev/null || break
+  sleep 2
+done
+tmux -L demo kill-server 2>/dev/null
+tmux -L rec kill-server 2>/dev/null
+sleep 1
+
+[ -s "$CAST" ] || { echo "NO CAST PRODUCED"; exit 1; }
+echo "cast: $(du -h "$CAST" | cut -f1)  duration: $(python3 -c "
+import json,sys
+last=0
+for l in open('$CAST'):
+    l=l.strip()
+    if l.startswith('['):
+        last=json.loads(l)[0]
+print(f'{last:.0f}s')")"
+
+"$SP/bin/agg" --font-size 14 --fps-cap 10 --idle-time-limit 2 --theme asciinema "$CAST" "$GIF"
+ffmpeg -y -loglevel error -i "$GIF" \
+  -movflags +faststart -pix_fmt yuv420p -c:v libx264 -crf 20 \
+  -vf "scale=trunc(iw/2)*2:trunc(ih/2)*2" "$MP4"
+echo "MP4: $MP4  ($(du -h "$MP4" | cut -f1))"
+ffprobe -v error -show_entries format=duration:stream=width,height -of default=nw=1 "$MP4"

demo/record_win.sh

@@ -0,0 +1,42 @@
+#!/usr/bin/env bash
+# Record the Windows rewind demo. NO idle compression: the video timeline must
+# equal wall clock so the narration can be aligned to the recorded beat marks.
+set -uo pipefail
+SP="$(cd "$(dirname "$0")" && pwd)"
+REPO=/home/justin/github/zerto-ai-rewind
+PY="$REPO/.venv/bin/python"
+VPG_ID=$(python3 -c "import json;print(json.load(open('$SP/demo_win.json'))['vpg_id'])")
+TAKE="${1:-win1}"
+CAST="$SP/$TAKE.cast"; GIF="$SP/$TAKE.gif"; MP4="$SP/$TAKE.mp4"
+
+tmux -L demo kill-server 2>/dev/null; tmux -L rec kill-server 2>/dev/null; sleep 1
+rm -f "$SP/marks.jsonl"
+
+tmux -L demo new-session -d -s demo -x 200 -y 50 -c "$REPO" \
+  "$PY $SP/windows_driver.py; echo; echo '  [take complete]'; sleep 2; tmux -L demo kill-server"
+tmux -L demo set-option -g status off
+tmux -L demo split-window -h -l 60 -t demo -c "$REPO" \
+  "JOURNAL_ROWS=10 $PY $SP/journal.py $VPG_ID ad1"
+tmux -L demo select-pane -t 0
+
+tmux -L rec new-session -d -s rec -x 200 -y 50 \
+  "$SP/recvenv/bin/asciinema rec '$CAST' --overwrite --quiet -c 'tmux -L demo attach -t demo'"
+
+echo "recording -> $CAST"
+for i in $(seq 1 450); do tmux -L rec has-session -t rec 2>/dev/null || break; sleep 2; done
+tmux -L demo kill-server 2>/dev/null; tmux -L rec kill-server 2>/dev/null; sleep 1
+[ -s "$CAST" ] || { echo "NO CAST"; exit 1; }
+
+echo "cast duration: $(python3 -c "
+import json
+last=0
+for l in open('$CAST'):
+    l=l.strip()
+    if l.startswith('['): last=json.loads(l)[0]
+print(f'{last:.1f}s')")"
+# idle-time-limit huge = no compression, so 1s of cast == 1s of video
+"$SP/bin/agg" --font-size 14 --fps-cap 10 --idle-time-limit 3600 --theme asciinema "$CAST" "$GIF"
+ffmpeg -y -loglevel error -i "$GIF" -movflags +faststart -pix_fmt yuv420p -c:v libx264 -crf 20 \
+  -vf "scale=trunc(iw/2)*2:trunc(ih/2)*2" "$MP4"
+echo "MP4: $MP4 ($(du -h "$MP4" | cut -f1))"
+ffprobe -v error -show_entries format=duration:stream=width,height -of default=nw=1 "$MP4"

demo/vo/build_narration.py

@@ -0,0 +1,35 @@
+"""Parse the approved markdown script into clips + durations."""
+import json, os, re, subprocess, sys
+from pathlib import Path
+SP = Path(__file__).parent
+SRC = Path(os.environ.get("NARRATION_MD", SP.parent / "narration.md"))
+KEY = Path(os.environ.get("XAI_KEY_FILE", "~/xai-api.key")).expanduser().read_text().strip()
+VOICE = os.environ.get("XAI_VOICE_ID", "")  # a custom voice id from /v1/custom-voices
+SPEED = float(sys.argv[1]) if len(sys.argv) > 1 else 1.0
+
+beats = re.findall(r"\[([id]\d+)\]\s*(.*?)(?=\n\n|\Z)", SRC.read_text(), re.S)
+lines = {k: " ".join(v.split()) for k, v in beats}
+(SP / "narration.json").write_text(json.dumps(lines, indent=2) + "\n")
+
+# Do NOT map an acronym to run-together phonetics: {"VM": "vee em"} is spoken
+# as one word, "vem". Expand it instead, or leave it alone.
+REPLACE = {"VM": "virtual machine"}
+out = {}
+for key, text in lines.items():
+    body = {"text": text, "voice_id": VOICE, "language": "en", "speed": SPEED,
+            "replace": REPLACE, "output_format": {"codec": "wav", "sample_rate": 24000}}
+    req = SP / "req.json"; req.write_text(json.dumps(body))
+    dest = SP / f"{key}.wav"
+    code = subprocess.run(["curl", "-sS", "-o", str(dest), "-w", "%{http_code}", "-X", "POST",
+        "https://api.x.ai/v1/tts", "-H", f"Authorization: Bearer {KEY}",
+        "-H", "Content-Type: application/json", "--data-binary", f"@{req}"],
+        capture_output=True, text=True).stdout.strip()
+    req.unlink()
+    if code != "200":
+        print(f"  {key}: HTTP {code} FAILED"); sys.exit(1)
+    d = float(subprocess.run(["ffprobe", "-v", "error", "-show_entries", "format=duration",
+        "-of", "default=nw=1:nk=1", str(dest)], capture_output=True, text=True).stdout.strip())
+    out[key] = round(d, 2)
+    print(f"  {key:5} {d:5.2f}s  {len(text.split()):3} words")
+(SP / "durations.json").write_text(json.dumps(out, indent=1))
+print(f"\n  {len(out)} clips, {sum(out.values()):.1f}s of narration")

demo/windows_driver.py

@@ -0,0 +1,204 @@
+"""Windows rewind demo. Beats hold for their narration length and log real start times."""
+from __future__ import annotations
+import json, os, sys, time
+from pathlib import Path
+
+import winrm
+import diagram
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+SP = Path(__file__).parent
+CFG = json.loads((SP / "demo_win.json").read_text())
+DUR = json.loads((SP / "vo" / "durations.json").read_text())
+REPO = "/home/justin/github/zerto-ai-rewind"
+MARKS = SP / "marks.jsonl"
+PAD = 0.8  # breath between beats
+
+RST, BOLD, DIM = "\033[0m", "\033[1m", "\033[2m"
+CYAN, GREEN, RED, YEL, MAG = "\033[36m", "\033[32m", "\033[31m", "\033[33m", "\033[35m"
+
+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()
+
+
+def type_out(s, delay=0.018, color=""):
+    sys.stdout.write(color)
+    for ch in s:
+        sys.stdout.write(ch); sys.stdout.flush(); time.sleep(delay)
+    sys.stdout.write(RST + "\n"); sys.stdout.flush()
+
+
+def begin(key, title=None):
+    """Mark the beat start so the narration can be aligned to it later."""
+    t = time.time() - T0
+    with MARKS.open("a") as fh:
+        fh.write(json.dumps({"key": key, "t": round(t, 3)}) + "\n")
+    if title:
+        STEP[0] += 1
+        w(); w(f"{BOLD}{CYAN}{'=' * 74}{RST}")
+        type_out(f"  STEP {STEP[0]}  {title}", 0.010, BOLD + CYAN)
+        w(f"{BOLD}{CYAN}{'=' * 74}{RST}")
+    return t
+
+
+def hold(start_t, key):
+    """Keep this beat on screen until its narration has finished."""
+    want = DUR.get(key, 6.0) + PAD
+    spent = (time.time() - T0) - start_t
+    if spent < want:
+        time.sleep(want - spent)
+
+
+def show(payload, keep=None, limit=18):
+    if keep:
+        payload = {k: payload[k] for k in keep if k in payload}
+    for line in json.dumps(payload, indent=2, default=str).splitlines()[:limit]:
+        w(f"{DIM}  |{RST} {line}")
+
+
+_sess = winrm.Session(f"http://{CFG['host']}:5985/wsman",
+                      auth=(CFG["user"], CFG["password"]), transport="ntlm")
+
+
+def guest_ps(ps, label):
+    w(f"{YEL}  PS {CFG['host']}> {label}{RST}")
+    r = _sess.run_ps(ps)
+    out = r.std_out.decode(errors="replace").strip()
+    for line in out.splitlines():
+        w(f"{DIM}  |{RST} {line}")
+    return out
+
+
+async def main():
+    import asyncio
+    MARKS.unlink(missing_ok=True)
+
+    for n, key in enumerate(("i1", "i2", "i3", "i4"), start=1):
+        t = begin(key)
+        sys.stdout.write(diagram.frame(n)); sys.stdout.write("\n"); sys.stdout.flush()
+        hold(t, key)
+
+    params = StdioServerParameters(
+        command=f"{REPO}/.venv/bin/zerto-rewind-mcp",
+        env={**os.environ, "ZERTO_REWIND_CONFIG": f"{REPO}/config.json"})
+    devnull = open(os.devnull, "w")
+    async with stdio_client(params, errlog=devnull) as (r, s):
+        async with ClientSession(r, s) as sess:
+            t = begin("d1", "Connect to the rewind MCP")
+            await sess.initialize()
+            tools = await sess.list_tools()
+            w(f"{DIM}  {len(tools.tools)} tools from zerto_rewind_mcp{RST}")
+            for x in tools.tools:
+                w(f"{DIM}    - {x.name}{RST}")
+            hold(t, "d1")
+
+            async def call(name, **args):
+                w(f"{MAG}  -> MCP call{RST} {BOLD}{name}{RST}")
+                for k, v in args.items():
+                    w(f"{DIM}       {k} = {v}{RST}")
+                res = await sess.call_tool(name, args)
+                return json.loads(res.content[0].text)
+
+            t = begin("d2", "The file the agent is about to break")
+            guest_ps(f"Get-Content '{CFG['guest_path']}'", f"Get-Content {CFG['guest_path']}")
+            hold(t, "d2")
+
+            t = begin("d3", "zerto_find_protection")
+            found = await call("zerto_find_protection", query=CFG["vm_query"])
+            show(found, keep=["ok", "outcome", "vm"])
+            if not found.get("ok"):
+                w(f"{RED}  not ok, stopping{RST}"); return 1
+            hold(t, "d3")
+
+            t = begin("d4", "zerto_guard_before_mutate")
+            guard = await call("zerto_guard_before_mutate", query=CFG["vm_query"],
+                               change_id=CFG["change_id"], agent="claude",
+                               action=f"edit {CFG['guest_path']}")
+            show(guard, keep=["ok", "tag", "tagged"])
+            if not guard.get("ok"):
+                w(f"{RED}  GUARD FAILED, refusing the change{RST}"); return 1
+            tg = guard["tagged"][0]
+            t_tag = time.strftime("%H:%M:%S")
+            w(f"{GREEN}{BOLD}  checkpoint {tg['checkpoint_id']} inserted {t_tag}  task={tg['task_state']}{RST}")
+            hold(t, "d4")
+
+            t = begin("d5", "Now the agent makes the bad change")
+            guest_ps(f"Set-Content -Path '{CFG['guest_path']}' -Value '{CFG['bad_content']}' -Encoding ASCII; "
+                     f"Get-Content '{CFG['guest_path']}'", f"Set-Content {CFG['guest_path']} ...")
+            t_bad = time.strftime("%H:%M:%S")
+            w(f"{RED}{BOLD}  the good config is gone from the guest ({t_bad}){RST}")
+            hold(t, "d5")
+
+            t = begin("d6", "zerto_recover_file without a human yes")
+            deny = await call("zerto_recover_file", vpg_identifier=CFG["vpg_id"],
+                              vm_identifier=CFG["vm_identifier"],
+                              checkpoint_identifier=tg["checkpoint_id"],
+                              guest_path=CFG["guest_path"], confirmed=False)
+            show(deny)
+            w(f"{YEL}  refused.{RST}")
+            hold(t, "d6")
+
+            t = begin("d7", "Human says yes: FLR from the pre-mutation checkpoint")
+            w(f"{BOLD}  provenance{RST}")
+            w(f"    {GREEN}cp {tg['checkpoint_id']} inserted {t_tag}{RST}  {DIM}<- guard, BEFORE{RST}")
+            w(f"    {RED}bad write        {t_bad}{RST}  {DIM}<- the mutation, AFTER{RST}")
+            rec = await call("zerto_recover_file", vpg_identifier=CFG["vpg_id"],
+                             vm_identifier=CFG["vm_identifier"],
+                             checkpoint_identifier=tg["checkpoint_id"],
+                             guest_path=CFG["guest_path"], confirmed=True,
+                             dest_dir=str(SP / "win_demo_out"))
+            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()
+            w(f"{YEL}  recovered file:{RST}")
+            for line in body.splitlines():
+                w(f"{GREEN}{BOLD}  | {line}{RST}")
+            hold(t, "d7")
+
+            t = begin("d8", "Copy it back to the guest")
+            # Ship the recovered bytes verbatim. Set-Content rewrites line
+            # endings and appends one, which makes the restored file differ
+            # from what Zerto handed back. WriteAllBytes does not.
+            import base64
+            blob = recovered_bytes(rec)
+            b64 = base64.b64encode(blob).decode()
+            guest_ps(
+                f"[IO.File]::WriteAllBytes('{CFG['guest_path']}',"
+                f"[Convert]::FromBase64String('{b64}'));"
+                f"Get-Content '{CFG['guest_path']}';"
+                f"'md5: ' + (Get-FileHash '{CFG['guest_path']}' -Algorithm MD5).Hash",
+                f"WriteAllBytes {CFG['guest_path']} <{len(blob)} recovered bytes>")
+            hold(t, "d8")
+
+            t = begin("d9")
+            w(); w(f"{BOLD}{YEL}  the same checkpoint, bigger blast radius{RST}")
+            type_out("  one file        ->  file level recovery   (what you just saw)", 0.012, DIM)
+            type_out("  the whole VM    ->  offsite clone, failover test, or failover", 0.012, DIM)
+            hold(t, "d9")
+
+            t = begin("d10")
+            w(); w(f"{BOLD}{GREEN}{'=' * 74}{RST}")
+            type_out("  REWOUND. The recovery point was the journal.", 0.018, BOLD + GREEN)
+            w(f"{BOLD}{GREEN}{'=' * 74}{RST}")
+            hold(t, "d10")
+    return 0
+
+import asyncio
+sys.exit(asyncio.run(main()))

docs/recover-ladder.md

@@ -11,7 +11,29 @@ known path (config, dropped file, one directory).
 
 FLR mounts a checkpoint and copies files out. The protected VM stays up.
 Official API: `POST /v1/flrs` then browse/download. This MCP writes the file
-to `recovery_dir` on the MCP host. Putting it back on the guest is a second
+to `recovery_dir` on the MCP host.
+
+**FLR runs at the VPG's recovery site**, because that is where the mount is
+created. A VPG replicating to a cloud ZCA must be recovered through that ZCA's
+API, not the protected ZVM's. A production server would hold credentials for
+every ZVM/ZCA in the estate and route the call; this one does not, so
+`zerto_recover_file` is gated to **locally replicated VPGs** (protected site ==
+recovery site) and refuses anything else while naming the site that owns the
+operation.
+
+Paths are rooted at partitions, and Linux and Windows are not symmetrical:
+
+| | guest path | FLR path |
+|---|---|---|
+| Linux | `/home/j/app.yaml` | `Volume2-Ext4%2fhome%2fj%2fapp.yaml` |
+| Windows | `C:\Users\j\app.conf` | `C%3a%2fUsers%2fj%2fapp.conf` |
+
+On Windows the drive letter **is** the partition name, so nothing is
+prepended. Browse form-encodes: `%2f` separator, `%3a` drive colon, and a
+space as `+` (`Program+Files`). A session reports mounted before volume
+enumeration settles, so the partition list must be polled until it stops
+changing -- an early read can show a restorable NTFS disk as
+`Volume4-Unknown`. Putting it back on the guest is a second
 step (scp/ssh). That copy-back is not Zerto; it is ordinary file transfer.
 
 Do not use FLR when:
@@ -24,6 +46,11 @@ Do not use FLR when:
 - OS-level dedup volumes
 - 10.9 FLR Operator role (broken; Administrator is the documented workaround)
 
+FLR sessions must be unmounted when done. `zerto_recover_file` does that
+itself and reports it in `unmount`, but that cleanup only runs if the MCP
+process survives the call. After a crash, list orphans with
+`zerto_list_flr_sessions` and end them with `zerto_end_flr_session`.
+
 A tagged checkpoint must already exist. Initial sync has an empty journal
 (`GET .../checkpoints` returns `[]`). Guard refuses until status is MeetingSLA
 (or NotMeetingSLA) and substatus is not a sync.

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

@@ -11,16 +11,33 @@ You talk to **one** MCP: `zerto_rewind_mcp`. Do not also require official ZVM MC
 
 ## Loop (mandatory)
 
-Before **every** guest-mutating tool call:
+Before **every** tool call that might touch a guest:
 
 1. Take the hostname / VM name / Zerto `vmIdentifier` from the tool args.
-2. Call `zerto_guard_before_mutate` with `change_id` and `action` (or `zerto_find_protection` then `zerto_create_tagged_checkpoint`).
-3. If `ok` is not true: **stop**. Do not mutate.
-4. Then run the mutating call.
+2. Call `zerto_check_tool(server, tool, vm)`. Act on the verdict:
 
-Reads skip the guard.
+| verdict | what you do |
+|---|---|
+| `read_only` | Run the tool. No checkpoint. |
+| `mutating` | Guard first. Do not ask — it is already known to change the guest. |
+| `unknown` | **Ask the human.** Do not assume it is safe, and do not silently guard. |
 
-Unlisted MCP tools pass through. If you are about to change a protected VM with a tool that is not in the catalog, call `zerto_add_mutating_tool` (server, tool, `vm_arg`) and then guard.
+3. For `mutating`, or for `unknown` where the human said yes: call
+   `zerto_guard_before_mutate` with `change_id` and `action`.
+4. If `ok` is not true: **stop**. Do not mutate.
+5. Then run the call.
+
+`unknown` is the normal case, not an edge case. The catalogs are short and the
+world of tools is not, so most tools are unclassified. Unknown means *nobody has
+said this is read-only* — it does not mean safe. Put the decision to the human:
+
+> `winrm/run_ps` is not a known read-only command. It may change `web01`.
+> Insert a Zerto tagged checkpoint first so this is reversible?
+
+If they say yes, guard, then run. If they say no, run it and tell them plainly
+that it is not reversible through Zerto. If they want it remembered, call
+`zerto_add_mutating_tool` (server, tool, `vm_arg`) so it is guarded
+automatically next time instead of asking again.
 
 ## find_protection outcomes
 
@@ -37,14 +54,26 @@ A VM can be in up to three VPGs (local backup + remote DR is common). Tag all of
 
 Human must confirm. Pass `confirmed=true` only after they say yes.
 
-- Bad config / dropped file: `zerto_recover_file` from **that tag**.
+- Bad config / dropped file: `zerto_recover_file` from **that tag**. Pass the guest path
+  (`/home/j/app.yaml` or `C:\Users\j\app.conf`); the server maps it into the FLR
+  namespace. **Locally replicated VPGs only** -- FLR happens at the recovery site, so a
+  cloud-replicated VPG must be recovered from that ZCA. The tool refuses and names the site.
 - Inspect a whole VM: `zerto_offsite_clone` or `zerto_start_failover_test`.
 - Never Failover Live. Never Move. Those are DR, not rewind.
 
+`zerto_recover_file` unmounts its own FLR session and reports the result in
+`unmount`. If `unmount.ok` is false, or a previous recovery died mid-flight,
+the mount is still up: FLR cannot run during clone, test, live failover or EJC,
+so a stuck session blocks the next recovery. Find it with
+`zerto_list_flr_sessions` and clear it with `zerto_end_flr_session`.
+
 ## 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/catalog.py

@@ -36,12 +36,42 @@ def entry_from_dict(raw: dict[str, Any]) -> CatalogEntry:
     )
 
 
+MUTATING = "mutating"
+READ_ONLY = "read_only"
+UNKNOWN = "unknown"
+
+
 class MutatingCatalog:
-    def __init__(self, entries: list[CatalogEntry] | None = None, path: Path | None = None):
+    def __init__(
+        self,
+        entries: list[CatalogEntry] | None = None,
+        path: Path | None = None,
+        read_only: list[CatalogEntry] | None = None,
+    ):
         self._entries: dict[tuple[str, str], CatalogEntry] = {}
+        self._read_only: dict[tuple[str, str], CatalogEntry] = {}
         self.path = path
         for entry in entries or []:
             self._entries[entry.key()] = entry
+        for entry in read_only or []:
+            self._read_only[entry.key()] = entry
+
+    def read_only(self) -> list[CatalogEntry]:
+        return sorted(self._read_only.values(), key=lambda e: e.key())
+
+    def classify(self, server: str, tool: str) -> str:
+        """mutating / read_only / unknown.
+
+        unknown is the important one. It does not mean safe: it means nobody has
+        said this tool is read-only, so the agent must ask a human before
+        changing a protected VM with it.
+        """
+        key = (server.lower(), tool.lower())
+        if key in self._entries:
+            return MUTATING
+        if key in self._read_only:
+            return READ_ONLY
+        return UNKNOWN
 
     def list(self) -> list[CatalogEntry]:
         return sorted(self._entries.values(), key=lambda e: e.key())
@@ -68,6 +98,9 @@ class MutatingCatalog:
 
     @classmethod
     def from_config(cls, data: dict[str, Any], path: Path | None = None) -> MutatingCatalog:
-        raw = data.get("mutating_tools") or []
-        entries = [entry_from_dict(item) for item in raw]
-        return cls(entries, path=path)
+        entries = [entry_from_dict(item) for item in data.get("mutating_tools") or []]
+        read_only = [
+            entry_from_dict({**item, "vm_arg": item.get("vm_arg") or "-"})
+            for item in data.get("read_only_tools") or []
+        ]
+        return cls(entries, path=path, read_only=read_only)

src/zerto_rewind_mcp/checkpoints.py

@@ -4,11 +4,13 @@ from __future__ import annotations
 
 import asyncio
 import re
+import statistics
 from datetime import UTC, datetime
 from typing import Any
 
 from zerto_rewind_mcp.client import ZertoClient, ZertoError
 from zerto_rewind_mcp.protection import FindResult
+from zerto_rewind_mcp.tasks import task_id_from, wait_for_task
 from zerto_rewind_mcp.util import pick
 
 _CTRL = re.compile(r"[\x00-\x1f\x7f]+")
@@ -22,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."""
@@ -73,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,
-    interval_s: float = 1.5,
+    timeout_s: float | None = None,
+    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)
-        for row in last:
+        await asyncio.sleep(interval_s)
+        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. "
-        "If the protected site is Azure or AWS, tagged checkpoints are not supported."
+        f"within {timeout_s:.0f}s (this VPG checkpoints about every {measured}). "
+        "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."
     )
 
 
@@ -108,16 +174,20 @@ async def tag_vpgs(
             "message": result.message,
             "find": result.as_dict(),
         }
-    # Insert then wait, one VPG at a time. Measured on 10.x: tagged checkpoint
-    # inserts fired back to back at the same VPG are silently dropped -- the POST
-    # returns 200 and queues a task, but only the first checkpoint ever appears.
-    # Do not turn this loop into an asyncio.gather().
+    # Insert then wait for the TASK, one VPG at a time. The POST returns a task
+    # id, not a result: 200 only means queued. Measured on 10.x, a second insert
+    # fired at the same VPG while the first is running comes back Failed (state
+    # 4) from GET /v1/tasks/{id} while the first is Completed (6). That failure
+    # is reported, not silent -- but only if you read the task. Do not turn this
+    # loop into an asyncio.gather().
     tagged: list[dict[str, Any]] = []
     skipped = [v.as_dict() for v in result.vm.vpgs if not v.can_tag]
     errors: list[str] = []
     for vpg in result.taggable_vpgs:
         try:
             insert = await client.insert_checkpoint(vpg.vpg_identifier, tag)
+            task_id = task_id_from(insert)
+            task = await wait_for_task(client, task_id)
             row = await wait_for_tag(client, vpg.vpg_identifier, tag)
             tagged.append(
                 {
@@ -125,7 +195,8 @@ async def tag_vpgs(
                     "vpg_name": vpg.vpg_name,
                     "checkpoint_id": checkpoint_id(row),
                     "tag": checkpoint_tag(row) or tag,
-                    "insert_result": insert if not isinstance(insert, dict) else "ok",
+                    "task_id": task_id,
+                    "task_state": task["label"],
                 }
             )
         except ZertoError as exc:

src/zerto_rewind_mcp/client.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import time
 from typing import Any
 from urllib.parse import urljoin
@@ -10,6 +11,9 @@ import httpx
 
 from zerto_rewind_mcp.util import pick
 
+TOKEN_FETCH_ATTEMPTS = 3
+TOKEN_RETRY_BACKOFF_S = 0.5
+
 
 class ZertoError(Exception):
     def __init__(self, message: str, status_code: int | None = None, body: str | None = None):
@@ -34,6 +38,14 @@ class ZertoClient:
         self.client_id = client_id
         self._token: str | None = None
         self._token_exp = 0.0
+        # Serialises token acquisition. Without it, concurrent tool calls each
+        # see a stale token and each POST to Keycloak; the ZVM rejects one of
+        # the simultaneous password grants and that call dies with HTTP 401.
+        self._token_lock = asyncio.Lock()
+        # Bumped on every successful fetch. A caller that got a 401 uses it to
+        # tell "my token is stale" from "another task already replaced it",
+        # so one expiry causes one re-auth, not one per in-flight request.
+        self._token_gen = 0
         self._http = httpx.AsyncClient(verify=verify_tls, timeout=timeout)
 
     async def aclose(self) -> None:
@@ -44,33 +56,83 @@ class ZertoClient:
             return path
         return urljoin(self.base_url + "/", path.lstrip("/"))
 
-    async def ensure_token(self) -> str:
-        if self._token and time.time() < self._token_exp - 30:
-            return self._token
+    def _token_is_fresh(self) -> bool:
+        return bool(self._token) and time.time() < self._token_exp - 30
+
+    async def _fetch_token(self) -> str:
+        """POST the password grant. Caller must hold _token_lock."""
         url = self._url("/auth/realms/zerto/protocol/openid-connect/token")
-        response = await self._http.post(
-            url,
-            data={
-                "grant_type": "password",
-                "username": self.username,
-                "password": self.password,
-                "client_id": self.client_id,
-                "scope": "openid",
-            },
-            headers={"Content-Type": "application/x-www-form-urlencoded"},
-        )
-        if response.status_code >= 400:
-            raise ZertoError(
-                f"Keycloak token failed HTTP {response.status_code}. "
-                "Check username/password/client_id "
-                "(10.x uses zerto-client; 9.x may use zerto-api).",
-                status_code=response.status_code,
-                body=response.text[:500],
+        data = {
+            "grant_type": "password",
+            "username": self.username,
+            "password": self.password,
+            "client_id": self.client_id,
+            "scope": "openid",
+        }
+        headers = {"Content-Type": "application/x-www-form-urlencoded"}
+        last: httpx.Response | None = None
+        for attempt in range(TOKEN_FETCH_ATTEMPTS):
+            try:
+                last = await self._http.post(url, data=data, headers=headers)
+            except httpx.HTTPError as exc:
+                if attempt == TOKEN_FETCH_ATTEMPTS - 1:
+                    raise ZertoError(f"Keycloak token request failed: {exc}") from exc
+                await asyncio.sleep(TOKEN_RETRY_BACKOFF_S * (attempt + 1))
+                continue
+            if last.status_code < 400:
+                payload = last.json()
+                self._token = payload["access_token"]
+                self._token_exp = time.time() + float(payload.get("expires_in") or 60)
+                self._token_gen += 1
+                return self._token
+            # 401/403 is the credentials themselves. Retrying hammers Keycloak
+            # and can trip its brute-force lockout on a real account, so do not.
+            if last.status_code in (401, 403) or attempt == TOKEN_FETCH_ATTEMPTS - 1:
+                break
+            await asyncio.sleep(TOKEN_RETRY_BACKOFF_S * (attempt + 1))
+        assert last is not None
+        # Keycloak names the cause: invalid_client is the wrong client_id,
+        # invalid_grant is the wrong username/password. Saying which saves
+        # chasing the wrong one.
+        detail = ""
+        try:
+            error = (last.json() or {}).get("error") or ""
+        except ValueError:
+            error = ""
+        if error == "invalid_client":
+            detail = (
+                f" Keycloak said invalid_client: client_id {self.client_id!r} is wrong "
+                "for this appliance (10.x uses zerto-client; 9.x may use zerto-api)."
             )
-        payload = response.json()
-        self._token = payload["access_token"]
-        self._token_exp = time.time() + float(payload.get("expires_in") or 60)
-        return self._token
+        elif error == "invalid_grant":
+            detail = " Keycloak said invalid_grant: the username or password is wrong."
+        elif error:
+            detail = f" Keycloak said {error}."
+        raise ZertoError(
+            f"Keycloak token failed HTTP {last.status_code}."
+            f"{detail or ' Check username/password/client_id.'}",
+            status_code=last.status_code,
+            body=last.text[:500],
+        )
+
+    async def ensure_token(self) -> str:
+        if self._token_is_fresh():
+            assert self._token is not None
+            return self._token
+        async with self._token_lock:
+            # Another task may have fetched one while we waited for the lock.
+            if self._token_is_fresh():
+                assert self._token is not None
+                return self._token
+            return await self._fetch_token()
+
+    async def _reauth(self, seen_gen: int) -> str:
+        """Re-auth after a 401, unless another task already did it."""
+        async with self._token_lock:
+            if self._token_gen != seen_gen and self._token:
+                return self._token
+            self._token = None
+            return await self._fetch_token()
 
     async def request(
         self,
@@ -81,6 +143,7 @@ class ZertoClient:
         json_body: Any = None,
     ) -> httpx.Response:
         token = await self.ensure_token()
+        gen = self._token_gen
         headers = {"Authorization": f"Bearer {token}"}
         if json_body is not None:
             headers["Content-Type"] = "application/json"
@@ -92,8 +155,7 @@ class ZertoClient:
             headers=headers,
         )
         if response.status_code == 401:
-            self._token = None
-            token = await self.ensure_token()
+            token = await self._reauth(gen)
             headers["Authorization"] = f"Bearer {token}"
             response = await self._http.request(
                 method,
@@ -151,7 +213,7 @@ class ZertoClient:
     async def get_vpg(self, vpg_identifier: str) -> dict[str, Any]:
         data = await self.json("GET", f"/v1/vpgs/{vpg_identifier}")
         if not isinstance(data, dict):
-            raise ZertoError("GET /v1/vpgs/{id} did not return an object")
+            raise ZertoError(f"GET /v1/vpgs/{vpg_identifier} did not return an object")
         return data
 
     async def list_checkpoints(self, vpg_identifier: str) -> list[dict[str, Any]]:
@@ -217,7 +279,7 @@ class ZertoClient:
 
     async def fetch_download(self, token: str) -> bytes:
         token = str(token).strip().strip('"')
-        if token.startswith("v1/") or token.startswith("/v1/"):
+        if token.startswith(("v1/", "/v1/")):
             paths = [token if token.startswith("/") else f"/{token}"]
         else:
             paths = [f"/v1/downloads/{token}", f"/v1/flrs/{token}"]
@@ -226,11 +288,25 @@ class ZertoClient:
             last = await self.request("GET", path)
             if last.status_code < 400:
                 return last.content
+        status = last.status_code if last else None
+        body = last.text[:300] if last else ""
         raise ZertoError(
-            f"FLR download HTTP {last.status_code if last else '?'}: {(last.text[:300] if last else '')}",
-            status_code=last.status_code if last else None,
+            f"FLR download HTTP {status if status is not None else '?'}: {body}",
+            status_code=status,
         )
 
+    async def get_localsite(self) -> dict[str, Any]:
+        data = await self.json("GET", "/v1/localsite")
+        return data if isinstance(data, dict) else {}
+
+    async def get_peersites(self) -> list[dict[str, Any]]:
+        data = await self.json("GET", "/v1/peersites")
+        return [r for r in data if isinstance(r, dict)] if isinstance(data, list) else []
+
+    async def list_flrs(self) -> Any:
+        """GET /v1/flrs. Every FLR session the ZVM currently knows about."""
+        return await self.json("GET", "/v1/flrs")
+
     async def end_flr(self, session_id: str) -> None:
         try:
             await self.json("DELETE", f"/v1/flrs/{session_id}")

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

src/zerto_rewind_mcp/recover.py

@@ -9,6 +9,9 @@ from typing import Any
 from zerto_rewind_mcp.client import ZertoClient, ZertoError
 from zerto_rewind_mcp.util import pick
 
+# Seconds between partition-enumeration polls. Module level so tests can shrink it.
+PARTITION_POLL_INTERVAL_S = 5.0
+
 READY_STATUSES = {
     "ready",
     "mounted",
@@ -98,8 +101,15 @@ async def wait_flr_ready(
 
 
 def _decode_flr_path(value: Any) -> str:
-    """Browse returns paths percent-encoded (%2f). Download wants them decoded."""
-    return urllib.parse.unquote(str(value or "")).replace("\\", "/")
+    """Decode a browse path for display and comparison.
+
+    Browse form-encodes: '%2f' is the separator, '%3a' the drive colon, and a
+    SPACE comes back as '+' ("Program+Files"). unquote alone leaves the '+',
+    so a basename compare against "Program Files" would never match.
+    Download accepts the raw and the decoded form, so we send the raw one and
+    only decode for matching and display.
+    """
+    return urllib.parse.unquote_plus(str(value or "")).replace("\\", "/")
 
 
 def path_items(payload: Any) -> list[dict[str, Any]]:
@@ -112,44 +122,136 @@ def path_items(payload: Any) -> list[dict[str, Any]]:
 
 
 async def browsable_partitions(client: ZertoClient, session_id: str) -> list[str]:
-    """FLR is rooted at partitions (Volume2-Ext4), not the guest's /.
+    """FLR is rooted at partitions, not the guest's /.
 
-    Volume1-Unknown and friends report IsBrowsable false and cannot be restored.
+    Linux: Volume2-Ext4. Windows: the drive letter is the partition name and
+    comes back percent-encoded, e.g. 'C%3a'. Partitions reporting IsBrowsable
+    false (FAT32, MicrosoftReservedPartition, Unknown) cannot be restored from.
     """
     rows = path_items(await client.browse_flr(session_id, path=""))
     return [str(r.get("Path")) for r in rows if r.get("IsBrowsable")]
 
 
-async def resolve_flr_path(client: ZertoClient, session_id: str, guest_path: str) -> str:
-    """Map a guest absolute path to the FLR namespace path the download API accepts.
+async def wait_partitions_stable(
+    client: ZertoClient,
+    session_id: str,
+    *,
+    timeout_s: float = 120.0,
+    interval_s: float | None = None,
+) -> list[str]:
+    """Wait until the partition list stops changing, then return it.
 
-    /home/justin/app-config.yaml -> Volume2-Ext4/home/justin/app-config.yaml
+    A session reports mounted before the ZVM has finished identifying volumes.
+    Browsing in that window returns a partial and MIS-LABELLED list: the same
+    Windows VM enumerated as 'Volume4-Unknown' with no C: drive, and moments
+    later as a browsable 'C%3a' holding the whole filesystem. Acting on the
+    early list makes a restorable disk look permanently unrestorable.
+    """
+    interval_s = PARTITION_POLL_INTERVAL_S if interval_s is None else interval_s
+    deadline = asyncio.get_event_loop().time() + timeout_s
+    previous: list[str] | None = None
+    while asyncio.get_event_loop().time() < deadline:
+        current = await browsable_partitions(client, session_id)
+        if current and previous is not None and current == previous:
+            return current
+        previous = current
+        await asyncio.sleep(interval_s)
+    if previous:
+        return previous
+    raise ZertoError(
+        "FLR mounted but no browsable partition appeared. Unsupported partition "
+        "type (FAT32, MicrosoftReservedPartition, LVM, unknown) cannot be restored."
+    )
+
+
+async def resolve_flr_path(client: ZertoClient, session_id: str, guest_path: str) -> str:
+    """Map a guest absolute path to the path the FLR download API accepts.
+
+    Linux  /home/justin/app-config.yaml -> Volume2-Ext4%2fhome%2fjustin%2f...
+    Windows C:\\Users\\x\\f.txt          -> C%3a%2fUsers%2fx%2ff.txt
+
+    The two are not symmetrical. On Linux the partition is a separate root that
+    must be prepended. On Windows the drive letter IS the partition, so the
+    guest path already carries it and prepending again yields 'C:/C:/Users'.
+
+    Returns the raw path exactly as browse reported it; download accepts that
+    verbatim, which avoids re-encoding the '+' and '%3a' back by hand.
     """
     rel = guest_path.replace("\\", "/").strip("/")
     if not rel:
         raise ZertoError("Empty guest_path")
     parts = rel.split("/")
-    name, parent = parts[-1], "/".join(parts[:-1])
+    name, parent_rel = parts[-1], "/".join(parts[:-1])
 
-    partitions = await browsable_partitions(client, session_id)
-    if not partitions:
-        raise ZertoError(
-            "FLR mounted but no browsable partition. Unsupported partition type "
-            "(LVM/unknown) cannot be restored by FLR."
-        )
-    tried = []
+    partitions = await wait_partitions_stable(client, session_id)
+    tried: list[str] = []
     for vol in partitions:
-        probe = f"{vol}/{parent}" if parent else vol
+        vol_dec = _decode_flr_path(vol).rstrip("/")
+        low_rel, low_vol = rel.lower(), vol_dec.lower()
+        if low_rel == low_vol or low_rel.startswith(low_vol + "/"):
+            # Windows: guest path already starts with the drive-letter partition
+            probe = parent_rel or vol_dec
+        else:
+            probe = f"{vol_dec}/{parent_rel}" if parent_rel else vol_dec
         tried.append(probe)
         try:
             rows = path_items(await client.browse_flr(session_id, path=probe))
         except ZertoError:
             continue
-        for row in rows:
-            decoded = _decode_flr_path(row.get("Path"))
-            if decoded.rsplit("/", 1)[-1] == name:
-                return decoded
+        # exact match first; Windows is case-insensitive, Linux is not, so only
+        # fall back to a case-insensitive match when nothing matched exactly.
+        for want_exact in (True, False):
+            for row in rows:
+                raw = str(row.get("Path") or "")
+                got = _decode_flr_path(raw).rsplit("/", 1)[-1]
+                if got == name if want_exact else got.lower() == name.lower():
+                    return raw
     raise ZertoError(
         f"{guest_path!r} not found in the FLR mount. Looked under {tried}. "
-        "The file may not have replicated into that checkpoint yet."
+        f"Browsable partitions were {partitions}. Either the path is wrong, or "
+        "the file had not replicated into that checkpoint yet."
     )
+
+
+def session_rows(payload: Any) -> list[dict[str, Any]]:
+    """Normalise GET /v1/flrs, which returns a list or a single object."""
+    if isinstance(payload, list):
+        return [r for r in payload if isinstance(r, dict)]
+    if isinstance(payload, dict):
+        return [payload]
+    return []
+
+
+def session_id_of(row: dict[str, Any]) -> str:
+    value = pick(
+        row,
+        "FlrSessionIdentifier",
+        "flrSessionIdentifier",
+        "SessionId",
+        "sessionId",
+        "Identifier",
+        "identifier",
+    )
+    return str(value) if value is not None else ""
+
+
+def session_summary(row: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "session_id": session_id_of(row),
+        "state": flr_status(row),
+        "vpg_name": pick(row, "VpgName", "vpgName"),
+        "vm_name": pick(row, "VmName", "vmName"),
+        "checkpoint_id": pick(row, "CheckpointIdentifier", "checkpointIdentifier"),
+        "mounted_at": pick(row, "MountedTime", "mountedTime", "StartTime", "startTime"),
+    }
+
+
+def is_live_session(row: dict[str, Any]) -> bool:
+    """A session still holding a mount on the recovery site.
+
+    Unmounted/ended sessions linger in GET /v1/flrs as history; they hold nothing.
+    """
+    state = flr_status(row).lower()
+    if not state:
+        return False
+    return "unmount" not in state and "fail" not in state and "end" not in state

src/zerto_rewind_mcp/server.py

@@ -2,21 +2,27 @@
 
 from __future__ import annotations
 
+import base64
+import hashlib
 import json
 from pathlib import Path
 from typing import Any
 
 from mcp.server.fastmcp import FastMCP
 
-from zerto_rewind_mcp.catalog import CatalogEntry, MutatingCatalog
+from zerto_rewind_mcp.catalog import MUTATING, READ_ONLY, CatalogEntry, MutatingCatalog
 from zerto_rewind_mcp.checkpoints import checkpoint_id, checkpoint_tag, make_tag, tag_vpgs
 from zerto_rewind_mcp.client import ZertoClient, ZertoError
 from zerto_rewind_mcp.config import load_catalog, load_config
 from zerto_rewind_mcp.protection import find_from_rows
 from zerto_rewind_mcp.recover import (
     download_token_from,
+    is_live_session,
     resolve_flr_path,
     session_id_from,
+    session_id_of,
+    session_rows,
+    session_summary,
     wait_flr_ready,
 )
 from zerto_rewind_mcp.util import pick
@@ -28,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)
 
@@ -129,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;
-    run this against the vSphere protected ZVM.
+    Works on Azure and AWS protected VPGs despite what the 9.0 API reference says,
+    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)
@@ -215,6 +236,69 @@ async def zerto_guard_before_mutate(
     )
 
 
+@mcp.tool(
+    name="zerto_check_tool",
+    annotations={
+        "title": "Is this tool safe to run on a protected VM?",
+        "readOnlyHint": True,
+        "destructiveHint": False,
+        "idempotentHint": True,
+        "openWorldHint": False,
+    },
+)
+async def zerto_check_tool(server: str, tool: str, vm: str = "") -> str:
+    """Classify a tool before you run it against a VM. Call this first.
+
+    Returns one of:
+
+    read_only  - known not to change the guest. Run it. No checkpoint.
+    mutating   - known to change the guest. Guard first, do not ask.
+    unknown    - NOT known to be read-only. Do not assume it is safe and do not
+                 silently guard either. ASK THE HUMAN whether to insert a
+                 tagged checkpoint before running it, then do what they say.
+
+    unknown is the normal case for any tool nobody has classified yet, which is
+    most of them. The catalogs are short; the world of tools is not.
+    """
+    verdict = get_catalog().classify(server, tool)
+    payload: dict[str, Any] = {
+        "ok": True,
+        "server": server,
+        "tool": tool,
+        "vm": vm or None,
+        "verdict": verdict,
+    }
+    target = vm or "<the VM this tool targets>"
+    if verdict == READ_ONLY:
+        payload["checkpoint_required"] = False
+        payload["next"] = "Run the tool. Reads do not need a checkpoint."
+    elif verdict == MUTATING:
+        payload["checkpoint_required"] = True
+        payload["next"] = (
+            f"Call zerto_guard_before_mutate(query={target!r}, change_id=..., "
+            "action=<what you are about to do>) and only run the tool if it "
+            "returns ok true."
+        )
+    else:
+        payload["checkpoint_required"] = None
+        payload["ask_user"] = True
+        payload["question"] = (
+            f"{server}/{tool} is not a known read-only command. It may change "
+            f"{target}. Insert a Zerto tagged checkpoint first so this is "
+            "reversible?"
+        )
+        payload["next"] = (
+            "Ask the human the question above and wait for an answer. Do not "
+            "guess. If they say yes, call zerto_guard_before_mutate("
+            f"query={target!r}, change_id=..., action=<what you are about to "
+            "do>) and run the tool only if it returns ok true. If they say no, "
+            "run the tool without a checkpoint and tell them it is not "
+            "reversible through Zerto. If they want this remembered, call "
+            "zerto_add_mutating_tool so it is guarded automatically next time."
+        )
+    return _dump(payload)
+
+
 @mcp.tool(
     name="zerto_list_mutating_tools",
     annotations={
@@ -226,9 +310,22 @@ async def zerto_guard_before_mutate(
     },
 )
 async def zerto_list_mutating_tools() -> str:
-    """Tools that must be guarded. Unlisted tools pass through."""
-    entries = [e.as_dict() for e in get_catalog().list()]
-    return _dump({"ok": True, "count": len(entries), "mutating_tools": entries})
+    """Both catalogs. A tool in neither is unknown: ask the human before mutating."""
+    cat = get_catalog()
+    entries = [e.as_dict() for e in cat.list()]
+    read_only = [e.as_dict() for e in cat.read_only()]
+    return _dump(
+        {
+            "ok": True,
+            "count": len(entries),
+            "mutating_tools": entries,
+            "read_only_tools": read_only,
+            "unlisted": (
+                "A tool in neither list is unknown, not safe. Call "
+                "zerto_check_tool and ask the human before changing a protected VM."
+            ),
+        }
+    )
 
 
 @mcp.tool(
@@ -256,6 +353,100 @@ async def zerto_add_mutating_tool(
     return _dump({"ok": True, "entry": entry.as_dict()})
 
 
+async def _site_name(client: ZertoClient, identifier: str | None) -> str:
+    if not identifier:
+        return "unknown site"
+    try:
+        local = await client.get_localsite()
+        if str(local.get("SiteIdentifier")) == identifier:
+            return str(local.get("SiteName") or identifier)
+        for peer in await client.get_peersites():
+            if str(peer.get("SiteIdentifier")) == identifier:
+                return str(peer.get("PeerSiteName") or identifier)
+    except ZertoError:
+        pass
+    return identifier
+
+
+async def _flr_site_gate(client: ZertoClient, vpg_identifier: str) -> dict[str, Any]:
+    """Refuse FLR on a VPG whose recovery site is not this ZVM.
+
+    FLR only exists at the VPG's RECOVERY site: the mount is created there. A
+    VPG replicating to a cloud ZCA has to be recovered from that ZCA's API, not
+    this one. Until this server can hold credentials for every ZVM/ZCA in an
+    estate and route the call, restrict FLR to local replication (protected
+    site == recovery site) and say plainly where the operation actually lives,
+    rather than letting it fail as a confusing path or mount error.
+    """
+    try:
+        vpg = await client.get_vpg(vpg_identifier)
+    except ZertoError as exc:
+        return {"ok": False, "message": f"Could not read VPG {vpg_identifier}: {exc}"}
+    protected = (vpg.get("ProtectedSite") or {}).get("identifier")
+    recovery = (vpg.get("RecoverySite") or {}).get("identifier")
+    if protected and recovery and str(protected) == str(recovery):
+        return {"ok": True}
+    where = await _site_name(client, str(recovery) if recovery else None)
+    return {
+        "ok": False,
+        "not_local_replication": True,
+        "vpg_name": vpg.get("VpgName"),
+        "recovery_site": where,
+        "message": (
+            f"FLR for VPG {vpg.get('VpgName')!r} lives at its recovery site "
+            f"({where}), not at this ZVM. This server only supports file "
+            "recovery for locally replicated VPGs (protected site == recovery "
+            "site). Point an MCP instance at that ZVM/ZCA, or use a bounded "
+            "whole-VM operation instead."
+        ),
+    }
+
+
+async def _live_session_ids(client: ZertoClient) -> set[str]:
+    try:
+        rows = session_rows(await client.list_flrs())
+    except ZertoError:
+        return set()
+    return {session_id_of(r) for r in rows if is_live_session(r) and session_id_of(r)}
+
+
+async def _teardown_flr(
+    client: ZertoClient,
+    session_id: str | None,
+    before: set[str],
+) -> dict[str, Any]:
+    """Unmount the FLR session and say what happened.
+
+    A swallowed unmount failure is how a mount silently wedges the recovery
+    site: FLR cannot run during clone/test/live/EJC, so a stuck session blocks
+    the next recovery. Report it instead.
+
+    If session_id is None the start may still have succeeded on the ZVM while
+    the response failed to parse, so reap anything new that appeared.
+    """
+    out: dict[str, Any] = {"attempted": False, "ok": True, "ended": [], "failed": []}
+    targets = [session_id] if session_id else []
+    if not targets:
+        orphans = sorted(await _live_session_ids(client) - before)
+        targets = orphans
+        out["orphans_reaped"] = orphans
+    for sid in targets:
+        out["attempted"] = True
+        try:
+            await client.end_flr(sid)
+            out["ended"].append(sid)
+        except ZertoError as exc:
+            out["ok"] = False
+            out["failed"].append({"session_id": sid, "message": str(exc)})
+    if not out["ok"]:
+        out["message"] = (
+            "FLR session may still be mounted on the recovery site. "
+            "List it with zerto_list_flr_sessions and end it with "
+            "zerto_end_flr_session; a stuck mount blocks the next FLR."
+        )
+    return out
+
+
 @mcp.tool(
     name="zerto_recover_file",
     annotations={
@@ -272,12 +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(
@@ -290,9 +489,17 @@ async def zerto_recover_file(
             }
         )
     client = get_client()
-    dest = Path(dest_dir or _settings.get("recovery_dir") or "./recovered")
+    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
+    # 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 = None
+    session_id: str | None = None
+    before = await _live_session_ids(client)
+    payload: dict[str, Any]
     try:
         started = await client.start_flr(
             vpg_identifier,
@@ -307,26 +514,118 @@ 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
-        out_path.write_bytes(blob)
-        return _dump(
-            {
-                "ok": True,
-                "path": str(out_path.resolve()),
+        limit = int(_settings.get("max_recover_bytes") or MAX_RECOVER_BYTES)
+        if len(blob) > limit:
+            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,
+                "bytes": len(blob),
+                "sha256": hashlib.sha256(blob).hexdigest(),
                 "session_id": session_id,
                 "flr_path": flr_path,
-                "message": f"Wrote {len(blob)} bytes to {out_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:
+        unmount = await _teardown_flr(client, session_id, before)
+    payload["unmount"] = unmount
+    if not unmount["ok"]:
+        # The bytes are on disk, so ok stays true, but the caller must be told
+        # the mount is still up rather than finding out at the next recovery.
+        payload["warning"] = unmount["message"]
+    return _dump(payload)
+
+
+@mcp.tool(
+    name="zerto_list_flr_sessions",
+    annotations={
+        "title": "List FLR sessions on the ZVM",
+        "readOnlyHint": True,
+        "destructiveHint": False,
+        "idempotentHint": True,
+        "openWorldHint": True,
+    },
+)
+async def zerto_list_flr_sessions(live_only: bool = True) -> str:
+    """Every FLR session the ZVM knows about, so orphaned mounts are visible.
+
+    zerto_recover_file tears its own session down, but that cleanup only runs if
+    this process survives the call. A crash, disconnect or timeout mid-recovery
+    leaves the mount up with nothing tracking it. FLR cannot run during clone,
+    test, live failover or EJC, so a stuck mount blocks the next recovery.
+
+    live_only keeps sessions still holding a mount. Pass false to see ended and
+    failed sessions too, which the ZVM keeps as history.
+    """
+    try:
+        rows = session_rows(await get_client().list_flrs())
     except ZertoError as exc:
         return _dump({"ok": False, "message": str(exc)})
-    finally:
-        if session_id:
-            try:
-                await client.end_flr(session_id)
-            except ZertoError:
-                pass
+    kept = [r for r in rows if is_live_session(r)] if live_only else rows
+    return _dump(
+        {
+            "ok": True,
+            "live_only": live_only,
+            "count": len(kept),
+            "total_known": len(rows),
+            "sessions": [session_summary(r) for r in kept],
+        }
+    )
+
+
+@mcp.tool(
+    name="zerto_end_flr_session",
+    annotations={
+        "title": "End an FLR session (unmount)",
+        "readOnlyHint": False,
+        "destructiveHint": True,
+        "idempotentHint": True,
+        "openWorldHint": True,
+    },
+)
+async def zerto_end_flr_session(session_id: str, confirmed: bool = False) -> str:
+    """Unmount an FLR session. Use to reap an orphan left by a crashed recovery.
+
+    Requires confirmed=true: ending a session that another operator is actively
+    pulling files from will interrupt them. Find the id with
+    zerto_list_flr_sessions.
+    """
+    if not confirmed:
+        return _dump(
+            {
+                "ok": False,
+                "needs_confirm": True,
+                "message": (
+                    "Set confirmed=true after a human yes. Ending a session that "
+                    "someone is actively recovering from will interrupt them."
+                ),
+            }
+        )
+    try:
+        await get_client().end_flr(session_id)
+    except ZertoError as exc:
+        return _dump({"ok": False, "session_id": session_id, "message": str(exc)})
+    return _dump(
+        {
+            "ok": True,
+            "session_id": session_id,
+            "message": f"Ended FLR session {session_id}.",
+        }
+    )
 
 
 @mcp.tool(

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

src/zerto_rewind_mcp/tasks.py

@@ -0,0 +1,90 @@
+"""Zerto async task polling.
+
+Most write operations return a task id, not a result. The POST returning 200
+only means the task was queued. Whether it actually happened is in
+GET /v1/tasks/{id}. Skipping that read is how a failed checkpoint insert looks
+like a success.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from zerto_rewind_mcp.client import ZertoClient, ZertoError
+
+# Zerto task states (9.0 Tasks API; mirrored by Zerto's published quickstarts).
+FIRST_UNUSED_VALUE = 0
+IN_PROGRESS = 1
+WAITING_FOR_USER_INPUT = 2
+PAUSED = 3
+FAILED = 4
+STOPPED = 5
+COMPLETED = 6
+CANCELLING = 7
+
+TASK_STATE_LABELS: dict[int, str] = {
+    FIRST_UNUSED_VALUE: "FirstUnusedValue",
+    IN_PROGRESS: "InProgress",
+    WAITING_FOR_USER_INPUT: "WaitingForUserInput",
+    PAUSED: "Paused",
+    FAILED: "Failed",
+    STOPPED: "Stopped",
+    COMPLETED: "Completed",
+    CANCELLING: "Cancelling",
+}
+TERMINAL_STATES = {FAILED, STOPPED, COMPLETED}
+
+
+def task_id_from(payload: Any) -> str:
+    """A write endpoint returns the task id as a bare (often quoted) string."""
+    if isinstance(payload, str):
+        value = payload.strip().strip('"')
+        if value:
+            return value
+    if isinstance(payload, dict):
+        for key in ("TaskIdentifier", "taskIdentifier", "Identifier", "identifier", "id"):
+            if payload.get(key):
+                return str(payload[key]).strip().strip('"')
+    raise ZertoError(f"Could not read a task id from {payload!r}")
+
+
+def task_state(payload: Any) -> int | None:
+    """Status is {"State": n, "Progress": n}, but can be a bare int."""
+    if not isinstance(payload, dict):
+        return None
+    status = payload.get("Status")
+    if isinstance(status, dict):
+        status = status.get("State")
+    return status if isinstance(status, bool) is False and isinstance(status, int) else None
+
+
+def task_label(state: int | None) -> str:
+    return TASK_STATE_LABELS.get(state, f"Unknown({state})") if state is not None else "Unknown"
+
+
+async def wait_for_task(
+    client: ZertoClient,
+    task_id: str,
+    *,
+    timeout_s: float = 120.0,
+    interval_s: float = 3.0,
+) -> dict[str, Any]:
+    """Poll until the task reaches a terminal state. Raise unless it Completed."""
+    deadline = asyncio.get_event_loop().time() + timeout_s
+    last: dict[str, Any] = {}
+    while asyncio.get_event_loop().time() < deadline:
+        last = await client.get_task(task_id)
+        state = task_state(last)
+        if state in TERMINAL_STATES:
+            if state == COMPLETED:
+                return {"task_id": task_id, "state": state, "label": task_label(state)}
+            raise ZertoError(
+                f"Zerto task {task_id} ({last.get('Type')}) finished as "
+                f"{task_label(state)}. The operation did not happen."
+            )
+        await asyncio.sleep(interval_s)
+    raise ZertoError(
+        f"Zerto task {task_id} ({last.get('Type')}) still "
+        f"{task_label(task_state(last))} after {timeout_s:.0f}s."
+    )

tests/test_catalog.py

@@ -31,3 +31,41 @@ def test_entry_requires_fields():
         raise AssertionError("expected ValueError")
     except ValueError:
         pass
+
+
+def test_example_config_covers_windows_and_linux():
+    """The starter catalog must not be Linux-only.
+
+    The catalog is opt-in: an unlisted tool passes through unguarded. A
+    Windows-only shop taking the shipped defaults would therefore mutate
+    protected guests with no checkpoint at all.
+    """
+    example = Path(__file__).resolve().parent.parent / "config.example.json"
+    data = json.loads(example.read_text(encoding="utf-8"))
+    cat = MutatingCatalog.from_config(data)
+    assert cat.get("ssh", "exec") is not None
+    assert cat.get("winrm", "run_command") is not None
+    assert cat.get("winrm", "run_ps") is not None
+    assert cat.get("powershell", "invoke_command") is not None
+    # every entry must name the arg holding the VM, or the guard cannot resolve one
+    for entry in cat.list():
+        assert entry.vm_arg, f"{entry.server}/{entry.tool} has no vm_arg"
+
+
+def test_classify_three_way():
+    data = {
+        "mutating_tools": [{"server": "winrm", "tool": "run_ps", "vm_arg": "host"}],
+        "read_only_tools": [{"server": "ssh", "tool": "read_file"}],
+    }
+    cat = MutatingCatalog.from_config(data)
+    assert cat.classify("winrm", "run_ps") == "mutating"
+    assert cat.classify("SSH", "Read_File") == "read_only"
+    # unknown is not safe; it means nobody classified it
+    assert cat.classify("anything", "else") == "unknown"
+
+
+def test_read_only_entries_need_no_vm_arg():
+    data = {"read_only_tools": [{"server": "ssh", "tool": "stat"}]}
+    cat = MutatingCatalog.from_config(data)
+    assert cat.classify("ssh", "stat") == "read_only"
+    assert [e.tool for e in cat.read_only()] == ["stat"]

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_client_auth.py

@@ -0,0 +1,120 @@
+import asyncio
+
+import pytest
+
+from zerto_rewind_mcp.client import ZertoClient, ZertoError
+
+
+class _FakeResponse:
+    def __init__(self, status_code, payload=None, text=""):
+        self.status_code = status_code
+        self._payload = payload
+        self.text = text
+
+    def json(self):
+        if self._payload is None:
+            raise ValueError("no json")
+        return self._payload
+
+
+class _FakeHttp:
+    """Counts token POSTs so we can prove they are serialised."""
+
+    def __init__(self, token_responses=None, delay=0.01):
+        self.token_posts = 0
+        self.delay = delay
+        self.token_responses = list(token_responses or [])
+
+    async def post(self, url, data=None, headers=None):
+        self.token_posts += 1
+        await asyncio.sleep(self.delay)  # widen the race window
+        if self.token_responses:
+            return self.token_responses.pop(0)
+        return _FakeResponse(200, {"access_token": f"tok{self.token_posts}", "expires_in": 300})
+
+
+def _client(http):
+    c = ZertoClient(base_url="https://zvm", username="u", password="p")
+    c._http = http
+    return c
+
+
+def test_concurrent_ensure_token_makes_one_request():
+    """The bug: N concurrent calls each POST to Keycloak and one gets 401."""
+    http = _FakeHttp()
+    client = _client(http)
+
+    async def run():
+        return await asyncio.gather(*(client.ensure_token() for _ in range(10)))
+
+    tokens = asyncio.run(run())
+    assert http.token_posts == 1, f"expected 1 token request, got {http.token_posts}"
+    assert set(tokens) == {"tok1"}
+
+
+def test_cached_token_is_reused_without_a_request():
+    http = _FakeHttp()
+    client = _client(http)
+    asyncio.run(client.ensure_token())
+    asyncio.run(client.ensure_token())
+    assert http.token_posts == 1
+
+
+def test_reauth_is_skipped_when_another_task_already_refreshed():
+    http = _FakeHttp()
+    client = _client(http)
+    first = asyncio.run(client.ensure_token())
+    stale_gen = client._token_gen - 1  # pretend we held the previous token
+
+    async def run():
+        return await client._reauth(stale_gen)
+
+    again = asyncio.run(run())
+    # someone else already moved past our generation: reuse, do not re-auth
+    assert again == first
+    assert http.token_posts == 1
+
+
+def test_reauth_refetches_when_our_token_is_the_current_one():
+    http = _FakeHttp()
+    client = _client(http)
+    asyncio.run(client.ensure_token())
+    gen = client._token_gen
+    token = asyncio.run(client._reauth(gen))
+    assert token == "tok2"
+    assert http.token_posts == 2
+
+
+def test_bad_credentials_are_not_retried():
+    """Retrying a 401 risks tripping Keycloak brute-force lockout."""
+    http = _FakeHttp(token_responses=[_FakeResponse(401, {"error": "invalid_grant"}, "denied")])
+    client = _client(http)
+    with pytest.raises(ZertoError) as err:
+        asyncio.run(client.ensure_token())
+    assert http.token_posts == 1
+    assert "invalid_grant" in str(err.value)
+    assert "username or password" in str(err.value)
+
+
+def test_wrong_client_id_is_named():
+    http = _FakeHttp(token_responses=[_FakeResponse(401, {"error": "invalid_client"}, "nope")])
+    client = _client(http)
+    with pytest.raises(ZertoError) as err:
+        asyncio.run(client.ensure_token())
+    assert "invalid_client" in str(err.value)
+    assert "zerto-client" in str(err.value)
+
+
+def test_transient_server_error_is_retried():
+    http = _FakeHttp(
+        token_responses=[
+            _FakeResponse(503, None, "busy"),
+            _FakeResponse(200, {"access_token": "ok", "expires_in": 300}),
+        ]
+    )
+    client = _client(http)
+    import zerto_rewind_mcp.client as mod
+
+    mod.TOKEN_RETRY_BACKOFF_S = 0
+    assert asyncio.run(client.ensure_token()) == "ok"
+    assert http.token_posts == 2

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

@@ -46,34 +46,254 @@ def test_decode_flr_path():
     )
 
 
-def test_resolve_flr_path_picks_browsable_partition():
+class _BrowseFake:
+    """Fake FLR mount. root is the partition list; tree maps probe path -> rows."""
+
+    def __init__(self, root, tree):
+        self.root, self.tree, self.seen = root, tree, []
+
+    async def browse_flr(self, session_id, path="", recursive=False):
+        self.seen.append(path)
+        if path == "":
+            return {"PathItems": self.root}
+        return {"PathItems": self.tree.get(path, [])}
+
+
+def _resolve(client, guest_path):
     import asyncio
 
-    from zerto_rewind_mcp.recover import resolve_flr_path
+    from zerto_rewind_mcp import recover
 
-    root = {
-        "PathItems": [
+    recover.PARTITION_POLL_INTERVAL_S = 0  # do not sleep in tests
+    return asyncio.run(recover.resolve_flr_path(client, "sess", guest_path))
+
+
+def test_resolve_flr_path_linux_prepends_partition():
+    client = _BrowseFake(
+        [
             {"Path": "Volume1-Unknown", "IsBrowsable": False},
             {"Path": "Volume2-Ext4", "IsBrowsable": True},
-        ]
-    }
-    listing = {
-        "PathItems": [
-            {"Path": "Volume2-Ext4%2fhome%2fjustin%2f.bashrc", "Type": "File"},
-            {"Path": "Volume2-Ext4%2fhome%2fjustin%2fapp-config.yaml", "Type": "File"},
-        ]
-    }
-
-    class FakeClient:
-        def __init__(self):
-            self.seen = []
-
-        async def browse_flr(self, session_id, path="", recursive=False):
-            self.seen.append(path)
-            return root if path == "" else listing
-
-    client = FakeClient()
-    got = asyncio.run(resolve_flr_path(client, "sess", "/home/justin/app-config.yaml"))
-    assert got == "Volume2-Ext4/home/justin/app-config.yaml"
-    # must not try the unrestorable partition
+        ],
+        {
+            "Volume2-Ext4/home/justin": [
+                {"Path": "Volume2-Ext4%2fhome%2fjustin%2f.bashrc", "Type": "File"},
+                {"Path": "Volume2-Ext4%2fhome%2fjustin%2fapp-config.yaml", "Type": "File"},
+            ]
+        },
+    )
+    # returns the raw path browse gave us; download accepts it verbatim
+    assert _resolve(client, "/home/justin/app-config.yaml") == (
+        "Volume2-Ext4%2fhome%2fjustin%2fapp-config.yaml"
+    )
     assert "Volume1-Unknown/home/justin" not in client.seen
+
+
+def test_resolve_flr_path_windows_does_not_double_the_drive_letter():
+    # On Windows the drive letter IS the partition: C%3a decodes to "C:".
+    client = _BrowseFake(
+        [
+            {"Path": "C%3a", "IsBrowsable": True},
+            {"Path": "Volume2-FAT32", "IsBrowsable": False},
+        ],
+        {"C:/Users/justin": [{"Path": "C%3a%2fUsers%2fjustin%2fapp.conf", "Type": "File"}]},
+    )
+    assert _resolve(client, r"C:\Users\justin\app.conf") == "C%3a%2fUsers%2fjustin%2fapp.conf"
+    # the bug this guards: probing C:/C:/Users/justin
+    assert not any(p.count("C:") > 1 for p in client.seen)
+
+
+def test_resolve_flr_path_windows_handles_spaces_encoded_as_plus():
+    client = _BrowseFake(
+        [{"Path": "C%3a", "IsBrowsable": True}],
+        {
+            "C:/Program Files/app": [
+                {"Path": "C%3a%2fProgram+Files%2fapp%2fmy+config.ini", "Type": "File"}
+            ]
+        },
+    )
+    got = _resolve(client, r"C:\Program Files\app\my config.ini")
+    assert got == "C%3a%2fProgram+Files%2fapp%2fmy+config.ini"
+
+
+def test_wait_partitions_stable_ignores_the_early_mislabelled_list():
+    import asyncio
+
+    from zerto_rewind_mcp import recover
+
+    recover.PARTITION_POLL_INTERVAL_S = 0
+    # first poll is the pre-enumeration list, then it settles on the real one
+    polls = [
+        [{"Path": "Volume4-Unknown", "IsBrowsable": False}],
+        [{"Path": "C%3a", "IsBrowsable": True}],
+        [{"Path": "C%3a", "IsBrowsable": True}],
+    ]
+
+    class Settling:
+        async def browse_flr(self, session_id, path="", recursive=False):
+            return {"PathItems": polls.pop(0) if polls else [{"Path": "C%3a", "IsBrowsable": True}]}
+
+    got = asyncio.run(recover.wait_partitions_stable(Settling(), "sess"))
+    assert got == ["C%3a"]
+
+
+def test_session_rows_and_id_shapes():
+    from zerto_rewind_mcp.recover import session_id_of, session_rows
+
+    assert session_rows([{"a": 1}]) == [{"a": 1}]
+    assert session_rows({"a": 1}) == [{"a": 1}]
+    assert session_rows(None) == []
+    assert session_id_of({"FlrSessionIdentifier": "s1"}) == "s1"
+    assert session_id_of({"sessionId": "s2"}) == "s2"
+    assert session_id_of({}) == ""
+
+
+def test_is_live_session():
+    from zerto_rewind_mcp.recover import is_live_session
+
+    assert is_live_session({"FlrSessionStatus": "MountCompletedSuccessfully"})
+    assert is_live_session({"FlrSessionStatus": "MountInProgress"})
+    # unmounted/ended/failed sessions linger as history and hold nothing
+    assert not is_live_session({"FlrSessionStatus": "UnmountCompletedSuccessfully"})
+    assert not is_live_session({"FlrSessionStatus": "MountFailed"})
+    assert not is_live_session({})
+
+
+class _FakeClient:
+    """Stands in for ZertoClient in teardown tests."""
+
+    def __init__(self, sessions=None, fail_end=False):
+        self._sessions = sessions or []
+        self.fail_end = fail_end
+        self.ended: list[str] = []
+
+    async def list_flrs(self):
+        return self._sessions
+
+    async def end_flr(self, session_id):
+        from zerto_rewind_mcp.client import ZertoError
+
+        if self.fail_end:
+            raise ZertoError("unmount refused")
+        self.ended.append(session_id)
+
+
+def test_teardown_reports_failure_instead_of_swallowing():
+    import asyncio
+
+    from zerto_rewind_mcp.server import _teardown_flr
+
+    client = _FakeClient(fail_end=True)
+    out = asyncio.run(_teardown_flr(client, "sess-1", set()))
+    assert out["ok"] is False
+    assert out["failed"][0]["session_id"] == "sess-1"
+    assert "still be mounted" in out["message"]
+
+
+def test_teardown_reaps_orphan_when_session_id_never_parsed():
+    import asyncio
+
+    from zerto_rewind_mcp.server import _teardown_flr
+
+    # start_flr succeeded on the ZVM but the response did not parse, so the
+    # caller never learned the id. The new live session must still be reaped.
+    live = [{"FlrSessionIdentifier": "new-1", "FlrSessionStatus": "MountCompletedSuccessfully"}]
+    client = _FakeClient(sessions=live)
+    out = asyncio.run(_teardown_flr(client, None, set()))
+    assert out["ended"] == ["new-1"]
+    assert out["orphans_reaped"] == ["new-1"]
+    assert client.ended == ["new-1"]
+
+
+def test_teardown_leaves_pre_existing_sessions_alone():
+    import asyncio
+
+    from zerto_rewind_mcp.server import _teardown_flr
+
+    live = [
+        {"FlrSessionIdentifier": "someone-else", "FlrSessionStatus": "MountCompletedSuccessfully"}
+    ]
+    client = _FakeClient(sessions=live)
+    out = asyncio.run(_teardown_flr(client, None, {"someone-else"}))
+    assert out["ended"] == []
+    assert client.ended == []
+
+
+class _SiteFake:
+    def __init__(self, protected, recovery):
+        self._vpg = {
+            "VpgName": "demo-vpg",
+            "ProtectedSite": {"identifier": protected},
+            "RecoverySite": {"identifier": recovery},
+        }
+
+    async def get_vpg(self, vpg_identifier):
+        return self._vpg
+
+    async def get_localsite(self):
+        return {"SiteIdentifier": "site-local", "SiteName": "VMware Site"}
+
+    async def get_peersites(self):
+        return [{"SiteIdentifier": "site-aws", "PeerSiteName": "aws-zca"}]
+
+
+def test_flr_gate_allows_local_replication():
+    import asyncio
+
+    from zerto_rewind_mcp.server import _flr_site_gate
+
+    out = asyncio.run(_flr_site_gate(_SiteFake("site-local", "site-local"), "v1"))
+    assert out["ok"] is True
+
+
+def test_flr_gate_refuses_remote_recovery_site_and_names_it():
+    import asyncio
+
+    from zerto_rewind_mcp.server import _flr_site_gate
+
+    out = asyncio.run(_flr_site_gate(_SiteFake("site-local", "site-aws"), "v1"))
+    assert out["ok"] is False
+    assert out["not_local_replication"] is True
+    # 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

tests/test_tasks.py

@@ -0,0 +1,69 @@
+import asyncio
+
+import pytest
+
+from zerto_rewind_mcp.client import ZertoError
+from zerto_rewind_mcp.tasks import (
+    COMPLETED,
+    FAILED,
+    task_id_from,
+    task_label,
+    task_state,
+    wait_for_task,
+)
+
+
+def test_task_id_from_shapes():
+    # a write endpoint returns the id as a bare, often quoted, string
+    assert task_id_from('"abc.def"') == "abc.def"
+    assert task_id_from("abc.def") == "abc.def"
+    assert task_id_from({"TaskIdentifier": "t1"}) == "t1"
+    with pytest.raises(ZertoError):
+        task_id_from(None)
+
+
+def test_task_state_handles_nested_and_bare():
+    assert task_state({"Status": {"State": 6, "Progress": 100}}) == COMPLETED
+    assert task_state({"Status": 4}) == FAILED
+    assert task_state({"Status": None}) is None
+    assert task_state("nope") is None
+
+
+def test_task_labels():
+    assert task_label(COMPLETED) == "Completed"
+    assert task_label(FAILED) == "Failed"
+
+
+class _TaskClient:
+    def __init__(self, states):
+        self.states = list(states)
+        self.calls = 0
+
+    async def get_task(self, task_id):
+        self.calls += 1
+        state = self.states.pop(0) if self.states else self.states
+        return {"Type": "InsertTaggedCP", "Status": {"State": state, "Progress": 100}}
+
+
+def test_wait_for_task_returns_on_completed():
+    client = _TaskClient([1, 1, COMPLETED])
+    out = asyncio.run(wait_for_task(client, "t1", interval_s=0))
+    assert out["state"] == COMPLETED
+    assert out["label"] == "Completed"
+    assert client.calls == 3
+
+
+def test_wait_for_task_raises_on_failed():
+    # this is the case that used to look like success: 200 queued, task Failed
+    client = _TaskClient([1, FAILED])
+    with pytest.raises(ZertoError) as err:
+        asyncio.run(wait_for_task(client, "t1", interval_s=0))
+    assert "Failed" in str(err.value)
+    assert "did not happen" in str(err.value)
+
+
+def test_wait_for_task_times_out_while_in_progress():
+    client = _TaskClient([1] * 50)
+    with pytest.raises(ZertoError) as err:
+        asyncio.run(wait_for_task(client, "t1", timeout_s=0.05, interval_s=0.01))
+    assert "InProgress" in str(err.value)