feat(snapshots): add --tail option to show rsync log

src/pobsync/cli.py

@@ -77,26 +77,20 @@ def build_parser() -> argparse.ArgumentParser:
     sn_list.add_argument("host", help="Host name")
     sn_list.add_argument("--kind", default="all", help="scheduled|manual|incomplete|all (default: all)")
     sn_list.add_argument("--limit", type=int, default=20, help="Max results (default: 20)")
-    sn_list.add_argument(
+    sn_list.add_argument("--include-incomplete", action="store_true", help="Include .incomplete when --kind=all (default: false)")
-        "--include-incomplete",
-        action="store_true",
-        help="Include .incomplete when --kind=all (default: false)",
-    )
     sn_list.set_defaults(_handler=cmd_snapshots_list)
 
     sn_show = sn_sub.add_parser("show", help="Show snapshot metadata")
     sn_show.add_argument("host", help="Host name")
     sn_show.add_argument("--kind", required=True, help="scheduled|manual|incomplete")
     sn_show.add_argument("dirname", help="Snapshot directory name (e.g. 20260202-223807Z__K3VQEVH7)")
+    sn_show.add_argument("--tail", type=int, default=None, help="Show last N lines of rsync.log")
     sn_show.set_defaults(_handler=cmd_snapshots_show)
 
     return p
 
 
 def parse_retention(s: str) -> dict[str, int]:
-    """
-    Parse format: daily=14,weekly=8,monthly=12,yearly=0
-    """
     out: dict[str, int] = {}
     parts = [p.strip() for p in s.split(",") if p.strip()]
     for part in parts:
@@ -111,7 +105,6 @@ def parse_retention(s: str) -> dict[str, int]:
         if n < 0:
             raise ValueError(f"Retention must be >= 0 for {k}")
         out[k] = n
-    # Ensure all keys exist (default missing to 0)
     for k in ("daily", "weekly", "monthly", "yearly"):
         out.setdefault(k, 0)
     return out
@@ -122,22 +115,18 @@ def _print(result: dict[str, Any], as_json: bool) -> None:
         print(json.dumps(to_json_safe(result), indent=2, sort_keys=False))
         return
 
-    # Minimal human output
     if result.get("ok") is True:
         print("OK")
     else:
         print("FAILED")
 
-    # Standard action list
     if "actions" in result:
         for a in result["actions"]:
             print(f"- {a}")
 
-    # Single action (e.g. init-host)
     if "action" in result:
         print(f"- {result['action']}")
 
-    # Doctor-style results list
     if "results" in result:
         for r in result["results"]:
             ok = r.get("ok", False)
@@ -158,7 +147,6 @@ def _print(result: dict[str, Any], as_json: bool) -> None:
                 line += f" {msg}"
             print(line)
 
-    # list-remotes style output
     if "hosts" in result:
         for h in result["hosts"]:
             print(h)
@@ -169,7 +157,6 @@ def _print(result: dict[str, Any], as_json: bool) -> None:
     if "base" in result and result["base"]:
         print(f"- base {result['base']}")
 
-    # snapshots list
     if "snapshots" in result:
         for s in result["snapshots"]:
             kind = s.get("kind", "?")
@@ -213,13 +200,16 @@ def cmd_init_host(args: argparse.Namespace) -> int:
         raise ConfigError("--address is required (or interactive input)")
 
     if args.retention is None:
-        # In phase 1 we require retention explicitly or via install default.
-        # We'll read global.yaml if present to fetch retention_defaults.
         from .config.load import load_global_config
 
         paths = PobsyncPaths(home=prefix)
         global_cfg = load_global_config(paths.global_config_path)
-        retention = global_cfg.get("retention_defaults") or {"daily": 14, "weekly": 8, "monthly": 12, "yearly": 0}
+        retention = global_cfg.get("retention_defaults") or {
+            "daily": 14,
+            "weekly": 8,
+            "monthly": 12,
+            "yearly": 0,
+        }
     else:
         retention = parse_retention(args.retention)
 
@@ -256,7 +246,12 @@ def cmd_show_config(args: argparse.Namespace) -> int:
 
 def cmd_doctor(args: argparse.Namespace) -> int:
     prefix = Path(args.prefix)
