"""Auto-update (Watchtower-style) orchestration. Runs once per image-update-check cycle (called from ``update_service.background_loop`` so it reuses the freshly-populated digest cache). For each enabled policy whose stack has a newer image available, either pulls + redeploys the stack or just notifies, recording the outcome. Central-only / DB-aware. Image resolution + digest comparison live in the DB-free ``update_service`` so the agent can answer ``/agent/stacks/{id}/updates`` with the same logic. """ from __future__ import annotations import logging from datetime import datetime, timezone from sqlmodel import Session, select from database import engine from models.agent import Agent from models.auto_update import AutoUpdate from models.setting import EVENT_PULL_FAILED, EVENT_STACK_AUTO_UPDATED from services import agent_service, compose_service, notify_service, update_service logger = logging.getLogger("stackpilot.autoupdate") # Stack states we'll act on. We never auto-start a stopped stack. _LIVE_STATES = {"running", "partial", "updating"} def _now() -> datetime: return datetime.now(timezone.utc) # --------------------------------------------------------------------------- # # Policy CRUD helpers (shared by the stacks + agents routers) # --------------------------------------------------------------------------- # def get_policy(session: Session, stack_id: str, agent_id: int | None = None) -> AutoUpdate | None: stmt = select(AutoUpdate).where(AutoUpdate.stack_id == stack_id) stmt = stmt.where(AutoUpdate.agent_id == agent_id) if agent_id is not None \ else stmt.where(AutoUpdate.agent_id.is_(None)) return session.exec(stmt).first() def upsert_policy( session: Session, stack_id: str, enabled: bool, redeploy: bool, agent_id: int | None = None ) -> AutoUpdate: policy = get_policy(session, stack_id, agent_id) if policy is None: policy = AutoUpdate(stack_id=stack_id, agent_id=agent_id) policy.enabled = enabled policy.redeploy = redeploy session.add(policy) session.commit() session.refresh(policy) return policy def to_read(session: Session, policy: AutoUpdate | None, stack_id: str, agent_id: int | None = None) -> dict: """Build an AutoUpdateRead-shaped dict, defaulting to disabled when absent.""" agent_name = None if agent_id is not None: agent = session.get(Agent, agent_id) agent_name = agent.name if agent else None if policy is None: return { "id": None, "stack_id": stack_id, "agent_id": agent_id, "agent_name": agent_name, "enabled": False, "redeploy": True, "last_run": None, "last_status": None, "last_result": None, } return { "id": policy.id, "stack_id": policy.stack_id, "agent_id": policy.agent_id, "agent_name": agent_name, "enabled": policy.enabled, "redeploy": policy.redeploy, "last_run": policy.last_run, "last_status": policy.last_status, "last_result": policy.last_result, } def _record(session: Session, policy: AutoUpdate, status: str, result: str = "") -> None: policy.last_run = _now() policy.last_status = status policy.last_result = result[:300] if result else None session.add(policy) session.commit() async def _run_local(session: Session, policy: AutoUpdate) -> None: stack_id = policy.stack_id try: state = compose_service.compute_status(stack_id) except Exception: # noqa: BLE001 state = "unknown" if state not in _LIVE_STATES: _record(session, policy, "skipped", f"stack not running ({state})") return summary = await update_service.stack_updates(stack_id, refresh=False) if not summary["update_available"]: _record(session, policy, "up-to-date") return stale = ", ".join(summary["stale_images"]) prev = policy.last_status if policy.redeploy: try: await compose_service.pull(stack_id) await compose_service.up(stack_id) except Exception as exc: # noqa: BLE001 _record(session, policy, "error", str(exc)) await _safe_notify(EVENT_PULL_FAILED, f"Auto-update of '{stack_id}' failed", str(exc), session) return _record(session, policy, "updated", stale) await _safe_notify( EVENT_STACK_AUTO_UPDATED, f"Stack '{stack_id}' auto-updated", f"Pulled and redeployed: {stale}.", session, ) else: _record(session, policy, "update-available", stale) if prev != "update-available": # notify once per transition, not every cycle await _safe_notify( EVENT_STACK_AUTO_UPDATED, f"Update available for '{stack_id}'", f"Newer images: {stale} (auto-redeploy is off).", session, ) async def _run_remote(session: Session, policy: AutoUpdate) -> None: agent = session.get(Agent, policy.agent_id) if not agent: _record(session, policy, "error", "agent not found") return stack_id = policy.stack_id try: summary = await agent_service.call( session, agent, "GET", f"/agent/stacks/{stack_id}/updates", params={"refresh": "true"}, ) except Exception as exc: # noqa: BLE001 _record(session, policy, "error", f"agent check failed: {exc}") return if not summary or not summary.get("update_available"): _record(session, policy, "up-to-date") return stale = ", ".join(summary.get("stale_images", [])) label = f"{agent.name}/{stack_id}" prev = policy.last_status if policy.redeploy: try: await agent_service.call(session, agent, "POST", f"/agent/stacks/{stack_id}/update") except Exception as exc: # noqa: BLE001 _record(session, policy, "error", str(exc)) await _safe_notify(EVENT_PULL_FAILED, f"Auto-update of '{label}' failed", str(exc), session) return _record(session, policy, "updated", stale) await _safe_notify( EVENT_STACK_AUTO_UPDATED, f"Stack '{label}' auto-updated", f"Pulled and redeployed: {stale}.", session, ) else: _record(session, policy, "update-available", stale) if prev != "update-available": await _safe_notify( EVENT_STACK_AUTO_UPDATED, f"Update available for '{label}'", f"Newer images: {stale} (auto-redeploy is off).", session, ) async def _safe_notify(event: str, title: str, message: str, session: Session) -> None: try: await notify_service.notify(event, title, message, session) except Exception as exc: # noqa: BLE001 - notifications are best-effort logger.debug("auto-update notify failed: %s", exc) async def run_policy(session: Session, policy: AutoUpdate) -> None: if policy.agent_id is None: await _run_local(session, policy) else: await _run_remote(session, policy) async def run_due() -> None: """Process every enabled policy. Best-effort: one failure never aborts the rest.""" with Session(engine) as session: policies = list(session.exec(select(AutoUpdate).where(AutoUpdate.enabled == True))) # noqa: E712 for policy in policies: try: with Session(engine) as session: fresh = session.get(AutoUpdate, policy.id) if fresh and fresh.enabled: await run_policy(session, fresh) except Exception as exc: # noqa: BLE001 logger.warning("Auto-update policy %s failed: %s", policy.id, exc)