feat: implement update backend API and client logic

Add complete update mechanism for client-side updates:
- backend/update.py: Core update logic (check, apply, rollback, status/logs)
- backend/app.py: REST API endpoints (GET /update/status, POST /update/check, POST /update/apply, POST /update/rollback, GET /update/logs)
- backend/models.py: Pydantic models for update API responses
- backend/settings.py: Update config (status/log file paths)
- scripts/rollback_client.sh: Rollback script for failed updates
- scripts/update_client.sh: Enhanced update client script
- CLAUDE.md: Documentation for future Claude Code instances

Complete US_000026-028 and TASK_000027-029:
- US_000026: Client pulls updates from remote service
- US_000027: Client verifies and applies updates atomically
- US_000028: Client reports update status to backend

All endpoints require authentication. Updates run asynchronously.
Documentation updated per SOP (CHANGELOG, PROJECT_STATUS, stories/tasks).

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
2025-12-30 11:18:34 +01:00
parent b64cc5981c
commit 107cdabe8d
15 changed files with 530 additions and 19 deletions

142
backend/update.py Normal file
View File

@ -0,0 +1,142 @@
import json
import os
import subprocess
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List
import httpx
from backend.settings import Settings
def _project_root() -> Path:
return Path(__file__).resolve().parents[1]
def _read_version() -> str:
try:
return (_project_root() / "VERSION").read_text(encoding="utf-8").strip()
except OSError:
return "unknown"
def _status_path(settings: Settings) -> Path:
return Path(settings.update_status_file)
def _log_path(settings: Settings) -> Path:
return Path(settings.update_log_file)
def _ensure_parent(path: Path) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
def _write_status(settings: Settings, status: str, version: str, error: str | None = None) -> None:
status_path = _status_path(settings)
_ensure_parent(status_path)
payload = {
"device_id": os.uname().nodename,
"version": version,
"status": status,
"error": error or "",
"timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
}
status_path.write_text(json.dumps(payload), encoding="utf-8")
_append_log(settings, payload)
def _append_log(settings: Settings, payload: Dict[str, Any]) -> None:
log_path = _log_path(settings)
_ensure_parent(log_path)
with log_path.open("a", encoding="utf-8") as handle:
handle.write(json.dumps(payload) + "\n")
def get_status(settings: Settings) -> Dict[str, Any]:
current_version = _read_version()
status_path = _status_path(settings)
if status_path.exists():
try:
data = json.loads(status_path.read_text(encoding="utf-8"))
except json.JSONDecodeError:
data = {}
else:
data = {}
return {
"current_version": current_version,
"last_status": data.get("status", "unknown"),
"last_error": data.get("error") or None,
"last_timestamp": data.get("timestamp"),
}
def _parse_version(value: str) -> List[int]:
return [int(part) for part in value.split(".")]
def check_update(settings: Settings) -> Dict[str, Any]:
headers = {}
if settings.update_token:
headers["Authorization"] = f"Bearer {settings.update_token}"
with httpx.Client(timeout=10.0) as client:
response = client.get(settings.update_url, headers=headers)
response.raise_for_status()
manifest = response.json()
latest_version = manifest.get("version", "")
artifact_url = manifest.get("artifact_url", "")
sha256 = manifest.get("sha256", "")
message = manifest.get("message")
available = False
current_version = _read_version()
try:
available = _parse_version(latest_version) > _parse_version(current_version)
except ValueError:
if latest_version and latest_version != current_version:
available = True
return {
"available": available,
"latest_version": latest_version,
"artifact_url": artifact_url,
"sha256": sha256,
"message": message,
}
def _run_async(script_path: Path, settings: Settings) -> None:
env = os.environ.copy()
env["SKD_UPDATE_STATUS_FILE"] = settings.update_status_file
env["SKD_UPDATE_LOG_FILE"] = settings.update_log_file
subprocess.Popen([str(script_path)], env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
def start_update(settings: Settings, version: str | None = None) -> None:
_write_status(settings, "in_progress", version or _read_version())
script = _project_root() / "scripts" / "update_client.sh"
_run_async(script, settings)
def start_rollback(settings: Settings) -> None:
_write_status(settings, "in_progress", _read_version())
script = _project_root() / "scripts" / "rollback_client.sh"
_run_async(script, settings)
def get_logs(settings: Settings, limit: int = 200) -> List[Dict[str, Any]]:
log_path = _log_path(settings)
if not log_path.exists():
return []
lines = log_path.read_text(encoding="utf-8").splitlines()
entries: List[Dict[str, Any]] = []
for line in lines[-limit:]:
try:
entries.append(json.loads(line))
except json.JSONDecodeError:
continue
return entries