feat: length-aware reinforcement + batch triage CLI + off-host backup

- Reinforcement matcher now handles paragraph-length memories via a
  dual-mode threshold: short memories keep the 70% overlap rule,
  long memories (>15 stems) require 12 absolute overlaps AND 35%
  fraction so organic paraphrase can still reinforce. Diagnosis:
  every active memory stayed at reference_count=0 because 40-token
  project summaries never hit 70% overlap on real responses.
- scripts/atocore_client.py gains batch-extract (fan out
  /interactions/{id}/extract over recent interactions) and triage
  (interactive promote/reject walker for the candidate queue),
  matching the Phase 9 reflection-loop review flow without pulling
  extraction into the capture hot path.
- deploy/dalidou/cron-backup.sh adds an optional off-host rsync step
  gated on ATOCORE_BACKUP_RSYNC, fail-open when the target is offline
  so a laptop being off at 03:00 UTC never reds the local backup.
- docs/next-steps.md records the retrieval-quality sweep: project
  state surfaces, chunks are on-topic but broad, active memories
  never reach the pack (reflection loop has no retrieval outlet yet).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

scripts/atocore_client.py

@@ -340,6 +340,22 @@ def build_parser() -> argparse.ArgumentParser:
     p = sub.add_parser("reject")
     p.add_argument("memory_id")
 
+    # batch-extract: fan out /interactions/{id}/extract?persist=true across
+    #   recent interactions. Idempotent — the extractor create_memory path
+    #   silently skips duplicates, so re-running is safe.
+    p = sub.add_parser("batch-extract")
+    p.add_argument("since", nargs="?", default="")
+    p.add_argument("project", nargs="?", default="")
+    p.add_argument("limit", nargs="?", type=int, default=100)
+    p.add_argument("persist", nargs="?", default="true")
+
+    # triage: interactive candidate review loop. Fetches the queue, shows
+    #   each candidate, accepts p/r/s (promote / reject / skip) / q (quit).
+    p = sub.add_parser("triage")
+    p.add_argument("memory_type", nargs="?", default="")
+    p.add_argument("project", nargs="?", default="")
+    p.add_argument("limit", nargs="?", type=int, default=50)
+
     return parser
 
 
@@ -474,10 +490,141 @@ def main() -> int:
                 {},
             )
         )
+    elif cmd == "batch-extract":
+        print_json(run_batch_extract(args.since, args.project, args.limit, args.persist))
+    elif cmd == "triage":
+        return run_triage(args.memory_type, args.project, args.limit)
     else:
         return 1
     return 0
 
 
+def run_batch_extract(since: str, project: str, limit: int, persist_flag: str) -> dict:
+    """Fetch recent interactions and run the extractor against each one.
+
+    Returns an aggregated summary. Safe to re-run: the server-side
+    persist path catches ValueError on duplicates and the endpoint
+    reports per-interaction candidate counts either way.
+    """
+    persist = persist_flag.lower() in {"1", "true", "yes", "y"}
+    query_parts: list[str] = []
+    if project:
+        query_parts.append(f"project={urllib.parse.quote(project)}")
+    if since:
+        query_parts.append(f"since={urllib.parse.quote(since)}")
+    query_parts.append(f"limit={int(limit)}")
+    query = "?" + "&".join(query_parts)
+
+    listing = request("GET", f"/interactions{query}")
+    interactions = listing.get("interactions", []) if isinstance(listing, dict) else []
+
+    processed = 0
+    total_candidates = 0
+    total_persisted = 0
+    errors: list[dict] = []
+    per_interaction: list[dict] = []
+
+    for item in interactions:
+        iid = item.get("id") or ""
+        if not iid:
+            continue
+        try:
+            result = request(
+                "POST",
+                f"/interactions/{urllib.parse.quote(iid, safe='')}/extract",
+                {"persist": persist},
+            )
+        except Exception as exc:  # pragma: no cover - network errors land here
+            errors.append({"interaction_id": iid, "error": str(exc)})
+            continue
+        processed += 1
+        count = int(result.get("candidate_count", 0) or 0)
+        persisted_ids = result.get("persisted_ids") or []
+        total_candidates += count
+        total_persisted += len(persisted_ids)
+        if count:
+            per_interaction.append(
+                {
+                    "interaction_id": iid,
+                    "candidate_count": count,
+                    "persisted_count": len(persisted_ids),
+                    "project": item.get("project") or "",
+                }
+            )
+
+    return {
+        "processed": processed,
+        "total_candidates": total_candidates,
+        "total_persisted": total_persisted,
+        "persist": persist,
+        "errors": errors,
+        "interactions_with_candidates": per_interaction,
+    }
+
+
+def run_triage(memory_type: str, project: str, limit: int) -> int:
+    """Interactive review of candidate memories.
+
+    Loads the queue once, walks through entries, prompts for
+    (p)romote / (r)eject / (s)kip / (q)uit. Stateless between runs —
+    re-running picks up whatever is still status=candidate.
+    """
+    query_parts = ["status=candidate"]
+    if memory_type:
+        query_parts.append(f"memory_type={urllib.parse.quote(memory_type)}")
+    if project:
+        query_parts.append(f"project={urllib.parse.quote(project)}")
+    query_parts.append(f"limit={int(limit)}")
+    listing = request("GET", "/memory?" + "&".join(query_parts))
+    memories = listing.get("memories", []) if isinstance(listing, dict) else []
+
+    if not memories:
+        print_json({"status": "empty_queue", "count": 0})
+        return 0
+
+    promoted = 0
+    rejected = 0
+    skipped = 0
+    stopped_early = False
+
+    print(f"Triage queue: {len(memories)} candidate(s)\n", file=sys.stderr)
+    for idx, mem in enumerate(memories, 1):
+        mid = mem.get("id", "")
+        print(f"[{idx}/{len(memories)}] {mem.get('memory_type','?')}  project={mem.get('project','')}  conf={mem.get('confidence','?')}", file=sys.stderr)
+        print(f"  id: {mid}", file=sys.stderr)
+        print(f"  {mem.get('content','')}", file=sys.stderr)
+        try:
+            choice = input("  (p)romote / (r)eject / (s)kip / (q)uit > ").strip().lower()
+        except EOFError:
+            stopped_early = True
+            break
+        if choice in {"q", "quit"}:
+            stopped_early = True
+            break
+        if choice in {"p", "promote"}:
+            request("POST", f"/memory/{urllib.parse.quote(mid, safe='')}/promote", {})
+            promoted += 1
+            print("  -> promoted", file=sys.stderr)
+        elif choice in {"r", "reject"}:
+            request("POST", f"/memory/{urllib.parse.quote(mid, safe='')}/reject", {})
+            rejected += 1
+            print("  -> rejected", file=sys.stderr)
+        else:
+            skipped += 1
+            print("  -> skipped", file=sys.stderr)
+
+    print_json(
+        {
+            "reviewed": promoted + rejected + skipped,
+            "promoted": promoted,
+            "rejected": rejected,
+            "skipped": skipped,
+            "stopped_early": stopped_early,
+            "remaining_in_queue": len(memories) - (promoted + rejected + skipped) - (1 if stopped_early else 0),
+        }
+    )
+    return 0
+
+
 if __name__ == "__main__":
     raise SystemExit(main())