-    result = run_doctor(prefix=prefix, host=args.host, connect=bool(args.connect), rsync_dry_run=bool(args.rsync_dry_run))
+    result = run_doctor(
+        prefix=prefix,
+        host=args.host,
+        connect=bool(args.connect),
+        rsync_dry_run=bool(args.rsync_dry_run),
+    )
     _print(result, as_json=bool(args.json))
     return 0 if result.get("ok") else 1
 
@@ -290,16 +285,26 @@ def cmd_snapshots_list(args: argparse.Namespace) -> int:
 
 def cmd_snapshots_show(args: argparse.Namespace) -> int:
     prefix = Path(args.prefix)
-    result = run_snapshots_show(prefix=prefix, host=args.host, kind=args.kind, dirname=args.dirname)
+    result = run_snapshots_show(
+        prefix=prefix,
+        host=args.host,
+        kind=args.kind,
+        dirname=args.dirname,
+        tail=args.tail,
+    )
 
     if args.json:
         _print(result, as_json=True)
     else:
-        # Similar to show-config: dump YAML for the meta block.
         print(dump_yaml(result.get("meta", {})).rstrip())
         if result.get("log_path"):
             print(f"\n# rsync.log: {result['log_path']}")
 
+        if result.get("log_tail"):
+            print("\n# rsync.log (tail)")
+            for line in result["log_tail"]:
+                print(line)
+
     return 0 if result.get("ok") else 1
 
 

src/pobsync/commands/snapshots_show.py

@@ -1,22 +1,49 @@
 from __future__ import annotations
 
 from pathlib import Path
-from typing import Any
+from typing import Any, List
 
 from ..config.load import load_global_config, load_host_config
 from ..config.merge import build_effective_config
 from ..errors import ConfigError
 from ..paths import PobsyncPaths
-from ..snapshot_meta import build_snapshot_ref, normalize_kind, read_snapshot_meta, resolve_host_root, snapshot_log_path
+from ..snapshot_meta import (
+    build_snapshot_ref,
+    normalize_kind,
+    read_snapshot_meta,
+    resolve_host_root,
+    snapshot_log_path,
+)
 from ..util import sanitize_host
 
 
-def run_snapshots_show(prefix: Path, host: str, kind: str, dirname: str) -> dict[str, Any]:
+def _tail_lines(path: Path, n: int) -> List[str]:
+    """
+    Read last n lines of a text file.
+    Simple and safe; rsync logs are not huge in normal cases.
+    """
+    try:
+        lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
+        return lines[-n:]
+    except OSError:
+        return []
+
+
+def run_snapshots_show(
+    prefix: Path,
+    host: str,
+    kind: str,
+    dirname: str,
+    tail: int | None,
+) -> dict[str, Any]:
     host = sanitize_host(host)
     k = normalize_kind(kind)
     if k == "all":
         raise ConfigError("kind must be scheduled, manual, or incomplete for show")
 
+    if tail is not None and tail < 1:
+        raise ConfigError("--tail must be >= 1")
+
     paths = PobsyncPaths(home=prefix)
 
     global_cfg = load_global_config(paths.global_config_path)
@@ -34,7 +61,11 @@ def run_snapshots_show(prefix: Path, host: str, kind: str, dirname: str) -> dict
         raise ConfigError(f"Snapshot not found: {k}/{dirname}")
 
     meta = read_snapshot_meta(ref.path)
-    log = snapshot_log_path(ref.path)
+    log_path = snapshot_log_path(ref.path)
+
+    log_tail = None
+    if tail is not None and log_path.exists():
+        log_tail = _tail_lines(log_path, tail)
 
     return {
         "ok": True,
@@ -43,7 +74,8 @@ def run_snapshots_show(prefix: Path, host: str, kind: str, dirname: str) -> dict
         "dirname": dirname,
         "path": str(ref.path),
         "meta_path": str(ref.path / "meta" / "meta.yaml"),
-        "log_path": str(log) if log.exists() else None,
+        "log_path": str(log_path) if log_path.exists() else None,
         "meta": meta,
+        "log_tail": log_tail,
     }