Phase 8: back up & restore remote (agent) stacks (0.8.0)

- Agent: GET /agent/stacks/{id}/backup + POST /agent/stacks/restore (reuse
  backup_service). backup_service gains backup_basename/backup_filename helpers.
- Main proxy streams agent <-> main <-> destination (creds stay central):
  agent_service download_to_file/upload_file; routers/agents.py backup download,
  backup/push, restore upload, restore-from.
- Schedules: BackupSchedule.agent_id; schedule_service downloads from the agent
  when set; per-host filename prefix isolates retention across hosts.
- Frontend: agents api backup/restore; BackupButton/RestoreButton agent-aware
  (Backup on remote stack detail, Restore per host section); schedule form host
  selector (local or an online agent) + host shown on schedule rows.

Rough-verified (per request): py_compile, frontend tsc build, image imports
(main 99 / agent 16 routes). Full live agent round-trip to be tested post-deploy.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/services/agent_service.py

@@ -101,6 +101,73 @@ async def call(
     return None
 
 
+def _handle_status(session: Session, agent: Agent, status_code: int, body_text: str = "") -> None:
+    """Update agent status from a response code; raise AgentError on failure."""
+    if status_code in (401, 403):
+        _mark(session, agent, "unauthorized")
+        raise AgentError(status_code, "agent_unauthorized", "Invalid agent token")
+    _mark(session, agent, "online")
+    if status_code >= 400:
+        raise AgentError(status_code, "agent_error", body_text[:500])
+
+
+async def download_to_file(
+    session: Session,
+    agent: Agent,
+    path: str,
+    dest_path: str,
+    *,
+    params: Optional[dict] = None,
+) -> None:
+    """Stream a GET from the agent into ``dest_path``."""
+    url = agent.url.rstrip("/") + path
+    headers = {"Authorization": f"Bearer {agent.token}"}
+    try:
+        async with httpx.AsyncClient(follow_redirects=True) as client:
+            async with client.stream("GET", url, headers=headers, params=params, timeout=None) as resp:
+                if resp.status_code >= 400:
+                    text = (await resp.aread()).decode("utf-8", "replace")
+                    _handle_status(session, agent, resp.status_code, text)
+                _handle_status(session, agent, resp.status_code)
+                with open(dest_path, "wb") as fh:
+                    async for chunk in resp.aiter_bytes(1024 * 256):
+                        fh.write(chunk)
+    except httpx.HTTPError as exc:
+        _mark(session, agent, "offline")
+        raise AgentError(502, "agent_unreachable", str(exc)) from exc
+
+
+async def upload_file(
+    session: Session,
+    agent: Agent,
+    path: str,
+    file_path: str,
+    filename: str,
+    data: dict,
+) -> Any:
+    """Stream a multipart POST (file + form fields) to the agent, returning JSON."""
+    url = agent.url.rstrip("/") + path
+    headers = {"Authorization": f"Bearer {agent.token}"}
+    try:
+        async with httpx.AsyncClient(follow_redirects=True) as client:
+            with open(file_path, "rb") as fh:
+                files = {"file": (filename, fh, "application/gzip")}
+                resp = await client.post(url, headers=headers, files=files, data=data, timeout=None)
+    except httpx.HTTPError as exc:
+        _mark(session, agent, "offline")
+        raise AgentError(502, "agent_unreachable", str(exc)) from exc
+
+    detail = ""
+    if resp.status_code >= 400:
+        try:
+            body = resp.json()
+            detail = body.get("detail") if isinstance(body, dict) else str(body)
+        except ValueError:
+            detail = resp.text[:500]
+    _handle_status(session, agent, resp.status_code, str(detail))
+    return resp.json() if resp.content else None
+
+
 async def ping(session: Session, agent: Agent) -> dict:
     """Health-check an agent and refresh its status + hostname. Never raises."""
     try:

backend/services/backup_service.py

@@ -38,6 +38,26 @@ class BackupError(Exception):
     pass
 
 
+# --------------------------------------------------------------------------- #
+# Backup filename convention
+# --------------------------------------------------------------------------- #
+
+
+def backup_basename(stack_id: str, prefix: Optional[str] = None) -> str:
+    """Filename stem used to group a stack's backups (and match for retention).
+
+    ``prefix`` (e.g. a remote host slug) keeps backups of same-named stacks on
+    different hosts from colliding / pruning each other on a shared destination.
+    """
+    return f"backup-{prefix}-{stack_id}" if prefix else f"backup-{stack_id}"
+
+
+def backup_filename(stack_id: str, include_volumes: bool, prefix: Optional[str] = None) -> str:
+    date = now().strftime("%Y%m%d-%H%M%S")
+    suffix = "full" if include_volumes else "config"
+    return f"{backup_basename(stack_id, prefix)}-{suffix}-{date}.tar.gz"
+
+
 # --------------------------------------------------------------------------- #
 # Helper container for volume I/O
 # --------------------------------------------------------------------------- #

backend/services/schedule_service.py

@@ -11,16 +11,19 @@ from __future__ import annotations
 import asyncio
 import logging
 import os
+import tempfile
 from datetime import datetime, timedelta, timezone
 
 from sqlmodel import Session, select
 
 from database import engine
+from models.agent import Agent
 from models.backup_destination import BackupDestination
 from models.backup_schedule import BackupSchedule
 from models.setting import EVENT_BACKUP_FAILED
 from models.stack import Stack
 from services import (
+    agent_service,
     backup_destination_service as dest_service,
     backup_service,
     compose_service,
@@ -58,17 +61,11 @@ def compute_next_run(
     return base
 
 
-def _backup_filename(stack_id: str, include_volumes: bool) -> str:
-    date = compose_service.now().strftime("%Y%m%d-%H%M%S")
-    suffix = "full" if include_volumes else "config"
-    return f"backup-{stack_id}-{suffix}-{date}.tar.gz"
-
-
-def _prune(dest: BackupDestination, stack_id: str, keep: int) -> int:
+def _prune(dest: BackupDestination, basename: str, keep: int) -> int:
     if keep <= 0:
         return 0
     items = dest_service.list_backups(dest)
-    mine = [i for i in items if i["name"].startswith(f"backup-{stack_id}-")]
+    mine = [i for i in items if i["name"].startswith(f"{basename}-")]
     mine.sort(key=lambda x: x.get("modified") or 0, reverse=True)
     removed = 0
     for old in mine[keep:]:
@@ -83,22 +80,40 @@ def _prune(dest: BackupDestination, stack_id: str, keep: int) -> int:
 async def run_schedule(session: Session, schedule: BackupSchedule) -> dict:
     """Execute one schedule now. Updates status + next_run. Returns a summary."""
     now = _now()
-    stack = session.get(Stack, schedule.stack_id)
     dest = session.get(BackupDestination, schedule.destination_id)
     result: dict = {"ok": False}
     path = None
     try:
-        if not stack:
-            raise RuntimeError(f"stack '{schedule.stack_id}' not found")
         if not dest:
             raise RuntimeError(f"destination {schedule.destination_id} not found")
-        path = await backup_service.create_backup(
-            schedule.stack_id, stack.name,
-            include_volumes=schedule.include_volumes, stop_first=schedule.stop_first,
-        )
-        filename = _backup_filename(schedule.stack_id, schedule.include_volumes)
+
+        # Produce the backup archive — locally or by streaming it from an agent.
+        if schedule.agent_id is not None:
+            agent = session.get(Agent, schedule.agent_id)
+            if not agent:
+                raise RuntimeError(f"agent {schedule.agent_id} not found")
+            prefix = compose_service.slugify(agent.name)
+            tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".tar.gz")
+            tmp.close()
+            path = tmp.name
+            await agent_service.download_to_file(
+                session, agent, f"/agent/stacks/{schedule.stack_id}/backup", path,
+                params={"include_volumes": schedule.include_volumes, "stop_first": schedule.stop_first},
+            )
+        else:
+            stack = session.get(Stack, schedule.stack_id)
+            if not stack:
+                raise RuntimeError(f"stack '{schedule.stack_id}' not found")
+            prefix = None
+            path = await backup_service.create_backup(
+                schedule.stack_id, stack.name,
+                include_volumes=schedule.include_volumes, stop_first=schedule.stop_first,
+            )
+
+        filename = backup_service.backup_filename(schedule.stack_id, schedule.include_volumes, prefix=prefix)
+        basename = backup_service.backup_basename(schedule.stack_id, prefix)
         await asyncio.to_thread(dest_service.upload, dest, path, filename)
-        pruned = await asyncio.to_thread(_prune, dest, schedule.stack_id, schedule.keep)
+        pruned = await asyncio.to_thread(_prune, dest, basename, schedule.keep)
         schedule.last_status = "ok"
         result = {"ok": True, "name": filename, "destination": dest.name, "pruned": pruned}
         logger.info("Scheduled backup %s → %s ok (pruned %d)", schedule.stack_id, dest.name, pruned)